Skip to content

Commit b130149

Browse files
author
Valtteri Heikkila
committed
More reafactoring of linux_connection http_linux.cpp
1 parent 49d0245 commit b130149

File tree

1 file changed

+96
-61
lines changed

1 file changed

+96
-61
lines changed

‎Release/src/http/client/http_linux.cpp

Lines changed: 96 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -61,18 +61,19 @@ namespace web { namespace http
6161
m_socket(io_service),
6262
m_pool_timer(io_service),
6363
m_pool_weak(pool_weak),
64-
m_is_reused(false)
64+
m_is_reused(false),
65+
m_keep_alive(true)
6566
{}
6667

67-
std::weak_ptr<linux_connection_pool> m_pool_weak;
68-
tcp::socket m_socket;
69-
boost::asio::deadline_timer m_pool_timer;
70-
bool m_is_reused;
71-
7268
void handle_pool_timer(const boost::system::error_code& ec);
7369

74-
bool close()
70+
bool close(bool cancel_timer)
7571
{
72+
if (cancel_timer)
73+
{
74+
m_pool_timer.cancel();
75+
}
76+
m_keep_alive = false;
7677
boost::system::error_code error;
7778
m_socket.shutdown(tcp::socket::shutdown_both, error);
7879
m_socket.close(error);
@@ -85,6 +86,33 @@ namespace web { namespace http
8586
m_socket.cancel(error);
8687
return !error;
8788
}
89+
90+
template <typename TimeoutHandler>
91+
void start_pool_timer(int timeout_secs, TimeoutHandler handler)
92+
{
93+
m_pool_timer.expires_from_now(boost::posix_time::milliseconds(timeout_secs * 1000));
94+
m_pool_timer.async_wait(handler);
95+
}
96+
97+
void start_reuse()
98+
{
99+
m_pool_timer.cancel();
100+
m_is_reused = true;
101+
}
102+
103+
bool is_reused() const { return m_is_reused; }
104+
105+
bool keep_alive() const { return m_keep_alive; }
106+
void set_keep_alive(bool keep_alive) { m_keep_alive = keep_alive; }
107+
108+
tcp::socket& socket() { return m_socket; }
109+
110+
private:
111+
tcp::socket m_socket;
112+
boost::asio::deadline_timer m_pool_timer;
113+
std::weak_ptr<linux_connection_pool> m_pool_weak;
114+
bool m_is_reused;
115+
bool m_keep_alive;
88116
};
89117

90118
class linux_connection_pool : public std::enable_shared_from_this<linux_connection_pool>
@@ -93,46 +121,51 @@ namespace web { namespace http
93121

94122
linux_connection_pool(boost::asio::io_service& io_service, utility::seconds idle_timeout) :
95123
m_io_service(io_service),
96-
m_timeout_secs(idle_timeout)
124+
m_timeout_secs(static_cast<int>(idle_timeout.count()))
97125
{}
98126

99127
~linux_connection_pool()
100128
{
101129
std::lock_guard<std::mutex> lock(m_connections_mutex);
102130
for (auto& connection : m_connections)
103131
{
104-
connection->m_pool_timer.cancel();
105-
connection->close();
132+
connection->close(true);
106133
}
107134
}
108135

109136
void release(std::shared_ptr<linux_connection> connection)
110137
{
111-
std::lock_guard<std::mutex> lock(m_connections_mutex);
112-
113-
const int secs = static_cast<int>(m_timeout_secs.count());
114-
connection->m_pool_timer.expires_from_now(boost::posix_time::milliseconds(secs * 1000));
115-
connection->m_pool_timer.async_wait(boost::bind(&linux_connection::handle_pool_timer, connection, boost::asio::placeholders::error));
138+
if (connection->keep_alive() && (m_timeout_secs > 0))
139+
{
140+
connection->cancel();
141+
// Remove idle connections from pool after timeout.
142+
connection->start_pool_timer(m_timeout_secs, boost::bind(&linux_connection::handle_pool_timer, connection, boost::asio::placeholders::error));
116143

117-
m_connections.insert(connection);
144+
{
145+
std::lock_guard<std::mutex> lock(m_connections_mutex);
146+
m_connections.insert(connection);
147+
}
148+
}
149+
else
150+
{
151+
// Connection is not returned to pool => will be destroyed.
152+
connection->close(false);
153+
}
118154
}
119155

120156
std::shared_ptr<linux_connection> obtain()
121157
{
122-
std::lock_guard<std::mutex> lock(m_connections_mutex);
123-
124-
if (m_connections.empty())
158+
if (is_connections_empty())
125159
{
160+
// No connections in pool => create new connection instance.
126161
return std::make_shared<linux_connection>(shared_from_this(), m_io_service);
127162
}
128163
else
129164
{
130-
(*m_connections.begin())->m_pool_timer.cancel();
131-
auto result(*m_connections.begin());
132-
result->m_is_reused = true;
133-
m_connections.erase(m_connections.begin());
134-
135-
return result;
165+
// Reuse connection from pool.
166+
auto connection(obtain_begin());
167+
connection->start_reuse();
168+
return connection;
136169
}
137170
}
138171

@@ -143,8 +176,22 @@ namespace web { namespace http
143176
}
144177

145178
private:
179+
bool is_connections_empty()
180+
{
181+
std::lock_guard<std::mutex> lock(m_connections_mutex);
182+
return m_connections.empty();
183+
}
184+
185+
std::shared_ptr<linux_connection> obtain_begin()
186+
{
187+
std::lock_guard<std::mutex> lock(m_connections_mutex);
188+
auto connection(*m_connections.begin());
189+
m_connections.erase(m_connections.begin());
190+
return connection;
191+
}
192+
146193
boost::asio::io_service& m_io_service;
147-
const utility::seconds m_timeout_secs;
194+
const int m_timeout_secs;
148195
std::unordered_set<std::shared_ptr<linux_connection> > m_connections;
149196
std::mutex m_connections_mutex;
150197
};
@@ -216,7 +263,6 @@ namespace web { namespace http
216263
bool m_timedout;
217264
boost::asio::streambuf m_body_buf;
218265
boost::asio::deadline_timer m_timeout_timer;
219-
bool m_close_socket_in_destructor;
220266

221267
~linux_client_request_context();
222268

@@ -271,7 +317,7 @@ namespace web { namespace http
271317
{
272318
boost::asio::ssl::context context(boost::asio::ssl::context::sslv23);
273319
context.set_default_verify_paths();
274-
ctx->m_ssl_stream.reset(new boost::asio::ssl::stream<boost::asio::ip::tcp::socket &>(ctx->m_connection->m_socket, context));
320+
ctx->m_ssl_stream.reset(new boost::asio::ssl::stream<boost::asio::ip::tcp::socket &>(ctx->m_connection->socket(), context));
275321
}
276322

277323
auto encoded_resource = uri_builder(m_uri).append(ctx->m_request.relative_uri()).to_uri().resource().to_string();
@@ -346,7 +392,7 @@ namespace web { namespace http
346392

347393
ctx->set_timer(static_cast<int>(client_config().timeout().count()));
348394

349-
if (ctx->m_connection->m_socket.is_open())
395+
if (ctx->m_connection->socket().is_open())
350396
{
351397
write_request(ctx);
352398
}
@@ -363,8 +409,7 @@ namespace web { namespace http
363409
ctx->m_cancellationRegistration = request_ctx->m_request._cancellation_token().register_callback([ctx]()
364410
{
365411
ctx->m_connection->cancel();
366-
ctx->m_connection->close();
367-
ctx->m_close_socket_in_destructor = true;
412+
ctx->m_connection->close(false);
368413
});
369414
}
370415
}
@@ -414,7 +459,7 @@ namespace web { namespace http
414459
ctx->m_ssl_stream->set_verify_mode(boost::asio::ssl::context::verify_none);
415460
}
416461
}
417-
ctx->m_connection->m_socket.async_connect(endpoint, boost::bind(&linux_client::handle_connect, shared_from_this(), boost::asio::placeholders::error, ++endpoints, ctx));
462+
ctx->m_connection->socket().async_connect(endpoint, boost::bind(&linux_client::handle_connect, shared_from_this(), boost::asio::placeholders::error, ++endpoints, ctx));
418463
}
419464
}
420465

