Handle signals with kqueue

We used sigsetjmp() and siglogjmp() for signal handling, which is very
hard to use correctly and too magical. Replace it with the kqueue(),
which can notify events on sockets and signals.

The get signals with kqueue, we need to block them. This has the nice
property that no function in any thread will fail with EINTR when we
receive a signal, which simplifies the code.

We setup signal handling before we open the pidfile, so receiving a
signal after the pidfile was created will always remove the pidfile.
Before this change we has a small window when receiving a signal would
terminate the process without removing the pidfile.

To wait for connection or signal, we wait for kqueue events. If we
receiving a signal we break the loop and exit normally, since
termination by signal is normal. This can help programs running
socket_vment that may be confused by exit code 1.

Notes:

- Keeping listen_fd in blocking mode to simplify the code. accept()
  should not block when we receive a EVFILT_READ event. We can change to
  non-blocking mode later if this assumption is wrong.

Signed-off-by: Nir Soffer <nirsof@gmail.com>

Author: nirs

main.c

@@ -3,10 +3,13 @@
 #include <errno.h>
 #include <grp.h>
 #include <sched.h>
-#include <setjmp.h>
+#include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
+#include <sys/event.h>
 #include <sys/stat.h>
+#include <sys/time.h>
+#include <sys/types.h>
 #include <sys/uio.h>
 #include <sys/un.h>
 #include <unistd.h>
@@ -19,6 +22,8 @@
 #error "Requires macOS 10.15 or later"
 #endif
 
+#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
+
 bool debug = false;
 
 static const char *vmnet_strerror(vmnet_return_t v) {
@@ -280,12 +285,6 @@ static interface_ref start(struct state *state, struct cli_options *cliopt) {
   return iface;
 }
 
-static sigjmp_buf jmpbuf;
-static void signalhandler(int signal) {
-  INFOF("Received signal: %s", strsignal(signal));
-  siglongjmp(jmpbuf, 1);
-}
-
 static void stop(struct state *state, interface_ref iface) {
   if (iface == NULL) {
     return;
@@ -391,13 +390,55 @@ static int create_pidfile(const char *pidfile)
   return fd;
 }
 
+static int setup_signals(int kq)
+{
+  struct kevent changes[] = {
+    { .ident=SIGHUP, .filter=EVFILT_SIGNAL, .flags=EV_ADD },
+    { .ident=SIGINT, .filter=EVFILT_SIGNAL, .flags=EV_ADD },
+    { .ident=SIGTERM, .filter=EVFILT_SIGNAL, .flags=EV_ADD },
+  };
+
+  // Block signals we want to receive via kqueue.
+  sigset_t mask;
+  sigemptyset(&mask);
+  for (size_t i = 0; i < ARRAY_SIZE(changes); i++) {
+    sigaddset(&mask, changes[i].ident);
+  }
+  if (sigprocmask(SIG_BLOCK, &mask, NULL) != 0) {
+    ERRORN("sigprocmask");
+    return -1;
+  }
+
+  // We will receive EPIPE on the socket.
+  signal(SIGPIPE, SIG_IGN);
+
+  if (kevent(kq, changes, ARRAY_SIZE(changes), NULL, 0, NULL) != 0) {
+    ERRORN("kevent");
+    return -1;
+  }
+  return 0;
+}
+
+static int add_listen_fd(int kq, int fd)
+{
+  struct kevent changes[] = {
+    { .ident=fd, .filter=EVFILT_READ, .flags=EV_ADD },
+  };
+  if (kevent(kq, changes, ARRAY_SIZE(changes), NULL, 0, NULL) != 0) {
+    ERRORN("kevent");
+    return -1;
+  }
+  return 0;
+}
+
 static void on_accept(struct state *state, int accept_fd, interface_ref iface);
 
 int main(int argc, char *argv[]) {
   debug = getenv("DEBUG") != NULL;
   int rc = 1;
   int listen_fd = -1;
   int pidfile_fd = -1;
+  int kq = -1;
   __block interface_ref iface = NULL;
 
   struct state state = {0};
@@ -411,6 +452,18 @@ int main(int argc, char *argv[]) {
     WARN("Seems running with SETUID. This is insecure and highly discouraged: See README.md");
   }
 
+  kq = kqueue();
+  if (kq == -1) {
+    ERRORN("kqueue");
+    goto done;
+  }
+
+  // Setup signals beofre creating the pidfile so the pidfile to ensure removal
+  // when terminating by signal.
+  if (setup_signals(kq)) {
+    goto done;
+  }
+
   if (cliopt->pidfile != NULL) {
     pidfile_fd = create_pidfile(cliopt->pidfile);
     if (pidfile_fd == -1) {
@@ -426,16 +479,6 @@ int main(int argc, char *argv[]) {
     goto done;
   }
 
-  if (sigsetjmp(jmpbuf, 1) != 0) {
-    goto done;
-  }
-  signal(SIGHUP, signalhandler);
-  signal(SIGINT, signalhandler);
-  signal(SIGTERM, signalhandler);
-
-  // We will receive EPIPE on the socket.
-  signal(SIGPIPE, SIG_IGN);
-
   state.sem = dispatch_semaphore_create(1);
 
   // Queue for vm connections, allowing processing vms requests in parallel.
@@ -452,16 +495,34 @@ int main(int argc, char *argv[]) {
     goto done;
   }
 
+  if (add_listen_fd(kq, listen_fd)) {
+    goto done;
+  }
+
   while (1) {
-    int accept_fd = accept(listen_fd, NULL, NULL);
-    if (accept_fd < 0) {
-      ERRORN("accept");
+    struct kevent events[1];
+    int n = kevent(kq, NULL, 0, events, 1, NULL);
+    if (n < 0) {
+      ERRORN("kevent");
       goto done;
     }
-    struct state *state_p = &state;
-    dispatch_async(state.vms_queue, ^{
-      on_accept(state_p, accept_fd, iface);
-    });
+
+    if (events[0].filter == EVFILT_SIGNAL) {
+      INFOF("Received signal %s", strsignal(events[0].ident));
+      break;
+    }
+
+    if (events[0].filter == EVFILT_READ) {
+      int accept_fd = accept(listen_fd, NULL, NULL);
+      if (accept_fd < 0) {
+        ERRORN("accept");
+        goto done;
+      }
+      struct state *state_p = &state;
+      dispatch_async(state.vms_queue, ^{
+        on_accept(state_p, accept_fd, iface);
+      });
+    }
   }
   rc = 0;
 done:
@@ -480,6 +541,9 @@ int main(int argc, char *argv[]) {
     dispatch_release(state.vms_queue);
   if (state.host_queue != NULL)
     dispatch_release(state.host_queue);
+  if (kq != -1) {
+    close(kq);
+  }
   cli_options_destroy(cliopt);
   return rc;
 }