Bug#23581389: SEMI-SYNC CAN'T RUN NORMALLY WHEN MANY CONNECTIONS

Problem & Analysis: Semisync Ack thread uses select() call to check if sockets are available for read. But select() uses FD_SET which is limited by FD_SETSIZE(1024 by default). Select() man page says: ####################### "An fd_set is a fixed size buffer. Executing FD_CLR() or FD_SET() with a value of fd that is negative or is equal to or larger than FD_SETSIZE will result in undefined behavior." ######################## Fix: Semisync ack thread is changed to use poll() instead of select(). On some systems, poll may not be available, hence the code is wrapped around HAVE_POLL define. On the systems that still use select(), Ack receive thread will generate an error and semi sync will be switched off. Windows systems is exception case because on windows this limitation does not exists.

Author: Venkatesh Duggirala

plugin/semisync/semisync_master_ack_receiver.cc

@@ -15,6 +15,7 @@
 
 #include "semisync_master.h"
 #include "semisync_master_ack_receiver.h"
+#include "semisync_master_socket_listener.h"
 
 extern ReplSemiSyncMaster repl_semisync;
 
@@ -186,22 +187,6 @@ inline void Ack_receiver::wait_for_slave_connection()
  mysql_cond_wait(&m_cond, &m_mutex);
 }
 