@@ -426,7 +471,7 @@ namespace web { namespace http
426471
}
427472
else
428473
{
429-
boost::asio::async_write(ctx->m_connection->m_socket, ctx->m_body_buf, boost::bind(&linux_client::handle_write_request, shared_from_this(), boost::asio::placeholders::error, ctx));
474+
boost::asio::async_write(ctx->m_connection->socket(), ctx->m_body_buf, boost::bind(&linux_client::handle_write_request, shared_from_this(), boost::asio::placeholders::error, ctx));
430475
}
431476
}
432477

@@ -444,15 +489,15 @@ namespace web { namespace http
444489
{
445490
ctx->m_timeout_timer.cancel();
446491

447-
ctx->m_connection->close();
492+
ctx->m_connection->close(false);
448493
ctx->m_connection = m_pool->obtain();
449494

450495
auto endpoint = *endpoints;
451496
if (ctx->m_ssl_stream)
452497
{
453498
boost::asio::ssl::context context(boost::asio::ssl::context::sslv23);
454499
context.set_default_verify_paths();
455-
ctx->m_ssl_stream.reset(new boost::asio::ssl::stream<boost::asio::ip::tcp::socket &>(ctx->m_connection->m_socket, context));
500+
ctx->m_ssl_stream.reset(new boost::asio::ssl::stream<boost::asio::ip::tcp::socket &>(ctx->m_connection->socket(), context));
456501

457502
// Check to turn off server certificate verification.
458503
if(client_config().validate_certificates())
@@ -466,7 +511,7 @@ namespace web { namespace http
466511
}
467512
}
468513

469-
ctx->m_connection->m_socket.async_connect(endpoint, boost::bind(&linux_client::handle_connect, shared_from_this(), boost::asio::placeholders::error, ++endpoints, ctx));
514+
ctx->m_connection->socket().async_connect(endpoint, boost::bind(&linux_client::handle_connect, shared_from_this(), boost::asio::placeholders::error, ++endpoints, ctx));
470515
}
471516
}
472517

