diff --git a/Source/Common/Async.h b/Source/Common/Async.h new file mode 100644 index 000000000..7035113ec --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +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 PollFDs; + + int AsyncStopRequest[2] = {-1, -1}; + +public: + ~poll_reactor() { + if (AsyncStopRequest[0]) { + ::close(AsyncStopRequest[0]); + ::close(AsyncStopRequest[1]); + } + } + + // Maps FD to callback + fextl::map> read_callbacks; + + struct Event { + pollfd FD; + bool Erase = false; + bool Insert = false; + }; + + std::vector 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 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(Duration); + Timespec.tv_sec = Seconds.count(); + Timespec.tv_nsec = std::chrono::duration_cast(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 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 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 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& 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, 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 +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 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(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)>; + +/** + * 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 +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 + requires std::is_invocable_r_v + 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 diff --git a/Source/Common/AsyncNet.h b/Source/Common/AsyncNet.h new file mode 100644 index 000000000..a5098c774 --- /dev/null +++ b/Source/Common/AsyncNet.h @@ -0,0 +1,276 @@ +// SPDX-License-Identifier: MIT +/** + * Socket wrappers for asynchronous programming with fasio + */ +#pragma once + +#include + +#include +#include + +namespace fasio { + +// TODO: Move to main header? +using read_callback = fextl::move_only_function)>; + +/** + * 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 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(&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)> 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(&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); +static_assert(!std::is_copy_assignable_v); + +} // namespace fasio