Add fd_event_signaller_t

fd_event_signaller_t exists to expose eventfd under Linux. This is a
more lightweight way of signalling events than using a pipe.

cmake/ConfigureChecks.cmake

@@ -107,6 +107,7 @@ check_include_file_cxx(sys/select.h HAVE_SYS_SELECT_H)
 check_include_files("sys/types.h;sys/sysctl.h" HAVE_SYS_SYSCTL_H)
 check_include_file_cxx(termios.h HAVE_TERMIOS_H) # Needed for TIOCGWINSZ
 
+check_cxx_symbol_exists(eventfd sys/eventfd.h HAVE_EVENTFD)
 check_cxx_symbol_exists(pipe2 unistd.h HAVE_PIPE2)
 check_cxx_symbol_exists(wcscasecmp wchar.h HAVE_WCSCASECMP)
 check_cxx_symbol_exists(wcsdup wchar.h HAVE_WCSDUP)

config_cmake.h.in

@@ -58,6 +58,9 @@
 /* Define to 1 if you have the <ncurses/term.h> header file. */
 #cmakedefine HAVE_NCURSES_TERM_H 1
 
+/* Define to 1 if you have the 'eventfd' function. */
+#cmakedefine HAVE_EVENTFD 1
+
 /* Define to 1 if you have the 'pipe2' function. */
 #cmakedefine HAVE_PIPE2 1
 

src/fds.cpp

@@ -14,6 +14,10 @@
 
 #include <fcntl.h>
 
+#ifdef HAVE_EVENTFD
+#include <sys/eventfd.h>
+#endif
+
 #if defined(__linux__)
 #include <sys/statfs.h>
 #endif
@@ -28,6 +32,83 @@ void autoclose_fd_t::close() {
     fd_ = -1;
 }
 
+#ifdef HAVE_EVENTFD
+// Note we do not want to use EFD_SEMAPHORE because we are binary (not counting) semaphore.
+fd_event_signaller_t::fd_event_signaller_t() {
+    int fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
+    if (fd < 0) {
+        wperror(L"eventfd");
+        exit_without_destructors(1);
+    }
+    fd_.reset(fd);
+};
+
+int fd_event_signaller_t::write_fd() const { return fd_.fd(); }
+
+#else
+// Implementation using pipes.
+fd_event_signaller_t::fd_event_signaller_t() {
+    auto pipes = make_autoclose_pipes();
+    if (!pipes) {
+        wperror(L"pipe");
+        exit_without_destructors(1);
+    }
+    DIE_ON_FAILURE(make_fd_nonblocking(pipes->read.fd()));
+    DIE_ON_FAILURE(make_fd_nonblocking(pipes->write.fd()));
+    fd_ = std::move(pipes->read);
+    write_ = std::move(pipes->write);
+}
+
+int fd_event_signaller_t::write_fd() const { return write_.fd(); }
+#endif
+
+bool fd_event_signaller_t::try_consume() {
+    // If we are using eventfd, we want to read a single uint64.
+    // If we are using pipes, read a lot; note this may leave data on the pipe if post has been
+    // called many more times. In no case do we care about the data which is read.
+#ifdef HAVE_EVENTFD
+    uint64_t buff[1];
+#else
+    uint8_t buff[1024];
+#endif
+    ssize_t ret;
+    do {
+        ret = read(read_fd(), buff, sizeof buff);
+    } while (ret < 0 && errno == EINTR);
+    if (ret < 0 && errno != EAGAIN) {
+        wperror(L"read");
+    }
+    return ret > 0;
+}
+
+void fd_event_signaller_t::post() {
+    // eventfd writes uint64; pipes write 1 byte.
+#ifdef HAVE_EVENTFD
+    const uint64_t c = 1;
+#else
+    const uint8_t c = 1;
+#endif
+    ssize_t ret;
+    do {
+        ret = write(write_fd(), &c, sizeof c);
+    } while (ret < 0 && errno == EINTR);
+    // EAGAIN occurs if either the pipe buffer is full or the eventfd overflows (very unlikely).
+    if (ret < 0 && errno != EAGAIN) {
+        wperror(L"write");
+    }
+}
+
+bool fd_event_signaller_t::poll(bool wait) const {
+    struct timeval timeout = {0, 0};
+    fd_set fds;
+    FD_ZERO(&fds);
+    FD_SET(read_fd(), &fds);
+    int res = select(read_fd() + 1, &fds, nullptr, nullptr, wait ? nullptr : &timeout);
+    return res > 0;
+}
+
+fd_event_signaller_t::~fd_event_signaller_t() = default;
+
 /// If the given fd is in the "user range", move it to a new fd in the "high range".
 /// zsh calls this movefd().
 /// \p input_has_cloexec describes whether the input has CLOEXEC already set, so we can avoid