-my_socket Ack_receiver::get_slave_sockets(fd_set *fds)
-{
- my_socket max_fd= INVALID_SOCKET;
- unsigned int i;
-
- FD_ZERO(fds);
- for (i= 0; i < m_slaves.size(); i++)
- {
- my_socket fd= m_slaves[i].sock_fd();
- max_fd= (fd > max_fd ? fd : max_fd);
- FD_SET(fd, fds);
- }
-
- return max_fd;
-}
-
 /* Auxilary function to initialize a NET object with given net buffer. */
 static void init_net(NET *net, unsigned char *buff, unsigned int buff_len)
 {
@@ -216,10 +201,12 @@ void Ack_receiver::run()
 {
  NET net;
  unsigned char net_buff[REPLY_MESSAGE_MAX_LENGTH];
-
- fd_set read_fds;
- my_socket max_fd= INVALID_SOCKET;
  uint i;
+#ifdef HAVE_POLL
+ Poll_socket_listener listener(m_slaves);
+#else
+ Select_socket_listener listener(m_slaves);
+#endif //HAVE_POLL
 
  sql_print_information("Starting ack receiver thread");
 
@@ -231,7 +218,6 @@ void Ack_receiver::run()
 
  while (1)
  {
- fd_set fds;
  Slave_vector_it it;
  int ret;
 
@@ -248,25 +234,19 @@ void Ack_receiver::run()
  mysql_mutex_unlock(&m_mutex);
  continue;
  }
-
- max_fd= get_slave_sockets(&read_fds);
+ if (!listener.init_slave_sockets())
+  goto end;
  m_slaves_changed= false;
- DBUG_PRINT("info", ("fd count %lu, max_fd %d", (ulong)m_slaves.size(),
- max_fd));
  }
-
- struct timeval tv= {1, 0};
- fds= read_fds;
- /* select requires max fd + 1 for the first argument */
- ret= select(max_fd+1, &fds, NULL, NULL, &tv);
+ ret= listener.listen_on_sockets();
  if (ret <= 0)
  {
  mysql_mutex_unlock(&m_mutex);
 
  ret= DBUG_EVALUATE_IF("rpl_semisync_simulate_select_error", -1, ret);
 
- if (ret == -1)
- sql_print_information("Failed to select() on semi-sync dump sockets, "
+ if (ret == -1 && errno != EINTR)
+ sql_print_information("Failed to wait on semi-sync dump sockets, "
  "error: errno=%d", socket_errno);
  /* Sleep 1us, so other threads can catch the m_mutex easily. */
  my_sleep(1);
@@ -277,7 +257,7 @@ void Ack_receiver::run()
  i= 0;
  while (i < m_slaves.size())
  {
- if (FD_ISSET(m_slaves[i].sock_fd(), &fds))
+ if (listener.is_socket_active(i))
  {
  ulong len;
 
@@ -289,7 +269,7 @@ void Ack_receiver::run()
  repl_semisync.reportReplyPacket(m_slaves[i].server_id(),
  net.read_pos, len);
  else if (net.last_errno == ER_NET_READ_ERROR)
- FD_CLR(m_slaves[i].sock_fd(), &read_fds);
+ listener.clear_socket_info(i);
  }
  i++;
  }

plugin/semisync/semisync_master_ack_receiver.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2015, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
 
  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
@@ -21,6 +21,18 @@
 #include "my_thread.h"
 #include "sql_class.h"
 
+struct Slave
+{
+ THD *thd;
+ Vio vio;
+
+ my_socket sock_fd() const { return vio.mysql_socket.fd; }
+ uint server_id() const { return thd->server_id; }
+};
+
+typedef std::vector<Slave> Slave_vector;
+typedef Slave_vector::iterator Slave_vector_it;
+
 /**
  Ack_receiver is responsible to control ack receive thread and maintain
  slave information used by ack receive thread.
@@ -100,20 +112,7 @@ class Ack_receiver : public ReplSemiSyncBase
  mysql_cond_t m_cond;
  /* If slave list is updated(add or remove). */
  bool m_slaves_changed;
-
- struct Slave
- {
- THD *thd;
- Vio vio;
-
- my_socket sock_fd() { return vio.mysql_socket.fd; }
- uint server_id() { return thd->server_id; }
- };
-
- typedef std::vector<Slave> Slave_vector;
- typedef Slave_vector::iterator Slave_vector_it;
  Slave_vector m_slaves;
-
  my_thread_handle m_pid;
 
 /* Declare them private, so no one can copy the object. */
@@ -122,7 +121,6 @@ class Ack_receiver : public ReplSemiSyncBase
 
  void set_stage_info(const PSI_stage_info &stage);
  void wait_for_slave_connection();
- my_socket get_slave_sockets(fd_set *fds);
 };
 
 extern Ack_receiver ack_receiver;

plugin/semisync/semisync_master_socket_listener.h

@@ -0,0 +1,124 @@
+/* Copyright (c) 2016 Oracle and/or its affiliates. All rights reserved.
+
+ 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-1301 USA */
+
+#ifndef SEMISYNC_MASTER_SOCKET_LISTENER
+#define SEMISYNC_MASTER_SOCKET_LISTENER
+#include "semisync_master_ack_receiver.h"
+
+#ifdef HAVE_POLL
+#include <sys/poll.h>
+#include <vector>
+
+class Poll_socket_listener
+{
+public:
+ Poll_socket_listener(const Slave_vector &slaves)
+ :m_slaves(slaves)
+ {
+ }
+
+ bool listen_on_sockets()
+ {
+ return poll(m_fds.data(), m_fds.size(), 1000 /*1 Second timeout*/);
+ }
+
+ bool is_socket_active(int index)
+ {
+ return m_fds[index].revents & POLLIN;
+ }
+
+ void clear_socket_info(int index)
+ {
+ m_fds[index].fd= -1;
+ m_fds[index].events= 0;
+ }
+
+ bool init_slave_sockets()
+ {
+ m_fds.clear();
+ for (uint i= 0; i < m_slaves.size(); i++)
+ {
+ pollfd poll_fd;
+ poll_fd.fd= m_slaves[i].sock_fd();
+ poll_fd.events= POLLIN;
+ m_fds.push_back(poll_fd);
+ }
+ return true;
+ }
+
+private:
+ const Slave_vector &m_slaves;
+ std::vector<pollfd> m_fds;
+};
+
+#else //NO POLL
+
+class Select_socket_listener
+{
+public:
+ Select_socket_listener(const Slave_vector &slaves)
+ :m_slaves(slaves), m_max_fd(INVALID_SOCKET)
+ {
+ }
+
+ bool listen_on_sockets()
+ {
+ /* Reinitialze the fds with active fds before calling select */
+ m_fds= m_init_fds;
+ struct timeval tv= {1,0};
+ /* select requires max fd + 1 for the first argument */
+ return select(m_max_fd+1, &m_fds, NULL, NULL, &tv);
+ }
+
+ bool is_socket_active(int index)
+ {
+ return FD_ISSET(m_slaves[index].sock_fd(), &m_fds);
+ }
+
+ void clear_socket_info(int index)
+ {
+ FD_CLR(m_slaves[index].sock_fd(), &m_init_fds);
+ }
+
+ bool init_slave_sockets()
+ {
+ FD_ZERO(&m_init_fds);
+ for (uint i= 0; i < m_slaves.size(); i++)
+ {
+ my_socket socket_id= m_slaves[i].sock_fd();
+ m_max_fd= (socket_id > m_max_fd ? socket_id : m_max_fd);
+#ifndef WINDOWS
+ if (socket_id > FD_SETSIZE)
+ {
+ sql_print_error("Semisync slave socket fd is %u. "
+ "select() cannot handle if the socket fd is "
+ "bigger than %u (FD_SETSIZE).", socket_id, FD_SETSIZE);
+ return false;
+ }
+#endif //WINDOWS
+ FD_SET(socket_id, &m_init_fds);
+ }
+ return true;
+ }
+
+private:
+ const Slave_vector &m_slaves;
+ my_socket m_max_fd;
+ fd_set m_init_fds;
+ fd_set m_fds;
+};
+
+#endif //HAVE_POLL
+#endif //SEMISYNC_MASTER_SOCKET_LISTENER