libstore: drop pervasive RemoteStore send buffering

we don't need to double-buffer commands. only the subframe protocol
needs a buffered backing, and connection setup is special *anyway*.

Change-Id: I596f2bf8e297c3c5dc2befae674deafcf559d9a9

lix/libstore/remote-store-connection.hh

@@ -5,6 +5,7 @@
 #include "lix/libstore/worker-protocol.hh"
 #include "lix/libstore/worker-protocol-impl.hh"
 #include "lix/libutil/async.hh"
+#include "lix/libutil/file-descriptor.hh"
 #include "lix/libutil/pool.hh"
 #include "lix/libutil/result.hh"
 #include "lix/libutil/serialise.hh"
@@ -27,7 +28,7 @@ struct RemoteStore::Connection
     /**
      * Send with this.
      */
-    std::unique_ptr<FdSink> to;
+    int toFD;
 
     /**
      * Receive with this.
@@ -97,7 +98,7 @@ struct RemoteStore::Connection
         return WorkerProto::WriteConn{*store, daemonVersion};
     }
 
-    virtual ~Connection();
+    virtual ~Connection() = default;
 
     // wrapper type for remote errors because `Result<std::exception_ptr>`
     // does not work very well and `Result<Result<void>>` is too confusing
@@ -155,12 +156,11 @@ struct RemoteStore::ConnectionHandle
         // 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 { *handle->to << std::declval<LastArgT>(); }) {
+        if constexpr (requires(StringSink s) { s << std::declval<LastArgT>(); }) {
             try {
                 StringSink msg;
                 ((msg << std::forward<Args>(args)), ...);
-                StringSource{msg.s}.drainInto(*handle->to);
-                handle->to->flush();
+                writeFull(handle->toFD, msg.s);
             } catch (...) {
                 handle.markBad();
                 throw;
@@ -177,8 +177,7 @@ struct RemoteStore::ConnectionHandle
                         std::get<Ids>(allArgs)
                     )),
                     ...);
-                    StringSource{msg.s}.drainInto(*handle->to);
-                    handle->to->flush();
+                    writeFull(handle->toFD, msg.s);
                 } catch (...) {
                     handle.markBad();
                     throw;

lix/libstore/remote-store.cc

@@ -82,8 +82,9 @@ try {
         conn.store = this;
         conn.from->specialEndOfFileError =
             "Nix daemon disconnected unexpectedly (maybe it crashed?)";
-        *conn.to << WORKER_MAGIC_1;
-        conn.to->flush();
+        FdSink to{conn.toFD};
+        to << WORKER_MAGIC_1;
+        to.flush();
 
         uint64_t magic = readLongLong(*conn.from);
         if (magic != WORKER_MAGIC_2)
@@ -94,14 +95,14 @@ try {
             throw Error("Nix daemon protocol version not supported");
         if (GET_PROTOCOL_MINOR(conn.daemonVersion) < MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION)
             throw Error("the Nix daemon version is too old");
-        *conn.to << PROTOCOL_VERSION;
+        to << PROTOCOL_VERSION;
 
         // Obsolete CPU affinity.
-        *conn.to << 0;
+        to << 0;
 
-        *conn.to << false; // obsolete reserveSpace
+        to << false; // obsolete reserveSpace
 
-        conn.to->flush();
+        to.flush();
         conn.daemonNixVersion = readString(*conn.from);
         conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(conn);
 
@@ -157,8 +158,7 @@ try {
     for (auto & i : overrides)
         command << i.first << i.second.value;
 
-    StringSource{command.s}.drainInto(*conn.to);
-    conn.to->flush();
+    writeFull(conn.toFD, command.s);
     auto ex = TRY_AWAIT(conn.processStderr());
     if (ex.e) {
         std::rethrow_exception(ex.e);
@@ -703,16 +703,6 @@ try {
     co_return result::current_exception();
 }
 
-
-RemoteStore::Connection::~Connection()
-{
-    try {
-        to->flush();
-    } catch (...) {
-        ignoreExceptionInDestructor();
-    }
-}
-
 kj::Promise<Result<box_ptr<AsyncInputStream>>> RemoteStore::narFromPath(const StorePath & path)
 try {
     auto conn(TRY_AWAIT(getConnection()));
@@ -833,8 +823,9 @@ kj::Promise<Result<void>> RemoteStore::ConnectionHandle::withFramedSinkAsync(
 )
 try {
     {
+        FdSink to{handle->toFD};
         FramedSinkHandler handler{*this, *handlerThreads.lock()};
-        FramedSink sink(*(*this)->to, handler.ex);
+        FramedSink sink(to, handler.ex);
         TRY_AWAIT(fun(sink));
         sink.flush();
     }

lix/libstore/ssh-store.cc

@@ -99,7 +99,7 @@ ref<RemoteStore::Connection> SSHStore::openConnection()
         command += " --store " + shellEscape(config_.remoteStore.get());
 
     conn->sshConn = ssh.startCommand(command);
-    conn->to = std::make_unique<FdSink>(conn->sshConn->socket.get());
+    conn->toFD = conn->sshConn->socket.get();
     conn->from = std::make_unique<FdSource>(conn->sshConn->socket.get());
     return conn;
 }

lix/libstore/uds-remote-store.cc

@@ -61,7 +61,7 @@ ref<RemoteStore::Connection> UDSRemoteStore::openConnection()
     nix::connect(conn->fd.get(), path ? *path : settings.nixDaemonSocketFile);
 
     conn->from = std::make_unique<FdSource>(conn->fd.get());
-    conn->to = std::make_unique<FdSink>(conn->fd.get());
+    conn->toFD = conn->fd.get();
 
     conn->startTime = std::chrono::steady_clock::now();
 

lix/nix/daemon.cc

@@ -410,7 +410,7 @@ static void daemonLoop(AsyncIoRoot & aio, std::optional<TrustedFlag> forceTrustC
 static void forwardStdioConnection(RemoteStore & store) {
     auto conn = store.openConnectionWrapper();
     int from = conn->from->fd;
-    int to = conn->to->fd;
+    int to = conn->toFD;
 
     auto nfds = std::max(from, STDIN_FILENO) + 1;
     while (true) {