poll-loop: Create Windows event handles for sockets automatically.

We currently have a poll_fd_wait_event(fd, wevent, events) function that
is used at places common to Windows and Linux where we have to wait on
sockets.  On Linux, 'wevent' is always set as zero. On Windows, for sockets,
when we send both 'fd' and 'wevent', we associate them with each other for
'events' and then wait on 'wevent'. Also on Windows, when we only send 'wevent'
to this function, we would simply wait for all events for that 'wevent'.

There is a disadvantage with this approach.
* Windows clients need to create a 'wevent' and then pass it along. This
means that at a lot of places where we create sockets, we also are forced
to create a 'wevent'.

With this commit, we pass the responsibility of creating a 'wevent' to
poll_fd_wait() in case of sockets. That way, a client using poll_fd_wait()
is only concerned about sockets and not about 'wevents'. There is a potential
disadvantage with this change in that we create events more often and that
may have a performance penalty. If that turns out to be the case, we will
eventually need to create a pool of wevents that can be re-used.

In Windows, there are cases where we want to wait on a event (not
associated with any sockets) and then control it using functions
like SetEvent() etc. For that purpose, introduce a new function
poll_wevent_wait(). For this function, the client needs to create a event
and then pass it along as an argument.

Signed-off-by: Gurucharan Shetty <[email protected]>
Acked-By: Ben Pfaff <[email protected]>

lib/daemon-windows.c

@@ -107,7 +107,7 @@ service_start(int *argcp, char **argvp[])
             VLOG_FATAL("Failed to create a event (%s).", msg_buf);
         }
 
-        poll_fd_wait_event(0, wevent, POLLIN);
+        poll_wevent_wait(wevent);
 
         /* Register the control handler. This function is called by the service
          * manager to stop the service. */
@@ -206,7 +206,7 @@ should_service_stop(void)
         if (service_status.dwCurrentState != SERVICE_RUNNING) {
             return true;
         } else {
-            poll_fd_wait_event(0, wevent, POLLIN);
+            poll_wevent_wait(wevent);
         }
     }
     return false;

lib/fatal-signal.c

@@ -59,9 +59,12 @@ static struct hook hooks[MAX_HOOKS];
 static size_t n_hooks;
 
 static int signal_fds[2];
-static HANDLE wevent;
 static volatile sig_atomic_t stored_sig_nr = SIG_ATOMIC_MAX;
 
+#ifdef _WIN32
+static HANDLE wevent;
+#endif
+
 static struct ovs_mutex mutex;
 
 static void call_hooks(int sig_nr);
@@ -215,7 +218,11 @@ void
 fatal_signal_wait(void)
 {
     fatal_signal_init();
-    poll_fd_wait_event(signal_fds[0], wevent, POLLIN);
+#ifdef _WIN32
+    poll_wevent_wait(wevent);
+#else
+    poll_fd_wait(signal_fds[0], POLLIN);
+#endif
 }
 
 void

lib/latch-unix.c

@@ -83,5 +83,5 @@ latch_is_set(const struct latch *latch)
 void
 latch_wait_at(const struct latch *latch, const char *where)
 {
-    poll_fd_wait_at(latch->fds[0], 0, POLLIN, where);
+    poll_fd_wait_at(latch->fds[0], POLLIN, where);
 }

lib/latch-windows.c

@@ -79,5 +79,5 @@ latch_is_set(const struct latch *latch)
 void
 latch_wait_at(const struct latch *latch, const char *where)
 {
-    poll_fd_wait_at(0, latch->wevent, POLLIN, where);
+    poll_wevent_wait_at(latch->wevent, where);
 }

lib/poll-loop.c

@@ -35,7 +35,7 @@
 
 VLOG_DEFINE_THIS_MODULE(poll_loop);
 
