Merge branch 'mysql-8.0' into mysql-trunk

Change-Id: I30be6c9461763b417e7f080f3eb1f0d963ad4a96

Author: weigon

router/src/routing/src/await_client_or_server.cc

@@ -48,9 +48,6 @@ stdx::expected<Processor::Result, std::error_code>
 AwaitClientOrServerProcessor::init() {
   stage(Stage::WaitBoth);
 
-  connection()->recv_from_either(
-      MysqlRoutingClassicConnectionBase::FromEither::Started);
-
   return Result::RecvFromBoth;
 }
 

router/src/routing/src/classic_connection_base.cc

@@ -234,39 +234,19 @@ void MysqlRoutingClassicConnectionBase::async_recv_client(Function next) {
   }
 
   ++active_work_;
-  this->socket_splicer()->async_recv_client([this, next](std::error_code ec,
-                                                         size_t transferred) {
-    (void)transferred;
-
-    --active_work_;
-
-    if (ec == std::errc::operation_canceled) {
-      // cancelled by:
-      //
-      // - request to shutdown
-      // - timer
-      // - read-from-client-xor-server
-      if (recv_from_either() ==
-          MysqlRoutingClassicConnectionBase::FromEither::RecvedFromServer) {
-        recv_from_either(MysqlRoutingClassicConnectionBase::FromEither::None);
-
-        return call_next_function(next);
-      }
-    } else {
-      read_timer().cancel();
-    }
+  this->socket_splicer()->async_recv_client(
+      [this, next](std::error_code ec, size_t transferred [[maybe_unused]]) {
+        --active_work_;
 
-    if (ec) return recv_client_failed(ec);
+        if (ec != std::errc::operation_canceled) {
+          read_timer().cancel();
+        }
 
-    if (recv_from_either() ==
-        MysqlRoutingClassicConnectionBase::FromEither::Started) {
-      recv_from_either(
-          MysqlRoutingClassicConnectionBase::FromEither::RecvedFromClient);
-    }
+        if (ec) return recv_client_failed(ec);
 
-    return trace_and_call_function(Tracer::Event::Direction::kClientToRouter,
-                                   "io::recv", next);
-  });
+        return trace_and_call_function(
+            Tracer::Event::Direction::kClientToRouter, "io::recv", next);
+      });
 }
 
 void MysqlRoutingClassicConnectionBase::async_send_server(Function next) {
@@ -302,6 +282,59 @@ void MysqlRoutingClassicConnectionBase::async_recv_server(Function next) {
   }
 
   ++active_work_;
+
+  this->socket_splicer()->async_recv_server(
+      [this, next](std::error_code ec, size_t transferred [[maybe_unused]]) {
+        --active_work_;
+
+        if (ec) return recv_server_failed(ec);
+
+        return trace_and_call_function(
+            Tracer::Event::Direction::kServerToRouter, "io::recv", next);
+      });
+}
+
+void MysqlRoutingClassicConnectionBase::async_recv_both(Function next) {
+  if (disconnect_requested()) {
+    return recv_client_failed(make_error_code(std::errc::operation_canceled));
+  }
+
+  recv_from_either(MysqlRoutingClassicConnectionBase::FromEither::Started);
+
+  ++active_work_;  // client
+  ++active_work_;  // server
+
+  this->socket_splicer()->async_recv_client([this, next](std::error_code ec,
+                                                         size_t transferred
+                                                         [[maybe_unused]]) {
+    --active_work_;
+
+    if (ec == std::errc::operation_canceled) {
+      // cancelled by:
+      //
+      // - request to shutdown
+      // - timer
+      // - read-from-client-xor-server
+      if (recv_from_either() ==
+          MysqlRoutingClassicConnectionBase::FromEither::RecvedFromServer) {
+        recv_from_either(MysqlRoutingClassicConnectionBase::FromEither::None);
+
+        return call_next_function(next);
+      }
+    }
+
+    if (ec) return recv_client_failed(ec);
+
+    if (recv_from_either() ==
+        MysqlRoutingClassicConnectionBase::FromEither::Started) {
+      recv_from_either(
+          MysqlRoutingClassicConnectionBase::FromEither::RecvedFromClient);
+    }
+
+    return trace_and_call_function(Tracer::Event::Direction::kClientToRouter,
+                                   "io::recv", next);
+  });
+
   this->socket_splicer()->async_recv_server([this, next](std::error_code ec,
                                                          size_t transferred) {
     (void)transferred;
@@ -764,9 +797,7 @@ void MysqlRoutingClassicConnectionBase::loop() {
       case Processor::Result::RecvFromServer:
         return async_recv_server(Function::kLoop);
       case Processor::Result::RecvFromBoth:
-        async_recv_client(Function::kLoop);
-        async_recv_server(Function::kLoop);
-        return;
+        return async_recv_both(Function::kLoop);
       case Processor::Result::SendToClient:
         return async_send_client(Function::kLoop);
       case Processor::Result::SendToServer:

router/src/routing/src/classic_connection_base.h

@@ -421,6 +421,8 @@ class MysqlRoutingClassicConnectionBase
 
   void async_recv_server(Function next);
 
+  void async_recv_both(Function next);
+
   void async_send_client_and_finish();
 
   void async_wait_client_closed();