MDEV-37316

rpl/remote_event_stream.cc

@@ -0,0 +1,169 @@
+/*
+  Copyright (c) 2026 MariaDB
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation; version 2 of the License.
+
+  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-1335 USA.
+*/
+
+#include "remote_event_stream.h"
+#include <my_global.h> // required for `sql_common.h`; don't know which part
+#include <m_string.h> // required for `sql_common.h` to fix "C++ in C linkage"
+#include <sql_common.h> // `mysql.h`, mysql_reconnect()
+#include <violite.h> // vio_close(), vio_shutdown()
+
+
+void Remote_event_stream::Connection_options::operator()(MYSQL *connector) const
+{
+  //TODO: Connector/C deprecated mysql_options() in favor of `mysql_optionsv()`.
+  /*
+  mysql_optionsv(connector, MARIADB_OPT_HOST, host);
+  mysql_optionsv(connector, MARIADB_OPT_USER, user);
+  mysql_optionsv(connector, MARIADB_OPT_PASSWORD, password);
+  mysql_optionsv(connector, MARIADB_OPT_PORT, &port);
+  */
+  connector->host= host; connector->user= user; connector->passwd= password;
+  connector->port= port;
+  if (ssl_options)
+  {
+    mysql_ssl_set(connector,
+      ssl_options->ssl_key, ssl_options->ssl_cert,
+      ssl_options->ssl_ca, ssl_options->ssl_capath, ssl_options->ssl_cipher);
+    mysql_options(connector, MYSQL_OPT_SSL_CRL, ssl_options->ssl_crl);
+    mysql_options(connector, MYSQL_OPT_SSL_CRLPATH, ssl_options->ssl_crlpath);
+    mysql_options(connector, MYSQL_OPT_SSL_VERIFY_SERVER_CERT,
+                  &ssl_options->ssl_verify_server_cert);
+  }
+  else
+    //mysql_optionsv(connector, MYSQL_OPT_SSL_ENFORCE, &use_ssl);
+    connector->options.use_ssl= false;
+  mysql_options(connector, MYSQL_SET_CHARSET_NAME, charset_name);
+  // In case the master asks for an external authentication plugin
+  mysql_options(connector, MYSQL_PLUGIN_DIR, plugin_dir);
+}
+
+
+Rpl_slave_connection::Rpl_slave_connection(
+  const Connection_options &options, uint32_t timeout)
+{
+  connector= mysql_init(nullptr);
+  if (!connector)
+    return;
+  mysql_options(connector, MYSQL_OPT_CONNECT_TIMEOUT, &timeout);
+  mysql_options(connector, MYSQL_OPT_READ_TIMEOUT, &timeout);
+  options(connector);
+}
+
+Remote_event_stream::Remote_event_stream(
+  const Connection_options &options, uint32_t timeout,
+  unsigned long max_allowed_packet): Rpl_slave_connection(options, timeout)
+{
+  if (!connector)
+    return;
+  unsigned char yes= true;
+  //@deprecated not applicable with Connector/C
+  mysql_options(connector, MYSQL_OPT_USE_THREAD_SPECIFIC_MEMORY, &yes);
+  //mysql_options(connector, MYSQL_OPT_MAX_ALLOWED_PACKET, &max_allowed_packet);
+  connector->options.max_allowed_packet= max_allowed_packet;
+  mysql_options(connector, MYSQL_OPT_RECONNECT, &yes);
+}
+
+Semi_sync_graceful_killer::Semi_sync_graceful_killer(
+  const Connection_options &options, uint32_t timeout):
+  Rpl_slave_connection(options, timeout)
+{
+  if (connector)
+    mysql_options(connector, MYSQL_OPT_WRITE_TIMEOUT, &timeout);
+}
+
+Rpl_slave_connection::~Rpl_slave_connection()
+{
+  if (connector)
+  {
+    vio_close(connector->net.vio); //@deprecated already covered by Connector/C
+    mysql_close(connector);
+  }
+}
+
+
+unsigned int Remote_event_stream::errnum() { return mysql_errno(connector); }
+const char *Remote_event_stream::errmsg() { return mysql_error(connector); }
+
+/** Call mysql_real_connect() with matching properties in the struct
+  TODO: Refactor Connector/C to avoid re-passing these parameters
+*/
+bool Rpl_slave_connection::connect(unsigned long flags)
+{
+  return !mysql_real_connect(connector,
+    connector->host, connector->user, connector->passwd, connector->db,
+    connector->port, connector->unix_socket, flags);
+}
+
+bool Remote_event_stream::connect(bool compress)
+{
+  if (do_reconnect)
+    return mysql_reconnect(connector);
+  do_reconnect= true;
+  return Rpl_slave_connection::connect(
+    CLIENT_REMEMBER_OPTIONS | (compress ? CLIENT_COMPRESS : 0));
+}
+
+
+bool Rpl_slave_connection::real_query(const char *query, size_t strlen)
+{ return mysql_real_query(connector, query, strlen); }
+
+unsigned long Remote_event_stream::master_version()
+{ return mysql_get_server_version(connector); }
+
+MYSQL_RES_P Remote_event_stream::store_result()
+{ return mysql_store_result(connector); }
+char **Remote_event_stream::fetch_row(MYSQL_RES_P query_result)
+{ return mysql_fetch_row(query_result); }
+void Remote_event_stream::free_result(MYSQL_RES_P query_result)
+{ return mysql_free_result(query_result); }
+
+bool Remote_event_stream::send_command(
+  int command, const unsigned char *args, size_t strlen, bool skip_check)
+{
+  return simple_command(connector, static_cast<enum_server_command>(command),
+    args, static_cast<unsigned long>(strlen), false);
+}
+
+unsigned long Remote_event_stream::thread_id()
+{ return connector->thread_id; }
+
+
+///TODO: Split this part from Connector/C's mariadb_rpl_fetch()
+std::basic_string_view<unsigned char> Remote_event_stream::next()
+{
+  auto strlen= static_cast<size_t>(cli_safe_read(connector));
+  return {connector->net.read_pos, strlen};
+}
+
+bool Remote_event_stream::semisync_ack(
+  const std::string_view log_name, uint64_t next_pos)
+{
+  constexpr size_t HEAD_SIZE= /* Semi-Sync Header */1 + sizeof(next_pos);
+  char payload[HEAD_SIZE + (FN_REFLEN+1)]= {'\xEF'};
+  int8store(&(payload[1]), next_pos);
+  log_name.copy(&(payload[HEAD_SIZE]), sizeof(payload)-HEAD_SIZE);
+  NET *net= &connector->net;
+  //@deprecated: not required in Connector/C
+  net->pkt_nr_can_be_reset= true;
+  // Connector/C might not require resetting; better be safe until confirmed.
+  net_clear(net, false);
+  return my_net_write(net, reinterpret_cast<const unsigned char *>(payload),
+    HEAD_SIZE + (log_name.size()+1)) || net_flush(net);
+}
+
+void Remote_event_stream::abort()
+{ vio_shutdown(connector->net.vio, SHUT_RDWR); }