src/fds.h

@@ -3,9 +3,12 @@
 #ifndef FISH_FDS_H
 #define FISH_FDS_H
 
+#include "config.h"  // IWYU pragma: keep
+
+#include <sys/types.h>
+
 #include <algorithm>
 #include <string>
-#include <sys/types.h>
 #include <vector>
 
 #include "maybe.h"
@@ -77,6 +80,49 @@ struct autoclose_pipes_t {
 /// \return pipes on success, none() on error.
 maybe_t<autoclose_pipes_t> make_autoclose_pipes();
 
+/// An event signaller implemented using a file descriptor, so it can plug into select().
+/// This is like a binary semaphore. A call to post() will signal an event, making the fd readable.
+/// Multiple calls to post() may be coalesced. On Linux this uses eventfd(); on other systems this
+/// uses a pipe. try_consume() may be used to consume the event. Importantly this is async signal
+/// safe. Of course it is CLO_EXEC as well.
+class fd_event_signaller_t {
+   public:
+    /// \return the fd to read from, for notification.
+    int read_fd() const { return fd_.fd(); }
+
+    /// If an event is signalled, consume it; otherwise return.
+    /// This does not block.
+    /// This retries on EINTR.
+    bool try_consume();
+
+    /// Mark that an event has been received. This may be coalesced.
+    /// This retries on EINTR.
+    void post();
+
+    /// Perform a poll to see if an event is received.
+    /// If \p wait is set, wait until it is readable; this does not consume the event
+    /// but guarantees that the next call to wait() will not block.
+    /// \return true if readable, false if not readable, or not interrupted by a signal.
+    bool poll(bool wait = false) const;
+
+    // The default constructor will abort on failure (fd exhaustion).
+    // This should only be used during startup.
+    fd_event_signaller_t();
+    ~fd_event_signaller_t();
+
+   private:
+    // \return the fd to write to.
+    int write_fd() const;
+
+    // Always the read end of the fd; maybe the write end as well.
+    autoclose_fd_t fd_;
+
+#ifndef HAVE_EVENTFD
+    // If using a pipe, then this is its write end.
+    autoclose_fd_t write_;
+#endif
+};
+
 /// Sets CLO_EXEC on a given fd according to the value of \p should_set.
 int set_cloexec(int fd, bool should_set = true);
 

src/fish_tests.cpp

@@ -6115,6 +6115,31 @@ static void test_pipes() {
     }
 }
 
+static void test_fd_event_signaller() {
+    say(L"Testing fd event signaller");
+    fd_event_signaller_t sema;
+    do_test(!sema.try_consume());
+    do_test(!sema.poll());
+
+    // Post once.
+    sema.post();
+    do_test(sema.poll());
+    do_test(sema.poll());
+    do_test(sema.try_consume());
+    do_test(!sema.poll());
+    do_test(!sema.try_consume());
+
+    // Posts are coalesced.
+    sema.post();
+    sema.post();
+    sema.post();
+    do_test(sema.poll());
+    do_test(sema.poll());
+    do_test(sema.try_consume());
+    do_test(!sema.poll());
+    do_test(!sema.try_consume());
+}
+
 static void test_timer_format() {
     say(L"Testing timer format");
     // This test uses numeric output, so we need to set the locale.
@@ -6351,6 +6376,7 @@ int main(int argc, char **argv) {
     if (should_test_function("topics")) test_topic_monitor();
     if (should_test_function("topics")) test_topic_monitor_torture();
     if (should_test_function("pipes")) test_pipes();
+    if (should_test_function("fd_event")) test_fd_event_signaller();
     if (should_test_function("timer_format")) test_timer_format();
     // history_tests_t::test_history_speed();