Use a socketpair for TriggerFd only on QNX

README.md

@@ -913,7 +913,7 @@ int value = resizable->VectorGet<int>(&vec_hdr, 0);  // Returns 42
 
 ### TriggerFd
 
-A file descriptor that can be used to trigger events. Implemented as an `eventfd` on Linux or a pipe on other systems.
+A file descriptor that can be used to trigger events. Implemented as an `eventfd` on Linux, a Unix domain `socketpair` on QNX, and a pipe on other POSIX systems, including macOS. QNX pipes go through a single-threaded pipe service, and a socketpair is slower than a pipe on macOS.
 
 **Header:** `toolbelt/triggerfd.h`
 

toolbelt/BUILD.bazel

@@ -81,6 +81,18 @@ cc_test(
         "@googletest//:gtest_main",
     ],
 )
+cc_test(
+    name = "triggerfd_test",
+    size = "small",
+    srcs = ["triggerfd_test.cc"],
+    deps = [
+        ":toolbelt",
+        "@abseil-cpp//absl/status",
+        "@abseil-cpp//absl/status:statusor",
+        "@googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "pipe_test",
     size = "medium",

toolbelt/triggerfd.cc

@@ -5,11 +5,18 @@
 #include "toolbelt/triggerfd.h"
 #include "absl/strings/str_format.h"
 
+// Linux has eventfd, which is the cheapest pollable trigger.  QNX pipes go
+// through a single-threaded pipe service, so QNX uses a Unix domain socketpair.
+// Every other POSIX system, including macOS, keeps the pipe.  A socketpair is
+// slower there, and its read end always reports POLLOUT.
+#include <unistd.h>
 #if defined(__linux__)
 #include <sys/eventfd.h>
+#elif defined(__QNX__) || defined(__QNXNTO__)
+#include <sys/socket.h>
+#include <sys/types.h>
 #else
 #include <fcntl.h>
-#include <unistd.h>
 #endif
 
 namespace toolbelt {
@@ -40,6 +47,21 @@ absl::Status TriggerFd::Open() {
   }
   poll_fd_.SetFd(fd);
   trigger_fd_ = poll_fd_;
+#elif defined(__QNX__) || defined(__QNXNTO__)
+  int fds[2];
+  if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == -1) {
+    return absl::InternalError(absl::StrFormat(
+        "Unable to create trigger fd socketpair: %s", strerror(errno)));
+  }
+  // Owned immediately so both ends are closed if setting nonblocking fails.
+  poll_fd_.SetFd(fds[0]);
+  trigger_fd_.SetFd(fds[1]);
+  if (absl::Status status = poll_fd_.SetNonBlocking(); !status.ok()) {
+    return status;
+  }
+  if (absl::Status status = trigger_fd_.SetNonBlocking(); !status.ok()) {
+    return status;
+  }
 #else
   int pipes[2];
   if (pipe(pipes) == -1) {
@@ -67,7 +89,7 @@ void TriggerFd::Trigger() {
   int64_t val = 1;
   (void)::write(trigger_fd_.Fd(), &val, 8);
 #else
-  // Pipe.  Write a single byte to it.
+  // Pipe, or a socketpair on QNX.  Write a single byte to it.
   char val = 'x';
   (void)::write(trigger_fd_.Fd(), &val, 1);
 #endif
@@ -81,8 +103,8 @@ bool TriggerFd::Clear() {
   ssize_t n = ::read(poll_fd_.Fd(), &val, 8);
   return (n > 0);
 #else
-  // Pipe.  Read all bytes from the pipe.  The pipe is nonblocking so we will
-  // get an EAGAIN when we've cleared it.
+  // Pipe, or a socketpair on QNX.  Read all pending bytes.  The fd is
+  // nonblocking, so we get EAGAIN once it has been drained.
   bool was_triggered = false;
   for (;;) {
     char buffer[256];

toolbelt/triggerfd.h

@@ -13,7 +13,8 @@ namespace toolbelt {
 
 // A TriggerFd is a file descriptor that can be used to trigger
 // an event.  If the facility is available, it is implemented
-// as an eventfd, otherwise it's a pipe.
+// as an eventfd on Linux, a Unix domain socketpair on QNX, and a pipe
+// everywhere else.
 class TriggerFd {
  public:
   TriggerFd() = default;

toolbelt/triggerfd_test.cc

@@ -0,0 +1,66 @@
+// Copyright 2023 David Allison
+// All Rights Reserved
+// See LICENSE file for licensing information.
+
+#include "toolbelt/triggerfd.h"
+#include <gtest/gtest.h>
+#include <poll.h>
+
+namespace {
+
+bool Readable(toolbelt::TriggerFd &t) {
+  struct pollfd pfd = t.GetPollFd().GetPollFd();
+  return ::poll(&pfd, 1, 0) == 1 && (pfd.revents & POLLIN) != 0;
+}
+
+TEST(TriggerFdTest, CreateAndClearUntriggered) {
+  auto t = toolbelt::TriggerFd::Create();
+  ASSERT_TRUE(t.ok());
+  ASSERT_TRUE(t->GetPollFd().Valid());
+  ASSERT_TRUE(t->GetTriggerFd().Valid());
+  ASSERT_FALSE(Readable(*t));
+  ASSERT_FALSE(t->Clear());
+}
+
+TEST(TriggerFdTest, TriggerThenClear) {
+  auto t = toolbelt::TriggerFd::Create();
+  ASSERT_TRUE(t.ok());
+  t->Trigger();
+  ASSERT_TRUE(Readable(*t));
+  ASSERT_TRUE(t->Clear());
+  ASSERT_FALSE(Readable(*t));
+  ASSERT_FALSE(t->Clear());
+}
+
+TEST(TriggerFdTest, MultipleTriggersClearedAtOnce) {
+  auto t = toolbelt::TriggerFd::Create();
+  ASSERT_TRUE(t.ok());
+  for (int i = 0; i < 10; i++) {
+    t->Trigger();
+  }
+  ASSERT_TRUE(t->Clear());
+  ASSERT_FALSE(Readable(*t));
+}
+
+TEST(TriggerFdTest, TriggerNeverBlocksWhenFull) {
+  auto t = toolbelt::TriggerFd::Create();
+  ASSERT_TRUE(t.ok());
+  // Far more triggers than any pipe/socket buffer can hold.  The trigger fd is
+  // nonblocking so this must not hang.
+  for (int i = 0; i < 1000000; i++) {
+    t->Trigger();
+  }
+  ASSERT_TRUE(t->Clear());
+  ASSERT_FALSE(Readable(*t));
+}
+
+TEST(TriggerFdTest, Move) {
+  auto t = toolbelt::TriggerFd::Create();
+  ASSERT_TRUE(t.ok());
+  toolbelt::TriggerFd moved = std::move(*t);
+  ASSERT_FALSE(t->GetPollFd().Valid());
+  moved.Trigger();
+  ASSERT_TRUE(moved.Clear());
+}
+
+} // namespace