Bug#36347485 - Remove libevent usage from X Plugin

Description
===========
MySQL Server source code, contains "net_ts" which may be used as alternative
for "libevent". The "net_ts" is maintained and Oracle homemade.

Fix
===
Refactored X Plugin code to use "net::io_context", instead "libevent".

Change-Id: I8f3945110feebf041c22b7abafa1b0a872661f4f

Author: lkotula

CMakeLists.txt

@@ -2348,10 +2348,11 @@ CONFIGURE_FILE(${CMAKE_SOURCE_DIR}/sql/sql_builtin.cc.in
     ${CMAKE_BINARY_DIR}/sql/sql_builtin.cc)
 
 # depends on mysql_version.h to exist
-IF(NOT WITHOUT_SERVER)
-  IF(WITH_ROUTER)
-    ADD_SUBDIRECTORY(router)
-  ENDIF()
+IF(NOT WITHOUT_SERVER AND WITH_ROUTER)
+  ADD_SUBDIRECTORY(router)
+ELSE()
+  # Define harness_net_ts target and public include directories.
+  ADD_SUBDIRECTORY(router/src/harness/include)
 ENDIF()
 
 IF(ENABLE_GCOV)

plugin/x/CMakeLists.txt

@@ -88,7 +88,7 @@ IF(WITH_MYSQLX)
     DEFAULT
     DEPENDENCIES xprotocol_tags
     LINK_LIBRARIES
-    ext::libevent
+    harness_net_ts
     ext::icu
     ${MYSQLX_PROTOCOL_LIB}
     extra::rapidjson

plugin/x/src/ngs/socket_events.cc

@@ -28,7 +28,8 @@
 #include <mutex>  // NOLINT(build/c++11)
 #include <new>
 
-#include "my_io.h"    // NOLINT(build/include_subdir)
+#include "my_io.h"  // NOLINT(build/include_subdir)
+#include "mysql/psi/mysql_socket.h"
 #include "violite.h"  // NOLINT(build/include_subdir)
 
 #include "plugin/x/src/interface/connection_acceptor.h"
@@ -37,31 +38,12 @@
 #include "plugin/x/src/ngs/socket_events.h"
 #include "plugin/x/src/operations_factory.h"
 
