http/client: Respect user-provided timeout

There are two places that can wait for undefined amount of time -- the
one that waits the connection pool to release a connection and the one
that establishes a new connection.

The former is simple, as it uses conditional variable, so just wait() on
it with the timeout and return back timeout error if it fired.

Tha latter is trickier. New connections come from factory and it's not
guaranteed that a factory obeys it (e.g. see scylladb#2302). So instead of
relying on the factory, check if connected_socket appeared soon enough
and return back timeout error if it didn't.

Signed-off-by: Pavel Emelyanov <[email protected]>

include/seastar/http/exception.hh

@@ -158,6 +158,17 @@ public:
  {}
 };
 
+/**
+ * Client-side exception to report that making request timed out
+ * prior to communicating to server
+ */
+class timeout_error : public base_exception {
+public:
+ timeout_error()
+ : base_exception("client timed out", http::reply::status_type::request_timeout)
+ {}
+};
+
 SEASTAR_MODULE_EXPORT_END
 }
 

src/http/client.cc

@@ -252,7 +252,14 @@ future<client::connection_ptr> client::get_connection(clock_type::time_point tim
  }
 
  if (_nr_connections >= _max_connections) {
- return _wait_con.wait().then([this, timeout] {
+ return _wait_con.wait(timeout).then_wrapped([this, timeout] (auto f) {
+ try {
+ f.get();
+ } catch (const condition_variable_timed_out&) {
+ return make_exception_future<connection_ptr>(httpd::timeout_error());
+ } catch (...) {
+ return current_exception_as_future<connection_ptr>();
+ }
  return get_connection(timeout);
  });
  }
@@ -262,9 +269,17 @@ future<client::connection_ptr> client::get_connection(clock_type::time_point tim
 
 future<client::connection_ptr> client::make_connection(clock_type::time_point timeout) {
  _total_new_connections++;
- return _new_connections->make().then([cr = internal::client_ref(this)] (connected_socket cs) mutable {
+ return _new_connections->make().then([this, timeout, cr = internal::client_ref(this)] (connected_socket cs) mutable {
  http_log.trace("created new http connection {}", cs.local_address());
  auto con = seastar::make_shared<connection>(std::move(cs), std::move(cr));
+ if (clock_type::now() > timeout) {
+ // Factory made new connection, though it's too late already. Don't throw
+ // this effort out, but don't use it to serve current request either.
+ return put_connection(std::move(con)).then_wrapped([] (auto f) {
+ f.ignore_ready_future();
+ return make_exception_future<connection_ptr>(httpd::timeout_error());
+ });
+ }
  return make_ready_future<connection_ptr>(std::move(con));
  });
 }