libutil: make the pool element factory a promise

processStderr of RemoteStore wants to be a promise and it must be used
from connection setup, so the pool factory callback must be a promise.

Change-Id: I9ac742b6048ae6dba0bfa5dcb58971386229690b

lix/libstore/legacy-ssh-store.cc

@@ -126,8 +126,8 @@ struct LegacySSHStore final : public Store
     {
     }
 
-    ref<Connection> openConnection()
-    {
+    kj::Promise<Result<ref<Connection>>> openConnection()
+    try {
         auto conn = make_ref<Connection>();
         conn->sshConn = ssh.startCommand(
             fmt("%s --serve --write", config_.remoteProgram)
@@ -154,7 +154,9 @@ struct LegacySSHStore final : public Store
             throw Error("cannot connect to '%1%'", host);
         }
 
-        return conn;
+        return {conn};
+    } catch (...) {
+        return {result::current_exception()};
     };
 
     std::string getUri() override

lix/libstore/remote-store.cc

@@ -35,25 +35,17 @@ namespace nix {
 RemoteStore::RemoteStore(const RemoteStoreConfig & config)
     : Store(config)
     , connections(make_ref<Pool<Connection>>(
-            std::max(1, (int) config.maxConnections),
-            [this]() {
-                auto conn = openConnectionWrapper();
-                try {
-                    initConnection(*conn);
-                } catch (...) {
-                    failed = true;
-                    throw;
-                }
-                return conn;
-            },
-            [this](const ref<Connection> & r) {
-                return
-                    r->to.good()
-                    && r->from.good()
-                    && std::chrono::duration_cast<std::chrono::seconds>(
-                        std::chrono::steady_clock::now() - r->startTime).count() < this->config().maxConnectionAge;
-            }
-            ))
+          std::max(1, (int) config.maxConnections),
+          [this]() { return openAndInitConnection(); },
+          [this](const ref<Connection> & r) {
+              return r->to.good() && r->from.good()
+                  && std::chrono::duration_cast<std::chrono::seconds>(
+                         std::chrono::steady_clock::now() - r->startTime
+                     )
+                         .count()
+                  < this->config().maxConnectionAge;
+          }
+      ))
 {
 }
 
@@ -70,6 +62,19 @@ ref<RemoteStore::Connection> RemoteStore::openConnectionWrapper()
     }
 }
 
+kj::Promise<Result<ref<RemoteStore::Connection>>> RemoteStore::openAndInitConnection()
+try {
+    auto conn = openConnection();
+    try {
+        initConnection(*conn);
+        co_return conn;
+    } catch (...) {
+        failed = true;
+        throw;
+    }
+} catch (...) {
+    co_return result::current_exception();
+}
 
 void RemoteStore::initConnection(Connection & conn)
 {

lix/libstore/remote-store.hh

@@ -179,6 +179,8 @@ protected:
 
     virtual ref<Connection> openConnection() = 0;
 
+    kj::Promise<Result<ref<Connection>>> openAndInitConnection();
+
     void initConnection(Connection & conn);
 
     ref<Pool<Connection>> connections;

lix/libutil/pool.hh

@@ -43,7 +43,7 @@ public:
     /**
      * A function that produces new instances of R on demand.
      */
-    typedef std::function<ref<R>()> Factory;
+    typedef std::function<kj::Promise<Result<ref<R>>>()> Factory;
 
     /**
      * A function that checks whether an instance of R is still
@@ -76,9 +76,17 @@ private:
 
 public:
 
-    Pool(size_t max = std::numeric_limits<size_t>::max(),
-        const Factory & factory = []() { return make_ref<R>(); },
-        const Validator & validator = [](ref<R> r) { return true; })
+    Pool(
+        size_t max = std::numeric_limits<size_t>::max(),
+        const Factory & factory = []() -> kj::Promise<Result<ref<R>>> {
+            try {
+                return {result::success(make_ref<R>())};
+            } catch (...) {
+                return {result::current_exception()};
+            }
+        },
+        const Validator & validator = [](ref<R> r) { return true; }
+    )
         : factory(factory)
         , validator(validator)
     {
@@ -192,7 +200,7 @@ public:
         /* We need to create a new instance. Because that might take a
            while, we don't hold the lock in the meantime. */
         try {
-            Handle h(*this, factory());
+            Handle h(*this, LIX_TRY_AWAIT(factory()));
             co_return h;
         } catch (...) {
             getFailed();

tests/unit/libutil/pool.cc

@@ -1,4 +1,5 @@
 #include "lix/libutil/pool.hh"
+#include "lix/libutil/result.hh"
 #include <gtest/gtest.h>
 #include <kj/async.h>
 
@@ -32,7 +33,9 @@ namespace nix {
 
     TEST_F(PoolTest, freshPoolHasZeroCountAndSpecifiedCapacity) {
         auto isGood = [](const ref<TestResource> & r) { return r->good; };
-        auto createResource = []() { return make_ref<TestResource>(); };
+        auto createResource = []() -> kj::Promise<Result<ref<TestResource>>> {
+            return {result::success(make_ref<TestResource>())};
+        };
 
         Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource, isGood);
 
@@ -42,7 +45,9 @@ namespace nix {
 
     TEST_F(PoolTest, freshPoolCanGetAResource) {
         auto isGood = [](const ref<TestResource> & r) { return r->good; };
-        auto createResource = []() { return make_ref<TestResource>(); };
+        auto createResource = []() -> kj::Promise<Result<ref<TestResource>>> {
+            return {result::success(make_ref<TestResource>())};
+        };
 
         Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource, isGood);
         ASSERT_EQ(pool.count(), 0);
@@ -57,7 +62,9 @@ namespace nix {
 
     TEST_F(PoolTest, capacityCanBeIncremented) {
         auto isGood = [](const ref<TestResource> & r) { return r->good; };
-        auto createResource = []() { return make_ref<TestResource>(); };
+        auto createResource = []() -> kj::Promise<Result<ref<TestResource>>> {
+            return {result::success(make_ref<TestResource>())};
+        };
 
         Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource, isGood);
         ASSERT_EQ(pool.capacity(), 1);
@@ -67,7 +74,9 @@ namespace nix {
 
     TEST_F(PoolTest, capacityCanBeDecremented) {
         auto isGood = [](const ref<TestResource> & r) { return r->good; };
-        auto createResource = []() { return make_ref<TestResource>(); };
+        auto createResource = []() -> kj::Promise<Result<ref<TestResource>>> {
+            return {result::success(make_ref<TestResource>())};
+        };
 
         Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource, isGood);
         ASSERT_EQ(pool.capacity(), 1);
@@ -78,7 +87,9 @@ namespace nix {
     // Test that the resources we allocate are being reused when they are still good.
     TEST_F(PoolTest, reuseResource) {
         auto isGood = [](const ref<TestResource> & r) { return true; };
-        auto createResource = []() { return make_ref<TestResource>(); };
+        auto createResource = []() -> kj::Promise<Result<ref<TestResource>>> {
+            return {result::success(make_ref<TestResource>())};
+        };
 
         Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource, isGood);
 
@@ -99,7 +110,9 @@ namespace nix {
     // Test that the resources we allocate are being thrown away when they are no longer good.
     TEST_F(PoolTest, badResourceIsNotReused) {
         auto isGood = [](const ref<TestResource> & r) { return false; };
-        auto createResource = []() { return make_ref<TestResource>(); };
+        auto createResource = []() -> kj::Promise<Result<ref<TestResource>>> {
+            return {result::success(make_ref<TestResource>())};
+        };
 
         Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource, isGood);