Change Parallel.For to ZeroAllocJobScheduler

This is more a proof of concept. When box2d v3 comes out having an actual job system will be useful. Also potentially as a replacement for Task.Run?

Robust.Client/GameController/GameController.cs

@@ -116,6 +116,7 @@ internal bool StartupContinue(DisplayMode displayMode)
                 Options.DefaultWindowTitle ?? _resourceManifest!.DefaultWindowTitle ?? "RobustToolbox");
 
             _taskManager.Initialize();
+            _parallelMgr.Initialize();
             _fontManager.SetFontDpi((uint)_configurationManager.GetCVar(CVars.DisplayFontDpi));
 
             // Load optional Robust modules.
@@ -357,7 +358,6 @@ internal bool StartupSystemSplash(
 
             ProfileOptSetup.Setup(_configurationManager);
 
-            _parallelMgr.Initialize();
             _prof.Initialize();
 
             _resourceCache.Initialize(Options.LoadConfigAndUserData ? userDataDir : null);

Robust.Server/BaseServer.cs

@@ -204,8 +204,6 @@ public bool Start(ServerOptions options, Func<ILogHandler>? logHandlerFactory =
 
             ProfileOptSetup.Setup(_config);
 
-            _parallelMgr.Initialize();
-
             //Sets up Logging
             _logHandlerFactory = logHandlerFactory;
 
@@ -268,6 +266,7 @@ public bool Start(ServerOptions options, Func<ILogHandler>? logHandlerFactory =
 
             // Has to be done early because this guy's in charge of the main thread Synchronization Context.
             _taskManager.Initialize();
+            _parallelMgr.Initialize();
 
             LoadSettings();
 

Robust.Shared/Physics/Systems/SharedPhysicsSystem.Contacts.cs

@@ -43,6 +43,7 @@
 using Robust.Shared.Physics.Dynamics.Contacts;
 using Robust.Shared.Physics.Events;
 using Robust.Shared.Utility;
+using Schedulers;
 
 namespace Robust.Shared.Physics.Systems;
 
@@ -582,24 +583,19 @@ internal void CollideContacts()
 
     private void BuildManifolds(Contact[] contacts, int count, ContactStatus[] status, Vector2[] worldPoints)
     {
-        var wake = ArrayPool<bool>.Shared.Rent(count);
-
-        if (count > ContactsPerThread * 2)
-        {
-            var batches = (int) Math.Ceiling((float) count / ContactsPerThread);
+        if (count == 0)
+            return;
 
-            Parallel.For(0, batches, i =>
-            {
-                var start = i * ContactsPerThread;
-                var end = Math.Min(start + ContactsPerThread, count);
-                UpdateContacts(contacts, start, end, status, wake, worldPoints);
-            });
+        var wake = ArrayPool<bool>.Shared.Rent(count);
 
-        }
-        else
+        _parallel.ProcessNow(new ManifoldsJob()
         {
-            UpdateContacts(contacts, 0, count, status, wake, worldPoints);
-        }
+            Physics = this,
+            Status = status,
+            WorldPoints = worldPoints,
+            Contacts = contacts,
+            Wake = wake,
+        }, count);
 
         // Can't do this during UpdateContacts due to IoC threading issues.
         for (var i = 0; i < count; i++)
@@ -620,6 +616,29 @@ private void BuildManifolds(Contact[] contacts, int count, ContactStatus[] statu
         ArrayPool<bool>.Shared.Return(wake);
     }
 
+    private record struct ManifoldsJob : IJobParallelFor
+    {
+        public int ThreadCount => 0;
+        public int BatchSize => ContactsPerThread;
+
+        public SharedPhysicsSystem Physics;
+
+        public Contact[] Contacts;
+        public ContactStatus[] Status;
+        public Vector2[] WorldPoints;
+        public bool[] Wake;
+
+        public void Execute(int index)
+        {
+            var end = index + 1;
+            Physics.UpdateContacts(Contacts, index, end, Status, Wake, WorldPoints);
+        }
+
+        public void Finish()
+        {
+        }
+    }
+
     private void UpdateContacts(Contact[] contacts, int start, int end, ContactStatus[] status, bool[] wake, Vector2[] worldPoints)
     {
         for (var i = start; i < end; i++)

Robust.Shared/Robust.Shared.csproj

@@ -27,6 +27,7 @@
     <ProjectReference Include="..\NetSerializer\NetSerializer\NetSerializer.csproj" />
     <ProjectReference Include="..\Robust.Physics\Robust.Physics.csproj" />
     <ProjectReference Include="..\Robust.Shared.Maths\Robust.Shared.Maths.csproj" />
+    <ProjectReference Include="..\ZeroAllocJobScheduler\JobScheduler\JobScheduler.csproj" />
   </ItemGroup>
   <ItemGroup>
     <EmbeddedResource Include="Utility\TypeAbbreviations.yaml">

Robust.Shared/Threading/ParallelManager.cs

@@ -1,10 +1,7 @@
 using System;
-using System.Threading;
-using System.Threading.Tasks;
 using Robust.Shared.Configuration;
 using Robust.Shared.IoC;
-using Robust.Shared.Log;
-using Robust.Shared.Utility;
+using Schedulers;
 
 namespace Robust.Shared.Threading;
 
@@ -18,6 +15,24 @@ public interface IParallelManager
     /// Add the delegate to <see cref="ParallelCountChanged"/> and immediately invoke it.
     /// </summary>
     void AddAndInvokeParallelCountChanged(Action changed);
+
+    /// <summary>
+    /// Takes in a job that gets flushed.
+    /// </summary>
+    /// <param name="job"></param>
+    JobHandle Process(IJob job);
+
+    /// <summary>
+    /// Takes in a parallel job and runs it the specified amount.
+    /// </summary>
+    void ProcessNow(IJobParallelFor jobs, int amount);
+
+    /// <summary>
+    /// Takes in a parallel job and runs it without blocking.
+    /// </summary>
+    JobHandle Process(IJobParallelFor jobs, int amount);
+
+    void Wait(JobHandle handle);
 }
 
 internal interface IParallelManagerInternal : IParallelManager
@@ -29,12 +44,22 @@ internal sealed class ParallelManager : IParallelManagerInternal
 {
     [Dependency] private readonly IConfigurationManager _cfg = default!;
 
+    private JobScheduler _scheduler = default!;
+
     public event Action? ParallelCountChanged;
     public int ParallelProcessCount { get; private set; }
 
     public void Initialize()
     {
         _cfg.OnValueChanged(CVars.ThreadParallelCount, UpdateCVar, true);
+
+        _scheduler = new JobScheduler(new JobScheduler.Config()
+        {
+            ThreadCount = ParallelProcessCount,
+            // Keep in mind parallel jobs count as 1.
+            MaxExpectedConcurrentJobs = Math.Max(32, ParallelProcessCount),
+            StrictAllocationMode = false,
+        });
     }
 
     public void AddAndInvokeParallelCountChanged(Action changed)
@@ -43,6 +68,13 @@ public void AddAndInvokeParallelCountChanged(Action changed)
         changed();
     }
 
+    public JobHandle Process(IJob job)
+    {
+        var handle = _scheduler.Schedule(job);
+        _scheduler.Flush();
+        return handle;
+    }
+
     private void UpdateCVar(int value)
     {
         var oldCount = ParallelProcessCount;
@@ -51,5 +83,24 @@ private void UpdateCVar(int value)
         if (oldCount != ParallelProcessCount)
             ParallelCountChanged?.Invoke();
     }
+
+    public void ProcessNow(IJobParallelFor job, int amount)
+    {
+        var handle = _scheduler.Schedule(job, amount);
+        _scheduler.Flush();
+        handle.Complete();
+    }
+
+    public JobHandle Process(IJobParallelFor job, int amount)
+    {
+        var handle = _scheduler.Schedule(job, amount);
+        _scheduler.Flush();
+        return handle;
+    }
+
+    public void Wait(JobHandle handle)
+    {
+        handle.Complete();
+    }
 }
 

Robust.UnitTesting/TestingParallelManager.cs

@@ -1,11 +1,12 @@
 using System;
 using Robust.Shared.Threading;
+using Schedulers;
 
 namespace Robust.UnitTesting;
 
 /// <summary>
 /// Only allows 1 parallel process for testing purposes.
-/// </summary>
+/// </summary>j
 public sealed class TestingParallelManager : IParallelManager
 {
     public event Action? ParallelCountChanged;
@@ -15,4 +16,25 @@ public void AddAndInvokeParallelCountChanged(Action changed)
         // Gottem
         return;
     }
+
+    public void ProcessNow(IJobParallelFor job, int amount)
+    {
+        for (var i = 0; i < amount; i++)
+        {
+            job.Execute(i);
+        }
+
+        job.Finish();
+    }
+
+    public JobHandle Process(IJobParallelFor job, int amount)
+    {
+        for (var i = 0; i < amount; i++)
+        {
+            job.Execute(i);
+        }
+
+        job.Finish();
+        return new JobHandle();
+    }
 }