From 3f6290531239530115be2eecfa1c5504a8d2add8 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Mon, 16 Jun 2025 18:51:59 +0200 Subject: [PATCH] libstore: instantiate RemoteStore FdSources as needed in the future we will want to instantiate either a sink, a source, both, or streams, depending on how the fd is used. to do this we need to share read buffers among sync and async readers. removing the FdSource we kept in the connection also helps prove that we always use this buffer for io Change-Id: Ib678e128ed6c4a07d6ce5ec1d3cde9eb3f5fc4ca --- lix/libstore/remote-store-connection.hh | 37 +++++------- lix/libstore/remote-store.cc | 79 +++++++++++++++---------- lix/libstore/ssh-store.cc | 7 ++- lix/libstore/uds-remote-store.cc | 3 - lix/libstore/uds-remote-store.hh | 5 ++ lix/nix/daemon.cc | 4 +- 6 files changed, 77 insertions(+), 58 deletions(-) diff --git a/lix/libstore/remote-store-connection.hh b/lix/libstore/remote-store-connection.hh index 867f57e47..3ff7bd620 100644 --- a/lix/libstore/remote-store-connection.hh +++ b/lix/libstore/remote-store-connection.hh @@ -6,6 +6,7 @@ #include "lix/libstore/worker-protocol-impl.hh" #include "lix/libutil/async.hh" #include "lix/libutil/file-descriptor.hh" +#include "lix/libutil/io-buffer.hh" #include "lix/libutil/pool.hh" #include "lix/libutil/result.hh" #include "lix/libutil/serialise.hh" @@ -20,20 +21,23 @@ namespace nix { * Bidirectional connection (send and receive) used by the Remote Store * implementation. * - * Contains `Source` and `Sink` for actual communication, along with + * Contains a socket fd and IO buffer for actual communication, along with * other information learned when negotiating the connection. */ struct RemoteStore::Connection { /** - * Send with this. + * Receive buffer, shared between sync Sources and async Streams. + * All buffered receiving sources or streams using the `getFD()` + * fd must use this buffer, or they will corrupt the connection. */ - int toFD; + ref fromBuf{make_ref()}; /** - * Receive with this. + * Returns the file descriptors socket backing this connection. A + * connection must be backed by a socket, not by a pair of pipes. */ - std::unique_ptr from; + virtual int getFD() const = 0; /** * The store this connection belongs to. @@ -72,19 +76,6 @@ struct RemoteStore::Connection */ std::chrono::time_point startTime; - /** - * Coercion to `WorkerProto::ReadConn`. This makes it easy to use the - * factored out worker protocol searlizers with a - * `RemoteStore::Connection`. - * - * The worker protocol connection types are unidirectional, unlike - * this type. - */ - operator WorkerProto::ReadConn () - { - return WorkerProto::ReadConn{*from, *store, daemonVersion}; - } - /** * Coercion to `WorkerProto::WriteConn`. This makes it easy to use the * factored out worker protocol searlizers with a @@ -160,7 +151,7 @@ struct RemoteStore::ConnectionHandle try { StringSink msg; ((msg << std::forward(args)), ...); - writeFull(handle->toFD, msg.s); + writeFull(handle->getFD(), msg.s); } catch (...) { handle.markBad(); throw; @@ -177,7 +168,7 @@ struct RemoteStore::ConnectionHandle std::get(allArgs) )), ...); - writeFull(handle->toFD, msg.s); + writeFull(handle->getFD(), msg.s); } catch (...) { handle.markBad(); throw; @@ -191,7 +182,11 @@ struct RemoteStore::ConnectionHandle co_return result::success(); } else { try { - co_return WorkerProto::Serialise::read(*handle); + FdSource from{handle->getFD(), handle->fromBuf}; + from.specialEndOfFileError = "Nix daemon disconnected while reading a response"; + co_return WorkerProto::Serialise::read( + {from, *handle->store, handle->daemonVersion} + ); } catch (...) { handle.markBad(); throw; diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index 046ec1991..a3f87f3cb 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -80,17 +80,18 @@ try { /* Send the magic greeting, check for the reply. */ try { conn.store = this; - conn.from->specialEndOfFileError = - "Nix daemon disconnected unexpectedly (maybe it crashed?)"; - FdSink to{conn.toFD}; + 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(); - uint64_t magic = readLongLong(*conn.from); + uint64_t magic = readLongLong(from); if (magic != WORKER_MAGIC_2) throw Error("protocol mismatch"); - *conn.from >> conn.daemonVersion; + from >> conn.daemonVersion; 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) @@ -103,8 +104,10 @@ try { to << false; // obsolete reserveSpace to.flush(); - conn.daemonNixVersion = readString(*conn.from); - conn.remoteTrustsUs = WorkerProto::Serialise>::read(conn); + conn.daemonNixVersion = readString(from); + conn.remoteTrustsUs = WorkerProto::Serialise>::read( + {from, *conn.store, conn.daemonVersion} + ); auto ex = TRY_AWAIT(conn.processStderr()); if (ex.e) { @@ -158,7 +161,7 @@ try { for (auto & i : overrides) command << i.first << i.second.value; - writeFull(conn.toFD, command.s); + writeFull(conn.getFD(), command.s); auto ex = TRY_AWAIT(conn.processStderr()); if (ex.e) { std::rethrow_exception(ex.e); @@ -705,16 +708,29 @@ try { kj::Promise>> RemoteStore::narFromPath(const StorePath & path) try { + struct NarStream : AsyncInputStream + { + ConnectionHandle conn; + AsyncFdIoStream rawStream; + AsyncBufferedInputStream bufferedIn; + box_ptr narCopier; + NarStream(ConnectionHandle conn) + : conn(std::move(conn)) + , rawStream(AsyncFdIoStream::shared_fd{}, this->conn->getFD()) + , bufferedIn(this->rawStream, this->conn->fromBuf) + , narCopier(copyNAR(bufferedIn)) + { + } + + kj::Promise> read(void * buffer, size_t size) override + { + return narCopier->read(buffer, size); + } + }; + auto conn(TRY_AWAIT(getConnection())); TRY_AWAIT(conn.sendCommand(WorkerProto::Op::NarFromPath, printStorePath(path))); - co_return make_box_ptr([](auto conn) -> WireFormatGenerator { - try { - co_yield copyNAR(*conn->from); - } catch (...) { - conn.handle.markBad(); - throw; - } - }(std::move(conn))); + co_return make_box_ptr(std::move(conn)); } catch (...) { co_return result::current_exception(); } @@ -745,34 +761,37 @@ kj::Promise> RemoteStore::Connectio try { while (true) { - auto msg = readNum(*from); + FdSource from{getFD(), fromBuf}; + from.specialEndOfFileError = "Nix daemon disconnected while waiting for a response"; + + auto msg = readNum(from); if (msg == STDERR_ERROR) { - co_return RemoteError{std::make_exception_ptr(readError(*from))}; + co_return RemoteError{std::make_exception_ptr(readError(from))}; } else if (msg == STDERR_NEXT) - printError(chomp(readString(*from))); + printError(chomp(readString(from))); else if (msg == STDERR_START_ACTIVITY) { - auto act = readNum(*from); - auto lvl = (Verbosity) readInt(*from); - auto type = (ActivityType) readInt(*from); - auto s = readString(*from); - auto fields = readFields(*from); - auto parent = readNum(*from); + auto act = readNum(from); + auto lvl = (Verbosity) readInt(from); + auto type = (ActivityType) readInt(from); + auto s = readString(from); + auto fields = readFields(from); + auto parent = readNum(from); logger->startActivity(act, lvl, type, s, fields, parent); } else if (msg == STDERR_STOP_ACTIVITY) { - auto act = readNum(*from); + auto act = readNum(from); logger->stopActivity(act); } else if (msg == STDERR_RESULT) { - auto act = readNum(*from); - auto type = (ResultType) readInt(*from); - auto fields = readFields(*from); + auto act = readNum(from); + auto type = (ResultType) readInt(from); + auto fields = readFields(from); logger->result(act, type, fields); } @@ -823,7 +842,7 @@ kj::Promise> RemoteStore::ConnectionHandle::withFramedSinkAsync( ) try { { - FdSink to{handle->toFD}; + FdSink to{handle->getFD()}; FramedSinkHandler handler{*this, *handlerThreads.lock()}; FramedSink sink(to, handler.ex); TRY_AWAIT(fun(sink)); diff --git a/lix/libstore/ssh-store.cc b/lix/libstore/ssh-store.cc index 193cbeb90..dad27f2ed 100644 --- a/lix/libstore/ssh-store.cc +++ b/lix/libstore/ssh-store.cc @@ -70,6 +70,11 @@ protected: struct Connection : RemoteStore::Connection { std::unique_ptr sshConn; + + int getFD() const override + { + return sshConn->socket.get(); + } }; ref openConnection() override; @@ -99,8 +104,6 @@ ref SSHStore::openConnection() command += " --store " + shellEscape(config_.remoteStore.get()); conn->sshConn = ssh.startCommand(command); - conn->toFD = conn->sshConn->socket.get(); - conn->from = std::make_unique(conn->sshConn->socket.get()); return conn; } diff --git a/lix/libstore/uds-remote-store.cc b/lix/libstore/uds-remote-store.cc index 4861bba5b..72a836e98 100644 --- a/lix/libstore/uds-remote-store.cc +++ b/lix/libstore/uds-remote-store.cc @@ -60,9 +60,6 @@ ref UDSRemoteStore::openConnection() nix::connect(conn->fd.get(), path ? *path : settings.nixDaemonSocketFile); - conn->from = std::make_unique(conn->fd.get()); - conn->toFD = conn->fd.get(); - conn->startTime = std::chrono::steady_clock::now(); return conn; diff --git a/lix/libstore/uds-remote-store.hh b/lix/libstore/uds-remote-store.hh index 733771a5c..a5365c11a 100644 --- a/lix/libstore/uds-remote-store.hh +++ b/lix/libstore/uds-remote-store.hh @@ -61,6 +61,11 @@ private: struct Connection : RemoteStore::Connection { AutoCloseFD fd; + + int getFD() const override + { + return fd.get(); + } }; ref openConnection() override; diff --git a/lix/nix/daemon.cc b/lix/nix/daemon.cc index 6fcfdc3b4..2a019eb20 100644 --- a/lix/nix/daemon.cc +++ b/lix/nix/daemon.cc @@ -409,8 +409,8 @@ static void daemonLoop(AsyncIoRoot & aio, std::optional forceTrustC */ static void forwardStdioConnection(RemoteStore & store) { auto conn = store.openConnectionWrapper(); - int from = conn->from->fd; - int to = conn->toFD; + int from = conn->getFD(); + int to = conn->getFD(); auto nfds = std::max(from, STDIN_FILENO) + 1; while (true) {