Changeset 68753 in vbox for trunk/src/VBox/NetworkServices/NAT

Timestamp:

Sep 14, 2017 2:15:39 AM (7 years ago)

Author:

vboxsync

Message:

Rewrite "channels". Use RTREQQUEUE to schedule requests on the
pollmgr thread and just one socket to wake up poll(2). To minimize
churn keep the old API, so no other files are affected.

Location:

trunk/src/VBox/NetworkServices/NAT

Files:

• proxy_pollmgr.c (modified) (9 diffs)
• proxy_pollmgr.h (modified) (1 diff)

trunk/src/VBox/NetworkServices/NAT/proxy_pollmgr.c

@@ -28 +28 @@
 #include <err.h>
 #include <errno.h>
+#include <fcntl.h>
 #include <poll.h>
 #include <stdio.h>
@@ -41 +42 @@
 #endif
 
+#include <iprt/req.h>
+#include <iprt/err.h>
+
+
 #define POLLMGR_GARBAGE (-1)
+
+
+enum {
+    POLLMGR_QUEUE = 0,
+
+    POLLMGR_SLOT_STATIC_COUNT,
+    POLLMGR_SLOT_FIRST_DYNAMIC = POLLMGR_SLOT_STATIC_COUNT
+};
+
+
+struct pollmgr_chan {
+    struct pollmgr_handler *handler;
+    void *arg;
+    bool arg_valid;
+};
 
 struct pollmgr {
@@ -53 +73 @@
 #define POLLMGR_CHFD_RD 0       /* - pollmgr side */
 #define POLLMGR_CHFD_WR 1       /* - client side */
+
+
+    /* emulate channels with request queue */
+    RTREQQUEUE queue;
+    struct pollmgr_handler queue_handler;
+    struct pollmgr_chan chan_handlers[POLLMGR_CHAN_COUNT];
 } pollmgr;
 
+
+static int pollmgr_queue_callback(struct pollmgr_handler *, SOCKET, int);
+static void pollmgr_chan_call_handler(int, void *);
 
 static void pollmgr_loop(void);
@@ -82 +111 @@
     struct pollmgr_handler **newhdls;
     nfds_t newcap;
-    int status;
+    int rc, status;
     nfds_t i;
+
+    rc = RTReqQueueCreate(&pollmgr.queue);
+    if (RT_FAILURE(rc))
+        return -1;
 
     pollmgr.fds = NULL;
@@ -97 +130 @@
     for (i = 0; i < POLLMGR_SLOT_STATIC_COUNT; ++i) {
 #ifndef RT_OS_WINDOWS
+        int j;
+
         status = socketpair(PF_LOCAL, SOCK_DGRAM, 0, pollmgr.chan[i]);
         if (status < 0) {
             DPRINTF(("socketpair: %R[sockerr]\n", SOCKERRNO()));
             goto cleanup_close;
+        }
+
+        /* now manually make them O_NONBLOCK */
+        for (j = 0; j < 2; ++j) {
+            int s = pollmgr.chan[i][j];
+            int sflags;
+
+            sflags = fcntl(s, F_GETFL, 0);
+            if (sflags < 0) {
+                DPRINTF0(("F_GETFL: %R[sockerr]\n", errno));
+                goto cleanup_close;
+            }
+
+            status = fcntl(s, F_SETFL, sflags | O_NONBLOCK);
+            if (status < 0) {
+                DPRINTF0(("O_NONBLOCK: %R[sockerr]\n", errno));
+                goto cleanup_close;
+            }
         }
 #else
@@ -141 +194 @@
     }
 
+    /* add request queue notification */
+    pollmgr.queue_handler.callback = pollmgr_queue_callback;
+    pollmgr.queue_handler.data = NULL;
+    pollmgr.queue_handler.slot = -1;
+
+    pollmgr_add_at(POLLMGR_QUEUE, &pollmgr.queue_handler,
+                   pollmgr.chan[POLLMGR_QUEUE][POLLMGR_CHFD_RD],
+                   POLLIN);
+
     return 0;
 
@@ -157 +219 @@
 
 /*
+ * Add new channel.  We now implement channels with request queue, so
+ * all channels get the same socket that triggers queue processing.
+ *
  * Must be called before pollmgr loop is started, so no locking.
  */
