libstore: have SSH use a socketpair, not two pipes

remote store async io will need to set O_NONBLOCK on the connection fds,
and right now the number of fds can vary between connection types: local
connections have one one fd for the sink/source pair since they use unix
sockets, but ssh connections have two because ssh uses pipes. this makes
it rather hard to manage flags correctly, and even harder to wait for io
readiness on both directions using kj. using sockets for ssh fixes this.

Change-Id: I0f563ece7627cd3fbd0f5ce21c25140469729e5a

lix/libstore/legacy-ssh-store.cc

@@ -135,8 +135,8 @@ struct LegacySSHStore final : public Store
                    ? ""
                    : " --store " + shellEscape(config_.remoteStore.get()))
         );
-        conn->to = FdSink(conn->sshConn->in.get());
-        conn->from = FdSource(conn->sshConn->out.get());
+        conn->to = FdSink(conn->sshConn->socket.get());
+        conn->from = FdSource(conn->sshConn->socket.get());
         conn->store = this;
 
         try {

lix/libstore/ssh-store.cc

@@ -97,8 +97,8 @@ ref<RemoteStore::Connection> SSHStore::openConnection()
         command += " --store " + shellEscape(config_.remoteStore.get());
 
     conn->sshConn = ssh.startCommand(command);
-    conn->to = FdSink(conn->sshConn->in.get());
-    conn->from = FdSource(conn->sshConn->out.get());
+    conn->to = FdSink(conn->sshConn->socket.get());
+    conn->from = FdSource(conn->sshConn->socket.get());
     return conn;
 }
 

lix/libstore/ssh.cc

@@ -1,10 +1,13 @@
 #include "lix/libutil/current-process.hh"
 #include "lix/libutil/environment-variables.hh"
 #include "lix/libstore/ssh.hh"
+#include "lix/libutil/error.hh"
+#include "lix/libutil/file-descriptor.hh"
 #include "lix/libutil/finally.hh"
 #include "lix/libutil/logging.hh"
 #include "lix/libutil/strings.hh"
 #include "lix/libstore/temporary-dir.hh"
+#include <sys/socket.h>
 
 namespace nix {
 
@@ -47,9 +50,23 @@ void SSH::addCommonSSHOpts(Strings & args)
 
 std::unique_ptr<SSH::Connection> SSH::startCommand(const std::string & command)
 {
-    Pipe in, out;
-    in.create();
-    out.create();
+    int sp[2];
+    // only linux and bsd support SOCK_CLOEXEC in socketpair type.
+#if __linux__ || __FreeBSD__
+    constexpr int sock_type = SOCK_STREAM | SOCK_CLOEXEC;
+#else
+    constexpr int sock_type = SOCK_STREAM;
+#endif
+    if (socketpair(AF_UNIX, sock_type, 0, sp) < 0) {
+        throw SysError("socketpair() for ssh");
+    }
+
+    AutoCloseFD parent(sp[0]), child(sp[1]);
+#if !(__linux__ || __FreeBSD__)
+    if (fcntl(parent.get(), F_SETFD, O_CLOEXEC) < 0 || fcntl(child.get(), F_SETFD, O_CLOEXEC) < 0) {
+        throw SysError("making socketpair O_CLOEXEC");
+    }
+#endif
 
     auto conn = std::make_unique<Connection>();
     ProcessOptions options;
@@ -64,13 +81,14 @@ std::unique_ptr<SSH::Connection> SSH::startCommand(const std::string & command)
     conn->sshPid = startProcess([&]() {
         restoreProcessContext();
 
-        close(in.writeSide.get());
-        close(out.readSide.get());
+        parent.close();
 
-        if (dup2(in.readSide.get(), STDIN_FILENO) == -1)
+        if (dup2(child.get(), STDIN_FILENO) == -1) {
             throw SysError("duping over stdin");
-        if (dup2(out.writeSide.get(), STDOUT_FILENO) == -1)
+        }
+        if (dup2(child.get(), STDOUT_FILENO) == -1) {
             throw SysError("duping over stdout");
+        }
         if (logFD != -1 && dup2(logFD, STDERR_FILENO) == -1)
             throw SysError("duping over stderr");
 
@@ -93,12 +111,9 @@ std::unique_ptr<SSH::Connection> SSH::startCommand(const std::string & command)
         throw SysError("unable to execute '%s'", args.front());
     }, options);
 
+    child.close();
 
-    in.readSide.reset();
-    out.writeSide.reset();
-
-    conn->out = std::move(out.readSide);
-    conn->in = std::move(in.writeSide);
+    conn->socket = std::move(parent);
 
     return conn;
 }

lix/libstore/ssh.hh

@@ -36,7 +36,7 @@ public:
     struct Connection
     {
         Pid sshPid;
-        AutoCloseFD out, in;
+        AutoCloseFD socket;
     };
 
     std::unique_ptr<Connection> startCommand(const std::string & command);