Bug#34781248 pooled connection not used if io.threads larger than 1

When connection pooling is used and a connection to a destination is
picked from the pool, it is ignored if the connection is belongs to
another io-thread.

- By default, as many io-threads are starts as there are CPU-threads.
- Each new connection is assigned another io-thread (round-robin)

E.g. a 24 core machine, with 2-threads per core, leads to 48 io-threads.

If a server-connection is taken from the pool for the current
client-connection then it must match the right:

- destination AND
- io-thread

With 3 possible destinations and 48 io-threads, the connection pool
has to contain 3 * 48 connections to have a good chance that there is 1
matching connection for this client-connection.

The higher the io-thread count, the lower the cache-hit rate.

Change
------

- ignore the io-thread when taking connections from the pool
- move pooled-connections to the current client connection's io-thread
  when taken from the pool.

Change-Id: I586e7b5b1904a3a0020d6c8784928d549a1675e4

Author: weigon

router/src/connection_pool/include/mysqlrouter/connection_base.h

@@ -70,6 +70,9 @@ class ConnectionBase {
   [[nodiscard]] virtual std::string endpoint() const = 0;
 
   [[nodiscard]] virtual stdx::expected<void, std::error_code> cancel() = 0;
+
+  [[nodiscard]] virtual stdx::expected<void, std::error_code> set_io_context(
+      net::io_context &new_ctx) = 0;
 };
 
 #endif

router/src/routing/src/basic_protocol_splicer.h

@@ -127,6 +127,25 @@ class BasicConnection : public ConnectionBase {
 
   net::io_context &io_ctx() override { return sock_.get_executor().context(); }
 
+  stdx::expected<void, std::error_code> set_io_context(
+      net::io_context &new_ctx) override {
+    // nothing to do.
+    if (sock_.get_executor() == new_ctx.get_executor()) return {};
+
+    return sock_.release().and_then(
+        [this, &new_ctx](
+            auto native_handle) -> stdx::expected<void, std::error_code> {
+          socket_type new_sock(new_ctx);
+
+          auto assign_res = new_sock.assign(ep_.protocol(), native_handle);
+          if (!assign_res) return assign_res;
+
+          std::swap(sock_, new_sock);
+
+          return {};
+        });
+  }
+
   void async_recv(recv_buffer_type &buf,
                   std::function<void(std::error_code ec, size_t transferred)>
                       completion) override {

router/src/routing/src/classic_connect.cc

@@ -224,9 +224,6 @@ ConnectProcessor::from_pool() {
     return Result::Again;
   }
 
-  auto &io_ctx =
-      connection()->socket_splicer()->client_conn().connection()->io_ctx();
-
   auto &pools = ConnectionPoolComponent::get_instance();
 
   if (auto pool = pools.get(ConnectionPoolComponent::default_pool_name())) {
@@ -243,19 +240,16 @@ ConnectProcessor::from_pool() {
 
     auto pool_res = pool->pop_if(
         [client_caps, ep = mysqlrouter::to_string(server_endpoint_),
-         my_executor = io_ctx.get_executor(),
          requires_tls = connection()->requires_tls()](const auto &pooled_conn) {
           auto pooled_caps = pooled_conn.shared_capabilities();
 
           pooled_caps.reset(classic_protocol::capabilities::pos::ssl)
               .reset(classic_protocol::capabilities::pos::compress)
               .reset(classic_protocol::capabilities::pos::compress_zstd);
 
-          return pooled_conn.endpoint() == ep &&  //
-                 client_caps == pooled_caps &&    //
-                 pooled_conn.connection()->io_ctx().get_executor() ==
-                     my_executor &&
-                 (requires_tls == (bool)pooled_conn.ssl());
+          return (pooled_conn.endpoint() == ep &&  //
+                  client_caps == pooled_caps &&    //
+                  (requires_tls == (bool)pooled_conn.ssl()));
         });
 
     if (pool_res) {
@@ -269,12 +263,16 @@ ConnectProcessor::from_pool() {
         trace(Tracer::Event().stage(
             "connect::from_pool: " +
             destination_id_from_endpoint(*endpoints_it_)));
+
         // if the socket would be closed, recv() would return 0 for "eof".
         //
         // socket is still alive. good.
         socket_splicer->server_conn() =
             make_connection_from_pooled(std::move(*pool_res));
 
+        (void)socket_splicer->server_conn().connection()->set_io_context(
+            socket_splicer->client_conn().connection()->io_ctx());
+
         stage(Stage::Connected);
         return Result::Again;
       }