Add lambda job and fixes

Author: emelrad12

JobScheduler.Benchmarks/Benchmark.cs

@@ -128,7 +128,6 @@ public class Benchmark
 {
     private const int jobCount = 2000;
     private const int loopCount = 1000;
-    private static JobScheduler jobScheduler = new();
 
     private static void CorrectnessTestJob()
     {
@@ -137,10 +136,10 @@ private static void CorrectnessTestJob()
         {
             TestCorrectnessJob.total = 0;
             TestCorrectnessJob.acceptsNewEntries = true;
-            var job = new ParallelJobProducer<TestCorrectnessJob>(0, jobCount, new(), jobScheduler);
+            var job = new ParallelJobProducer<TestCorrectnessJob>(0, jobCount, new());
             job.CheckAndSplit();
-            jobScheduler.Flush(job.GetHandle());
-            jobScheduler.Wait(job.GetHandle());
+            ParallelForJobCommon.GlobalScheduler.Flush(job.GetHandle());
+            ParallelForJobCommon.GlobalScheduler.Wait(job.GetHandle());
             // Thread.Sleep(1);
             // Console.WriteLine($"UnfinishedJobs {job.GetHandle().UnfinishedJobs} total {TestCorrectnessJob.total}");
             // TestCorrectnessJob.acceptsNewEntries = false;
@@ -182,7 +181,7 @@ private static void BenchC()
         var timer = new JobTimer();
         for (var sindex = 0; sindex < loopCount; sindex++)
         {
-            var job = new ParallelJobProducer<HeavyCalculationJob>(0, jobCount, new(), jobScheduler);
+            var job = new ParallelJobProducer<HeavyCalculationJob>(0, jobCount, new());
             jobScheduler.Wait(job.GetHandle());
         }
 
@@ -208,21 +207,22 @@ private static long BenchVector(bool useVector)
     {
         var timer = new JobTimer();
         var data = new VectorCalculationJob { a = new float[jobCount], b = new float[jobCount], result = new float[jobCount], Repetitions = 500 };
-        var parentJob = jobScheduler.Schedule();
+        var parentJob = ParallelForJobCommon.GlobalScheduler.Schedule();
         for (var sindex = 0; sindex < loopCount; sindex++)
         {
-            var job = new ParallelJobProducer<VectorCalculationJob>(0, jobCount, data, jobScheduler, 16, !useVector);
+            var job = new ParallelJobProducer<VectorCalculationJob>(0, jobCount, data, 16, !useVector);
             job.CheckAndSplit();
             job.GetHandle().SetParent(parentJob);
-            jobScheduler.Flush(job.GetHandle());
+            ParallelForJobCommon.GlobalScheduler.Flush(job.GetHandle());
         }
-        jobScheduler.Flush(parentJob);
-        jobScheduler.Wait(parentJob);
+        ParallelForJobCommon.GlobalScheduler.Flush(parentJob);
+        ParallelForJobCommon.GlobalScheduler.Wait(parentJob);
         return timer.End(jobCount * loopCount, $"Use vector: {useVector}");
     }
 
     private static void Main(string[] args)
     {
+        ParallelForJobCommon.SetScheduler(new());
         // new JobHierarchyTest();
         // return;
 
@@ -246,7 +246,7 @@ private static void Main(string[] args)
             var nonVectorized = BenchVector(false);
             Console.WriteLine($"Ratio {(double)nonVectorized / vectorized}");
         }
-        jobScheduler.Dispose();
+        ParallelForJobCommon.DisposeScheduler();
 
         //using var jobScheduler = new JobScheduler();
 

JobScheduler.Test/ParallelForTests.cs

@@ -79,4 +79,29 @@ public void TestParallelForEachMultiple()
         }
         Thread.Sleep(100);
     }
+
+    [Test]
+    public void TestLambdaJob()
+    {
+        var autoresetEvent = new AutoResetEvent(false);
+        LambdaJob.CreateFlushAndWait(() =>
+        {
+            autoresetEvent.Set();
+        });
+        Assert.That(autoresetEvent.WaitOne(1000), Is.True, "Lambda job did not complete in time.");
+    }
+
+    [Test]
+    public void NestedJobs()
+    {
+        var triggers = 0;
+        LambdaJob.CreateFlushAndWait(() =>
+        {
+            ParallelForJob.CreateFlushAndWait(0, 100, i =>
+            {
+                Interlocked.Increment(ref triggers);
+            });
+        });
+        Assert.That(triggers, Is.EqualTo(0));
+    }
 }

