Upgraded to JobScheduler 1.1.1

genaray · Jan 31, 2024 · 878381f · 878381f
1 parent e859b9d
commit 878381f
Show file tree

Hide file tree

Showing 11 changed files with 86 additions and 53 deletions.
diff --git a/src/Arch.Benchmarks/Arch.Benchmarks.csproj b/src/Arch.Benchmarks/Arch.Benchmarks.csproj
@@ -64,7 +64,7 @@
     <PackageReference Include="BenchmarkDotNet" Version="0.13.2" />
     <PackageReference Include="BenchmarkDotNet.Diagnostics.Windows" Version="0.13.2" />
     <PackageReference Include="Microsoft.Extensions.ObjectPool" Version="7.0.0" />
-    <PackageReference Include="ZeroAllocJobScheduler" Version="1.0.2" />
+    <PackageReference Include="ZeroAllocJobScheduler" Version="1.1.1" />
   </ItemGroup>
 
 </Project>
diff --git a/src/Arch.Samples/Game.cs b/src/Arch.Samples/Game.cs
@@ -4,6 +4,7 @@
 using Microsoft.Xna.Framework;
 using Microsoft.Xna.Framework.Graphics;
 using Microsoft.Xna.Framework.Input;
+using Schedulers;
 
 namespace Arch.Samples;
 
