netdev-dpdk: fix mbuf leaks
[cascardo/ovs.git] / lib / poll-loop.c
index 0f45d98..e83d989 100644 (file)
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008, 2009, 2010, 2011, 2012, 2013 Nicira, Inc.
+ * Copyright (c) 2008, 2009, 2010, 2011, 2012, 2013, 2014 Nicira, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
 #include "fatal-signal.h"
 #include "list.h"
 #include "ovs-thread.h"
+#include "seq.h"
 #include "socket-util.h"
 #include "timeval.h"
-#include "vlog.h"
+#include "openvswitch/vlog.h"
+#include "hmap.h"
+#include "hash.h"
 
 VLOG_DEFINE_THIS_MODULE(poll_loop);
 
-COVERAGE_DEFINE(poll_fd_wait);
+COVERAGE_DEFINE(poll_create_node);
 COVERAGE_DEFINE(poll_zero_timeout);
 
+struct poll_node {
+    struct hmap_node hmap_node;
+    struct pollfd pollfd;       /* Events to pass to time_poll(). */
+    HANDLE wevent;              /* Events for WaitForMultipleObjects(). */
+    const char *where;          /* Where poll_node was created. */
+};
+
 struct poll_loop {
     /* All active poll waiters. */
-    struct pollfd *pollfds;     /* Events to pass to poll(). */
-    const char **where;         /* Where each pollfd was created. */
-    size_t n_waiters;           /* Number of elems in 'where' and 'pollfds'. */
-    size_t allocated_waiters;   /* Allocated elems in 'where' and 'pollfds'. */
+    struct hmap poll_nodes;
 
     /* Time at which to wake up the next call to poll_block(), LLONG_MIN to
      * wake up immediately, or LLONG_MAX to wait forever. */
@@ -50,10 +57,85 @@ struct poll_loop {
 
 static struct poll_loop *poll_loop(void);
 
+/* Look up the node with same fd or wevent. */
+static struct poll_node *
+find_poll_node(struct poll_loop *loop, int fd, HANDLE wevent)
+{
+    struct poll_node *node;
+
+    /* Both 'fd' and 'wevent' cannot be set. */
+    ovs_assert(!fd != !wevent);
+
+    HMAP_FOR_EACH_WITH_HASH (node, hmap_node,
+                             hash_2words(fd, (uint32_t)wevent),
+                             &loop->poll_nodes) {
+        if ((fd && node->pollfd.fd == fd)
+            || (wevent && node->wevent == wevent)) {
+            return node;
+        }
+    }
+    return NULL;
+}
+
+/* On Unix based systems:
+ *
+ *     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.
+ *
+ * On Windows system:
+ *
+ *     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'.) */
+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_create_node);
+
+    /* Both 'fd' and 'wevent' cannot be set. */
+    ovs_assert(!fd != !wevent);
+
+    /* 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, (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.
@@ -64,22 +146,26 @@ static struct poll_loop *poll_loop(void);
 void
 poll_fd_wait_at(int fd, short int events, const char *where)
 {
-    struct poll_loop *loop = poll_loop();
-
-    COVERAGE_INC(poll_fd_wait);
-    if (loop->n_waiters >= loop->allocated_waiters) {
-        loop->where = x2nrealloc(loop->where, &loop->allocated_waiters,
-                                 sizeof *loop->where);
-        loop->pollfds = xrealloc(loop->pollfds,
-                                 (loop->allocated_waiters
-                                  * sizeof *loop->pollfds));
-    }
+    poll_create_node(fd, 0, events, where);
+}
 
-    loop->where[loop->n_waiters] = where;
-    loop->pollfds[loop->n_waiters].fd = fd;
-    loop->pollfds[loop->n_waiters].events = events;
-    loop->n_waiters++;
+#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()
@@ -167,7 +253,9 @@ log_wakeup(const char *where, const struct pollfd *pollfd, int timeout)
     cpu_usage = get_cpu_usage();
     if (VLOG_IS_DBG_ENABLED()) {
         level = VLL_DBG;
-    } else if (cpu_usage > 50 && !VLOG_DROP_INFO(&rl)) {
+    } else if (cpu_usage > 50
+               && !thread_is_pmd()
+               && !VLOG_DROP_INFO(&rl)) {
         level = VLL_INFO;
     } else {
         return;
@@ -207,6 +295,23 @@ log_wakeup(const char *where, const struct pollfd *pollfd, int timeout)
     ds_destroy(&s);
 }
 
+static void
+free_poll_nodes(struct poll_loop *loop)
+{
+    struct poll_node *node, *next;
+
+    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);
+    }
+}
+
 /* Blocks until one or more of the events registered with poll_fd_wait()
  * occurs, or until the minimum duration registered with poll_timer_wait()
  * elapses, or not at all if poll_immediate_wake() has been called. */
@@ -214,8 +319,12 @@ void
 poll_block(void)
 {
     struct poll_loop *loop = poll_loop();
+    struct poll_node *node;
+    struct pollfd *pollfds;
+    HANDLE *wevents = NULL;
     int elapsed;
     int retval;
+    int i;
 
     /* Register fatal signal events before actually doing any real work for
      * poll_block. */
@@ -225,7 +334,34 @@ poll_block(void)
         COVERAGE_INC(poll_zero_timeout);
     }
 
-    retval = time_poll(loop->pollfds, loop->n_waiters,
+    timewarp_run();
+    pollfds = xmalloc(hmap_count(&loop->poll_nodes) * sizeof *pollfds);
+
+#ifdef _WIN32
+    wevents = xmalloc(hmap_count(&loop->poll_nodes) * sizeof *wevents);
+#endif
+
+    /* Populate with all the fds and events. */
+    i = 0;
+    HMAP_FOR_EACH (node, hmap_node, &loop->poll_nodes) {
+        pollfds[i] = node->pollfd;
+#ifdef _WIN32
+        wevents[i] = node->wevent;
+        if (node->pollfd.fd && node->wevent) {
+            short int wsa_events = 0;
+            if (node->pollfd.events & POLLIN) {
+                wsa_events |= FD_READ | FD_ACCEPT | FD_CLOSE;
+            }
+            if (node->pollfd.events & POLLOUT) {
+                wsa_events |= FD_WRITE | FD_CONNECT | FD_CLOSE;
+            }
+            WSAEventSelect(node->pollfd.fd, node->wevent, wsa_events);
+        }
+#endif
+        i++;
+    }
+
+    retval = time_poll(pollfds, hmap_count(&loop->poll_nodes), wevents,
                        loop->timeout_when, &elapsed);
     if (retval < 0) {
         static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
@@ -233,21 +369,25 @@ poll_block(void)
     } else if (!retval) {
         log_wakeup(loop->timeout_where, NULL, elapsed);
     } else if (get_cpu_usage() > 50 || VLOG_IS_DBG_ENABLED()) {
-        size_t i;
-
-        for (i = 0; i < loop->n_waiters; i++) {
-            if (loop->pollfds[i].revents) {
-                log_wakeup(loop->where[i], &loop->pollfds[i], 0);
+        i = 0;
+        HMAP_FOR_EACH (node, hmap_node, &loop->poll_nodes) {
+            if (pollfds[i].revents) {
+                log_wakeup(node->where, &pollfds[i], 0);
             }
+            i++;
         }
     }
 
+    free_poll_nodes(loop);
     loop->timeout_when = LLONG_MAX;
     loop->timeout_where = NULL;
-    loop->n_waiters = 0;
+    free(pollfds);
+    free(wevents);
 
     /* Handle any pending signals before doing anything else. */
     fatal_signal_run();
+
+    seq_woke();
 }
 \f
 static void
@@ -255,8 +395,8 @@ free_poll_loop(void *loop_)
 {
     struct poll_loop *loop = loop_;
 
-    free(loop->pollfds);
-    free(loop->where);
+    free_poll_nodes(loop);
+    hmap_destroy(&loop->poll_nodes);
     free(loop);
 }
 
@@ -275,7 +415,8 @@ poll_loop(void)
     loop = pthread_getspecific(key);
     if (!loop) {
         loop = xzalloc(sizeof *loop);
-        pthread_setspecific(key, loop);
+        hmap_init(&loop->poll_nodes);
+        xpthread_setspecific(key, loop);
     }
     return loop;
 }