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) {