-COVERAGE_DEFINE(poll_fd_wait);
+COVERAGE_DEFINE(poll_create_node);
 COVERAGE_DEFINE(poll_zero_timeout);
 
 struct poll_node {
@@ -59,11 +59,12 @@ static struct poll_loop *poll_loop(void);
 
 /* Look up the node with same fd and wevent. */
 static struct poll_node *
-find_poll_node(struct poll_loop *loop, int fd, uint32_t wevent)
+find_poll_node(struct poll_loop *loop, int fd, HANDLE wevent)
 {
     struct poll_node *node;
 
-    HMAP_FOR_EACH_WITH_HASH (node, hmap_node, hash_2words(fd, wevent),
+    HMAP_FOR_EACH_WITH_HASH (node, hmap_node,
+                             hash_2words(fd, (uint32_t)wevent),
                              &loop->poll_nodes) {
         if (node->pollfd.fd == fd && node->wevent == wevent) {
             return node;
@@ -77,54 +78,91 @@ find_poll_node(struct poll_loop *loop, int fd, uint32_t wevent)
  *     Registers 'fd' as waiting for the specified 'events' (which should be
  *     POLLIN or POLLOUT or POLLIN | POLLOUT).  The following call to
  *     poll_block() will wake up when 'fd' becomes ready for one or more of the
- *     requested events.  the 'fd's are given to poll() function later.
+ *     requested events. The 'fd's are given to poll() function later.
  *
  * On Windows system:
  *
- *     If both 'wevent' handle and 'fd' is specified, associate the 'fd' with
- *     with that 'wevent' for 'events' (implemented in poll_block()).
- *     In case of no 'fd' specified, wake up on any event on that 'wevent'.
- *     These wevents are given to the WaitForMultipleObjects() to be polled.
- *     The event registration is one-shot: only the following call to
- *     poll_block() is affected.  The event will need to be re-registered after
- *     poll_block() is called if it is to persist.
+ *     If 'fd' is specified, create a new 'wevent'. Association of 'fd' and
+ *     'wevent' for 'events' happens in poll_block(). If 'wevent' is specified,
+ *     it is assumed that it is unrelated to any sockets and poll_block()
+ *     will wake up on any event on that 'wevent'. It is an error to pass
+ *     both 'wevent' and 'fd'.
+ *
+ * The event registration is one-shot: only the following call to
+ * poll_block() is affected.  The event will need to be re-registered after
+ * poll_block() is called if it is to persist.
  *
  * ('where' is used in debug logging.  Commonly one would use poll_fd_wait() to
  * automatically provide the caller's source file and line number for
  * 'where'.) */
-void
-poll_fd_wait_at(int fd, HANDLE wevent, short int events, const char *where)
+static void
+poll_create_node(int fd, HANDLE wevent, short int events, const char *where)
 {
     struct poll_loop *loop = poll_loop();
     struct poll_node *node;
 
-    COVERAGE_INC(poll_fd_wait);
+    COVERAGE_INC(poll_create_node);
 
-#ifdef _WIN32
-    /* Null event cannot be polled. */
-    if (wevent == 0) {
-        VLOG_ERR("No event to wait fd %d", fd);
-        return;
-    }
-#else
-    wevent = 0;
-#endif
+    /* Both 'fd' and 'wevent' cannot be set. */
+    ovs_assert(!fd != !wevent);
 
-    /* Check for duplicate.  If found, "or" the event. */
+    /* Check for duplicate.  If found, "or" the events. */
     node = find_poll_node(loop, fd, wevent);
     if (node) {
         node->pollfd.events |= events;
     } else {
         node = xzalloc(sizeof *node);
         hmap_insert(&loop->poll_nodes, &node->hmap_node,
-                    hash_2words(fd, wevent));
+                    hash_2words(fd, (uint32_t)wevent));
         node->pollfd.fd = fd;
         node->pollfd.events = events;
+#ifdef _WIN32
+        if (!wevent) {
+            wevent = CreateEvent(NULL, FALSE, FALSE, NULL);
+        }
+#endif
         node->wevent = wevent;
         node->where = where;
     }
 }
 
+/* Registers 'fd' as waiting for the specified 'events' (which should be POLLIN
+ * or POLLOUT or POLLIN | POLLOUT).  The following call to poll_block() will
+ * wake up when 'fd' becomes ready for one or more of the requested events.
+ *
+ * On Windows, 'fd' must be a socket.
+ *
+ * The event registration is one-shot: only the following call to poll_block()
+ * is affected.  The event will need to be re-registered after poll_block() is
+ * called if it is to persist.
+ *
+ * ('where' is used in debug logging.  Commonly one would use poll_fd_wait() to
+ * automatically provide the caller's source file and line number for
+ * 'where'.) */
+void
+poll_fd_wait_at(int fd, short int events, const char *where)
+{
+    poll_create_node(fd, 0, events, where);
+}
+
+#ifdef _WIN32
+/* Registers for the next call to poll_block() to wake up when 'wevent' is
+ * signaled.
+ *
+ * The event registration is one-shot: only the following call to poll_block()
+ * is affected.  The event will need to be re-registered after poll_block() is
+ * called if it is to persist.
+ *
+ * ('where' is used in debug logging.  Commonly one would use
+ * poll_wevent_wait() to automatically provide the caller's source file and
+ * line number for 'where'.) */
+void
+poll_wevent_wait_at(HANDLE wevent, const char *where)
+{
+    poll_create_node(0, wevent, 0, where);
+}
+#endif /* _WIN32 */
+
 /* Causes the following call to poll_block() to block for no more than 'msec'
  * milliseconds.  If 'msec' is nonpositive, the following call to poll_block()
  * will not block at all.
@@ -258,6 +296,12 @@ free_poll_nodes(struct poll_loop *loop)
 
     HMAP_FOR_EACH_SAFE (node, next, hmap_node, &loop->poll_nodes) {
         hmap_remove(&loop->poll_nodes, &node->hmap_node);
+#ifdef _WIN32
+        if (node->wevent && node->pollfd.fd) {
+            WSAEventSelect(node->pollfd.fd, NULL, 0);
+            CloseHandle(node->wevent);
+        }
+#endif
         free(node);
     }
 }

lib/poll-loop.h

@@ -50,12 +50,12 @@ extern "C" {
  * caller to supply a location explicitly, which is useful if the caller's own
  * caller would be more useful in log output.  See timer_wait_at() for an
  * example. */
-void poll_fd_wait_at(int fd, HANDLE wevent, short int events, const char *where);
-#ifndef _WIN32
-#define poll_fd_wait(fd, events) poll_fd_wait_at(fd, 0, events, SOURCE_LOCATOR)
-#endif
-#define poll_fd_wait_event(fd, wevent, events)  \
-    poll_fd_wait_at(fd, wevent, events, SOURCE_LOCATOR)
+void poll_fd_wait_at(int fd, short int events, const char *where);
+#define poll_fd_wait(fd, events) poll_fd_wait_at(fd, events, SOURCE_LOCATOR)
+
+#ifdef _WIN32
+#define poll_wevent_wait(wevent) poll_wevent_wait_at(wevent, SOURCE_LOCATOR)
+#endif /* _WIN32 */
 
 void poll_timer_wait_at(long long int msec, const char *where);
 #define poll_timer_wait(msec) poll_timer_wait_at(msec, SOURCE_LOCATOR)

lib/stream-fd-windows.c

@@ -39,7 +39,6 @@ struct stream_fd
 {
     struct stream stream;
     int fd;
-    HANDLE wevent;
 };
 
 static const struct stream_class stream_fd_class;
@@ -61,7 +60,6 @@ new_fd_stream(const char *name, int fd, int connect_status,
     s = xmalloc(sizeof *s);
     stream_init(&s->stream, &stream_fd_class, connect_status, name);
     s->fd = fd;
-    s->wevent = CreateEvent(NULL, FALSE, FALSE, NULL);
     *streamp = &s->stream;
     return 0;
 }
@@ -77,8 +75,6 @@ static void
 fd_close(struct stream *stream)
 {
     struct stream_fd *s = stream_fd_cast(stream);
-    WSAEventSelect(s->fd, NULL, 0);
-    CloseHandle(s->wevent);
     closesocket(s->fd);
     free(s);
 }
@@ -130,11 +126,11 @@ fd_wait(struct stream *stream, enum stream_wait_type wait)
     switch (wait) {
     case STREAM_CONNECT:
     case STREAM_SEND:
-        poll_fd_wait_event(s->fd, s->wevent, POLLOUT);
+        poll_fd_wait(s->fd, POLLOUT);
         break;
 
     case STREAM_RECV:
-        poll_fd_wait_event(s->fd, s->wevent, POLLIN);
+        poll_fd_wait(s->fd, POLLIN);
         break;
 
     default:
@@ -161,7 +157,6 @@ struct fd_pstream
 {
     struct pstream pstream;
     int fd;
-    HANDLE wevent;
     int (*accept_cb)(int fd, const struct sockaddr_storage *, size_t ss_len,
                      struct stream **);
     int (*set_dscp_cb)(int fd, uint8_t dscp);
@@ -201,7 +196,6 @@ new_fd_pstream(const char *name, int fd,
     struct fd_pstream *ps = xmalloc(sizeof *ps);
     pstream_init(&ps->pstream, &fd_pstream_class, name);
     ps->fd = fd;
-    ps->wevent = CreateEvent(NULL, FALSE, FALSE, NULL);
     ps->accept_cb = accept_cb;
     ps->set_dscp_cb = set_dscp_cb;
     ps->unlink_path = unlink_path;
@@ -213,8 +207,6 @@ static void
 pfd_close(struct pstream *pstream)
 {
     struct fd_pstream *ps = fd_pstream_cast(pstream);
-    WSAEventSelect(ps->fd, NULL, 0);
-    CloseHandle(ps->wevent);
     closesocket(ps->fd);
     if (ps->unlink_path) {
         fatal_signal_unlink_file_now(ps->unlink_path);
@@ -254,7 +246,7 @@ static void
 pfd_wait(struct pstream *pstream)
 {
     struct fd_pstream *ps = fd_pstream_cast(pstream);
-    poll_fd_wait_event(ps->fd, ps->wevent, POLLIN);
+    poll_fd_wait(ps->fd, POLLIN);
 }
 
 static int