rpl/remote_event_stream.h

@@ -0,0 +1,130 @@
+/*
+  Copyright (c) 2026 MariaDB
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation; version 2 of the License.
+
+  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-1335 USA.
+*/
+
+#include <cstdint>
+#include <optional> // Remote_event_stream::Connection_options::ssl_options
+#include <string_view> // Return type of Remote_event_stream::next()
+
+using MYSQL= struct st_mysql;
+using MYSQL_RES_P= struct st_mysql_res *;
+
+
+struct Rpl_slave_connection
+{
+  struct Connection_options
+  {
+    char *host, *user, *password;
+    unsigned int port;
+    struct SSL_options
+    {
+      const char
+        *ssl_ca, *ssl_capath,
+        *ssl_cert,
+        *ssl_crl, *ssl_crlpath,
+        *ssl_key,
+        *ssl_cipher;
+      bool ssl_verify_server_cert;
+    };
+    std::optional<SSL_options> ssl_options; ///< Disable SSL with std::nullopt
+    const char *charset_name;
+    /// In case the master asks for an external authentication plugin
+    const char *plugin_dir;
+    void operator ()(MYSQL *connector) const;
+  };
+
+  operator bool() { return connector; }
+  ///@deprecated TODO: hide this behind reusable helpers @{
+    ///@return `false` on success, or `true` on error
+    bool connect(unsigned long flags= 0);
+    bool real_query(const char *query, size_t strlen);
+  /// }@
+protected:
+  /**TODO:
+    Replace this opaque pointer with `std::optional<MYSQL>`
+    when `../sql-common` is completely phased out.
+  */
+  MYSQL *connector;
+  /**
+    @note This does not invoke the setup methods; do so before calling next().
+      (TODO: clean up the rest of `sql/slave.cc`'s `handle_slave_io()` &
+       sub-procedures so they don't have to be manual.)
+    @post operator bool() is `true` on success or `false` on OOM.
+  */
+  Rpl_slave_connection(const Connection_options &options, uint32_t timeout);
+  ~Rpl_slave_connection();
+};
+
+
+struct Remote_event_stream: Rpl_slave_connection
+{
+  bool do_reconnect= false;
+  Remote_event_stream(const Connection_options &options,
+                      uint32_t timeout, unsigned long max_allowed_packet);
+
+  /** Ordered setup commands and error status getters
+    @deprecated TODO: hide these behind helper
+    methods of protected Chain of Command classes @{
+  */
+    unsigned int errnum();
+    const char *errmsg();
+    /**
+      (Re)connect to the remote
+      @return `false` on success, or `true` on error
+    */
+    bool connect(bool compress= false);
+    ///@deprecated MySQL 4.x and before are long EOL.
+    unsigned long master_version();
+    MYSQL_RES_P store_result();
+    static char **fetch_row(MYSQL_RES_P query_result);
+    static void free_result(MYSQL_RES_P query_result);
+    /**TODO:
+      merge `register_slave_on_master()` & `request_dump()`
+      in `sql/slave.cc` and `dump_remote_log_entries()` in
+      `client/mysqlbinlog.cc` with Connector/C's mariadb_rpl_open()
+      @return `false` on success, or `true` on error
+    */
+    bool send_command(
+      int command, const unsigned char *args, size_t strlen, bool skip_check);
+  /// }@
+
+  unsigned long thread_id(); /// for Semi_sync_graceful_killer
+  /** @return a connector-managed string, which can be
+    * the next event
+    * an EOF packet
+    * an empty string, which represents error
+  */
+  std::basic_string_view<unsigned char> next();
+  /**
+    @return `false` on success, or `true` on error
+    @note Acknowledgement comes after the caller has safely recorded the
+          event from next(); this method is therefore separate from next().
+  */
+  bool semisync_ack(const std::string_view log_name, uint64_t next_pos);
+  /**
+    Force-close the stream. It remains existing for recovery,
+    but all ongoing and subsequent operations will error.
+    @note Call this from another thread _with additional mutexing_
+  */
+  void abort();
+};
+
+
+struct Semi_sync_graceful_killer: Rpl_slave_connection
+{
+  Semi_sync_graceful_killer(
+    const Connection_options &options, uint32_t timeout);
+};

