Merge branch 'mysql-8.0' into mysql-trunk

Change-Id: Ib43ba2dc405d166387427b638fc674195811706f

Author: weigon

router/src/routing/src/CMakeLists.txt

@@ -56,8 +56,9 @@ ADD_LIBRARY(routing SHARED
   connection.cc
   blocked_endpoints.cc
 
-  processor.cc
+  await_client_or_server.cc
   forwarding_processor.cc
+  processor.cc
 
   classic_auth.cc
   classic_auth_cleartext.cc

router/src/routing/src/await_client_or_server.cc

@@ -0,0 +1,129 @@
+/*
+  Copyright (c) 2023, Oracle and/or its affiliates.
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License, version 2.0,
+  as published by the Free Software Foundation.
+
+  This program is also distributed with certain software (including
+  but not limited to OpenSSL) that is licensed under separate terms,
+  as designated in a particular file or component or in included license
+  documentation.  The authors of MySQL hereby grant you an additional
+  permission to link the program and your derivative works with the
+  separately licensed software that they have included with MySQL.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+*/
+
+#include "await_client_or_server.h"
+
+#include "classic_connection_base.h"
+
+stdx::expected<Processor::Result, std::error_code>
+AwaitClientOrServerProcessor::process() {
+  switch (stage()) {
+    case Stage::Init:
+      return init();
+    case Stage::WaitBoth:
+      return wait_both();
+    case Stage::WaitClientCancelled:
+      return wait_client_cancelled();
+    case Stage::WaitServerCancelled:
+      return wait_server_cancelled();
+    case Stage::Done:
+      return Result::Done;
+  }
+
+  harness_assert_this_should_not_execute();
+}
+
+stdx::expected<Processor::Result, std::error_code>
+AwaitClientOrServerProcessor::init() {
+  stage(Stage::WaitBoth);
+
+  connection()->recv_from_either(
+      MysqlRoutingClassicConnectionBase::FromEither::Started);
+
+  return Result::RecvFromBoth;
+}
+
+/**
+ * wait for an read-event from client and server at the same time.
+ *
+ * two async-reads have been started, which both will call wait_both(). Only one
+ * of the two should continue.
+ *
+ * To ensure that event handlers are properly synchronized:
+ *
+ * - the first returning event, cancels the other waiter and leaves without
+ *   "returning" (::Void)
+ * - the cancelled side, continues with executing.
+ */
+stdx::expected<Processor::Result, std::error_code>
+AwaitClientOrServerProcessor::wait_both() {
+  auto *socket_splicer = connection()->socket_splicer();
+
+  switch (connection()->recv_from_either()) {
+    case MysqlRoutingClassicConnectionBase::FromEither::RecvedFromServer: {
+      // server side sent something.
+      //
+      // - cancel the client side
+      // - read from server in ::wait_client_cancelled
+
+      stage(Stage::WaitClientCancelled);
+
+      (void)socket_splicer->client_conn().cancel();
+
+      // end this execution branch.
+      return Result::Void;
+    }
+    case MysqlRoutingClassicConnectionBase::FromEither::RecvedFromClient: {
+      // client side sent something
+      //
+      // - cancel the server side
+      // - read from client in ::wait_server_cancelled
+      stage(Stage::WaitServerCancelled);
+
+      (void)socket_splicer->server_conn().cancel();
+
+      // end this execution branch.
+      return Result::Void;
+    }
+    case MysqlRoutingClassicConnectionBase::FromEither::None:
+    case MysqlRoutingClassicConnectionBase::FromEither::Started:
+      break;
+  }
+
+  harness_assert_this_should_not_execute();
+}
+
+stdx::expected<Processor::Result, std::error_code>
+AwaitClientOrServerProcessor::wait_server_cancelled() {
+  stage(Stage::Done);
+
+  on_done_(AwaitResult::ClientReadable);
+
+  return Result::Again;
+}
+
+/**
+ * read-event from server while waiting for client command.
+ *
+ * - either a connection-close by the server or
+ * - ERR packet before connection-close.
+ */
+stdx::expected<Processor::Result, std::error_code>
+AwaitClientOrServerProcessor::wait_client_cancelled() {
+  stage(Stage::Done);
+
+  on_done_(AwaitResult::ServerReadable);
+
+  return Result::Again;
+}

router/src/routing/src/await_client_or_server.h

@@ -0,0 +1,66 @@
+/*
+  Copyright (c) 2023, Oracle and/or its affiliates.
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License, version 2.0,
+  as published by the Free Software Foundation.
+
+  This program is also distributed with certain software (including
+  but not limited to OpenSSL) that is licensed under separate terms,
+  as designated in a particular file or component or in included license
+  documentation.  The authors of MySQL hereby grant you an additional
+  permission to link the program and your derivative works with the
+  separately licensed software that they have included with MySQL.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+*/
+
+#ifndef ROUTING_CLASSIC_AWAIT_CLIENT_OR_SERVER_PROCESSOR_INCLUDED
+#define ROUTING_CLASSIC_AWAIT_CLIENT_OR_SERVER_PROCESSOR_INCLUDED
+
+#include "processor.h"
+
+class AwaitClientOrServerProcessor : public BasicProcessor {
+ public:
+  enum class AwaitResult {
+    ClientReadable,
+    ServerReadable,
+  };
+
+  AwaitClientOrServerProcessor(
+      MysqlRoutingClassicConnectionBase *conn,
+      std::function<void(stdx::expected<AwaitResult, std::error_code>)> on_done)
+      : BasicProcessor(conn), on_done_(std::move(on_done)) {}
+
+  stdx::expected<Result, std::error_code> process() override;
+
+ private:
+  enum class Stage {
+    Init,
+    WaitBoth,
+    WaitClientCancelled,
+    WaitServerCancelled,
+    Done,
+  };
+
+  void stage(Stage stage) { stage_ = stage; }
+  Stage stage() const { return stage_; }
+
+  stdx::expected<Result, std::error_code> init();
+  stdx::expected<Result, std::error_code> wait_both();
+  stdx::expected<Result, std::error_code> wait_client_cancelled();
+  stdx::expected<Result, std::error_code> wait_server_cancelled();
+
+  Stage stage_{Stage::Init};
+
+  std::function<void(stdx::expected<AwaitResult, std::error_code>)> on_done_;
+};
+
+#endif

router/src/routing/src/classic_command.cc

@@ -28,6 +28,7 @@
 #include <memory>  // make_unique
 #include <string>
 
+#include "await_client_or_server.h"
 #include "classic_binlog_dump_forwarder.h"
 #include "classic_change_user_forwarder.h"
 #include "classic_clone_forwarder.h"
@@ -55,6 +56,7 @@
 #include "harness_assert.h"
 #include "hexify.h"
 #include "mysql/harness/logging/logging.h"
+#include "mysql/harness/stdx/expected.h"
 #include "mysql/harness/tls_error.h"
 #include "mysqld_error.h"
 #include "mysqlrouter/connection_pool.h"
@@ -71,10 +73,6 @@ stdx::expected<Processor::Result, std::error_code> CommandProcessor::process() {
       return is_authed();
     case Stage::WaitBoth:
       return wait_both();
-    case Stage::WaitClientCancelled:
-      return wait_client_cancelled();
-    case Stage::WaitServerCancelled:
-      return wait_server_cancelled();
     case Stage::Command:
       return command();
     case Stage::Done:
@@ -317,74 +315,43 @@ class SelectSessionCollationConnectionHandler : public QuerySender::Handler {
  */
 stdx::expected<Processor::Result, std::error_code>
 CommandProcessor::wait_both() {
-  auto *socket_splicer = connection()->socket_splicer();
-
-  if (connection()->recv_from_either() ==
-      MysqlRoutingClassicConnectionBase::FromEither::RecvedFromServer) {
-    // server side sent something.
-    //
-    // - cancel the client side
-    // - read from server in ::wait_client_cancelled
-
-    stage(Stage::WaitClientCancelled);
+  if (wait_both_result_) {
+    switch (*wait_both_result_) {
+      case AwaitClientOrServerProcessor::AwaitResult::ClientReadable:
+        stage(Stage::Command);
 
-    (void)socket_splicer->client_conn().cancel();
+        return Result::Again;
+      case AwaitClientOrServerProcessor::AwaitResult::ServerReadable: {
+        auto *socket_splicer = connection()->socket_splicer();
 
-    // end this execution branch.
-    return Result::Void;
-  } else if (connection()->recv_from_either() ==
-             MysqlRoutingClassicConnectionBase::FromEither::RecvedFromClient) {
-    // client side sent something
-    //
-    // - cancel the server side
-    // - read from client in ::wait_server_cancelled
-    stage(Stage::WaitServerCancelled);
-
-    (void)socket_splicer->server_conn().cancel();
-
-    // end this execution branch.
-    return Result::Void;
-  }
+        auto *src_channel = socket_splicer->server_channel();
+        auto *src_protocol = connection()->server_protocol();
 
-  harness_assert_this_should_not_execute();
-}
-
-stdx::expected<Processor::Result, std::error_code>
-CommandProcessor::wait_server_cancelled() {
-  stage(Stage::Command);
-
-  return Result::Again;
-}
+        auto read_res =
+            ClassicFrame::ensure_has_msg_prefix(src_channel, src_protocol);
+        if (!read_res) return recv_server_failed(read_res.error());
 
-/**
- * read-event from server while waiting for client command.
- *
- * - either a connection-close by the server or
- * - ERR packet before connection-close.
- */
-stdx::expected<Processor::Result, std::error_code>
-CommandProcessor::wait_client_cancelled() {
-  auto *socket_splicer = connection()->socket_splicer();
+        stage(Stage::Done);
 
-  auto dst_channel = socket_splicer->server_channel();
-  auto dst_protocol = connection()->server_protocol();
+        if (auto &tr = tracer()) {
+          tr.trace(Tracer::Event().stage("server::error"));
+        }
 
-  auto read_res =
-      ClassicFrame::ensure_has_msg_prefix(dst_channel, dst_protocol);
-  if (!read_res) return recv_server_failed(read_res.error());
+        // should be a Error packet.
+        return forward_server_to_client();
+      }
+    }
 
-  if (auto &tr = tracer()) {
-    tr.trace(Tracer::Event().stage("server::error"));
+    harness_assert_this_should_not_execute();
+  } else {
+    return stdx::make_unexpected(wait_both_result_.error());
   }
-
-  // should be a Error packet.
-  return forward_server_to_client();
 }
 
 stdx::expected<Processor::Result, std::error_code> CommandProcessor::command() {
   auto *socket_splicer = connection()->socket_splicer();
-  auto src_channel = socket_splicer->client_channel();
-  auto src_protocol = connection()->client_protocol();
+  auto *src_channel = socket_splicer->client_channel();
+  auto *src_protocol = connection()->client_protocol();
   auto &server_conn = socket_splicer->server_conn();
 
   if (connection()->disconnect_requested()) {
@@ -470,12 +437,14 @@ stdx::expected<Processor::Result, std::error_code> CommandProcessor::command() {
         //
         // watch server-side for connection-close
 
-        stage(Stage::WaitBoth);
+        connection()->push_processor(
+            std::make_unique<AwaitClientOrServerProcessor>(
+                connection(),
+                [this](auto result) { wait_both_result_ = result; }));
 
-        connection()->recv_from_either(
-            MysqlRoutingClassicConnectionBase::FromEither::Started);
+        stage(Stage::WaitBoth);
 
-        return Result::RecvFromBoth;
+        return Result::Again;
       }
     }
 

router/src/routing/src/classic_command.h

@@ -27,15 +27,15 @@
 
 #include "forwarding_processor.h"
 
+#include "await_client_or_server.h"
+
 class CommandProcessor : public ForwardingProcessor {
  public:
   using ForwardingProcessor::ForwardingProcessor;
 
   enum class Stage {
     IsAuthed,
     WaitBoth,
-    WaitClientCancelled,
-    WaitServerCancelled,
     Command,
     Done,
   };
@@ -48,13 +48,14 @@ class CommandProcessor : public ForwardingProcessor {
  private:
   stdx::expected<Result, std::error_code> is_authed();
   stdx::expected<Result, std::error_code> wait_both();
-  stdx::expected<Result, std::error_code> wait_client_cancelled();
-  stdx::expected<Result, std::error_code> wait_server_cancelled();
   stdx::expected<Result, std::error_code> command();
 
   void client_idle_timeout();
 
   Stage stage_{Stage::IsAuthed};
+
+  stdx::expected<AwaitClientOrServerProcessor::AwaitResult, std::error_code>
+      wait_both_result_{};
 };
 
 #endif