@@ -162 +227 @@
 pollmgr_add_chan(int slot, struct pollmgr_handler *handler)
 {
-    if (slot >= POLLMGR_SLOT_FIRST_DYNAMIC) {
-        handler->slot = -1;
-        return INVALID_SOCKET;
-    }
-
-    pollmgr_add_at(slot, handler, pollmgr.chan[slot][POLLMGR_CHFD_RD], POLLIN);
-    return pollmgr.chan[slot][POLLMGR_CHFD_WR];
+    AssertReturn(0 <= slot && slot < POLLMGR_CHAN_COUNT, INVALID_SOCKET);
+    AssertReturn(handler != NULL && handler->callback != NULL, INVALID_SOCKET);
+
+    handler->slot = slot;
+    pollmgr.chan_handlers[slot].handler = handler;
+    return pollmgr.chan[POLLMGR_QUEUE][POLLMGR_CHFD_WR];
+}
+
+
+/*
+ * This used to actually send data over the channel's socket.  Now we
+ * queue a request and send single byte notification over shared
+ * POLLMGR_QUEUE socket.
+ */
+ssize_t
+pollmgr_chan_send(int slot, void *buf, size_t nbytes)
+{
+    static const char notification = 0x5a;
+
+    void *ptr;
+    SOCKET fd;
+    ssize_t nsent;
+    int rc;
+
+    AssertReturn(0 <= slot && slot < POLLMGR_CHAN_COUNT, -1);
+
+    /* 
+     * XXX: Hack alert.  We only ever "sent" single pointer which was
+     * simultaneously both the wakeup event for the poll and the
+     * argument for the channel handler that it read from the channel.
+     * So now we pass this pointer to the request and arrange for the
+     * handler to "read" it when it asks for it.
+     */
+    if (nbytes != sizeof(void *)) {
+        return -1;
+    }
+
+    ptr = *(void **)buf;
+
+    rc = RTReqQueueCallEx(pollmgr.queue, NULL, 0,
+                          RTREQFLAGS_VOID | RTREQFLAGS_NO_WAIT,
+                          (PFNRT)pollmgr_chan_call_handler, 2,
+                          slot, ptr);
+
+    fd = pollmgr.chan[POLLMGR_QUEUE][POLLMGR_CHFD_WR];
+    nsent = send(fd, &notification, 1, 0);
+    if (nsent == SOCKET_ERROR) {
+        DPRINTF(("send on chan %d: %R[sockerr]\n", slot, SOCKERRNO()));
+        return -1;
+    }
+    else if ((size_t)nsent != 1) {
+        DPRINTF(("send on chan %d: datagram truncated to %u bytes",
+                 slot, (unsigned int)nsent));
+        return -1;
+    }
+
+    /* caller thinks it's sending the pointer */
+    return sizeof(void *);
+}
+
+
+/*
+ * pollmgr_chan_send() sent us a notification, process the queue.
+ */
+static int
+pollmgr_queue_callback(struct pollmgr_handler *handler, SOCKET fd, int revents)
+{
+    ssize_t nread;
+    int sockerr;
+    int rc;
+
+    RT_NOREF(handler, revents);
+    Assert(pollmgr.queue != NIL_RTREQQUEUE);
+
+    nread = recv(fd, (char *)pollmgr_udpbuf, sizeof(pollmgr_udpbuf), 0);
+    sockerr = SOCKERRNO();      /* save now, may be clobbered */
+
+    if (nread == SOCKET_ERROR) {
+        DPRINTF0(("%s: recv: %R[sockerr]\n", __func__, sockerr));
+        return POLLIN;
+    }
+
+    DPRINTF2(("%s: read %zd\n", __func__, nread));
+    if (nread == 0) {
+        return POLLIN;
+    }
+
+    rc = RTReqQueueProcess(pollmgr.queue, 0);
+    if (RT_UNLIKELY(rc != VERR_TIMEOUT && RT_FAILURE_NP(rc))) {
+        DPRINTF0(("%s: RTReqQueueProcess: %Rrc\n", __func__, rc));
+    }
+
+    return POLLIN;
+}
+
+
+/*
+ * Queued requests use this function to emulate the call to the
+ * handler's callback.
+ */
+static void
+pollmgr_chan_call_handler(int slot, void *arg)
+{
+    struct pollmgr_handler *handler;
+    int nevents;
+
+    AssertReturnVoid(0 <= slot && slot < POLLMGR_CHAN_COUNT);
+
+    handler = pollmgr.chan_handlers[slot].handler;
+    AssertReturnVoid(handler != NULL && handler->callback != NULL);
+
+    /* arrange for pollmgr_chan_recv_ptr() to "receive" the arg */
+    pollmgr.chan_handlers[slot].arg = arg;
+    pollmgr.chan_handlers[slot].arg_valid = true;
+
+    nevents = handler->callback(handler, -1, POLLIN);
+    if (nevents != POLLIN) {
+        DPRINTF2(("%s: nevents=0x%x!\n", nevents));
+    }
+}
+
+
+/*
+ * "Receive" a pointer "sent" over poll manager channel.
+ */
+void *
+pollmgr_chan_recv_ptr(struct pollmgr_handler *handler, SOCKET fd, int revents)
+{
+    int slot;
+    void *ptr;
+
+    RT_NOREF(fd);
+
+    slot = handler->slot;
+    Assert(0 <= slot && slot < POLLMGR_CHAN_COUNT);
+
+    if (revents & POLLNVAL) {
+        errx(EXIT_FAILURE, "chan %d: fd invalid", (int)handler->slot);
+        /* NOTREACHED */
+    }
+
+    if (revents & (POLLERR | POLLHUP)) {
+        errx(EXIT_FAILURE, "chan %d: fd error", (int)handler->slot);
+        /* NOTREACHED */
+    }
+
+    LWIP_ASSERT1(revents & POLLIN);
+
+    if (!pollmgr.chan_handlers[slot].arg_valid) {
+        err(EXIT_FAILURE, "chan %d: recv", (int)handler->slot);
+        /* NOTREACHED */
+    }
+
+    ptr = pollmgr.chan_handlers[slot].arg;
+    pollmgr.chan_handlers[slot].arg_valid = false;
+
+    return ptr;
 }
 
