From 9efa51b7132603f4450f373567082630e87549e0 Mon Sep 17 00:00:00 2001 From: ManlyMarco <39247311+ManlyMarco@users.noreply.github.com> Date: Fri, 11 Feb 2022 03:16:42 +0100 Subject: [PATCH] Add ThreadingExtensions.ForEachParallel --- BepInEx.Unity/ThreadingHelper.cs | 67 +++++++++++++++++++++++++++++++- 1 file changed, 65 insertions(+), 2 deletions(-) diff --git a/BepInEx.Unity/ThreadingHelper.cs b/BepInEx.Unity/ThreadingHelper.cs index c84c87d..a6ca40d 100644 --- a/BepInEx.Unity/ThreadingHelper.cs +++ b/BepInEx.Unity/ThreadingHelper.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.ComponentModel; using System.Linq; @@ -214,7 +214,7 @@ public static class ThreadingExtensions /// Type of the output values. /// Input values for the work function. /// Function to apply to the data on multiple threads at once. - /// Number of worker threads. By default SystemInfo.processorCount is used. + /// Number of worker threads. By default Environment.ProcessorCount is used. /// /// An exception was thrown inside one of the threads, and the operation was /// aborted. @@ -297,4 +297,67 @@ public static class ThreadingExtensions throw new TargetInvocationException("An exception was thrown inside one of the threads", exceptionThrown); } + + /// + /// Apply a function to a collection of data by spreading the work on multiple threads. + /// Lower overhead than RunParallel but it blocks the main thread until all work is completed or an exception has been thrown. + /// + /// Type of the input values. + /// Input values for the work function. + /// Function to apply to the data on multiple threads at once. + /// Number of worker threads. By default Environment.ProcessorCount is used. + /// An exception was thrown inside one of the threads, and the operation was aborted. + /// Need at least 1 workerCount. + public static void ForEachParallel(this IList data, Action work, int workerCount = -1) + { + if (workerCount < 0) + workerCount = Mathf.Max(2, Environment.ProcessorCount); + else if (workerCount == 0) + throw new ArgumentException("Need at least 1 worker", nameof(workerCount)); + + var currentIndex = data.Count; + + var are = new ManualResetEvent(false); + var runningCount = workerCount; + Exception exceptionThrown = null; + + void DoWork(object _) + { + try + { + while (true) + { + if (exceptionThrown != null) + return; + + var decrementedIndex = Interlocked.Decrement(ref currentIndex); + if (decrementedIndex < 0) + return; + + work(data[decrementedIndex]); + } + } + catch (Exception ex) + { + exceptionThrown = ex; + } + finally + { + var decCount = Interlocked.Decrement(ref runningCount); + if (decCount <= 0) + are.Set(); + } + } + + // Start threads to process the data + for (var i = 0; i < workerCount - 1; i++) + ThreadPool.QueueUserWorkItem(DoWork); + + DoWork(null); + + are.WaitOne(); + + if (exceptionThrown != null) + throw new TargetInvocationException("An exception was thrown inside one of the threads", exceptionThrown); + } }