sql-common/client.c

@@ -731,9 +731,6 @@ void end_server(MYSQL *mysql)
     mysql->connector_fd = 0;
 
     DBUG_PRINT("info",("Net: %s", vio_description(mysql->net.vio)));
-#ifdef MYSQL_SERVER
-    slave_io_thread_detach_vio();
-#endif
     vio_delete(mysql->net.vio);
     mysql->net.vio= 0;          /* Marker */
     mysql_prune_stmt_list(mysql);

sql/CMakeLists.txt

@@ -124,7 +124,7 @@ SET (SQL_SOURCE
                protocol.cc records.cc repl_failsafe.cc rpl_filter.cc
                session_tracker.cc
                sql_path.cc
-               set_var.cc
+               set_var.cc ../rpl/remote_event_stream.cc
                slave.cc sp.cc sp_cache.cc sp_head.cc sp_pcontext.cc
                sp_rcontext.cc sp_cursor.cc
                spatial.cc sql_acl.cc sql_analyse.cc sql_base.cc

sql/client_settings.h

@@ -42,11 +42,3 @@
 #define mysql_server_init(a,b,c) mysql_client_plugin_init()
 #define mysql_server_end()       mysql_client_plugin_deinit()
 
-#ifdef HAVE_REPLICATION
-C_MODE_START
-void slave_io_thread_detach_vio();
-C_MODE_END
-#else
-#define slave_io_thread_detach_vio()
-#endif
-