From 6b86f0bf4989af97345fa6f68efe3e803c42719c Mon Sep 17 00:00:00 2001
From: eldritch horrors <pennae@lix.systems>
Date: Sun, 9 Feb 2025 17:30:04 +0100
Subject: [PATCH] libutil: make aio roots usable in processGraph

addMultipleToStore calls addToStore through processGraph, so for now we
need an aio root in processGraph callbacks. eventually we'll drop them,
but that can only be done once we know that the single-threaded runtime
will not get deadlocked. the threads guarantee progress, for time being

Change-Id: I62869a72ef8085c508609a8d4274330e004ab685
---
 lix/libutil/thread-pool.hh | 31 +++++++++++++++++++++++--------
 1 file changed, 23 insertions(+), 8 deletions(-)

diff --git a/lix/libutil/thread-pool.hh b/lix/libutil/thread-pool.hh
index 0e35fb27e..bb090c00f 100644
--- a/lix/libutil/thread-pool.hh
+++ b/lix/libutil/thread-pool.hh
@@ -93,8 +93,8 @@ template<typename T>
 void processGraph(
     const char *poolName,
     const std::set<T> & nodes,
-    std::function<std::set<T>(const T &)> getEdges,
-    std::function<void(const T &)> processNode)
+    std::function<std::set<T>(AsyncIoRoot &, const T &)> getEdges,
+    std::function<void(AsyncIoRoot &, const T &)> processNode)
 {
     struct Graph {
         std::set<T> left;
@@ -103,14 +103,14 @@ void processGraph(
 
     Sync<Graph> graph_(Graph{nodes, {}, {}});
 
-    std::function<void(const T &)> worker;
+    std::function<void(AsyncIoRoot &, const T &)> worker;
 
     /* Create pool last to ensure threads are stopped before other destructors
      * run */
     ThreadPool pool{poolName};
 
 
-    worker = [&](const T & node) {
+    worker = [&](AsyncIoRoot & aio, const T & node) {
 
         {
             auto graph(graph_.lock());
@@ -122,7 +122,7 @@ void processGraph(
 
     getRefs:
         {
-            auto refs = getEdges(node);
+            auto refs = getEdges(aio, node);
             refs.erase(node);
 
             {
@@ -140,7 +140,7 @@ void processGraph(
         return;
 
     doWork:
-        processNode(node);
+        processNode(aio, node);
 
         /* Enqueue work for all nodes that were waiting on this one
            and have no unprocessed dependencies. */
@@ -152,7 +152,7 @@ void processGraph(
                 assert(i != refs.end());
                 refs.erase(i);
                 if (refs.empty())
-                    pool.enqueue(std::bind(worker, rref));
+                    pool.enqueueWithAio(std::bind(worker, std::placeholders::_1, rref));
             }
             graph->left.erase(node);
             graph->refs.erase(node);
@@ -161,7 +161,7 @@ void processGraph(
     };
 
     for (auto & node : nodes)
-        pool.enqueue(std::bind(worker, std::ref(node)));
+        pool.enqueueWithAio(std::bind(worker, std::placeholders::_1, std::ref(node)));
 
     pool.process();
 
@@ -169,4 +169,19 @@ void processGraph(
         throw Error("graph processing incomplete (cyclic reference?)");
 }
 
+template<typename T>
+void processGraph(
+    const char *poolName,
+    const std::set<T> & nodes,
+    std::function<std::set<T>(const T &)> getEdges,
+    std::function<void(const T &)> processNode)
+{
+    processGraph<T>(
+        poolName,
+        nodes,
+        [&](AsyncIoRoot &, const T & node) { return getEdges(node); },
+        [&](AsyncIoRoot &, const T & node) { processNode(node); }
+    );
+}
+
 }