libstore/libfetchers: do more async lazy init

while these two are currently safe using synchronous mutexes we can't
easily guarantee this forever. since we have async mutexes too now we
may as well use them; the overhead they incur is small enough that we
can just ignore it here. this can be revisited later, when necessary.

Change-Id: Ia4cb2361d5638aa48e3b982a05e75ae9c4f254cd

lix/libfetchers/registry.cc

@@ -1,6 +1,7 @@
 #include "lix/libfetchers/registry.hh"
 #include "lix/libfetchers/fetchers.hh"
 #include "lix/libutil/async.hh"
+#include "lix/libutil/sync.hh"
 #include "lix/libutil/types.hh"
 #include "lix/libutil/users.hh"
 #include "lix/libstore/globals.hh"
@@ -12,7 +13,6 @@
 #include <memory>
 #include <mutex>
 #include <nlohmann/json.hpp>
-#include <shared_mutex>
 
 namespace nix::fetchers {
 
@@ -160,21 +160,16 @@ void overrideRegistry(
 
 static kj::Promise<Result<std::shared_ptr<Registry>>> getGlobalRegistry(ref<Store> store)
 try {
-    static std::shared_mutex mtx;
-    static std::shared_ptr<Registry> reg;
+    static Sync<std::shared_ptr<Registry>, AsyncMutex> reg;
 
-    if (std::shared_lock l(mtx); reg) {
-        co_return reg;
-    }
+    auto lk = co_await reg.lock();
 
-    std::scoped_lock l(mtx);
-
-    if (!reg) {
+    if (!*lk) {
         auto path = fetchSettings.flakeRegistry.get();
         if (path == "") {
-            reg = std::make_shared<Registry>(Registry::Global); // empty registry
+            *lk = std::make_shared<Registry>(Registry::Global); // empty registry
         } else if (path == "vendored") {
-            reg = Registry::read(settings.nixDataDir + "/flake-registry.json", Registry::Global);
+            *lk = Registry::read(settings.nixDataDir + "/flake-registry.json", Registry::Global);
         } else {
             if (!path.starts_with("/")) {
                 warn(
@@ -193,11 +188,11 @@ try {
                 path = store->toRealPath(storePath);
             }
 
-            reg = Registry::read(path, Registry::Global);
+            *lk = Registry::read(path, Registry::Global);
         }
     };
 
-    co_return reg;
+    co_return *lk;
 } catch (...) {
     co_return result::current_exception();
 }

lix/libstore/store-api.cc

@@ -5,6 +5,7 @@
 #include "lix/libstore/nar-info-disk-cache.hh"
 #include "lix/libutil/async.hh"
 #include "lix/libutil/result.hh"
+#include "lix/libutil/sync.hh"
 #include "lix/libutil/thread-pool.hh"
 #include "lix/libutil/url.hh"
 #include "lix/libutil/archive.hh"
@@ -19,7 +20,6 @@
 #include <mutex>
 #include <nlohmann/json.hpp>
 #include <regex>
-#include <shared_mutex>
 
 using json = nlohmann::json;
 
@@ -1598,34 +1598,29 @@ try {
 
 kj::Promise<Result<std::list<ref<Store>>>> getDefaultSubstituters()
 try {
-    static std::shared_mutex mtx;
-    static std::optional<std::list<ref<Store>>> stores;
+    static Sync<std::optional<std::list<ref<Store>>>, AsyncMutex> stores;
 
-    if (std::shared_lock l(mtx); stores.has_value()) {
-        co_return *stores;
-    }
+    auto lk = co_await stores.lock();
 
-    std::lock_guard l(mtx);
-
-    if (!stores.has_value()) {
+    if (!lk->has_value()) {
         StringSet done;
 
-        stores.emplace();
+        lk->emplace();
         for (auto uri : settings.substituters.get()) {
             if (!done.insert(uri).second) continue;
             try {
-                stores->push_back(TRY_AWAIT(openStore(uri)));
+                (*lk)->push_back(TRY_AWAIT(openStore(uri)));
             } catch (Error & e) {
                 logWarning(e.info());
             }
         }
 
-        stores->sort([](ref<Store> & a, ref<Store> & b) {
+        (*lk)->sort([](ref<Store> & a, ref<Store> & b) {
             return a->config().priority < b->config().priority;
         });
     }
 
-    co_return *stores;
+    co_return **lk;
 } catch (...) {
     co_return result::current_exception();
 }