Bug#26695357 SOME SELECTS STILL PRESENT IN XCOM CODE

Description:
-----------
After fixing parent issue BUG#25892493, one noticed that selects are still
present in XCom client code, which is used to connect from GR/GCS to the XCom
running thread.

Fix:
---
This patch removes the two 'select()' calls that were still present in XCom:
- In timed_connect(), the select is replaced by a poll function;
- All code enclosed by '#ifdef USE_SELECT' has been removed, including a
select() call contained inside one function implemented within one such
conditional block.

Author: Andre Negrao

rapid/plugin/group_replication/libmysqlgcs/src/bindings/xcom/xcom/task.c

@@ -73,10 +73,8 @@
 #include "plugin/group_replication/libmysqlgcs/src/bindings/xcom/xcom/xcom_cfg.h"
 
 #ifndef _WIN32
-#ifndef USE_SELECT
 #include <poll.h>
 #endif
-#endif
 
 #include "plugin/group_replication/libmysqlgcs/src/bindings/xcom/xcom/retry.h"
 #include "plugin/group_replication/libmysqlgcs/src/bindings/xcom/xcom/xdr_utils.h"
@@ -88,15 +86,6 @@ task_arg null_arg = {a_end, {0}};
 struct iotasks;
 typedef struct iotasks iotasks;
 