-// Surpressing numerous warnings generated by libevent on Windows.
-#ifdef WIN32
-#pragma warning(push)
-#pragma warning(disable : 4005)
-#endif                     // WIN32
-#include <event2/event.h>  // libevent
-#include <event2/event_compat.h>
-#include <event2/event_struct.h>
-#include <event2/thread.h>
-#ifdef WIN32
-#pragma warning(pop)
-#endif  // WIN32
-
-#ifdef EVTHREAD_USE_WINDOWS_THREADS_IMPLEMENTED
-#define XPL_EVTHREAD_INITIALIZE() evthread_use_windows_threads()
-#elif EVTHREAD_USE_PTHREADS_IMPLEMENTED
-#define XPL_EVTHREAD_INITIALIZE() evthread_use_pthreads()
-#else
-#define XPL_EVTHREAD_INITIALIZE() \
-  do {                            \
-  } while (0)
-#endif
-
 namespace ngs {
 
+using TcpAcceptor = net::ip::tcp::acceptor;
+using Socket = TcpAcceptor::socket_type;
+using Endpoint_type = TcpAcceptor::endpoint_type;
+
 class Connection_acceptor_socket : public xpl::iface::Connection_acceptor {
  public:
   using Socket_ptr = std::shared_ptr<xpl::iface::Socket>;
@@ -112,65 +94,55 @@ class Connection_acceptor_socket : public xpl::iface::Connection_acceptor {
   static const int MAX_ACCEPT_REATTEMPT = 10;
 };
 
-struct Socket_events::Timer_data {
-  std::function<bool()> callback;
-  event ev;
-  timeval tv;
-  Socket_events *self;
+class Socket_events::EntryTimer {
+ public:
+  EntryTimer(net::io_context &io) : timer{io} {}
 
-  static void free(Timer_data *data) {
-    evtimer_del(&data->ev);
-    free_object(data);
-  }
+ public:
+  std::function<bool()> callback;
+  std::chrono::milliseconds duration;
+  net::steady_timer timer;
 };
 
-struct Socket_events::Socket_data {
+class Socket_events::EntryAcceptingSocket {
+ public:
+  EntryAcceptingSocket(net::io_context &io) : acceptor{io} {}
+  ~EntryAcceptingSocket() { acceptor.release(); }
+
+ public:
   std::function<void(xpl::iface::Connection_acceptor &)> callback;
-  event ev;
   std::shared_ptr<xpl::iface::Socket> socket;
-
-  static void free(Socket_data *data) {
-    event_del(&data->ev);
-    free_object(data);
-  }
+  TcpAcceptor acceptor;
 };
 
-Socket_events::Socket_events() {
-  static std::once_flag flag_event_threads_initialized;
-
-  std::call_once(flag_event_threads_initialized,
-                 []() { XPL_EVTHREAD_INITIALIZE(); });
-
-  m_evbase = event_base_new();
-
-  if (!m_evbase) throw std::bad_alloc();
-}
+Socket_events::Socket_events() {}
 
 Socket_events::~Socket_events() {
   std::for_each(m_timer_events.begin(), m_timer_events.end(),
-                &Timer_data::free);
+                &free_object<EntryTimer>);
 
   std::for_each(m_socket_events.begin(), m_socket_events.end(),
-                &Socket_data::free);
-
-  event_base_free(m_evbase);
+                &free_object<EntryAcceptingSocket>);
 }
 
 bool Socket_events::listen(
     std::shared_ptr<xpl::iface::Socket> sock,
     std::function<void(xpl::iface::Connection_acceptor &)> callback) {
-  m_socket_events.push_back(allocate_object<Socket_data>());
-  Socket_data *socket_event = m_socket_events.back();
+  m_socket_events.push_back(
+      allocate_object<EntryAcceptingSocket>(m_io_context));
+  EntryAcceptingSocket *socket_event = m_socket_events.back();
 
   socket_event->callback = callback;
   socket_event->socket = sock;
+  socket_event->acceptor.assign(Endpoint_type().protocol(),
+                                sock->get_socket_fd());
 
-  event_set(&socket_event->ev, static_cast<int>(sock->get_socket_fd()),
-            EV_READ | EV_PERSIST, &Socket_events::socket_data_avaiable,
-            socket_event);
-  event_base_set(m_evbase, &socket_event->ev);
+  socket_event->acceptor.async_wait(Socket::wait_read,
+                                    [this, socket_event](std::error_code ec) {
+                                      callback_accept_socket(socket_event, ec);
+                                    });
 
-  return 0 == event_add(&socket_event->ev, nullptr);
+  return true;
 }
 
 /** Register a callback to be executed in a fixed time interval.
@@ -182,53 +154,61 @@ NOTE: This method may only be called from the same thread as the event loop.
 */
 void Socket_events::add_timer(const std::size_t delay_ms,
                               std::function<bool()> callback) {
-  Timer_data *data = allocate_object<Timer_data>();
-  data->tv.tv_sec = static_cast<long>(delay_ms / 1000);
-  data->tv.tv_usec = (delay_ms % 1000) * 1000;
-  data->callback = callback;
-  data->self = this;
-  // XXX use persistent timer events after switch to libevent2
-  evtimer_set(&data->ev, timeout_call, data);
-  event_base_set(m_evbase, &data->ev);
-  evtimer_add(&data->ev, &data->tv);
-
-  MUTEX_LOCK(lock, m_timers_mutex);
-  m_timer_events.push_back(data);
-}
-
-void Socket_events::loop() { event_base_loop(m_evbase, 0); }
+  EntryTimer *timer_entry = allocate_object<EntryTimer>(m_io_context);
+  timer_entry->duration = std::chrono::milliseconds{delay_ms};
+  timer_entry->callback = callback;
+  {
+    MUTEX_LOCK(lock, m_timers_mutex);
+    m_timer_events.push_back(timer_entry);
+  }
 
-void Socket_events::break_loop() { event_base_loopexit(m_evbase, nullptr); }
+  timer_entry->timer.expires_after(timer_entry->duration);
+  timer_entry->timer.async_wait([this, timer_entry](std::error_code ec) {
+    callback_timeout(timer_entry, ec);
+  });
+}
 
-void Socket_events::timeout_call(socket_type, short, void *arg) {
-  Timer_data *data = static_cast<Timer_data *>(arg);
+void Socket_events::loop() { m_io_context.run(); }
 
-  if (!data->callback()) {
-    evtimer_del(&data->ev);
+void Socket_events::break_loop() { m_io_context.stop(); }
 
+void Socket_events::callback_timeout(EntryTimer *timer_entry,
+                                     std::error_code ec) {
+  if (ec || !timer_entry->callback()) {
     {
-      MUTEX_LOCK(timer_lock, data->self->m_timers_mutex);
-      data->self->m_timer_events.erase(
-          std::remove(data->self->m_timer_events.begin(),
-                      data->self->m_timer_events.end(), data),
-          data->self->m_timer_events.end());
+      MUTEX_LOCK(timer_lock, m_timers_mutex);
+
+      m_timer_events.erase(std::remove(m_timer_events.begin(),
+                                       m_timer_events.end(), timer_entry),
+                           m_timer_events.end());
     }
 
-    free_object(data);
+    free_object(timer_entry);
   } else {
     // schedule for another round
-    evtimer_add(&data->ev, &data->tv);
+    timer_entry->timer.expires_after(timer_entry->duration);
+    timer_entry->timer.async_wait([this, timer_entry](std::error_code ec) {
+      callback_timeout(timer_entry, ec);
+    });
   }
 }
 
-void Socket_events::socket_data_avaiable(socket_type, short, void *arg) {
-  Socket_data *data = static_cast<Socket_data *>(arg);
-  xpl::Operations_factory operations_factory;
-  std::shared_ptr<xpl::iface::System> system_interface(
-      operations_factory.create_system_interface());
-  Connection_acceptor_socket acceptor(data->socket, *system_interface);
-
-  data->callback(acceptor);
+void Socket_events::callback_accept_socket(
+    EntryAcceptingSocket *acceptors_entry, std::error_code ec) {
+  if (!ec) {
+    xpl::Operations_factory operations_factory;
+    std::shared_ptr<xpl::iface::System> system_interface(
+        operations_factory.create_system_interface());
+    Connection_acceptor_socket vio_socket_forge(acceptors_entry->socket,
+                                                *system_interface);
+
+    acceptors_entry->callback(vio_socket_forge);
+
+    acceptors_entry->acceptor.async_wait(
+        Socket::wait_read, [this, acceptors_entry](std::error_code ec) {
+          callback_accept_socket(acceptors_entry, ec);
+        });
+  }
 }
 
 }  // namespace ngs

plugin/x/src/ngs/socket_events.h

@@ -29,22 +29,18 @@
 #include <memory>
 #include <vector>
 
+#include "mysql/harness/net_ts.h"
 #include "plugin/x/src/helper/multithread/mutex.h"
 #include "plugin/x/src/interface/socket_events.h"
 #include "plugin/x/src/xpl_performance_schema.h"
 
-struct event_base;
-
 namespace ngs {
 
 class Socket_events : public xpl::iface::Socket_events {
+ private:
+  using Socket = net::ip::tcp::socket;
+
  public:
-#ifdef _WIN32
-  // mimick evutil_socket_t in libevent-2.x
-  using socket_type = intptr_t;
-#else
-  using socket_type = int;
-#endif
   Socket_events();
   ~Socket_events() override;
 
@@ -58,14 +54,16 @@ class Socket_events : public xpl::iface::Socket_events {
   void break_loop() override;
 
  private:
-  static void timeout_call(socket_type sock, short which, void *arg);
-  static void socket_data_avaiable(socket_type sock, short which, void *arg);
+  class EntryTimer;
+  class EntryAcceptingSocket;
+
+  void callback_timeout(EntryTimer *timer_entry, std::error_code ec);
+  void callback_accept_socket(EntryAcceptingSocket *acceptors_entry,
+                              std::error_code ec);
 
-  struct Timer_data;
-  struct Socket_data;
-  struct event_base *m_evbase;
-  std::vector<Socket_data *> m_socket_events;
-  std::vector<Timer_data *> m_timer_events;
+  net::io_context m_io_context;
+  std::vector<EntryAcceptingSocket *> m_socket_events;
+  std::vector<EntryTimer *> m_timer_events;
   xpl::Mutex m_timers_mutex{KEY_mutex_x_socket_events_timers};
 };
 

router/src/harness/CMakeLists.txt

@@ -24,7 +24,7 @@
 ADD_SUBDIRECTORY(src)
 ADD_SUBDIRECTORY(include)
 
-IF(WITH_UNIT_TESTS)
+IF(WITH_UNIT_TESTS AND WITH_ROUTER)
   ADD_SUBDIRECTORY(shared)
   ADD_SUBDIRECTORY(tests)
 ENDIF()

router/src/harness/include/CMakeLists.txt

@@ -53,7 +53,6 @@ TARGET_SOURCES(harness_net_ts
   ${CMAKE_CURRENT_SOURCE_DIR}/mysql/harness/net_ts/timer.h
   ${CMAKE_CURRENT_SOURCE_DIR}/mysql/harness/net_ts/win32_named_pipe.h
   )
-TARGET_LINK_LIBRARIES(harness_net_ts INTERFACE harness_stdx)
 
 IF(CMAKE_SYSTEM_NAME STREQUAL "SunOS")
   TARGET_LINK_LIBRARIES(harness_net_ts INTERFACE socket;nsl)

router/src/harness/include/mysql/harness/net_ts/buffer.h

@@ -902,11 +902,12 @@ read(SyncReadStream &stream, DynamicBuffer &&b, CompletionCondition cond) {
 
       // if socket was non-blocking and some bytes where already read, return
       // the success
-      const auto ec = res.error();
-      if ((ec == make_error_condition(
-                     std::errc::resource_unavailable_try_again) ||
-           ec == make_error_condition(std::errc::operation_would_block) ||
-           ec == net::stream_errc::eof) &&
+      const auto error_code = res.error();
+      if ((error_code == make_error_condition(
+                             std::errc::resource_unavailable_try_again) ||
+           error_code ==
+               make_error_condition(std::errc::operation_would_block) ||
+           error_code == net::stream_errc::eof) &&
           transferred != 0) {
         return transferred;
       }

router/src/harness/include/mysql/harness/net_ts/io_context.h

@@ -499,9 +499,10 @@ class io_context : public execution_context {
         // code should be similar to ::cancel(fd)
         std::lock_guard<std::mutex> lk(mtx_);
 
-        if (auto op = active_ops_.extract_first(fd, static_cast<short>(wt))) {
-          op->cancel();
-          cancelled_ops_.push_back(std::move(op));
+        if (auto async_op =
+                active_ops_.extract_first(fd, static_cast<short>(wt))) {
+          async_op->cancel();
+          cancelled_ops_.push_back(std::move(async_op));
         }
       }
     }
@@ -1175,7 +1176,7 @@ inline io_context::count_type io_context::do_one(
     // timer
     std::chrono::milliseconds min_duration{0};
     {
-      std::lock_guard<std::mutex> lk(mtx_);
+      std::lock_guard<std::mutex> lock(mtx_);
       // check the smallest timestamp of all timer-queues
       for (auto q : timer_queues_) {
         const auto duration = q->next();
@@ -1198,16 +1199,16 @@ inline io_context::count_type io_context::do_one(
 
     if (auto op = [this]() -> std::unique_ptr<async_op> {
           // handle all the cancelled ops without polling first
-          std::lock_guard<std::mutex> lk(mtx_);
+          std::lock_guard<std::mutex> lock(mtx_);
 
           // ops have all cancelled operators at the front
           if (!cancelled_ops_.empty() &&
               cancelled_ops_.front()->is_cancelled()) {
-            auto op = std::move(cancelled_ops_.front());
+            auto cancelled_op = std::move(cancelled_ops_.front());
 
             cancelled_ops_.pop_front();
 
-            return op;
+            return cancelled_op;
           }
 
           return {};
@@ -1273,7 +1274,7 @@ inline io_context::count_type io_context::do_one(
 
     if (auto op = [this](native_handle_type fd,
                          short events) -> std::unique_ptr<async_op> {
-          std::lock_guard<std::mutex> lk(mtx_);
+          std::lock_guard<std::mutex> lock(mtx_);
 
           return active_ops_.extract_first(fd, events);
         }(res->fd, res->event)) {