JobScheduler/JobHandle.cs

@@ -60,7 +60,7 @@ internal JobHandlePool()
             DependenciesBackingArray[i] = new ushort[DependencyCount];
             for (var j = 0; j < DependencyCount; j++)
             {
-                DependenciesBackingArray[i][j] = JobHandle.NullHandle;
+                DependenciesBackingArray[i][j] = JobHandle.NullHandleId;
             }
         }
 
@@ -94,12 +94,11 @@ internal JobHandle RentJobHandle(IJob? iJob = null)
             Thread.Yield();
         }
 
-        return new()
+        return new(index)
         {
             Generation = ++Generation[index],
-            Index = index,
             Job = iJob,
-            Parent = JobHandle.NullHandle,
+            Parent = JobHandle.NullHandleId,
             UnfinishedJobs = 2, // 2 unfinished, because 1 is the job itself, and 1 is whether the job can be scheduled at all yet
             Dependents = null,
         };
@@ -116,7 +115,7 @@ public void ReturnHandle(JobHandle handle)
         {
             for (var i = 0; i < handle.Dependents!.Length; i++)
             {
-                handle.Dependents[i] = JobHandle.NullHandle;
+                handle.Dependents[i] = JobHandle.NullHandleId;
             }
         }
 
@@ -139,20 +138,48 @@ public struct JobHandle
     /// <summary>
     /// Invalid handle value, used to indicate that a <see cref="JobHandle"/> is not valid.
     /// </summary>
-    public const ushort NullHandle = ushort.MaxValue;
+    public const ushort NullHandleId = ushort.MaxValue;
 
     /// <summary>
     /// The index of this <see cref="JobHandle"/>, pointing towards its data inside the <see cref="Pool"/>.
     /// </summary>
     public ushort Index;
 
+    /// <summary>
+    /// The generation of the handle.
+    /// It is only used to check whether the handle is still valid or has been recycled.
+    /// Which is only important when waiting for the handle.
+    /// Internally we can safely store handles without generation, as if there is a reference to it, it means it is still valid and has not been recycled yet.
+    /// Hence, in that case the generation is irrelevant as it is always the same as the current generation of the pool.
+    /// </summary>
+    internal ushort Generation;
+
+    private readonly bool _isInitialized;
+    /// <summary>
+    /// Indicates whether this <see cref="JobHandle"/> is null.
+    /// Default or Just JobHandle() is null, meaning it has not been initialized yet.
+    /// </summary>
+    public readonly bool IsNull
+    {
+        get => !_isInitialized;
+    }
+
     /// <summary>
     /// Creates a new <see cref="JobHandle"/>.
     /// </summary>
     /// <param name="id">Its id.</param>
     public JobHandle(ushort id)
     {
         Index = id;
+        _isInitialized = true;
+    }
+
+    /// <summary>
+    /// Gets a null handle.
+    /// </summary>
+    public static JobHandle Null
+    {
+        get => default;
     }
 
     /// <summary>
@@ -190,10 +217,15 @@ public ref int UnfinishedJobs
 
     /// <summary>
     /// Sets a <see cref="JobHandle"/> that is to be the parent.
+    /// If a null handle is passed it will be ignored.
     /// </summary>
     /// <param name="parent">The parent job handle</param>
     public void SetParent(JobHandle parent)
     {
+        if (parent.IsNull)
+        {
+            return;
+        }
         Interlocked.Increment(ref parent.UnfinishedJobs);
         Parent = parent.Index;
     }
@@ -245,13 +277,32 @@ public void SetDependsOn(JobHandle source)
     }
 
     /// <summary>
