From b0edb262b2bfac5d0aab86a4217cff722e492b65 Mon Sep 17 00:00:00 2001
From: eldritch horrors <pennae@lix.systems>
Date: Mon, 16 Jun 2025 18:51:59 +0200
Subject: [PATCH] libutil: add bidirectional async fd streams

this may as well be called AsyncSocketStream since that will be what we
use it for, but hopefully it will not exist for long enough to need any
other socket functions to actually justify such highly specific naming.

Change-Id: Icf2fe88cf345405218e4b1bd440267e7f132f5c7
---
 lix/libutil/async-io.cc | 57 +++++++++++++++++++++++++++++++++++++++++
 lix/libutil/async-io.hh | 30 ++++++++++++++++++++++
 2 files changed, 87 insertions(+)

diff --git a/lix/libutil/async-io.cc b/lix/libutil/async-io.cc
index 22d943437..7488c3a62 100644
--- a/lix/libutil/async-io.cc
+++ b/lix/libutil/async-io.cc
@@ -1,4 +1,11 @@
 #include "async-io.hh"
+#include "async.hh"
+#include "error.hh"
+#include "file-descriptor.hh"
+#include "result.hh"
+#include <cerrno>
+#include <exception>
+#include <fcntl.h>
 
 namespace nix {
 kj::Promise<Result<void>> AsyncInputStream::drainInto(Sink & sink)
@@ -133,4 +140,54 @@ try {
 } catch (...) {
     co_return result::current_exception();
 }
+
+AsyncFdIoStream::AsyncFdIoStream(AutoCloseFD fd) : AsyncFdIoStream(shared_fd{}, fd.get())
+{
+    ownedFd = std::move(fd);
+}
+
+AsyncFdIoStream::AsyncFdIoStream(shared_fd, int fd)
+    : fd(fd)
+    , observer(AIO().unixEventPort, fd, kj::UnixEventPort::FdObserver::OBSERVE_READ_WRITE)
+{
+    oldFlags = fcntl(fd, F_GETFL, 0);
+    if (oldFlags == -1 || fcntl(fd, F_SETFL, oldFlags | O_NONBLOCK)) {
+        throw SysError("making file descriptor non-blocking");
+    }
+}
+
+AsyncFdIoStream::~AsyncFdIoStream() noexcept(false)
+{
+    if (fcntl(fd, F_SETFL, oldFlags)) {
+        try {
+            throw SysError("restoring file descriptor flags");
+        } catch (...) {
+            ignoreExceptionInDestructor();
+        }
+    }
+}
+
+kj::Promise<Result<size_t>> AsyncFdIoStream::read(void * tgt, size_t size)
+{
+    auto got = ::read(fd, tgt, size);
+    if (got >= 0) {
+        return {result::success(got)};
+    } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
+        return observer.whenBecomesReadable().then([=, this] { return read(tgt, size); });
+    } else {
+        return {result::failure(std::make_exception_ptr(SysError(errno, "read failed")))};
+    }
+}
+
+kj::Promise<Result<size_t>> AsyncFdIoStream::write(const void * src, size_t size)
+{
+    auto got = ::write(fd, src, size);
+    if (got >= 0) {
+        return {result::success(got)};
+    } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
+        return observer.whenBecomesWritable().then([=, this] { return write(src, size); });
+    } else {
+        return {result::failure(std::make_exception_ptr(SysError(errno, "write failed")))};
+    }
+}
 }
diff --git a/lix/libutil/async-io.hh b/lix/libutil/async-io.hh
index d71b043b0..680307ca5 100644
--- a/lix/libutil/async-io.hh
+++ b/lix/libutil/async-io.hh
@@ -3,10 +3,13 @@
 
 #include "lix/libutil/async.hh"
 #include "lix/libutil/box_ptr.hh"
+#include "lix/libutil/file-descriptor.hh"
 #include "lix/libutil/io-buffer.hh"
 #include "lix/libutil/ref.hh"
 #include "lix/libutil/result.hh"
 #include "lix/libutil/serialise.hh"
+#include <kj/async-io.h>
+#include <kj/async-unix.h>
 #include <kj/async.h>
 #include <kj/common.h>
 #include <memory>
@@ -152,4 +155,31 @@ public:
     kj::Promise<Result<size_t>> write(const void * src, size_t size) override;
     kj::Promise<Result<void>> flush();
 };
+
+class AsyncStream : public AsyncInputStream, public AsyncOutputStream
+{};
+
+class AsyncFdIoStream : public AsyncStream
+{
+    int fd, oldFlags;
+    AutoCloseFD ownedFd; // only for closing automatically, must equal fd if set
+    kj::UnixEventPort::FdObserver observer;
+
+public:
+    struct shared_fd
+    {};
+
+    explicit AsyncFdIoStream(AutoCloseFD fd);
+    AsyncFdIoStream(shared_fd, int fd);
+
+    ~AsyncFdIoStream() noexcept(false);
+
+    int getFD() const
+    {
+        return fd;
+    }
+
+    kj::Promise<Result<size_t>> read(void * tgt, size_t size) override;
+    kj::Promise<Result<size_t>> write(const void * src, size_t size) override;
+};
 }