Add framework for multiplexing IO on network sockets and other file descriptors

The design leans heavily on Boost.Asio, a battle-tested library that's widely
used and that forms the basis of upcoming C++ networking support.

Source/Common/Async.h

@@ -0,0 +1,374 @@
+// SPDX-License-Identifier: MIT
+/**
+ * Helper framework to enable asynchronous IO operations on file descriptor objects (networking, files).
+ *
+ * Strongly inspired by Boost.Asio.
+ */
+#pragma once
+
+#include <algorithm>
+#include <cassert>
+#include <chrono>
+#include <optional>
+#include <poll.h>
+#include <span>
+#include <utility>
+#include <vector>
+
+#include <FEXCore/fextl/functional.h>
+#include <FEXCore/fextl/map.h>
+#include <FEXCore/fextl/vector.h>
+
+namespace fasio {
+
+enum class error {
+  success,
+  timeout,      // User-specified timeout expired
+  eof,          // Permanently reached end of data stream (e.g. because socket connection was closed by peer)
+  invalid,      // Invalid input parameters
+  generic_errno // Read errno for details
+};
+
+/**
+ * This selects which action to trigger when returning from a reactor callback.
+ * The default (drop) will drop the callback so that the caller can register
+ * a new one.
+ */
+enum class post_callback {
+  drop,         // Drop the callback
+  repeat,       // Continue using the same callback
+  stop_reactor, // Triggers exit from run()
+};
+
+/**
+ * Core event loop for asynchronous code. Corresponds to asio::io_context,
+ * specialized for multiplexing file descriptors via ppoll().
+ *
+ * A reactor tracks a set of file descriptors and calls user-provided callbacks
+ * when they become ready. For example, the callback for a network socket will
+ * be called when data is ready to be reveived on the socket.
+ *
+ */
+struct poll_reactor {
+private:
+  std::vector<pollfd> PollFDs;
+
+  int AsyncStopRequest[2] = {-1, -1};
+
+public:
+  ~poll_reactor() {
+    if (AsyncStopRequest[0]) {
+      ::close(AsyncStopRequest[0]);
+      ::close(AsyncStopRequest[1]);
+    }
+  }
+
+  // Maps FD to callback
+  fextl::map<int, fextl::move_only_function<post_callback(error)>> read_callbacks;
+
+  struct Event {
+    pollfd FD;
+    bool Erase = false;
+    bool Insert = false;
+  };
+
+  std::vector<Event> QueuedEvents;
+
+  // Adds an internal FD to wake up and exit the reactor when stop_async() is called from any thread.
+  void enable_async_stop() {
+    ::pipe(AsyncStopRequest);
+    PollFDs.push_back(pollfd {.fd = AsyncStopRequest[0], .events = POLLHUP, .revents = 0});
+    read_callbacks[AsyncStopRequest[0]] = [](error) {
+      return post_callback::stop_reactor;
+    };
+  }
+
+  void stop_async() {
+    if (AsyncStopRequest[1] == -1) {
+      ERROR_AND_DIE_FMT("Tried to use stop_async without calling enable_async_stop during setup");
+    }
+    // Wake up run() thread by closing this pipe endpoint
+    close(AsyncStopRequest[1]);
+  }
+
+  error run(std::optional<std::chrono::nanoseconds> Timeout = std::nullopt) {
+    // Process events queued before entering wait loop
+    update_fd_list();
+
+    timespec ts = to_timespec(Timeout.value_or(std::chrono::nanoseconds {0}));
+
+    while (true) {
+      int Result = ppoll(PollFDs.data(), PollFDs.size(), Timeout ? &ts : nullptr, nullptr);
+
+      if (Result < 0) {
+        return error::generic_errno;
+      } else if (Result == 0) {
+        return error::timeout;
+      } else {
+        bool exit_requested = false;
+
+        // Walk the FDs and see if we got any results
+        for (auto& ActiveFD : PollFDs) {
+          if (ActiveFD.revents == 0) {
+            continue;
+          }
+          if (ActiveFD.revents & POLLIN) {
+            // NOTE: For sockets, this is triggered on close, too. Pipes only report POLLHUP, however.
+            auto Callback = std::move(read_callbacks[ActiveFD.fd]);
+            if (!Callback) {
+              ERROR_AND_DIE_FMT("Data available for reading on FD {} but no read callback registered", ActiveFD.fd);
+            }
+            auto Ret = Callback(error::success);
+            if (Ret == post_callback::repeat) {
+              read_callbacks[ActiveFD.fd] = std::move(Callback);
+            } else if (Ret == post_callback::stop_reactor) {
+              exit_requested = true;
+            }
+          }
+          if (ActiveFD.revents & (POLLHUP | POLLERR | POLLNVAL | POLLRDHUP)) {
+            auto Callback = std::move(read_callbacks[ActiveFD.fd]);
+            if (Callback) {
+              exit_requested |= (Callback(error::eof) == post_callback::stop_reactor);
+            }
+            // Error or hangup, close the socket and erase it from our list
+            QueuedEvents.push_back(Event {.FD = {.fd = ActiveFD.fd}, .Erase = true});
+          }
+
+          ActiveFD.revents = 0;
+          if (--Result == 0) {
+            break;
+          }
+        }
+
+        if (exit_requested) {
+          return error::success;
+        }
+
+        update_fd_list();
+      }
+    }
+  }
+
+private:
+  timespec to_timespec(std::chrono::nanoseconds Duration) {
+    timespec Timespec {};
+    auto Seconds = std::chrono::duration_cast<std::chrono::seconds>(Duration);
+    Timespec.tv_sec = Seconds.count();
+    Timespec.tv_nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(Duration - Seconds).count();
+    return Timespec;
+  }
+
+  void update_fd_list() {
+    for (auto& Event : QueuedEvents) {
+      if (Event.Erase) {
+        auto Index = std::find_if(PollFDs.begin(), PollFDs.end(), [&](pollfd& FD) { return FD.fd == Event.FD.fd; }) - PollFDs.begin();
+        if (Index == PollFDs.size()) {
+          ERROR_AND_DIE_FMT("bla");
+        }
+        close(Event.FD.fd);
+        PollFDs.erase(PollFDs.begin() + Index);
+        read_callbacks.erase(Event.FD.fd);
+      }
+
+      if (Event.Insert) {
+        PollFDs.push_back(Event.FD);
+      }
+    }
+    QueuedEvents.clear();
+  }
+};
+
+/**
+ * Corresponds to asio::mutable_buffer.
+ */
+struct mutable_buffer {
+  std::span<std::byte> Data;
+  mutable_buffer* Next = nullptr;
+
+  // Optional FD to send/receive via ancillary buffer.
+  // This may only be used with non-empty data, and there may only be up to one FD per buffer chain
+  std::optional<int*> FD;
+
+  size_t size() const {
+    size_t Ret = 0;
+    const mutable_buffer* Current = this;
+    do {
+      Ret += Current->Data.size_bytes();
+      Current = Current->Next;
+    } while (Current);
+
+    if (Ret == 0) {
+      assert(!FD);
+    }
+    return Ret;
+  }
+
+  int consume_fd() {
+    assert(FD);
+    return **std::exchange(FD, std::nullopt);
+  }
+
+  mutable_buffer& operator+=(size_t NumBytes) {
+    mutable_buffer* Current = this;
+    while (Current->Next && NumBytes >= Current->Data.size_bytes()) {
+      NumBytes -= Data.size_bytes();
+      Current = Current->Next;
+      assert(Current->FD == std::nullopt);
+    }
+    auto FD = std::exchange(this->FD, std::nullopt);
+    *this = *Current;
+    Data = Data.subspan(std::min(Data.size_bytes(), NumBytes));
+    this->FD = FD;
+    return *Current;
+  }
+
+  size_t count_chunks() const {
+    size_t Ret = 1;
+    const mutable_buffer* Current = this;
+    while (Current->Next) {
+      Current = Current->Next;
+      ++Ret;
+    }
+    return Ret;
+  }
+};
+
+inline mutable_buffer Chained(std::span<mutable_buffer> Buffers) {
+  for (size_t i = 0; i + 1 < Buffers.size(); ++i) {
+    Buffers[i].Next = &Buffers[i + 1];
+  }
+  return Buffers[0];
+}
+
+/**
+ * Corresponds to asio::dynamic_vector_buffer.
+ */
+struct dynamic_vector_buffer {
+  fextl::vector<std::byte>& Data;
+
+  // Maximum number of bytes to grow to
+  size_t max_size = Data.capacity();
+};
+
+/**
+ * Asynchronously reads data from the given stream until MatchPredicate reports a match. The read
+ * is queued to the stream's reactor and will progress whenever data is available.
+ *
+ * MatchPredicate must have the signature pair<Iter, bool>(Iter, Iter):
+ * - The input iterators provide the range of new data bytes
+ * - The returned boolean indicates if a match was found
+ * - The returned iterator is the match location or the location at which to continue testing after the next read
+ *
+ * The read data will be appended to Buffers. Data past the match returned from the last read data will also be included.
+ *
+ * Corresponds to asio::async_read_until.
+ */
+template<typename AsyncReadStream, typename MatchPredicate>
+void async_read_until(AsyncReadStream& Stream, dynamic_vector_buffer Buffers, MatchPredicate Predicate, auto token) {
+  struct Callback {
+    size_t BeginPos;
+    size_t EndPos;
+    AsyncReadStream& Stream;
+    dynamic_vector_buffer Buffers;
+    MatchPredicate Predicate;
+    decltype(token) Token;
+
+    void operator()(error Err, size_t BytesRead, std::optional<int> FD) {
+      if (Err != error::success) {
+        Token(Err, 0);
+        return;
+      }
+
+      // Start with the predicate check to avoid fetching data unnecessarily
+      EndPos += BytesRead;
+      if (EndPos != BeginPos) {
+        auto Begin = Buffers.Data.begin() + BeginPos;
+        auto End = Buffers.Data.begin() + EndPos;
+        auto [It, Found] = Predicate(Begin, End);
+        BeginPos = It - Buffers.Data.begin();
+        if (Found) {
+          Buffers.Data.resize(EndPos); // Shrink down to size of data actually received
+          Token(error::success, BeginPos);
+          return;
+        }
+      }
+
+      // Fill the entire remaining capacity, or resize for a minimum of 512 bytes
+      auto BytesToRead = std::max<size_t>(std::min(Buffers.Data.capacity(), Buffers.max_size) - EndPos, 512);
+      if (Buffers.Data.size() + BytesToRead > Buffers.max_size) {
+        ERROR_AND_DIE_FMT("Out of buffer space");
+      }
+
+      Buffers.Data.resize(EndPos + BytesToRead);
+
+      // Queue data read.
+      // On completion, Reader will check if enough data was received and will queue more reads if needed.
+      Stream.async_read_some(mutable_buffer {std::span {Buffers.Data}}, *this);
+    }
+  };
+
+  // Check existing data for a predicate match, then initiate async reading if necessary
+  Callback {0, Buffers.Data.size(), Stream, Buffers, std::move(Predicate), std::move(token)}(error::success, 0, std::nullopt);
+}
+
+using read_callback = fextl::move_only_function<void(error, size_t, std::optional<int>)>;
+
+/**
+ * Synchronously writes fixed-length data to the given Stream.
+ *
+ * The length is inferred from the size of the input buffer(s).
+ *
+ * Corresponds to asio::write.
+ */
+template<typename AsyncReadStream>
+std::size_t write(AsyncReadStream& Stream, mutable_buffer Buffers, error& ec) {
+  size_t TotalBytesWritten = 0;
+  while (Buffers.size() != 0 || Buffers.FD) {
+    auto BytesWritten = Stream.write_some(Buffers, ec);
+    TotalBytesWritten += BytesWritten;
+    if (Buffers.FD) {
+      (void)Buffers.consume_fd();
+    }
+    Buffers += BytesWritten;
+    if (ec != error::success) {
+      return TotalBytesWritten;
+    }
+  }
+  ec = error::success;
+  return TotalBytesWritten;
+}
+
+/**
+ * Non-owning wrapper around a file descriptor, which is registered to the
+ * reactor on construction.
+ *
+ * Corresponds to asio::posix::descriptor.
+ */
+struct posix_descriptor {
+  poll_reactor* Reactor = nullptr;
+  int FD = -1;
+
+  posix_descriptor(poll_reactor& Reactor, int FD)
+    : Reactor(&Reactor)
+    , FD(FD) {
+    Reactor.QueuedEvents.push_back(fasio::poll_reactor::Event {.FD =
+                                                                 pollfd {
+                                                                   .fd = FD,
+                                                                   .events = POLLIN,
+                                                                   .revents = 0,
+                                                                 },
+                                                               .Insert = true});
+  }
+
+  /**
+   * Wait until there is data available to read on this object, then execute the given callback
+   */
+  template<typename Fn>
+  requires std::is_invocable_r_v<post_callback, Fn, error>
+  void async_wait(Fn Callback) {
+    [[maybe_unused]] auto Previous = std::exchange(Reactor->read_callbacks[FD], std::move(Callback));
+    assert(!Previous && "May not queue multiple async operations");
+  }
+};
+
+} // namespace fasio

