From 22cf5eb9890983da830865d689ac730f4fa9afe3 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Tue, 5 Aug 2025 19:49:11 +0200 Subject: [PATCH] libstore: use async io for remote store IO this means both the worker protocol and the serve protocol, i.e. ssh-ng/local connections and legacy ssh connections. now we have no blocking reads left anywhere in our client store connections. Change-Id: I2f628d4d2e71ef0a7006918f175192f3f58eea95 --- lix/libstore/legacy-ssh-store.cc | 13 +++---- lix/libstore/remote-store-connection.hh | 17 ++++----- lix/libstore/remote-store.cc | 49 +++++++++++++------------ 3 files changed, 39 insertions(+), 40 deletions(-) diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc index 2d05d8dd7..db98b026e 100644 --- a/lix/libstore/legacy-ssh-store.cc +++ b/lix/libstore/legacy-ssh-store.cc @@ -170,8 +170,9 @@ struct LegacySSHStore final : public Store } }); + AsyncFdIoStream stream(AsyncFdIoStream::shared_fd{}, sshConn->socket.get()); + { - AsyncFdIoStream stream(AsyncFdIoStream::shared_fd{}, sshConn->socket.get()); StringSink buffer; // can't use TRY_AWAIT here because macros break with variadic templates. sigh. ((co_await sendArg(stream, buffer, std::forward(args))).value(), ...); @@ -182,12 +183,10 @@ struct LegacySSHStore final : public Store invalidateOnCancel.cancel(); co_return result::success(); } else { - // NOTE while no async streams are using the fd it is fully synchronous. - // we need either sync sources or async sources, and async sources would - // require a *lot* of code duplication. response messages are mostly not - // large enough to block us for long, so we just accept the hit for now. - FdSource from{sshConn->socket.get(), fromBuf}; - auto result = ServeProto::Serialise::read({from, *store, remoteVersion}); + AsyncBufferedInputStream from{stream, fromBuf}; + auto result = TRY_AWAIT(ServeProto::readAsync( + from, *store, remoteVersion, ServeProto::Serialise::read + )); invalidateOnCancel.cancel(); co_return result; } diff --git a/lix/libstore/remote-store-connection.hh b/lix/libstore/remote-store-connection.hh index 64bccde6d..6ec14eda5 100644 --- a/lix/libstore/remote-store-connection.hh +++ b/lix/libstore/remote-store-connection.hh @@ -10,6 +10,7 @@ #include "lix/libutil/io-buffer.hh" #include "lix/libutil/pool.hh" #include "lix/libutil/result.hh" +#include "lix/libutil/serialise-async.hh" #include "lix/libutil/serialise.hh" #include "lix/libutil/signals.hh" #include @@ -143,13 +144,14 @@ struct RemoteStore::ConnectionHandle } }); + AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()}; + // if the last argument can be serialized normally we will serialize *all* // arguments at once and hand off to the remote. if the last argument does // not have a serializer we assume it's a callback for a subframe protocol // and serialize all *preceding* arguments normally before handing over to // the subframing layer (which is then responsible for any error handling) if constexpr (requires(StringSink s) { s << std::declval(); }) { - AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()}; try { StringSink msg; ((msg << std::forward(args)), ...); @@ -163,7 +165,6 @@ struct RemoteStore::ConnectionHandle using ImmediateArgsIdxs = std::make_index_sequence; AllArgsT allArgs(std::forward(args)...); - AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()}; try { StringSink msg; [&](std::integer_sequence) { @@ -186,14 +187,10 @@ struct RemoteStore::ConnectionHandle co_return result::success(); } else { try { - // NOTE while no async streams are using the fd it is fully synchronous. - // we need either sync sources or async sources, and async sources would - // require a *lot* of code duplication. response messages are mostly not - // large enough to block us for long, so we just accept the hit for now. - FdSource from{handle->getFD(), handle->fromBuf}; - from.specialEndOfFileError = "Nix daemon disconnected while reading a response"; - auto result = - WorkerProto::Serialise::read({from, *handle->store, handle->daemonVersion}); + AsyncBufferedInputStream from{stream, handle->fromBuf}; + auto result = LIX_TRY_AWAIT(WorkerProto::readAsync( + from, *handle->store, handle->daemonVersion, WorkerProto::Serialise::read + )); invalidateOnCancel.cancel(); co_return result; } catch (...) { diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index a18a5720c..8362df1c3 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -5,6 +5,7 @@ #include "lix/libutil/error.hh" #include "lix/libutil/file-descriptor.hh" #include "lix/libutil/result.hh" +#include "lix/libutil/serialise-async.hh" #include "lix/libutil/serialise.hh" #include "lix/libutil/signals.hh" #include "lix/libstore/path-with-outputs.hh" @@ -83,41 +84,43 @@ try { kj::Promise> RemoteStore::initConnection(Connection & conn) try { /* Send the magic greeting, check for the reply. */ - // NOTE: this is synchronous until we call processStderr. this is intentional; - // reading the response would be synchronous anyway, and making sending of the - // greeting synchronous also makes this code path significantly more readable. try { - conn.store = this; - FdSource from{conn.getFD(), conn.fromBuf}; - from.specialEndOfFileError = - "Nix daemon connection broke during setup phase (is it reachable?)"; - FdSink to{conn.getFD()}; - to << WORKER_MAGIC_1; - to.flush(); + AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, conn.getFD()}; + AsyncBufferedInputStream from{stream, conn.fromBuf}; - uint64_t magic = readNum(from); + conn.store = this; + { + StringSink packet; + packet << WORKER_MAGIC_1; + TRY_AWAIT(stream.writeFull(packet.s.data(), packet.s.size())); + } + + uint64_t magic = TRY_AWAIT(readNum(from)); if (magic != WORKER_MAGIC_2) throw Error("protocol mismatch"); - conn.daemonVersion = readNum(from); + conn.daemonVersion = TRY_AWAIT(readNum(from)); if (GET_PROTOCOL_MAJOR(conn.daemonVersion) != GET_PROTOCOL_MAJOR(PROTOCOL_VERSION)) throw Error("Nix daemon protocol version not supported"); if (GET_PROTOCOL_MINOR(conn.daemonVersion) < MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION) throw Error("The remote Nix daemon version is too old"); - to << PROTOCOL_VERSION; - // Obsolete CPU affinity. - to << 0; + { + StringSink packet; + packet << PROTOCOL_VERSION; + packet << 0; // Obsolete CPU affinity. + packet << false; // obsolete reserveSpace + TRY_AWAIT(stream.writeFull(packet.s.data(), packet.s.size())); + } - to << false; // obsolete reserveSpace + conn.daemonNixVersion = TRY_AWAIT(readString(from)); + conn.remoteTrustsUs = TRY_AWAIT(WorkerProto::readAsync( + from, + *conn.store, + conn.daemonVersion, + WorkerProto::Serialise>::read + )); - to.flush(); - conn.daemonNixVersion = readString(from); - conn.remoteTrustsUs = WorkerProto::Serialise>::read( - {from, *conn.store, conn.daemonVersion} - ); - - AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, conn.getFD()}; auto ex = TRY_AWAIT(conn.processStderr(stream)); if (ex.e) { std::rethrow_exception(ex.e);