-    /// A shortcut to <see cref="JobScheduler.Flush(JobHandle)"/>.
+    /// A shortcut to <see cref="JobScheduler.Wait(JobHandle)"/>.
     /// Calls the global scheduler wait the job.
     /// In case you want to use other scheduler then use <see cref="JobScheduler.Wait(JobHandle)"/> instead.
     /// </summary>
-    public void Wait()
+    public JobHandle Wait()
     {
         ParallelForJobCommon.GlobalScheduler!.Wait(this);
+        return this;
+    }
+
+    /// <summary>
+    /// Flushes the job handle.
+    /// </summary>
+    public JobHandle Flush()
+    {
+        ParallelForJobCommon.GlobalScheduler!.Flush(this);
+        return this;
+    }
+
+    /// <summary>
+    /// Combines <see cref="Wait"/> and <see cref="Flush"/>.
+    /// </summary>
+    public void FlushAndWait()
+    {
+        Flush();
+        Wait();
     }
 
     /// <summary>
@@ -296,7 +347,7 @@ private bool NextNodeExist()
             throw new InvalidOperationException("Dependents are not initialized. Please set them before using this method.");
         }
 
-        return Dependents[^1] != NullHandle;
+        return Dependents[^1] != NullHandleId;
     }
 
     /// <summary>
@@ -310,7 +361,7 @@ private bool IsAnySpaceAvailable()
             throw new InvalidOperationException("Dependents are not initialized. Please set them before using this method.");
         }
 
-        return Dependents[^2] == NullHandle;
+        return Dependents[^2] == NullHandleId;
     }
 
     /// <summary>
@@ -325,21 +376,12 @@ private int GetFreeDependentsIndex()
 
         for (var i = 0; i < Dependents.Length; i++)
         {
-            if (Dependents[i] == NullHandle)
+            if (Dependents[i] == NullHandleId)
             {
                 return i;
             }
         }
 
         return -1;
     }
-
-    /// <summary>
-    /// The generation of the handle.
-    /// It is only used to check whether the handle is still valid or has been recycled.
-    /// Which is only important when waiting for the handle.
-    /// Internally we can safely store handles without generation, as if there is a reference to it, it means it is still valid and has not been recycled yet.
-    /// Hence, in that case the generation is irrelevant as it is always the same as the current generation of the pool.
-    /// </summary>
-    internal ushort Generation;
 }

JobScheduler/JobScheduler.cs

@@ -71,6 +71,17 @@ public JobHandle Schedule(IJob? iJob, JobHandle parent)
         return job;
     }
 
+    /// <summary>
+    /// Schedule a new <see cref="JobHandle"/> with a parent <see cref="JobHandle"/>.
+    /// Simple a helper function to avoid writing (null, parent) every time.
+    /// </summary>
+    public JobHandle Schedule(JobHandle parent)
+    {
+        var job = JobHandle.Pool.RentJobHandle();
+        job.SetParent(parent);
+        return job;
+    }
+
     /// <summary>
     /// Creates a dependency between two <see cref="JobHandle"/>s.
     /// The dependent <see cref="JobHandle"/>s is executed after the target <see cref="JobHandle"/> has finished.
@@ -143,11 +154,11 @@ public void Wait(JobHandle job)
             {
                 for (var i = 0; i < Workers.Count; i++)
                 {
+                    nextJob = Workers[i].Queue.TrySteal(out stolenJob);
                     if (nextJob)
                     {
                         break;
                     }
-                    nextJob = Workers[i].Queue.TrySteal(out stolenJob);
                 }
             }
 
@@ -195,7 +206,7 @@ internal void Finish(JobHandle handle)
             throw new($"Finish called on a job with {handle.UnfinishedJobs} != 1 unfinished jobs, id:{handle.Index} job: {handle.Job == null}");
         }
 
-        if (handle.Parent != JobHandle.NullHandle)
+        if (handle.Parent != JobHandle.NullHandleId)
         {
             OnChildFinish(new(handle.Parent));
         }