@@ -540,12 +585,12 @@ namespace web { namespace http
540585
{
541586
if (readSize != 0)
542587
{
543-
boost::asio::async_write(ctx->m_connection->m_socket, ctx->m_body_buf,
588+
boost::asio::async_write(ctx->m_connection->socket(), ctx->m_body_buf,
544589
boost::bind(&linux_client::handle_write_chunked_body, shared_from_this(), boost::asio::placeholders::error, ctx));
545590
}
546591
else
547592
{
548-
boost::asio::async_write(ctx->m_connection->m_socket, ctx->m_body_buf,
593+
boost::asio::async_write(ctx->m_connection->socket(), ctx->m_body_buf,
549594
boost::bind(&linux_client::handle_write_body, shared_from_this(), boost::asio::placeholders::error, ctx));
550595
}
551596
}
@@ -600,7 +645,7 @@ namespace web { namespace http
600645
}
601646
else
602647
{
603-
boost::asio::async_write(ctx->m_connection->m_socket, ctx->m_body_buf,
648+
boost::asio::async_write(ctx->m_connection->socket(), ctx->m_body_buf,
604649
boost::bind(&linux_client::handle_write_large_body, shared_from_this(), boost::asio::placeholders::error, ctx));
605650
}
606651
});
@@ -653,7 +698,7 @@ namespace web { namespace http
653698
}
654699
else
655700
{
656-
boost::asio::async_read_until(ctx->m_connection->m_socket, ctx->m_body_buf, CRLF+CRLF,
701+
boost::asio::async_read_until(ctx->m_connection->socket(), ctx->m_body_buf, CRLF+CRLF,
657702
boost::bind(&linux_client::handle_status_line, shared_from_this(), boost::asio::placeholders::error, ctx));
658703
}
659704
}
@@ -695,12 +740,11 @@ namespace web { namespace http
695740
const bool socket_was_closed((boost::asio::error::eof == ec)
696741
|| (boost::asio::error::connection_reset == ec)
697742
|| (boost::asio::error::connection_aborted == ec));
698-
if (socket_was_closed && ctx->m_connection->m_is_reused && ctx->m_connection->m_socket.is_open())
743+
if (socket_was_closed && ctx->m_connection->is_reused() && ctx->m_connection->socket().is_open())
699744
{
700745
// Connection was closed by the server for some reason during the connection was
701746
// being pooled. We re-send the request to get a new connection.
702-
ctx->m_connection->close();
703-
ctx->m_close_socket_in_destructor = true;
747+
ctx->m_connection->close(false);
704748

705749
auto new_ctx = details::linux_client_request_context::create_request_context(ctx->m_http_client, ctx->m_request);
706750
new_ctx->m_request_completion = ctx->m_request_completion;
@@ -740,10 +784,11 @@ namespace web { namespace http
740784

741785
if (boost::iequals(name, header_names::connection))
742786
{
743-
// This assumes server uses HTTP/1.1 so that 'Keep-Alive' is the default.
744-
// We don't handle HTTP/1.0 server explicitly here. HTTP/1.0 server would need
745-
// to respond with 'Connection: Keep-Alive' to resume connection.
746-
ctx->m_close_socket_in_destructor = boost::iequals(value, U("close"));
787+
// This assumes server uses HTTP/1.1 so that 'Keep-Alive' is the default,
788+
// so connection is explicitly closed only if we get "Connection: close".
789+
// We don't handle HTTP/1.0 server here. HTTP/1.0 server would need
790+
// to respond using 'Connection: Keep-Alive' every time.
791+
ctx->m_connection->set_keep_alive(!boost::iequals(value, U("close")));
747792
}
748793

749794
ctx->m_response.headers().add(std::move(name), std::move(value));
@@ -792,7 +837,7 @@ namespace web { namespace http
792837
}
793838
else
794839
{
795-
boost::asio::async_read_until(ctx->m_connection->m_socket, ctx->m_body_buf, CRLF,
840+
boost::asio::async_read_until(ctx->m_connection->socket(), ctx->m_body_buf, CRLF,
796841
boost::bind(&linux_client::handle_chunk_header, shared_from_this(), boost::asio::placeholders::error, ctx));
797842
}
798843
}
@@ -814,7 +859,7 @@ namespace web { namespace http
814859
}
815860
else
816861
{
817-
boost::asio::async_read(ctx->m_connection->m_socket, ctx->m_body_buf, boost::asio::transfer_at_least(size_to_read), handler);
862+
boost::asio::async_read(ctx->m_connection->socket(), ctx->m_body_buf, boost::asio::transfer_at_least(size_to_read), handler);
818863
}
819864
}
820865

@@ -909,7 +954,7 @@ namespace web { namespace http
909954
}
910955
else
911956
{
912-
boost::asio::async_read_until(ctx->m_connection->m_socket, ctx->m_body_buf, CRLF,
957+
boost::asio::async_read_until(ctx->m_connection->socket(), ctx->m_body_buf, CRLF,
913958
boost::bind(&linux_client::handle_chunk_header, shared_from_this(), boost::asio::placeholders::error, ctx));
914959
}
915960
});
@@ -1024,7 +1069,6 @@ namespace web { namespace http
10241069
, m_current_size(0)
10251070
, m_timeout_timer(crossplat::threadpool::shared_instance().service())
10261071
, m_connection(connection)
1027-
, m_close_socket_in_destructor(false)
10281072
{}
10291073

10301074
std::shared_ptr<request_context> linux_client_request_context::create_request_context(
@@ -1038,16 +1082,7 @@ namespace web { namespace http
10381082
linux_client_request_context::~linux_client_request_context()
10391083
{
10401084
m_timeout_timer.cancel();
1041-
1042-
if (m_close_socket_in_destructor)
1043-
{
1044-
m_connection->close();
1045-
}
1046-
else
1047-
{
1048-
m_connection->cancel();
1049-
std::static_pointer_cast<linux_client>(m_http_client)->m_pool->release(m_connection);
1050-
}
1085+
std::static_pointer_cast<linux_client>(m_http_client)->m_pool->release(m_connection);
10511086
}
10521087

10531088
void linux_connection::handle_pool_timer(const boost::system::error_code& ec)

0 commit comments

Comments
 (0)