Add a concurrentForEach method to start a concurrent task for each element in a collection

This is simpler to reason about than creating a `TaskGroup`.

Author: ahoppen

Sources/SemanticIndex/SemanticIndexManager.swift

@@ -339,13 +339,7 @@ package final actor SemanticIndexManager {
   }
 
   private func waitForBuildGraphGenerationTasks() async {
-    await withTaskGroup(of: Void.self) { taskGroup in
-      for generateBuildGraphTask in scheduleIndexingTasks.values {
-        taskGroup.addTask {
-          await generateBuildGraphTask.value
-        }
-      }
-    }
+    await scheduleIndexingTasks.values.concurrentForEach { await $0.value }
   }
 
   /// Wait for all in-progress index tasks to finish.
@@ -355,18 +349,13 @@ package final actor SemanticIndexManager {
     // can await the index tasks below.
     await waitForBuildGraphGenerationTasks()
 
-    await withTaskGroup(of: Void.self) { taskGroup in
-      for (_, status) in inProgressIndexTasks {
-        switch status {
-        case .waitingForPreparation(preparationTaskID: _, indexTask: let indexTask),
-          .preparing(preparationTaskID: _, indexTask: let indexTask),
-          .updatingIndexStore(updateIndexStoreTask: _, indexTask: let indexTask):
-          taskGroup.addTask {
-            await indexTask.value
-          }
-        }
+    await inProgressIndexTasks.concurrentForEach { (_, status) in
+      switch status {
+      case .waitingForPreparation(preparationTaskID: _, indexTask: let indexTask),
+        .preparing(preparationTaskID: _, indexTask: let indexTask),
+        .updatingIndexStore(updateIndexStoreTask: _, indexTask: let indexTask):
+        await indexTask.value
       }
-      await taskGroup.waitForAll()
     }
     index.pollForUnitChangesAndWait()
     logger.debug("Done waiting for up-to-date index")
@@ -774,17 +763,9 @@ package final actor SemanticIndexManager {
       }
       indexTasksWereScheduled(newIndexTasks)
     }
-    let indexTasksImmutable = indexTasks
 
     return Task(priority: priority) {
-      await withTaskGroup(of: Void.self) { taskGroup in
-        for indexTask in indexTasksImmutable {
-          taskGroup.addTask {
-            await indexTask.value
-          }
-        }
-        await taskGroup.waitForAll()
-      }
+      await indexTasks.concurrentForEach { await $0.value }
     }
   }
 }

Sources/SwiftExtensions/AsyncUtils.swift

@@ -165,6 +165,17 @@ extension Collection where Element: Sendable {
       count = indexedResults.count
     }
   }
+
+  /// Invoke `body` for every element in the collection and wait for all calls of `body` to finish
+  package func concurrentForEach(_ body: @escaping @Sendable (Element) async -> Void) async {
+    await withTaskGroup(of: Void.self) { taskGroup in
+      for element in self {
+        taskGroup.addTask {
+          await body(element)
+        }
+      }
+    }
+  }
 }
 
 package struct TimeoutError: Error, CustomStringConvertible {