@@ -14,7 +15,7 @@ public sealed class Game : Microsoft.Xna.Framework.Game
 {
     // The world and a job scheduler for multithreading.
     private World _world;
-    private JobScheduler.JobScheduler _jobScheduler;
+    private JobScheduler _jobScheduler;
 
     // Our systems processing entities.
     private MovementSystem _movementSystem;
@@ -61,9 +62,20 @@ protected override void BeginRun()
     {
         base.BeginRun();
 
-        // Create world & systems
+        // Create world & Job Scheduler
         _world = World.Create();
-        _jobScheduler = new("SampleWorkerThreads");
+        _jobScheduler = new(
+                new JobScheduler.Config
+                {
+                    ThreadPrefixName = "Arch.Samples",
+                    ThreadCount = 0,
+                    MaxExpectedConcurrentJobs = 64,
+                    StrictAllocationMode = false,
+                }
+        );
+        World.SharedJobScheduler = _jobScheduler;
+
+        // Create systems
         _movementSystem = new MovementSystem(_world, GraphicsDevice.Viewport.Bounds);
         _colorSystem = new ColorSystem(_world);
         _drawSystem = new DrawSystem(_world, _spriteBatch);

diff --git a/src/Arch.SourceGen/Queries/InlineParallelQuery.cs b/src/Arch.SourceGen/Queries/InlineParallelQuery.cs
@@ -37,19 +37,20 @@ public static void AppendHpParallelQuery(this StringBuilder builder, int amount)
                         job.Size = range.Length;
                         job.Chunks = archetype.Chunks;
                         job.Instance = innerJob;
+
+                        var jobHandle = SharedJobScheduler.Schedule(job);
                         JobsCache.Add(job);
+                        JobHandles.Add(jobHandle);
                     }
 
-                    IJob.Schedule(JobsCache, JobHandles);
-                    JobScheduler.JobScheduler.Instance.Flush();
-                    JobHandle.Complete(JobHandles);
-                    JobHandle.Return(JobHandles);
-
                     // Return jobs to pool
-                    for (var jobIndex = 0; jobIndex < JobsCache.Count; jobIndex++)
+                    SharedJobScheduler.Flush();
+                    JobHandle.CompleteAll(JobHandles.Span);
+
+                    for (var index = 0; index < JobsCache.Count; index++)
                     {
-                        var job = Unsafe.As<ChunkIterationJob<IForEachJob<T,{{generics}}>>>(JobsCache[jobIndex]);
-                        pool.Return(job);
+                          var job = Unsafe.As<ChunkIterationJob<IForEachJob<T,{{generics}}>>>(JobsCache[index]);
+                          pool.Return(job);
                     }
 
                     JobHandles.Clear();
@@ -96,18 +97,19 @@ public static void AppendHpeParallelQuery(this StringBuilder builder, int amount
                         job.Size = range.Length;
                         job.Chunks = archetype.Chunks;
                         job.Instance = innerJob;
+
+                        var jobHandle = SharedJobScheduler.Schedule(job);
                         JobsCache.Add(job);
+                        JobHandles.Add(jobHandle);
                     }
 
-                    IJob.Schedule(JobsCache, JobHandles);
-                    JobScheduler.JobScheduler.Instance.Flush();
-                    JobHandle.Complete(JobHandles);
-                    JobHandle.Return(JobHandles);
-
                     // Return jobs to pool
-                    for (var jobIndex = 0; jobIndex < JobsCache.Count; jobIndex++)
+                    SharedJobScheduler.Flush();
+                    JobHandle.CompleteAll(JobHandles.Span);
+
+                    for (var index = 0; index < JobsCache.Count; index++)
                     {
-                        var job = Unsafe.As<ChunkIterationJob<IForEachWithEntityJob<T,{{generics}}>>>(JobsCache[jobIndex]);
+                        var job = Unsafe.As<ChunkIterationJob<IForEachWithEntityJob<T,{{generics}}>>>(JobsCache[index]);
                         pool.Return(job);
                     }
 

diff --git a/src/Arch.SourceGen/Queries/ParallelQuery.cs b/src/Arch.SourceGen/Queries/ParallelQuery.cs
@@ -37,18 +37,18 @@ public static StringBuilder AppendParallelQuery(this StringBuilder sb, int amoun
                         job.Size = range.Length;
                         job.Chunks = archetype.Chunks;
                         job.Instance = innerJob;
+
+                        var jobHandle = SharedJobScheduler.Schedule(job);
                         JobsCache.Add(job);
+                        JobHandles.Add(jobHandle);
                     }
 
-                    IJob.Schedule(JobsCache, JobHandles);
-                    JobScheduler.JobScheduler.Instance.Flush();
-                    JobHandle.Complete(JobHandles);
-                    JobHandle.Return(JobHandles);
+                    SharedJobScheduler.Flush();
+                    JobHandle.CompleteAll(JobHandles.Span);
 
-                    // Return jobs to pool
-                    for (var jobIndex = 0; jobIndex < JobsCache.Count; jobIndex++)
+                    for (var index = 0; index < JobsCache.Count; index++)
                     {
-                        var job = Unsafe.As<ChunkIterationJob<ForEachJob<{{generics}}>>>(JobsCache[jobIndex]);
+                        var job = Unsafe.As<ChunkIterationJob<ForEachJob<{{generics}}>>>(JobsCache[index]);
                         pool.Return(job);
                     }
 
@@ -97,19 +97,19 @@ public static StringBuilder AppendParallelEntityQuery(this StringBuilder sb, int
                         job.Size = range.Length;
                         job.Chunks = archetype.Chunks;
                         job.Instance = innerJob;
+
+                        var jobHandle = SharedJobScheduler.Schedule(job);
                         JobsCache.Add(job);
+                        JobHandles.Add(jobHandle);
                     }
 
-                    IJob.Schedule(JobsCache, JobHandles);
-                    JobScheduler.JobScheduler.Instance.Flush();
-                    JobHandle.Complete(JobHandles);
-                    JobHandle.Return(JobHandles);
+                    SharedJobScheduler.Flush();
+                    JobHandle.CompleteAll(JobHandles.Span);
 
-                    // Return jobs to pool
-                    for (var jobIndex = 0; jobIndex < JobsCache.Count; jobIndex++)
+                    for (var index = 0; index < JobsCache.Count; index++)
                     {
-                        var job = Unsafe.As<ChunkIterationJob<ForEachWithEntityJob<{{generics}}>>>(JobsCache[jobIndex]);
-                        pool.Return(job);
+                      var job = Unsafe.As<ChunkIterationJob<ForEachWithEntityJob<{{generics}}>>>(JobsCache[index]);
+                      pool.Return(job);
                     }
 
                     JobHandles.Clear();

diff --git a/src/Arch.SourceGen/QueryGenerator.cs b/src/Arch.SourceGen/QueryGenerator.cs
@@ -56,7 +56,7 @@ public void Initialize(IncrementalGeneratorInitializationContext context)
             var accessors = new StringBuilder();
             accessors.AppendLine("using System;");
             accessors.AppendLine("using System.Runtime.CompilerServices;");
-            accessors.AppendLine("using JobScheduler;");
+            accessors.AppendLine("using Schedulers;");
             accessors.AppendLine("using Arch.Core.Utils;");
             accessors.AppendLine("using System.Diagnostics.Contracts;");
             accessors.AppendLine("using Arch.Core.Extensions;");

diff --git a/src/Arch.Tests/CommandBufferTest.cs b/src/Arch.Tests/CommandBufferTest.cs
@@ -1,6 +1,7 @@
 using Arch.CommandBuffer;
 using Arch.Core;
 using Arch.Core.Utils;
+using Schedulers;
 using static NUnit.Framework.Assert;
 
 namespace Arch.Tests;
@@ -222,12 +223,18 @@ public void CommandBufferCombined()
 public partial class CommandBufferTest
 {
 
-    private JobScheduler.JobScheduler _jobScheduler;
+    private JobScheduler _jobScheduler;
 
     [OneTimeSetUp]
     public void Setup()
     {
-        _jobScheduler = new JobScheduler.JobScheduler("CommandBuffer");
+        _jobScheduler = new JobScheduler(
+            new JobScheduler.Config{
+            ThreadPrefixName = "CommandBuffer",
+            ThreadCount = 0,
+            MaxExpectedConcurrentJobs = 64,
+            StrictAllocationMode = false,
+        });
     }
 
     [OneTimeTearDown]

diff --git a/src/Arch.Tests/QueryTest.cs b/src/Arch.Tests/QueryTest.cs
@@ -1,13 +1,14 @@
 using Arch.Core;
 using Arch.Core.Utils;
+using Schedulers;
 using static NUnit.Framework.Assert;
 
 namespace Arch.Tests;
 
 [TestFixture]
 public sealed partial class QueryTest
 {
-    private JobScheduler.JobScheduler _jobScheduler;
+    private JobScheduler _jobScheduler;
     private World? _world;
 
     private static readonly ComponentType[] _entityGroup = { typeof(Transform), typeof(Rotation) };
@@ -19,7 +20,12 @@ public sealed partial class QueryTest
     [OneTimeSetUp]
     public void Setup()
     {
-        _jobScheduler = new JobScheduler.JobScheduler("Test");
+        _jobScheduler = new JobScheduler(new JobScheduler.Config {
+                ThreadPrefixName = "Arch.Samples",
+                ThreadCount = 0,
+                MaxExpectedConcurrentJobs = 64,
+                StrictAllocationMode = false,
+        });
     }
 
     [OneTimeTearDown]

diff --git a/src/Arch/Arch.csproj b/src/Arch/Arch.csproj
@@ -96,8 +96,9 @@ Fixed .Add when a newly non registered component was added. </PackageReleaseNote
     <PackageReference Include="Arch.LowLevel" Version="1.1.0" />
     <PackageReference Include="Collections.Pooled" Version="2.0.0-preview.27" />
     <PackageReference Include="CommunityToolkit.HighPerformance" Version="7.1.2" />
+    <PackageReference Include="Microsoft.Extensions.ObjectPool" Version="7.0.0" />
     <PackageReference Include="System.Runtime.CompilerServices.Unsafe" Version="6.0.0" />
-    <PackageReference Include="ZeroAllocJobScheduler" Version="1.0.1" />
+    <PackageReference Include="ZeroAllocJobScheduler" Version="1.1.1" />
   </ItemGroup>
 
 </Project>
diff --git a/src/Arch/Core/Jobs/Jobs.cs b/src/Arch/Core/Jobs/Jobs.cs
@@ -1,6 +1,6 @@
 using CommunityToolkit.HighPerformance;
-using JobScheduler;
 using Microsoft.Extensions.ObjectPool;
+using Schedulers;
 
 namespace Arch.Core;
 

diff --git a/src/Arch/Core/Jobs/World.Jobs.cs b/src/Arch/Core/Jobs/World.Jobs.cs
@@ -1,6 +1,6 @@
 using Arch.Core.Utils;
 using Collections.Pooled;
-using IJob = JobScheduler.IJob;
+using Schedulers;
 
 // ReSharper disable once CheckNamespace
 namespace Arch.Core;
@@ -11,9 +11,9 @@ public partial class World
 {
 
     /// <summary>
-    ///     A list of <see cref="JobScheduler.JobHandle"/> which are pooled to avoid allocs.
+    ///     A list of <see cref="JobHandle"/> which are pooled to avoid allocs.
     /// </summary>
-    private PooledList<JobScheduler.JobHandle> JobHandles { get; }
+    private PooledList<JobHandle> JobHandles { get; }
 
     /// <summary>
     ///     A cache used for the parallel queries to prevent list allocations.
@@ -86,7 +86,7 @@ public void InlineParallelQuery<T>(in QueryDescription queryDescription, in IFor
     public void InlineParallelChunkQuery<T>(in QueryDescription queryDescription, in T innerJob) where T : struct, IChunkJob
     {
         // Job scheduler needs to be initialized.
-        if (JobScheduler.JobScheduler.Instance is null)
+        if (SharedJobScheduler is null)
         {
             throw new Exception("JobScheduler was not initialized, create one instance of JobScheduler. This creates a singleton used for parallel iterations.");
         }
@@ -105,19 +105,19 @@ public void InlineParallelChunkQuery<T>(in QueryDescription queryDescription, in
                 job.Size = range.Length;
                 job.Chunks = archetype.Chunks;
                 job.Instance = innerJob;
+
+                var jobHandle = SharedJobScheduler.Schedule(job);
                 JobsCache.Add(job);
+                JobHandles.Add(jobHandle);
             }
 
             // Schedule, flush, wait, return.
-            IJob.Schedule(JobsCache, JobHandles);
-            JobScheduler.JobScheduler.Instance.Flush();
-            JobScheduler.JobHandle.Complete(JobHandles);
-            JobScheduler.JobHandle.Return(JobHandles);
+            SharedJobScheduler.Flush();
+            JobHandle.CompleteAll(JobHandles.Span);
 
-            // Return jobs to pool.
-            for (var jobIndex = 0; jobIndex < JobsCache.Count; jobIndex++)
+            for (var index = 0; index < JobsCache.Count; index++)
             {
-                var job = Unsafe.As<ChunkIterationJob<T>>(JobsCache[jobIndex]);
+                var job = Unsafe.As<ChunkIterationJob<T>>(JobsCache[index]);
                 pool.Return(job);
             }
 

diff --git a/src/Arch/Core/World.cs b/src/Arch/Core/World.cs
@@ -4,7 +4,7 @@
 using Arch.Core.Extensions.Internal;
 using Arch.Core.Utils;
 using Collections.Pooled;
-using JobScheduler;
+using Schedulers;
 using Component = Arch.Core.Utils.Component;
 
 namespace Arch.Core;
@@ -83,6 +83,11 @@ public partial class World
     /// </summary>
     public static int WorldSize { [MethodImpl(MethodImplOptions.AggressiveInlining)] get; [MethodImpl(MethodImplOptions.AggressiveInlining)] private set; }
 
+    /// <summary>
+    ///     The shared static <see cref="JobScheduler"/> used for Multithreading.
+    /// </summary>
+    public static JobScheduler? SharedJobScheduler { get; set; }
+
     /// <summary>
     ///     Creates a <see cref="World"/> instance.
     /// </summary>