Source/Common/AsyncNet.h

@@ -0,0 +1,276 @@
+// SPDX-License-Identifier: MIT
+/**
+ * Socket wrappers for asynchronous programming with fasio
+ */
+#pragma once
+
+#include <Common/Async.h>
+
+#include <sys/socket.h>
+#include <sys/un.h>
+
+namespace fasio {
+
+// TODO: Move to main header?
+using read_callback = fextl::move_only_function<void(error, size_t, std::optional<int>)>;
+
+/**
+ * Non-owning wrapper around a socket, which is registered to the
+ * reactor on construction.
+ *
+ * Corresponds to asio::local::stream_protocol::socket.
+ */
+struct tcp_socket {
+  poll_reactor& Reactor;
+  int FD;
+
+  tcp_socket(poll_reactor& Reactor_, int FD_)
+    : Reactor(Reactor_)
+    , FD(FD_) {
+    Reactor.QueuedEvents.push_back(poll_reactor::Event {.FD =
+                                                          pollfd {
+                                                            .fd = FD,
+                                                            .events = POLLIN | POLLPRI | POLLRDHUP,
+                                                            .revents = 0,
+                                                          },
+                                                        .Insert = true});
+  }
+
+  /**
+   * Queues an asynchronous operation that will run the completion token once
+   * at least one byte of data was received
+   */
+  void async_read_some(mutable_buffer Buffers, read_callback token) {
+    auto Callback = [Buffers, Socket = FD, token = std::move(token)](error ec) {
+      if (ec != error::success) {
+        token(ec, 0, std::nullopt);
+        return post_callback::drop;
+      }
+
+      auto BytesRead = read_some_from_fd(Buffers, ec, Socket);
+      if (ec != error::success) {
+        token(ec, BytesRead, std::nullopt);
+      } else {
+        token(ec, BytesRead, Buffers.FD ? std::optional {**Buffers.FD} : std::nullopt);
+      }
+      return post_callback::drop;
+    };
+
+    [[maybe_unused]] auto Previous = std::exchange(Reactor.read_callbacks[FD], std::move(Callback));
+    assert(!Previous && "May not queue multiple async operations");
+  }
+
+  /**
+   * Blocks until at least one byte of data was received
+   */
+  size_t read_some(const mutable_buffer& Buffers, error& ec) {
+    return read_some_from_fd(Buffers, ec, FD);
+  }
+
+  /**
+   * Blocks until at least one byte of data was sent
+   */
+  size_t write_some(const mutable_buffer& Buffers, error& ec) {
+    auto iov = (iovec*)alloca(sizeof(mutable_buffer) * Buffers.count_chunks());
+    size_t NumIovs = 0;
+    for (auto Buffer = &Buffers; Buffer; Buffer = Buffer->Next) {
+      iov[NumIovs].iov_base = Buffer->Data.data();
+      iov[NumIovs].iov_len = Buffer->Data.size_bytes();
+      ++NumIovs;
+    }
+    msghdr msg {
+      .msg_name = nullptr,
+      .msg_namelen = 0,
+      .msg_iov = iov,
+      .msg_iovlen = NumIovs,
+    };
+
+    // Setup the ancillary buffer. This is where we will be getting pipe FDs
+    // We only need 4 bytes for the FD
+    constexpr size_t CMSG_SIZE = CMSG_SPACE(sizeof(int));
+    union AncillaryBuffer {
+      cmsghdr Header;
+      uint8_t Buffer[CMSG_SIZE];
+    };
+    AncillaryBuffer AncBuf {};
+
+    if (Buffers.FD) {
+      // Enable ancillary buffer
+      msg.msg_control = AncBuf.Buffer;
+      msg.msg_controllen = CMSG_SIZE;
+
+      // Now we need to setup the ancillary buffer data. We are only sending an FD
+      cmsghdr* cmsg = CMSG_FIRSTHDR(&msg);
+      cmsg->cmsg_len = CMSG_LEN(sizeof(int));
+      cmsg->cmsg_level = SOL_SOCKET;
+      cmsg->cmsg_type = SCM_RIGHTS;
+
+      // We are giving the daemon the write side of the pipe
+      memcpy(CMSG_DATA(cmsg), Buffers.FD.value(), sizeof(int));
+    }
+
+    auto Ret = sendmsg(FD, &msg, 0);
+    if (Ret < 0) {
+      ec = error::generic_errno;
+      return 0;
+    }
+    ec = error::success;
+    return Ret;
+  }
+
+private:
+  static size_t read_some_from_fd(const mutable_buffer& Buffers, error& ec, int FD) {
+    auto iov = (iovec*)alloca(sizeof(mutable_buffer) * Buffers.count_chunks());
+    size_t NumIovs = 0;
+    for (auto Buffer = &Buffers; Buffer; Buffer = Buffer->Next) {
+      iov[NumIovs].iov_base = Buffer->Data.data();
+      iov[NumIovs].iov_len = Buffer->Data.size_bytes();
+      ++NumIovs;
+    }
+    msghdr msg {
+      .msg_name = nullptr,
+      .msg_namelen = 0,
+      .msg_iov = iov,
+      .msg_iovlen = NumIovs,
+    };
+
+    // If requested, set up a 4-byte ancillary buffer for receiving a file descriptor
+    constexpr size_t CMSG_SIZE = CMSG_SPACE(sizeof(int));
+    union AncillaryBuffer {
+      cmsghdr Header;
+      uint8_t Buffer[CMSG_SIZE];
+    };
+    AncillaryBuffer AncBuf {};
+
+    if (Buffers.FD) {
+      // Enable ancillary buffer
+      msg.msg_control = AncBuf.Buffer;
+      msg.msg_controllen = CMSG_SIZE;
+    }
+
+    ssize_t BytesRead = recvmsg(FD, &msg, 0);
+    if (BytesRead < 0) {
+      if (errno != 0) {
+        ec = error::generic_errno;
+        return 0;
+      }
+    } else if (BytesRead == 0) {
+      ec = error::eof;
+      return 0;
+    }
+
+    if (Buffers.FD && msg.msg_controllen != CMSG_SIZE) {
+      ec = error::invalid;
+      return 0;
+    }
+
+    if (Buffers.FD) {
+      memcpy(*Buffers.FD, AncBuf.Buffer, sizeof(FD));
+    }
+
+    ec = error::success;
+    return BytesRead;
+  }
+};
+
+/**
+ * Owning wrapper around a server socket that listens for connections after
+ * creation. Clients can be accepted asynchronously using async_accept().
+ *
+ * Corresponds to asio::local::stream_protocol::acceptor.
+ */
+struct tcp_acceptor {
+  poll_reactor& Reactor;
+  int FD;
+
+  tcp_acceptor(tcp_acceptor&& other)
+    : Reactor(other.Reactor)
+    , FD(other.FD) {
+    other.FD = -1;
+  }
+
+  ~tcp_acceptor() {
+    if (FD != -1) {
+      close(FD);
+    }
+  }
+
+  tcp_acceptor& operator=(tcp_acceptor&& other) {
+    FD = std::exchange(other.FD, -1);
+    return *this;
+  }
+
+  static std::optional<tcp_acceptor> create(poll_reactor& Reactor, bool abstract, std::string_view Name, int MaxPending = SOMAXCONN) {
+    // Create the initial unix socket
+    int FD = socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0);
+    if (FD == -1) {
+      return {};
+    }
+
+    sockaddr_un addr {};
+    addr.sun_family = AF_UNIX;
+
+    if (Name.size() > sizeof(addr.sun_path) - 1) {
+      ERROR_AND_DIE_FMT("Invalid FEXServer socket name: {}", Name);
+    }
+
+    auto NameEnd = addr.sun_path;
+    if (!abstract) {
+      // sun_path is null-terminated
+      NameEnd = std::copy(Name.begin(), Name.end(), addr.sun_path);
+      *NameEnd++ = 0;
+    } else {
+      // Abstract AF_UNIX sockets start with \0 but aren't null-terminated
+      addr.sun_path[0] = 0;
+      NameEnd = std::copy(Name.begin(), Name.end(), addr.sun_path + 1);
+    }
+
+    // Bind the socket to the path
+    int Result = bind(FD, reinterpret_cast<sockaddr*>(&addr), sizeof(addr.sun_family) + (NameEnd - addr.sun_path));
+    if (Result == -1) {
+      close(FD);
+      return {};
+    }
+
+    Result = listen(FD, MaxPending);
+    if (Result == -1) {
+      close(FD);
+      return {};
+    }
+
+    Reactor.QueuedEvents.push_back(poll_reactor::Event {.FD =
+                                                          {
+                                                            .fd = FD,
+                                                            .events = POLLIN,
+                                                            .revents = 0,
+                                                          },
+                                                        .Insert = true});
+    return tcp_acceptor(Reactor, FD);
+  }
+
+  void async_accept(fextl::move_only_function<post_callback(error, std::optional<tcp_socket>)> OnAccept) {
+    Reactor.read_callbacks[FD] = [ServerFD = FD, &Reactor = Reactor, OnAccept = std::move(OnAccept)](error ec) mutable {
+      if (ec != error::success) {
+        return post_callback::drop;
+      }
+
+      sockaddr_storage Addr {};
+      socklen_t AddrSize {};
+      int NewFD = accept(ServerFD, reinterpret_cast<sockaddr*>(&Addr), &AddrSize);
+      if (NewFD < 0) {
+        return OnAccept(error::generic_errno, std::nullopt);
+      }
+
+      return OnAccept(error::success, tcp_socket {Reactor, NewFD});
+    };
+  }
+
+private:
+  tcp_acceptor(poll_reactor& Reactor_, int FD_)
+    : Reactor(Reactor_)
+    , FD(FD_) {}
+};
+static_assert(!std::is_copy_constructible_v<tcp_acceptor>);
+static_assert(!std::is_copy_assignable_v<tcp_acceptor>);
+
+} // namespace fasio