@@ -239 +454 @@
 
     handler->slot = slot;
-}
-
-
-ssize_t
-pollmgr_chan_send(int slot, void *buf, size_t nbytes)
-{
-    SOCKET fd;
-    ssize_t nsent;
-
-    if (slot >= POLLMGR_SLOT_FIRST_DYNAMIC) {
-        return -1;
-    }
-
-    fd = pollmgr.chan[slot][POLLMGR_CHFD_WR];
-    nsent = send(fd, buf, (int)nbytes, 0);
-    if (nsent == SOCKET_ERROR) {
-        DPRINTF(("send on chan %d: %R[sockerr]\n", slot, SOCKERRNO()));
-        return -1;
-    }
-    else if ((size_t)nsent != nbytes) {
-        DPRINTF(("send on chan %d: datagram truncated to %u bytes",
-                 slot, (unsigned int)nsent));
-        return -1;
-    }
-
-    return nsent;
-}
-
-
-/**
- * Receive a pointer sent over poll manager channel.
- */
-void *
-pollmgr_chan_recv_ptr(struct pollmgr_handler *handler, SOCKET fd, int revents)
-{
-    void *ptr;
-    ssize_t nread;
-    NOREF(handler);
-
-    if (revents & POLLNVAL) {
-        errx(EXIT_FAILURE, "chan %d: fd invalid", (int)handler->slot);
-        /* NOTREACHED */
-    }
-
-    if (revents & (POLLERR | POLLHUP)) {
-        errx(EXIT_FAILURE, "chan %d: fd error", (int)handler->slot);
-        /* NOTREACHED */
-    }
-
-    LWIP_ASSERT1(revents & POLLIN);
-    nread = recv(fd, (char *)&ptr, sizeof(ptr), 0);
-
-    if (nread == SOCKET_ERROR) {
-        err(EXIT_FAILURE, "chan %d: recv", (int)handler->slot);
-        /* NOTREACHED */
-    }
-    if (nread != sizeof(ptr)) {
-        errx(EXIT_FAILURE, "chan %d: recv: read %d bytes",
-             (int)handler->slot, (int)nread);
-        /* NOTREACHED */
-    }
-
-    return ptr;
 }
 

trunk/src/VBox/NetworkServices/NAT/proxy_pollmgr.h

@@ -36 +36 @@
     POLLMGR_CHAN_PORTFWD,       /* add/remove port forwarding rules */
 
-    POLLMGR_SLOT_STATIC_COUNT,
-    POLLMGR_SLOT_FIRST_DYNAMIC = POLLMGR_SLOT_STATIC_COUNT
+    POLLMGR_CHAN_COUNT
 };