Merge pull request #887 from capnproto/fix-rt-signals-and-more

Fix handling of queued RT signals, plus some other crap

c++/src/capnp/rpc-twoparty.c++

@@ -382,7 +382,9 @@ TwoPartyClient::TwoPartyClient(kj::AsyncCapabilityStream& connection, uint maxFd
       rpcSystem(network, bootstrapInterface) {}
 
 Capability::Client TwoPartyClient::bootstrap() {
-  MallocMessageBuilder message(4);
+  capnp::word scratch[4];
+  memset(&scratch, 0, sizeof(scratch));
+  capnp::MallocMessageBuilder message(scratch);
   auto vatId = message.getRoot<rpc::twoparty::VatId>();
   vatId.setSide(network.getSide() == rpc::twoparty::Side::CLIENT
                 ? rpc::twoparty::Side::SERVER

c++/src/kj/async-io-unix.c++

@@ -405,6 +405,12 @@ private:
               }
             }
           }
+
+          if (spaceLeft >= CMSG_LEN(0) && spaceLeft >= cmsg->cmsg_len) {
+            spaceLeft -= cmsg->cmsg_len;
+          } else {
+            spaceLeft = 0;
+          }
         }
 
 #ifndef MSG_CMSG_CLOEXEC

c++/src/kj/async-unix-test.c++

@@ -62,6 +62,10 @@ void captureSignals() {
     UnixEventPort::captureSignal(SIGURG);
     UnixEventPort::captureSignal(SIGIO);
 
+#ifdef SIGRTMIN
+    UnixEventPort::captureSignal(SIGRTMIN);
+#endif
+
     UnixEventPort::captureChildExit();
   }
 }
@@ -878,6 +882,50 @@ KJ_TEST("UnixEventPort poll for signals") {
   promise2.wait(waitScope);
 }
 
+#if defined(SIGRTMIN) && !__CYGWIN__
+// TODO(someday): Figure out why RT signals don't seem to work correctly on Cygwin. It looks like
+//   only the first signal is delivered, like how non-RT signals work. Is it possible Cygwin
+//   advertites RT signal support but doesn't actually implement them correctly? I can't find any
+//   information on the internet about this and TBH I don't care about Cygwin enough to dig in.
+
+void testRtSignals(UnixEventPort& port, WaitScope& waitScope, bool doPoll) {
+  union sigval value;
+  memset(&value, 0, sizeof(value));
+
+  // Queue three copies of the signal upfront.
+  for (uint i = 0; i < 3; i++) {
+    value.sival_int = 123 + i;
+    KJ_SYSCALL(sigqueue(getpid(), SIGRTMIN, value));
+  }
+
+  // Now wait for them.
+  for (uint i = 0; i < 3; i++) {
+    auto promise = port.onSignal(SIGRTMIN);
+    if (doPoll) {
+      KJ_ASSERT(promise.poll(waitScope));
+    }
+    auto info = promise.wait(waitScope);
+    KJ_EXPECT(info.si_value.sival_int == 123 + i);
+  }
+
+  KJ_EXPECT(!port.onSignal(SIGRTMIN).poll(waitScope));
+}
+
+KJ_TEST("UnixEventPort can receive multiple queued instances of an RT signal") {
+  captureSignals();
+  UnixEventPort port;
+  EventLoop loop(port);
+  WaitScope waitScope(loop);
+
+  testRtSignals(port, waitScope, true);
+
+  // Test again, but don't poll() the promises. This may test a different code path, if poll() and
+  // wait() are very different in how they read signals. (For the poll(2)-based implementation of
+  // UnixEventPort, they are indeed pretty different.)
+  testRtSignals(port, waitScope, false);
+}
+#endif
+
 }  // namespace
 }  // namespace kj
 

c++/src/kj/async-unix.c++

@@ -601,6 +601,19 @@ bool UnixEventPort::doEpollWait(int timeout) {
         KJ_ASSERT(n == sizeof(siginfo));
 
         gotSignal(toRegularSiginfo(siginfo));
+
+#ifdef SIGRTMIN
+        if (siginfo.ssi_signo >= SIGRTMIN) {
+          // This is an RT signal. There could be multiple copies queued. We need to remove it from
+          // the signalfd's signal mask before we continue, to avoid accidentally reading and
+          // discarding the extra copies.
+          // TODO(perf): If high throughput of RT signals is desired then perhaps we should read
+          //   them all into userspace and queue them here. Maybe we even need a better interface
+          //   than onSignal() for receiving high-volume RT signals.
+          KJ_SYSCALL(sigdelset(&signalFdSigset, siginfo.ssi_signo));
+          KJ_SYSCALL(signalfd(signalFd, &signalFdSigset, SFD_NONBLOCK | SFD_CLOEXEC));
+        }
+#endif
       }
     } else if (events[i].data.u64 == 1) {
       // Someone called wake() from another thread. Consume the event.

c++/src/kj/compat/http-test.c++

@@ -126,6 +126,7 @@ KJ_TEST("HttpHeaders::parseRequest") {
       "Content-Length: 123\r\n"
       "DATE:     early\r\n"
       "other-Header: yep\r\n"
+      "with.dots: sure\r\n"
       "\r\n");
   auto result = headers.tryParseRequest(text.asArray()).get<HttpHeaders::Request>();
 
@@ -143,12 +144,13 @@ KJ_TEST("HttpHeaders::parseRequest") {
   headers.forEach([&](kj::StringPtr name, kj::StringPtr value) {
     KJ_EXPECT(unpackedHeaders.insert(std::make_pair(name, value)).second);
   });
-  KJ_EXPECT(unpackedHeaders.size() == 5);
+  KJ_EXPECT(unpackedHeaders.size() == 6);
   KJ_EXPECT(unpackedHeaders["Content-Length"] == "123");
   KJ_EXPECT(unpackedHeaders["Host"] == "example.com");
   KJ_EXPECT(unpackedHeaders["Date"] == "early");
   KJ_EXPECT(unpackedHeaders["Foo-Bar"] == "Baz");
   KJ_EXPECT(unpackedHeaders["other-Header"] == "yep");
+  KJ_EXPECT(unpackedHeaders["with.dots"] == "sure");
 
   KJ_EXPECT(headers.serializeRequest(result.method, result.url) ==
       "POST /some/path HTTP/1.1\r\n"
@@ -157,6 +159,7 @@ KJ_TEST("HttpHeaders::parseRequest") {
       "Date: early\r\n"
       "Foo-Bar: Baz\r\n"
       "other-Header: yep\r\n"
+      "with.dots: sure\r\n"
       "\r\n");
 }