Fix workers

Author: emelrad12

JobScheduler.Benchmarks/Benchmark.cs

@@ -126,12 +126,12 @@ public long End(int jobs, string type)
 
 public class Benchmark
 {
-    private const int jobCount = 2000;
-    private const int loopCount = 100;
+    private const int jobCount = 20000;
+    private const int loopCount = 1000;
+    private static JobScheduler jobScheduler = new();
 
     private static void CorrectnessTestJob()
     {
-        using var jobScheduler = new JobScheduler();
         var timer = new JobTimer();
         for (var sindex = 0; sindex < loopCount; sindex++)
         {
@@ -206,12 +206,12 @@ private static void BenchD()
 
     private static long BenchVector(bool useVector)
     {
-        using var jobScheduler = new JobScheduler();
         var timer = new JobTimer();
         var data = new VectorCalculationJob { a = new float[jobCount], b = new float[jobCount], result = new float[jobCount], Repetitions = 500 };
         for (var sindex = 0; sindex < loopCount; sindex++)
         {
             var job = new ParallelJobProducer<VectorCalculationJob>(0, jobCount, data, jobScheduler, 16, !useVector);
+            job.CheckAndSplit();
             jobScheduler.Flush(job.GetHandle());
             jobScheduler.Wait(job.GetHandle());
         }
@@ -221,8 +221,8 @@ private static long BenchVector(bool useVector)
 
     private static void Main(string[] args)
     {
-        new JobHierarchyTest();
-        return;
+        // new JobHierarchyTest();
+        // return;
 
         // var config = DefaultConfig.Instance.AddJob(Job.Default
         //     .WithWarmupCount(2)

JobScheduler/JobScheduler.cs

@@ -83,37 +83,33 @@ public void AddDependency(JobHandle dependency, JobHandle dependOn)
         dependency.SetDependsOn(dependOn);
     }
 
-    public float CalculateThreadUsages()
+    /// <summary>
+    /// A function that calculates the current thread usage of the <see cref="Workers"/>.
+    /// It is very simple and just counts the amount of <see cref="Worker"/>s that are currently working.
+    /// </summary>
+    /// <returns>a value between 0 - 1</returns>
+    public float CalculateThreadUsage()
     {
         var total = 0f;
         foreach (var worker in Workers)
         {
-            total += worker.IsCurentlyWorking ? 1 : 0;
+            total += worker.IsCurrentlyWorking ? 1 : 0;
         }
 
+        total /= Workers.Count;
         return total;
     }
 
     private void FlushDirect(JobHandle job)
     {
-        var workerIndex = NextWorkerIndex;
         // Round Robin,
-        var failsBeforeForce = Workers.Count - 1; // We will try to enqueue the job on all workers before we force it.
+        var workerIndex = NextWorkerIndex;
         while (!Workers[workerIndex].Enqueue(job))
         {
             NextWorkerIndex = (NextWorkerIndex + 1) % Workers.Count;
-            failsBeforeForce--;
-            if (failsBeforeForce == 0)
-            {
-                break;
-            }
         }
 
-        // Force it on whatever worker we are on if we failed to enqueue it on all workers.
-        if (failsBeforeForce == 0)
-        {
-            Workers[workerIndex].ForceEnqueue(job);
-        }
+        NextWorkerIndex = (NextWorkerIndex + 1) % Workers.Count;
     }
 
     /// <summary>

JobScheduler/Worker.cs

@@ -11,13 +11,15 @@ internal class Worker
     private readonly int _workerId;
     private readonly Thread _thread;
 
-    private readonly UnorderedThreadSafeQueue<JobHandle> _incomingQueue;
+    private static readonly UnorderedThreadSafeQueue<JobHandle> _incomingQueue = new();
     private readonly WorkStealingDeque<JobHandle> _queue;
 
     private readonly JobScheduler _jobScheduler;
     private volatile CancellationTokenSource _cancellationToken;
-    public bool IsCurentlyWorking { get; private set; } = false;
-    private readonly SemaphoreSlim _workAvailable = new(0);
+    public bool IsCurrentlyWorking { get; private set; } = false;
+    // use a high spin count to avoid sleeping the thread under variable load.
+    // 2047 is the maximum value for a ManualResetEventSlim spin count.
+    private readonly ManualResetEventSlim _workAvailable = new(false, 2047);
 
     /// <summary>
     /// Creates a new <see cref="Worker"/>.
@@ -28,7 +30,6 @@ public Worker(JobScheduler jobScheduler, int id)
     {
         _workerId = id;
 
-        _incomingQueue = new();
         _queue = new(32);
 
         _jobScheduler = jobScheduler;
@@ -41,7 +42,7 @@ public Worker(JobScheduler jobScheduler, int id)
     public bool Enqueue(JobHandle handle)
     {
         var result = _incomingQueue.TryEnqueue(handle);
-        _workAvailable.Release();
+        _workAvailable.Set();
         return result;
     }
 
@@ -94,11 +95,13 @@ private void Run(CancellationToken token)
         {
             while (!token.IsCancellationRequested)
             {
-                IsCurentlyWorking = true;
+                IsCurrentlyWorking = true;
+                var noWorkFound = true;
                 // Pass jobs to the local queue
                 while (_queue.Size() < 32 && _incomingQueue.TryDequeue(out var jobHandle))
                 {
                     _queue.PushBottom(jobHandle);
+                    noWorkFound = false;
                 }
 
                 // Process job in own queue
@@ -107,6 +110,7 @@ private void Run(CancellationToken token)
                 {
                     job.Job?.Execute();
                     _jobScheduler.Finish(job);
+                    noWorkFound = false;
                 }
                 else
                 {
@@ -126,22 +130,23 @@ private void Run(CancellationToken token)
 
                         job.Job?.Execute();
                         _jobScheduler.Finish(job);
+                        noWorkFound = false;
                         break;
                     }
-
-                    if (!exists)
+                }
+                if (noWorkFound)
+                {
+                    IsCurrentlyWorking = false;
+                    // Keep a small fleet of threads always ready to work.
+                    // But don't use all the cpu power on yields.
+                    if (_workerId == 0)
                     {
-                        IsCurentlyWorking = false;
-                        // Keep a small fleet of threads always ready to work.
-                        // But don't use all the cpu power on yields.
-                        if (_workerId < 3)
-                        {
-                            Thread.Yield();
-                        }
-                        else
-                        {
-                            _workAvailable.Wait(token);
-                        }
+                        Thread.Yield();
+                    }
+                    else
+                    {
+                        _workAvailable.Reset();
+                        _workAvailable.Wait(1, token);
                     }
                 }
             }