@@ -205,7 +216,7 @@ internal void Finish(JobHandle handle)
             var dependencies = handle.GetDependents();
             foreach (var id in dependencies)
             {
-                if (id == JobHandle.NullHandle)
+                if (id == JobHandle.NullHandleId)
                 {
                     continue;
                 }

JobScheduler/Utils/BulkQueue.cs

@@ -6,7 +6,7 @@ namespace Schedulers.Utils;
 /// Thread-safe bulk queue implementation.
 /// Provides around 10-20% performance in jobs compared to ordinary flush.
 /// </summary>
-public class BulkQueue<T>
+internal class BulkQueue<T>
 {
     public BulkQueue(int segmentCount, int segmentSize)
     {

JobScheduler/Utils/IParallelJobProducer.cs

@@ -35,12 +35,11 @@ public void RunSingle(int index)
 /// </summary>
 /// <typeparam name="T">Type of the </typeparam>
 /// <remarks>The reason we pass T instead of just IParallelJobProducer is because the compiler otherwise cannot inline it</remarks>
-public class ParallelJobProducer<T> : IJob where T : struct, IParallelJobProducer
+public struct ParallelJobProducer<T> : IJob where T : struct, IParallelJobProducer
 {
     private int _from;
     private readonly int _to;
     private readonly T _producer;
-    private readonly JobScheduler _scheduler;
     private readonly JobHandle _selfHandle;
     private readonly bool _onlySingle;
     private readonly int _loopSize;
@@ -54,33 +53,29 @@ public class ParallelJobProducer<T> : IJob where T : struct, IParallelJobProduce
     /// <param name="scheduler">The scheduler where the jobs should be put</param>
     /// <param name="loopSize">Size of the loop, useful for when you want to use vectorization</param>
     /// <param name="onlySingle">Makes sure you can ignore the end parameter in the IParallelJobProducer, and use start as index</param>
-    public ParallelJobProducer(int from, int to, T producer, JobScheduler scheduler, int loopSize = 16, bool onlySingle = false)
+    public ParallelJobProducer(int from, int to, T producer, int loopSize = 16, bool onlySingle = false)
     {
         _from = from;
         _to = to;
         _producer = producer;
-        _scheduler = scheduler;
         _onlySingle = onlySingle;
         _loopSize = loopSize;
-        _selfHandle = _scheduler.Schedule(this);
+        _selfHandle = ParallelForJobCommon.GlobalScheduler!.Schedule(this);
     }
 
     //Only used to spawn sub-jobs
-    private ParallelJobProducer(T producer, JobScheduler scheduler, JobHandle parent, int start, int end, int loopSize, bool onlySingle)
+    private ParallelJobProducer(T producer, JobHandle parent, int start, int end, int loopSize, bool onlySingle)
     {
         _producer = producer;
-        _scheduler = scheduler;
         _from = start;
         _to = end;
         _loopSize = loopSize;
         _onlySingle = onlySingle;
-        _selfHandle = _scheduler.Schedule(this, parent);
+        _selfHandle = ParallelForJobCommon.GlobalScheduler.Schedule(this, parent);
     }
 
     /// <summary>
     /// Executes the job.
-    /// If external thread overwrites the <see cref="_shouldSplitWhenAvailable"/> to true, it will split the job into multiple children.
-    /// The current job will stop executing and the children will be scheduled.
     /// </summary>
     public void Execute()
     {
@@ -154,8 +149,12 @@ private void Split()
                 throw new($"Invalid range from {start} to {end}");
             }
 
-            var pjob = new ParallelJobProducer<T>(_producer, _scheduler, _selfHandle, start, end, _loopSize, _onlySingle);
-            segment.Enqueue(pjob._selfHandle);
+            var pjob = new ParallelJobProducer<T>(_producer, _selfHandle, start, end, _loopSize, _onlySingle);
+            if (!segment.Enqueue(pjob._selfHandle))
+            {
+                throw new InvalidOperationException("Failed to enqueue job in segment. Probably segment size was changed without changing the code using it.");
+            }
+            //Simulate flushing but without using the scheduler
             pjob._selfHandle.UnfinishedJobs--;
         }
         Worker.EnqueueFullSegment(segment);