-#ifdef USE_SELECT
-struct iotasks {
-  int maxfd;
-  fd_set read_set;
-  fd_set write_set;
-  fd_set err_set;
-  linkage tasks; /* OHKFIX Should be one each for read and write */
-};
-#else
 typedef struct {
   u_int pollfd_array_len;
   pollfd *pollfd_array_val;
@@ -117,7 +106,7 @@ struct iotasks {
   pollfd_array fd;
   task_env_p_array tasks;
 };
-#endif
+
 int task_errno = 0;
 static task_env *extract_first_delayed();
 static task_env *task_ref(task_env *t);
@@ -593,149 +582,7 @@ static task_env *extract_first_delayed() {
 }
 
 static iotasks iot;
-#ifdef USE_SELECT
-static void iotasks_init(iotasks *iot) {
-  iot->maxfd = 0;
-  FD_ZERO(&iot->read_set);
-  FD_ZERO(&iot->write_set);
-  FD_ZERO(&iot->err_set);
-  link_init(&iot->tasks, type_hash("task_env"));
-}
-
-static void iotasks_deinit(iotasks *iot)
-{
-	DBGOUT(FN);
-}
-
-#if TASK_DBUG_ON
-static void poll_debug() MY_ATTRIBUTE((unused));
-static void poll_debug() {
-  GET_GOUT;
-  if (!IS_XCOM_DEBUG_WITH(XCOM_DEBUG_TRACE))
-    return;
-#if 0
-	NDBG(FD_SETSIZE, d);
-	PTREXP(&iot.tasks);
-	NDBG(iot.tasks.type, u);
-	NDBG(cardinal(&iot.tasks), d);
-	PTREXP(iot.tasks.suc);
-	PTREXP(iot.tasks.pred);
-#endif
-  FWD_ITER(&iot.tasks, task_env, STRLIT("->"); PTREXP(link_iter);
-           PTREXP(link_iter->l.suc); PTREXP(link_iter->l.pred);
-           NDBG(link_iter->waitfd, d);
-           NDBG(FD_ISSET(link_iter->waitfd, &iot.read_set), d);
-           NDBG(FD_ISSET(link_iter->waitfd, &iot.write_set), d);
-           NDBG(FD_ISSET(link_iter->waitfd, &iot.err_set), d););
-  PRINT_GOUT;
-  FREE_GOUT;
-}
-#endif
-
-static int check_completion(task_env *t, fd_set *r, fd_set *w, fd_set *e) {
-  int interrupt = 0;
-  assert(&t->l != &iot.tasks);
-  /* MAY_DBG(FN;
-          STREXP(t->name);
-          NDBG(t->waitfd,d);
-      NDBG(FD_ISSET(t->waitfd,r),d);
-      NDBG(FD_ISSET(t->waitfd,w),d);
-      NDBG(FD_ISSET(t->waitfd,e),d);
-  ); */
-  if (FD_ISSET(t->waitfd, e)) abort(); /* Close file here instead? */
-  interrupt = (t->time != 0.0 && t->time < task_now());
-  if (interrupt || /* timeout ? */
-      FD_ISSET(t->waitfd, r) || FD_ISSET(t->waitfd, w)) {
-    FD_CLR(t->waitfd, &iot.read_set);
-    FD_CLR(t->waitfd, &iot.write_set);
-    FD_CLR(t->waitfd, &iot.err_set);
-    t->interrupt = interrupt;
-    activate(t);
-    if (iot.maxfd - 1 == t->waitfd)
-      iot.maxfd = t->waitfd; /* Shrink set of watched files */
-    return 1;
-  } else {
-    return 0;
-  }
-}
-
-static int poll_wait(int ms) {
-  /* Wait at most ms milliseconds */
-  int wake = 0;
-  struct timeval select_timeout;
-  MAY_DBG(FN; NDBG(ms, d));
-  if (ms < 0 || ms > 1000) ms = 1000; /* Wait at most 1000 ms */
-  /* convert milliseconds to seconds and microseconds */
-  select_timeout.tv_sec = ms / 1000;
-  select_timeout.tv_usec = (ms % 1000) * 1000;
-  {
-    result nfds = {0, 0};
-    fd_set r = iot.read_set;
-    fd_set w = iot.write_set;
-    fd_set e = iot.err_set;
-    MAY_DBG(FN; poll_debug());
-    SET_OS_ERR(0);
-    while ((nfds.val = select(iot.maxfd, &r, &w, &e, &select_timeout)) == -1) {
-      nfds.funerr = to_errno(GET_OS_ERR);
-      if (hard_select_err(nfds.funerr)) {
-        task_dump_err(nfds.funerr);
-        DBGOUT(STRLIT("select failed"); NDBG(iot.maxfd, d));
-        return 0;
-      }
-      SET_OS_ERR(0);
-      r = iot.read_set;
-      w = iot.write_set;
-      e = iot.err_set;
-    }
-    /* MAY_DBG(FN; poll_debug()); */
-    /* Wake up ready tasks */
-    /* if (nfds.val > 0) { */
-    /*   FWD_ITER(&iot.tasks, task_env,  */
-    /*            nfds.val -= check_completion(link_iter, &r, &w, &e);  */
-    /*            if (nfds.val == 0) break); */
-    /* } */
-    FWD_ITER(&iot.tasks, task_env,
-             if (check_completion(link_iter, &r, &w, &e)) wake = 1);
-  }
-  return wake;
-}
 
-static void add_fd(task_env *t, int fd, int op) {
-  MAY_DBG(FN; PTREXP(t); STREXP(t->name); NDBG(fd, d); NDBG(op, c));
-  assert(fd >= 0);
-  t->waitfd = fd;
-  if (fd >= iot.maxfd) iot.maxfd = fd + 1;
-  FD_CLR(fd, &iot.err_set);
-  if ('r' == op)
-    FD_SET(fd, &iot.read_set);
-  else
-    FD_SET(fd, &iot.write_set);
-  task_wait(t, &iot.tasks);
-  /* MAY_DBG(FN; poll_debug()); */
-}
-
-static void unpoll(int i) {
-  FD_CLR(i, &iot.read_set);
-  FD_CLR(i, &iot.write_set);
-  FD_CLR(i, &iot.err_set);
-}
-
-static void wake_all_io() {
-  FWD_ITER(&iot.tasks, task_env, unpoll(link_iter->waitfd);
-           activate(link_iter););
-}
-
-void remove_and_wakeup(int fd) {
-  MAY_DBG(FN; NDBG(fd, d));
-  FWD_ITER(&iot.tasks, task_env, if (fd == link_iter->waitfd) {
-    unpoll(link_iter->waitfd);
-    activate(link_iter);
-    if (iot.maxfd - 1 == link_iter->waitfd)
-      iot.maxfd = link_iter->waitfd; /* Shrink set of watched files */
-  });
-}
-
-#else
 static void iotasks_init(iotasks *iot)
  {
   DBGOUT(FN);
@@ -864,7 +711,6 @@ void remove_and_wakeup(int fd) {
   }
 }
 
-#endif
 task_env *stack = NULL;
 
 task_env *wait_io(task_env *t, int fd, int op) {
@@ -1131,13 +977,8 @@ void task_loop() {
        Wait until something happens.
     */
 #ifdef DEBUG_TASKS
-#ifdef USE_SELECT
-    MAY_DBG(FN; STRLIT("waiting tasks time "); NDBG(seconds(), f);
-            NDBG(cardinal(&iot.tasks), d); NDBG(task_time_q.curn, d));
-#else
     MAY_DBG(FN; STRLIT("waiting tasks time "); NDBG(seconds(), f);
             NDBG(iot.nwait, d); NDBG(task_time_q.curn));
-#endif
 #endif
     {
       double time = seconds();

rapid/plugin/group_replication/libmysqlgcs/src/bindings/xcom/xcom/task.h

@@ -40,7 +40,6 @@ extern "C" {
         Nonblocking IO and event handling need to be rewritten for each new OS.
 */
 
-/* #define USE_SELECT */
 #ifdef TASK_EVENT_TRACE
 void add_base_event(double when, char const *file, int state);
 #define ADD_BASE_EVENT                             \

rapid/plugin/group_replication/libmysqlgcs/src/bindings/xcom/xcom/task_os.h

@@ -34,11 +34,12 @@ extern "C" {
 #define SOCK_EAGAIN WSAEINPROGRESS
 #define SOCK_EWOULDBLOCK WSAEWOULDBLOCK
 #define SOCK_EINPROGRESS WSAEINPROGRESS
+#define SOCK_EALREADY WSAEALREADY
+#define SOCK_ECONNREFUSED WSAECONNREFUSED
 #define SOCK_ERRNO task_errno
 #define SOCK_OPT_REUSEADDR SO_EXCLUSIVEADDRUSE
 #define GET_OS_ERR WSAGetLastError()
 #define SET_OS_ERR(x) WSASetLastError(x)
-#define SOCK_ECONNREFUSED WSAECONNREFUSED
 #define CLOSESOCKET(x) closesocket(x)
 #define SOCK_SHUT_RDWR SD_BOTH
 
@@ -63,6 +64,11 @@ static inline int poll(pollfd * fds, nfds_t nfds, int timeout) {
   return WSAPoll(fds, nfds, timeout);
 }
 
+static inline int is_socket_error(int x)
+{
+	return x == SOCKET_ERROR || x < 0;
+}
+
 #else
 #include <errno.h>
 #include <netdb.h>
@@ -82,11 +88,12 @@ static inline int poll(pollfd * fds, nfds_t nfds, int timeout) {
 #define SOCK_EAGAIN EAGAIN
 #define SOCK_EWOULDBLOCK EWOULDBLOCK
 #define SOCK_EINPROGRESS EINPROGRESS
+#define SOCK_EALREADY EALREADY
+#define SOCK_ECONNREFUSED ECONNREFUSED
 #define SOCK_ERRNO task_errno
 #define SOCK_OPT_REUSEADDR SO_REUSEADDR
 #define GET_OS_ERR errno
 #define SET_OS_ERR(x) errno = (x)
-#define SOCK_ECONNREFUSED ECONNREFUSED
 #define CLOSESOCKET(x) close(x)
 #define SOCK_SHUT_RDWR (SHUT_RD | SHUT_WR)
 
@@ -100,6 +107,11 @@ static inline int hard_select_err(int err) {
 
 typedef struct pollfd pollfd;
 
+static inline int is_socket_error(int x)
+{
+	return x < 0;
+}
+
 #endif
 
 extern void remove_and_wakeup(int fd);