Commit 347378e6 authored by knowledge4igor's avatar knowledge4igor

Fix data races on Client desctruction

parent 3f298999
......@@ -368,6 +368,7 @@ private:
Lock queuesLock;
std::unordered_map<std::string, MPMCQueue<std::shared_ptr<Connection::RequestData>, 2048>> requestsQueues;
bool stopProcessPequestsQueues;
RequestBuilder prepareRequest(const std::string& resource, Http::Method method);
......
......@@ -704,9 +704,11 @@ Client::Client()
, ioIndex(0)
, queuesLock()
, requestsQueues()
, stopProcessPequestsQueues(false)
{ }
Client::~Client() {
assert(stopProcessPequestsQueues == true && "You must explicitly call shutdown method of Client object");
}
Client::Options
......@@ -726,6 +728,8 @@ Client::init(const Client::Options& options) {
void
Client::shutdown() {
reactor_->shutdown();
Guard guard(queuesLock);
stopProcessPequestsQueues = true;
}
RequestBuilder
......@@ -821,6 +825,9 @@ void
Client::processRequestQueue() {
Guard guard(queuesLock);
if (stopProcessPequestsQueues)
return;
for (auto& queues: requestsQueues) {
for (;;) {
const auto& domain = queues.first;
......
......@@ -203,7 +203,7 @@ TEST(http_client_test, timeout_reject)
ASSERT_TRUE(is_reject);
}
TEST(http_client_test, one_client_with_multiple_requests_and_one_connection_per_host)
TEST(http_client_test, one_client_with_multiple_requests_and_one_connection_per_host_and_two_threads)
{
const Pistache::Address address("localhost", Pistache::Port(0));
......@@ -248,3 +248,49 @@ TEST(http_client_test, one_client_with_multiple_requests_and_one_connection_per_
ASSERT_TRUE(response_counter == RESPONSE_SIZE);
}
TEST(http_client_test, one_client_with_multiple_requests_and_two_connections_per_host_and_one_thread)
{
const Pistache::Address address("localhost", Pistache::Port(0));
Http::Endpoint server(address);
auto flags = Tcp::Options::InstallSignalHandler | Tcp::Options::ReuseAddr;
auto server_opts = Http::Endpoint::options().flags(flags);
server.init(server_opts);
server.setHandler(Http::make_handler<HelloHandler>());
server.serveThreaded();
const std::string server_address = "localhost:" + server.getPort().toString();
std::cout << "Server address: " << server_address << "\n";
Http::Client client;
auto opts = Http::Client::options().maxConnectionsPerHost(2).threads(1);
client.init(opts);
std::vector<Async::Promise<Http::Response>> responses;
const int RESPONSE_SIZE = 6;
std::atomic<int> response_counter(0);
auto rb = client.get(server_address);
for (int i = 0; i < RESPONSE_SIZE; ++i)
{
auto response = rb.send();
response.then([&](Http::Response rsp)
{
if (rsp.code() == Http::Code::Ok)
++response_counter;
},
Async::IgnoreException);
responses.push_back(std::move(response));
}
auto sync = Async::whenAll(responses.begin(), responses.end());
Async::Barrier<std::vector<Http::Response>> barrier(sync);
barrier.wait_for(std::chrono::seconds(5));
server.shutdown();
client.shutdown();
ASSERT_TRUE(response_counter == RESPONSE_SIZE);
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment