Bug#24303169 XPLUGIN SOCKET/TCP CONNECT ISSUES FOR CONCCURENT CONNECTIONS

Description
  There are two issues in the bug. First one related to limited number
  of file descriptors and X Plugin prints trace errors (it only occurs
  with UNIX sockets). Second one is about using invalid file descriptor
  by X Plugin.

Analysis of first problem
  The limit of descriptors was set to 1024 which is less than  system
  variable `max_connection` (900) and should be sufficient for the test.
  X Plugin accesses user table at startup (mysql.user) which also creates
  a descriptor. We get two descriptors per connection, thus the limit of
  descriptors should be set to 1800.
  The issue isn't reproducable with TCP sockets because the sockets are
  much slower and they get sufficient time not to release the mysql.user
  file descriptor (not are opened simultaneously).

Analysis of second problem
  `Connecton_listener_interface` can return a Vio structure with invalid
  socket inside. `ngs::Server::on_accept` didn't check if `Vio` is valid.

Solution
  Fixed second problem by validating the socket and returning NULL pointer
  to `Vio` object, `ngs::Server` counts such objects as accept failure.
  To prevent trace error overflow added additional counting of accept
  failures and printing the error when `num_of_accept_error %244 == 0`.

RB: 13538
Reviewed-by: Grzegorz Szwarc <[email protected]>

Author: lkotula

rapid/plugin/x/ngs/include/ngs/interface/listener_factory_interface.h

@@ -33,11 +33,10 @@ typedef Memory_new<Listener_interface>::Unique_ptr Listener_interface_ptr;
 class Listener_factory_interface
 {
 public:
-
   virtual ~Listener_factory_interface() {}
 
-  virtual Listener_interface_ptr create_unix_socket_listener(const std::string &unix_socket_path, Time_and_socket_events &event) = 0;
-  virtual Listener_interface_ptr create_tcp_socket_listener(const unsigned short port, Time_and_socket_events &event) = 0;
+  virtual Listener_interface_ptr create_unix_socket_listener(const std::string &unix_socket_path, Time_and_socket_events &event, const uint32 backlog) = 0;
+  virtual Listener_interface_ptr create_tcp_socket_listener(const unsigned short port, Time_and_socket_events &event, const uint32 backlog) = 0;
 };
 
 } // namespace ngs

rapid/plugin/x/ngs/include/ngs/server.h

@@ -147,6 +147,7 @@ class Server: public Server_interface
 
   bool m_timer_running;
   bool m_skip_name_resolve;
+  uint32 m_errors_while_accepting;
 
   boost::shared_ptr<Server_acceptors> m_acceptors;
   boost::shared_ptr<Scheduler_dynamic> m_accept_scheduler;

rapid/plugin/x/ngs/include/ngs/server_acceptors.h

@@ -41,7 +41,8 @@ class Server_acceptors
 
   Server_acceptors(Listener_factory_interface &listener_factory,
                    const unsigned short tcp_port,
-                   const std::string &unix_socket_file_or_named_pipe);
+                   const std::string &unix_socket_file_or_named_pipe,
+                   const uint32 backlog);
 
   bool prepare(On_connection on_connection, const bool skip_networking, const bool use_unix_sockets_or_named_pipes);
   void abort();

rapid/plugin/x/ngs/include/ngs_common/connection_vio.h

@@ -111,8 +111,8 @@ class Connection_vio
 
   static my_socket accept(my_socket sock, struct sockaddr* addr, socklen_t& len, int& err, std::string& strerr);
 
-  static my_socket create_and_bind_socket(const unsigned short port, std::string &error_message);
-  static my_socket create_and_bind_socket(const std::string &unix_socket_file, std::string &error_message);
+  static my_socket create_and_bind_socket(const unsigned short port, std::string &error_message, const uint32 backlog);
+  static my_socket create_and_bind_socket(const std::string &unix_socket_file, std::string &error_message, const uint32 backlog);
 
   static void get_error(int& err, std::string& strerr);
 
@@ -123,7 +123,6 @@ class Connection_vio
 private:
   static bool create_lockfile(const std::string &unix_socket_file, std::string &error_message);
   static std::string get_lockfile_name(const std::string &unix_socket_file);
-
   friend class Ssl_context;
 
   Mutex m_shutdown_mutex;

rapid/plugin/x/ngs/ngs_common/connection_vio.cc

@@ -411,7 +411,7 @@ my_socket Connection_vio::accept(my_socket sock, struct sockaddr* addr, socklen_
   return res;
 }
 
-my_socket Connection_vio::create_and_bind_socket(const unsigned short port, std::string &error_message)
+my_socket Connection_vio::create_and_bind_socket(const unsigned short port, std::string &error_message, const uint32 backlog)
 {
   int err;
   std::string errstr;
@@ -450,7 +450,7 @@ my_socket Connection_vio::create_and_bind_socket(const unsigned short port, std:
     return INVALID_SOCKET;
   }
 
-  if (m_socket_operations->listen(result, 9999) < 0)
+  if (m_socket_operations->listen(result, backlog) < 0)
   {
     // lets decide later if its an error or not
     get_error(err, errstr);
@@ -467,7 +467,7 @@ my_socket Connection_vio::create_and_bind_socket(const unsigned short port, std:
   return result;
 }
 
-my_socket Connection_vio::create_and_bind_socket(const std::string &unix_socket_file, std::string &error_message)
+my_socket Connection_vio::create_and_bind_socket(const std::string &unix_socket_file, std::string &error_message, const uint32 backlog)
 {
 #if defined(HAVE_SYS_UN_H)
   struct sockaddr_un addr;
@@ -535,7 +535,7 @@ my_socket Connection_vio::create_and_bind_socket(const std::string &unix_socket_
   umask(old_mask);
 
   // listen
-  if (m_socket_operations->listen(listener_socket, 9999) < 0)
+  if (m_socket_operations->listen(listener_socket, backlog) < 0)
   {
     get_error(err, errstr);
 
@@ -563,7 +563,7 @@ bool Connection_vio::create_lockfile(const std::string &unix_socket_file, std::s
   char buffer[8];
   const char x_prefix = 'X';
   const pid_t cur_pid= m_system_operations->getpid();
-  std::string lock_filename= get_lockfile_name(unix_socket_file);
+  const std::string lock_filename= get_lockfile_name(unix_socket_file);
 
   compile_time_assert(sizeof(pid_t) == 4);
   int retries= 3;

rapid/plugin/x/ngs/src/server.cc

@@ -44,6 +44,7 @@ Server::Server(boost::shared_ptr<Server_acceptors>  acceptors,
                boost::shared_ptr<Protocol_config> config)
 : m_timer_running(false),
   m_skip_name_resolve(false),
+  m_errors_while_accepting(0),
   m_acceptors(acceptors),
   m_accept_scheduler(accept_scheduler),
   m_worker_scheduler(work_scheduler),
@@ -270,18 +271,21 @@ void Server::on_accept(Connection_acceptor_interface &connection_acceptor)
   if (m_state.is(State_terminating))
     return;
 
-  int err = 0;
-  std::string strerr;
   Vio *vio = connection_acceptor.accept();
 
   if (NULL == vio)
   {
     m_delegate->did_reject_client(Server_delegate::AcceptError);
 
-    // error accepting client
-    log_error("Error accepting client: "
-              "Error code: %s (%d)",
-              strerr.c_str(), err);
+    if (0 == (m_errors_while_accepting++ & 255))
+    {
+      // error accepting client
+      log_error("Error accepting client");
+    }
+    const time_t microseconds_to_sleep = 100000;
+
+    my_sleep(microseconds_to_sleep);
+
     return;
   }
 

rapid/plugin/x/ngs/src/server_acceptors.cc

@@ -120,10 +120,10 @@ class Server_acceptors::Server_task_time_and_event: public Server_task_interface
 Server_acceptors::Server_acceptors(
     Listener_factory_interface &listener_factory,
     const unsigned short tcp_port,
-    const std::string &unix_socket_file_or_named_pipe)
-: m_tcp_socket(listener_factory.create_tcp_socket_listener(tcp_port, m_event)),
+    const std::string &unix_socket_file_or_named_pipe, const uint32 backlog)
+: m_tcp_socket(listener_factory.create_tcp_socket_listener(tcp_port, m_event, backlog)),
 #if defined(HAVE_SYS_UN_H)
-  m_unix_socket(listener_factory.create_unix_socket_listener(unix_socket_file_or_named_pipe, m_event)),
+  m_unix_socket(listener_factory.create_unix_socket_listener(unix_socket_file_or_named_pipe, m_event, backlog)),
 #endif
   m_time_and_event_state(State_listener_initializing),
   m_time_and_event_task(new Server_task_time_and_event(m_event, m_time_and_event_state))

rapid/plugin/x/ngs/src/time_socket_events.cc

@@ -34,26 +34,37 @@ class Connection_acceptor_socket : public Connection_acceptor_interface
   {
     Vio *vio;
     sockaddr_storage accept_address;
-    socklen_t accept_len = sizeof(accept_address);
     int err = 0;
     std::string strerr;
+    my_socket sock = INVALID_SOCKET;
 
-    my_socket sock = Connection_vio::accept(m_socket_listener, (struct sockaddr*)&accept_address, accept_len, err, strerr);
-    const bool is_tcpip = (accept_address.ss_family == AF_INET || accept_address.ss_family == AF_INET6);
+    for (int i = 0; i < MAX_ACCEPT_REATTEMPT; ++i)
+    {
+      socklen_t accept_len = sizeof(accept_address);
+      sock = Connection_vio::accept(m_socket_listener, (struct sockaddr*)&accept_address, accept_len, err, strerr);
+
+      if (INVALID_SOCKET != sock)
+        break;
+    }
 
+    if (INVALID_SOCKET == sock)
+      return NULL;
+
+    const bool is_tcpip = (accept_address.ss_family == AF_INET || accept_address.ss_family == AF_INET6);
     vio = vio_new(sock, is_tcpip ? VIO_TYPE_TCPIP : VIO_TYPE_SOCKET, 0);
     if (!vio)
       throw std::bad_alloc();
+
     // enable TCP_NODELAY
     vio_fastsend(vio);
-
     vio_keepalive(vio, TRUE);
 
     return vio;
   }
 
 private:
   my_socket m_socket_listener;
+  static const int MAX_ACCEPT_REATTEMPT = 10;
 };
 
 

rapid/plugin/x/src/xpl_listener_factory.cc

@@ -31,12 +31,12 @@ namespace details
 class Tcp_listener: public ngs::Listener_interface
 {
 public:
-  Tcp_listener(const unsigned short port, ngs::Time_and_socket_events &event)
+  Tcp_listener(const unsigned short port, ngs::Time_and_socket_events &event, const uint32 backlog)
   : m_state(ngs::State_listener_initializing),
     m_port(port),
     m_event(event)
   {
-    m_tcp_socket = ngs::Connection_vio::create_and_bind_socket(port, m_last_error);
+    m_tcp_socket = ngs::Connection_vio::create_and_bind_socket(port, m_last_error, backlog);
   }
 
   ~Tcp_listener()
@@ -119,15 +119,16 @@ class Tcp_listener: public ngs::Listener_interface
 class Unix_socket_listener: public ngs::Listener_interface
 {
 public:
-  Unix_socket_listener(const std::string &unix_socket_path, ngs::Time_and_socket_events &event)
+  Unix_socket_listener(const std::string &unix_socket_path, ngs::Time_and_socket_events &event, const uint32 backlog)
   : m_state(ngs::State_listener_initializing),
     m_unix_socket_path(unix_socket_path),
+    m_unix_socket(INVALID_SOCKET),
     m_event(event)
   {
 #if !defined(HAVE_SYS_UN_H)
     m_state.set(ngs::State_listener_stopped);
 #else
-    m_unix_socket = ngs::Connection_vio::create_and_bind_socket(unix_socket_path, m_last_error);
+    m_unix_socket = ngs::Connection_vio::create_and_bind_socket(unix_socket_path, m_last_error, backlog);
 #endif // !defined(HAVE_SYS_UN_H)
   }
 
@@ -209,14 +210,14 @@ class Unix_socket_listener: public ngs::Listener_interface
 } // namespace details
 
 
-ngs::Listener_interface_ptr Listener_factory::create_unix_socket_listener(const std::string &unix_socket_path, ngs::Time_and_socket_events &event)
+ngs::Listener_interface_ptr Listener_factory::create_unix_socket_listener(const std::string &unix_socket_path, ngs::Time_and_socket_events &event, const uint32 backlog)
 {
   return ngs::Listener_interface_ptr(
-      new details::Unix_socket_listener(unix_socket_path, event));
+      new details::Unix_socket_listener(unix_socket_path, event, backlog));
 }
 
-ngs::Listener_interface_ptr Listener_factory::create_tcp_socket_listener(const unsigned short port, ngs::Time_and_socket_events &event)
+ngs::Listener_interface_ptr Listener_factory::create_tcp_socket_listener(const unsigned short port, ngs::Time_and_socket_events &event, const uint32 backlog)
 {
   return ngs::Listener_interface_ptr(
-      new details::Tcp_listener(port, event));
+      new details::Tcp_listener(port, event, backlog));
 }

rapid/plugin/x/src/xpl_listener_factory.h

@@ -29,8 +29,8 @@ namespace xpl
 class Listener_factory: public ngs::Listener_factory_interface
 {
 public:
-  ngs::Listener_interface_ptr create_unix_socket_listener(const std::string &unix_socket_path, ngs::Time_and_socket_events &event);
-  ngs::Listener_interface_ptr create_tcp_socket_listener(const unsigned short port, ngs::Time_and_socket_events &event);
+  ngs::Listener_interface_ptr create_unix_socket_listener(const std::string &unix_socket_path, ngs::Time_and_socket_events &event, const uint32 backlog);
+  ngs::Listener_interface_ptr create_tcp_socket_listener(const unsigned short port, ngs::Time_and_socket_events &event, const uint32 backlog);
 };
 
 } // namespace xpl

rapid/plugin/x/src/xpl_server.cc

@@ -271,6 +271,10 @@ int xpl::Server::main(MYSQL_PLUGIN p)
 {
   xpl::plugin_handle = p;
 
+  uint32 listen_backlog = 50 + Plugin_system_variables::max_connections / 5;
+  if (listen_backlog > 900)
+    listen_backlog= 900;
+
   try
   {
     Global_status_variables::instance().reset();
@@ -284,7 +288,7 @@ int xpl::Server::main(MYSQL_PLUGIN p)
 
     Listener_factory listener_factory;
     boost::shared_ptr<ngs::Server_acceptors> acceptors(
-        new ngs::Server_acceptors(listener_factory, Plugin_system_variables::port, Plugin_system_variables::socket));
+        new ngs::Server_acceptors(listener_factory, Plugin_system_variables::port, Plugin_system_variables::socket, listen_backlog));
 
     instance_rwl.wlock();
 

rapid/unittest/gunit/xplugin/TestGroups.txt

@@ -36,5 +36,7 @@ add_test(InstantiationNegativeTest/Query_string_builder_multiple_too_many_tags_p
 add_test(Valid_ip_mask_addresses/User_verification_dbuser_param_valid_test xplugin_unit_tests --gtest_filter=Valid_ip_mask_addresses/User_verification_dbuser_param_valid_test.*)
 add_test(Invalid_ip_mask_addresses/User_verification_dbuser_param_notvalid_test xplugin_unit_tests --gtest_filter=Invalid_ip_mask_addresses/User_verification_dbuser_param_notvalid_test.*)
 add_test(Range_from_0_to_9/User_verification_param_test xplugin_unit_tests --gtest_filter=Range_from_0_to_9/User_verification_param_test.*)
+add_test(Supported_connection_type_require_transport_combinations/User_verification_param_test_with_supported_combinations xplugin_unit_tests --gtest_filter=Supported_connection_type_require_transport_combinations/User_verification_param_test_with_supported_combinations.*)
+add_test(Unsupported_connection_type_require_transport_combinations/User_verification_param_test_with_unsupported_combinations xplugin_unit_tests --gtest_filter=Unsupported_connection_type_require_transport_combinations/User_verification_param_test_with_unsupported_combinations.*)
 add_test(InstantiationNegativeTests/Getter_any_type_testsuite xplugin_unit_tests --gtest_filter=InstantiationNegativeTests/Getter_any_type_testsuite.*)
 add_test(InstantiationNegativeTests/Getter_scalar_type_testsuite xplugin_unit_tests --gtest_filter=InstantiationNegativeTests/Getter_scalar_type_testsuite.*)