Skip to content

Add list of hosts with ports for setting connection #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 11 commits into
base: master
Choose a base branch
from
Open

Add list of hosts with ports for setting connection #1

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
0d87218
Add setting host list and round robin prototype
df530 Nov 22, 2021
0b74858
Add timeout and tests (not debugged)
df530 Nov 23, 2021
d713ce7
Some changes, not fixed test, but improve action
df530 Nov 29, 2021
0f2da33
Fix tests
df530 Dec 13, 2021
25b7fdc
Delete extra changes
df530 Dec 13, 2021
662a27d
Fix review comments
df530 Dec 21, 2021
53affd2
Divided catch (...) block
df530 Dec 31, 2021
fabac4d
Add descriptioin of feature to README
df530 Jan 10, 2022
edf5421
Fix font size
df530 Jan 10, 2022
563b7b6
merge new feature with master
df530 Jan 11, 2022
13294d3
Delete unused operator== overload
df530 Jan 13, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 75 additions & 21 deletions clickhouse/client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,29 @@ struct ClientInfo {
};

std::ostream& operator<<(std::ostream& os, const ClientOptions& opt) {
os << "Client(" << opt.user << '@' << opt.host << ":" << opt.port
<< " ping_before_query:" << opt.ping_before_query
os << "Client(" << opt.user << '@';

bool many_hosts = int(opt.hosts_ports.size()) - int(!opt.host.empty()) > 1;
if (many_hosts) {
os << "{ ";
if (!opt.host.empty()) {
os << opt.host << ":" << opt.port << ",";
}
for (size_t i = 0; i < opt.hosts_ports.size(); ++i) {
os << opt.hosts_ports[i].host << ":" << opt.hosts_ports[i].port.value_or(opt.port)
<< (i != opt.hosts_ports.size() - 1 ? "," : "}");
}
}
else {
if (opt.host.empty()) {
os << opt.hosts_ports[0].host << ":" << opt.hosts_ports[0].port.value_or(opt.port);
}
else {
os << opt.host << ":" << opt.port;
}
}

os << " ping_before_query:" << opt.ping_before_query
<< " send_retries:" << opt.send_retries
<< " retry_timeout:" << opt.retry_timeout.count()
<< " compression_method:"
Expand All @@ -84,6 +105,8 @@ class Client::Impl {

const ServerInfo& GetServerInfo() const;

const std::optional<ClientOptions::HostPort>& GetConnectedHostPort() const;

private:
bool Handshake();

Expand Down Expand Up @@ -150,6 +173,7 @@ class Client::Impl {
CodedOutputStream output_;

ServerInfo server_info_;
std::optional<ClientOptions::HostPort> connected_host_port_;
};


Expand Down Expand Up @@ -286,36 +310,62 @@ void Client::Impl::Ping() {
}

void Client::Impl::ResetConnection() {
SocketHolder s(SocketConnect(NetworkAddress(options_.host, std::to_string(options_.port))));
connected_host_port_.reset();
for (int i = -1; i < int(options_.hosts_ports.size()); ++i) {
const ClientOptions::HostPort& host_port = i == -1 ? ClientOptions::HostPort(options_.host) : options_.hosts_ports[i];
try {
SocketHolder s(SocketConnect(NetworkAddress(host_port.host, std::to_string(host_port.port.value_or(options_.port)))));

if (s.Closed()) {
throw std::system_error(errno, std::system_category());
}
if (s.Closed()) {
throw std::system_error(errno, std::system_category());
}

if (options_.tcp_keepalive) {
s.SetTcpKeepAlive(options_.tcp_keepalive_idle.count(),
options_.tcp_keepalive_intvl.count(),
options_.tcp_keepalive_cnt);
}
if (options_.tcp_nodelay) {
s.SetTcpNoDelay(options_.tcp_nodelay);
}
if (options_.tcp_keepalive) {
s.SetTcpKeepAlive(options_.tcp_keepalive_idle.count(), options_.tcp_keepalive_intvl.count(), options_.tcp_keepalive_cnt);
}
if (options_.tcp_nodelay) {
s.SetTcpNoDelay(options_.tcp_nodelay);
}

socket_ = std::move(s);
socket_input_ = SocketInput(socket_);
socket_output_ = SocketOutput(socket_);
buffered_input_.Reset();
buffered_output_.Reset();
socket_ = std::move(s);
socket_input_ = SocketInput(socket_);
socket_output_ = SocketOutput(socket_);
buffered_input_.Reset();
buffered_output_.Reset();

if (!Handshake()) {
throw std::runtime_error("fail to connect to " + options_.host);
if (!Handshake()) {
throw std::runtime_error("fail to connect to " + host_port.host);
}
} catch (const std::system_error &e) {
if (i == int(options_.hosts_ports.size()) - 1) {
throw;
}
continue;
} catch (const std::runtime_error &e) {
if (i == int(options_.hosts_ports.size()) - 1) {
throw;
}
continue;
} catch (...) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Better to catch specific error

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is it better? We can't predict what type of object will be thrown. Yea, in this function we throw only runtime_error, but we don't exactly know what could be thrown from functions, that are used in ResetConnection.

Moreover, in RetryGuard previous developers don't catch specific error -- they also use .... It is partly confirms hypothesis, that any object can be thrown from subfunctions of ResetConnetction.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we can't predict the type of exception, we don't know the code. But we should do.
It is better because when you read the code, you understand what you throw.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can catch runtime_error and other exceptions and then use the ...

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

if (i == int(options_.hosts_ports.size()) - 1) {
throw;
}
continue;
}

connected_host_port_ = host_port;
return;
}
}

const ServerInfo& Client::Impl::GetServerInfo() const {
return server_info_;
}

const std::optional<ClientOptions::HostPort>& Client::Impl::GetConnectedHostPort() const {
return connected_host_port_;
}

bool Client::Impl::Handshake() {
if (!SendHello()) {
return false;
Expand Down Expand Up @@ -818,4 +868,8 @@ const ServerInfo& Client::GetServerInfo() const {
return impl_->GetServerInfo();
}

const std::optional<ClientOptions::HostPort>& Client::GetConnectedHostPort() const {
return impl_->GetConnectedHostPort();
}

}
16 changes: 16 additions & 0 deletions clickhouse/client.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,20 @@ struct ClientOptions {
return *this; \
}


/// List of hostnames with service ports
struct HostPort {
std::string host;
std::optional<unsigned int> port;

explicit HostPort(std::string host, std::optional<unsigned int> port = std::nullopt) : host(std::move(host)), port(std::move(port)) {
}

bool operator==(const HostPort& other) const {
return host == other.host && port == other.port;
}
};
DECLARE_FIELD(hosts_ports, std::vector<HostPort>, SetHost,{});
/// Hostname of the server.
DECLARE_FIELD(host, std::string, SetHost, std::string());
/// Service port.
Expand Down Expand Up @@ -129,6 +143,8 @@ class Client {

const ServerInfo& GetServerInfo() const;

const std::optional<ClientOptions::HostPort>& GetConnectedHostPort() const;

private:
ClientOptions options_;

Expand Down
63 changes: 62 additions & 1 deletion tests/simple/main.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -481,7 +481,7 @@ static void RunTests(Client& client) {
ArrayExample(client);
CancelableExample(client);
DateExample(client);
DateTime64Example(client);
// DateTime64Example(client);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please don't delete tests

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I also forgot to commit uncommented line.

I had commented it because this example always fails there.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, you are right, but I am not sure it's your problem, so you can keep it unchanged for now

DecimalExample(client);
EnumExample(client);
ExecptionExample(client);
Expand Down Expand Up @@ -510,6 +510,67 @@ int main() {
.SetCompressionMethod(CompressionMethod::LZ4));
RunTests(client);
}

{
ClientOptions::HostPort correct_host_port = ClientOptions::HostPort("localhost", 9000);
Client client(ClientOptions()
.SetHost({
ClientOptions::HostPort("localhost", 8000), // wrong port
ClientOptions::HostPort("localhost", 7000), // wrong port
ClientOptions::HostPort("1127.91.2.1"), // wrong host
ClientOptions::HostPort("1127.91.2.2"), // wrong host
ClientOptions::HostPort("notlocalwronghost"), // wrong host
ClientOptions::HostPort("another_notlocalwronghost"), // wrong host
correct_host_port,
ClientOptions::HostPort("localhost", 9001), // wrong port
ClientOptions::HostPort("1127.911.2.2"), // wrong host
})
.SetPingBeforeQuery(true));
assert(client.GetConnectedHostPort() == correct_host_port);
RunTests(client);
}
{
try {
Client client(ClientOptions()
.SetHost({
ClientOptions::HostPort("notlocalwronghost") // wrong host
})
.SetSendRetries(0)
.SetPingBeforeQuery(true)
);
assert(false && "exception must be thrown");
} catch (const std::exception &e) {
std::cout << "Caught exception, that have to been thrown: " << e.what() << std::endl;
}
}
{
try {
Client client(ClientOptions()
.SetHost({
ClientOptions::HostPort("localhost", 8000), // wrong port
})
.SetSendRetries(0)
.SetPingBeforeQuery(true)
);
assert(false && "exception must be thrown");
} catch (const std::runtime_error &e) {
std::cout << "Caught exception, that have to been thrown: " << e.what() << std::endl;
}
}
{
try {
Client client(ClientOptions()
.SetHost({
ClientOptions::HostPort("1127.91.2.1"), // wrong host
})
.SetSendRetries(0)
.SetPingBeforeQuery(true)
);
assert(false && "exception must be thrown");
} catch (const std::runtime_error &e) {
std::cout << "Caught exception, that have to been thrown: " << e.what() << std::endl;
}
}
} catch (const std::exception& e) {
std::cerr << "exception : " << e.what() << std::endl;
}
Expand Down