Fixed Client::stop problem with more than one requests on threads

This commit is contained in:
yhirose 2020-06-12 11:04:37 -04:00
parent ec00fe5d5b
commit 5af7222217
2 changed files with 71 additions and 42 deletions

View File

@ -194,6 +194,7 @@ using socket_t = int;
#include <mutex> #include <mutex>
#include <random> #include <random>
#include <regex> #include <regex>
#include <set>
#include <string> #include <string>
#include <sys/stat.h> #include <sys/stat.h>
#include <thread> #include <thread>
@ -834,7 +835,8 @@ protected:
bool process_request(Stream &strm, const Request &req, Response &res, bool process_request(Stream &strm, const Request &req, Response &res,
bool last_connection, bool &connection_close); bool last_connection, bool &connection_close);
std::atomic<socket_t> sock_; std::set<socket_t> cli_socks_;
std::mutex cli_socks_mutex_;
const std::string host_; const std::string host_;
const int port_; const int port_;
@ -911,6 +913,7 @@ protected:
private: private:
socket_t create_client_socket() const; socket_t create_client_socket() const;
bool create_and_connect_socket(socket_t &sock);
bool read_response_line(Stream &strm, Response &res); bool read_response_line(Stream &strm, Response &res);
bool write_request(Stream &strm, const Request &req, bool last_connection); bool write_request(Stream &strm, const Request &req, bool last_connection);
bool redirect(const Request &req, Response &res); bool redirect(const Request &req, Response &res);
@ -1397,7 +1400,9 @@ public:
#endif #endif
private: private:
#ifdef CPPHTTPLIB_OPENSSL_SUPPORT
bool is_ssl_ = false; bool is_ssl_ = false;
#endif
std::shared_ptr<Client> cli_; std::shared_ptr<Client> cli_;
}; };
@ -4309,7 +4314,7 @@ inline Client::Client(const std::string &host, int port)
inline Client::Client(const std::string &host, int port, inline Client::Client(const std::string &host, int port,
const std::string &client_cert_path, const std::string &client_cert_path,
const std::string &client_key_path) const std::string &client_key_path)
: sock_(INVALID_SOCKET), host_(host), port_(port), : /*cli_sock_(INVALID_SOCKET),*/ host_(host), port_(port),
host_and_port_(host_ + ":" + std::to_string(port_)), host_and_port_(host_ + ":" + std::to_string(port_)),
client_cert_path_(client_cert_path), client_key_path_(client_key_path) {} client_cert_path_(client_cert_path), client_key_path_(client_key_path) {}
@ -4328,6 +4333,20 @@ inline socket_t Client::create_client_socket() const {
connection_timeout_usec_, interface_); connection_timeout_usec_, interface_);
} }
inline bool Client::create_and_connect_socket(socket_t &sock) {
sock = create_client_socket();
if (sock == INVALID_SOCKET) { return false; }
#ifdef CPPHTTPLIB_OPENSSL_SUPPORT
if (is_ssl() && !proxy_host_.empty()) {
Response res;
bool error;
if (!connect(sock, res, error)) { return error; }
}
#endif
return true;
}
inline bool Client::read_response_line(Stream &strm, Response &res) { inline bool Client::read_response_line(Stream &strm, Response &res) {
std::array<char, 2048> buf; std::array<char, 2048> buf;
@ -4347,54 +4366,58 @@ inline bool Client::read_response_line(Stream &strm, Response &res) {
} }
inline bool Client::send(const Request &req, Response &res) { inline bool Client::send(const Request &req, Response &res) {
sock_ = create_client_socket(); socket_t sock = INVALID_SOCKET;
if (sock_ == INVALID_SOCKET) { return false; } if (!create_and_connect_socket(sock)) { return false; }
#ifdef CPPHTTPLIB_OPENSSL_SUPPORT {
if (is_ssl() && !proxy_host_.empty()) { std::lock_guard<std::mutex> guard(cli_socks_mutex_);
bool error; cli_socks_.insert(sock);
if (!connect(sock_, res, error)) { return error; }
} }
#endif
return process_and_close_socket( auto ret = process_and_close_socket(
sock_, 1, sock, 1, [&](Stream &strm, bool last_connection, bool &connection_close) {
[&](Stream &strm, bool last_connection, bool &connection_close) {
return handle_request(strm, req, res, last_connection, return handle_request(strm, req, res, last_connection,
connection_close); connection_close);
}); });
{
std::lock_guard<std::mutex> guard(cli_socks_mutex_);
cli_socks_.erase(sock);
}
return ret;
} }
inline bool Client::send(const std::vector<Request> &requests, inline bool Client::send(const std::vector<Request> &requests,
std::vector<Response> &responses) { std::vector<Response> &responses) {
size_t i = 0; size_t i = 0;
while (i < requests.size()) { while (i < requests.size()) {
sock_ = create_client_socket(); socket_t sock = INVALID_SOCKET;
if (sock_ == INVALID_SOCKET) { return false; } if (!create_and_connect_socket(sock)) { return false; }
#ifdef CPPHTTPLIB_OPENSSL_SUPPORT {
if (is_ssl() && !proxy_host_.empty()) { std::lock_guard<std::mutex> guard(cli_socks_mutex_);
Response res; cli_socks_.insert(sock);
bool error;
if (!connect(sock_, res, error)) { return false; }
} }
#endif
if (!process_and_close_socket(sock_, requests.size() - i, auto ret = process_and_close_socket(
sock, requests.size() - i,
[&](Stream &strm, bool last_connection, [&](Stream &strm, bool last_connection,
bool &connection_close) -> bool { bool &connection_close) -> bool {
auto &req = requests[i++]; auto &req = requests[i++];
auto res = Response(); auto res = Response();
auto ret = handle_request(strm, req, res, auto ret =
last_connection, handle_request(strm, req, res, last_connection, connection_close);
connection_close); if (ret) { responses.emplace_back(std::move(res)); }
if (ret) {
responses.emplace_back(std::move(res));
}
return ret; return ret;
})) { });
return false;
{
std::lock_guard<std::mutex> guard(cli_socks_mutex_);
cli_socks_.erase(sock);
} }
if (!ret) { return false; }
} }
return true; return true;
@ -5062,11 +5085,12 @@ inline std::shared_ptr<Response> Client::Options(const char *path,
} }
inline void Client::stop() { inline void Client::stop() {
if (sock_ != INVALID_SOCKET) { std::lock_guard<std::mutex> guard(cli_socks_mutex_);
std::atomic<socket_t> sock(sock_.exchange(INVALID_SOCKET)); for (auto &sock : cli_socks_) {
detail::shutdown_socket(sock); detail::shutdown_socket(sock);
detail::close_socket(sock); detail::close_socket(sock);
} }
cli_socks_.clear();
} }
inline void Client::set_timeout_sec(time_t timeout_sec) { inline void Client::set_timeout_sec(time_t timeout_sec) {

View File

@ -1766,15 +1766,20 @@ TEST_F(ServerTest, GetStreamedEndless) {
} }
TEST_F(ServerTest, ClientStop) { TEST_F(ServerTest, ClientStop) {
thread t = thread([&]() { std::vector<std::thread> threads;
for (auto i = 0; i < 10; i++) {
threads.emplace_back(thread([&]() {
auto res = cli_.Get("/streamed-cancel", auto res = cli_.Get("/streamed-cancel",
[&](const char *, uint64_t) { return true; }); [&](const char *, uint64_t) { return true; });
ASSERT_TRUE(res == nullptr); ASSERT_TRUE(res == nullptr);
}); }));
}
std::this_thread::sleep_for(std::chrono::seconds(1)); std::this_thread::sleep_for(std::chrono::seconds(1));
cli_.stop(); cli_.stop();
for (auto& t: threads) {
t.join(); t.join();
} }
}
TEST_F(ServerTest, GetWithRange1) { TEST_F(ServerTest, GetWithRange1) {
auto res = cli_.Get("/with-range", {{make_range_header({{3, 5}})}}); auto res = cli_.Get("/with-range", {{make_range_header({{3, 5}})}});