Skip to content

Commit 53d3fcc

Browse files
emelrad12emelrad12
committed
Fix hierachy and add lots of other things
Add ability to wait a job handle Optimize workers Fix workers Fix parallel for jobs not being parallel Improve threads Update to support source jobs Add lambda job and fixes
1 parent eec44f8 commit 53d3fcc

20 files changed

+1260
-267
lines changed

‎JobScheduler.Benchmarks/Benchmark.cs‎

Lines changed: 32 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,5 @@
11
using System.Diagnostics;
22
using System.Numerics;
3-
using BenchmarkDotNet.Jobs;
4-
using BenchmarkDotNet.Toolchains.CsProj;
5-
using BenchmarkDotNet.Toolchains.InProcess.NoEmit;
6-
using CommunityToolkit.HighPerformance;
73
using Schedulers;
84
using Schedulers.Benchmarks;
95
using Schedulers.Utils;
@@ -14,7 +10,6 @@ public struct EmptyJob : IJob
1410
{
1511
public void Execute()
1612
{
17-
1813
}
1914
}
2015

@@ -69,7 +64,7 @@ public HeavyCalculationJob(int first, int second)
6964

7065
public void Execute()
7166
{
72-
for (var i = 0; i < 100; i++)
67+
for (var i = 0; i < 1; i++)
7368
{
7469
_first = double.Sqrt(_second);
7570
_second = double.Sqrt(_first) + 1;
@@ -105,7 +100,7 @@ public void RunVectorized(int index, int end)
105100

106101
public void RunSingle(int index)
107102
{
108-
if (!acceptsNewEntries) throw new("Should not accept new entries");
103+
if (!acceptsNewEntries) throw new($"Should not accept new entries {index}");
109104
var newValue = Interlocked.Increment(ref total);
110105
// Console.WriteLine($" {index} {newValue}");
111106
}
@@ -131,24 +126,27 @@ public long End(int jobs, string type)
131126

132127
public class Benchmark
133128
{
134-
private const int jobCount = 200000;
135-
private const int loopCount = 100;
129+
private const int jobCount = 2000;
130+
private const int loopCount = 1000;
136131

137132
private static void CorrectnessTestJob()
138133
{
139-
using var jobScheduler = new JobScheduler();
140134
var timer = new JobTimer();
141135
for (var sindex = 0; sindex < loopCount; sindex++)
142136
{
143137
TestCorrectnessJob.total = 0;
144138
TestCorrectnessJob.acceptsNewEntries = true;
145-
var job = new ParallelJobProducer<TestCorrectnessJob>(jobCount, new(), jobScheduler);
146-
jobScheduler.Wait(job.GetHandle());
147-
TestCorrectnessJob.acceptsNewEntries = false;
148-
var expected = jobCount;
149-
if (TestCorrectnessJob.total != expected)
139+
var job = new ParallelJobProducer<TestCorrectnessJob>(0, jobCount, new());
140+
job.CheckAndSplit();
141+
ParallelForJobCommon.GlobalScheduler.Flush(job.GetHandle());
142+
ParallelForJobCommon.GlobalScheduler.Wait(job.GetHandle());
143+
// Thread.Sleep(1);
144+
// Console.WriteLine($"UnfinishedJobs {job.GetHandle().UnfinishedJobs} total {TestCorrectnessJob.total}");
145+
// TestCorrectnessJob.acceptsNewEntries = false;
146+
var total = TestCorrectnessJob.total;
147+
if (total != jobCount)
150148
{
151-
throw new($"{TestCorrectnessJob.total} != {expected}");
149+
throw new($"{total} != {jobCount}");
152150
}
153151
}
154152

@@ -166,8 +164,7 @@ private static void BenchB()
166164
{
167165
var job = new HeavyCalculationJob(index, index);
168166
var handle = jobScheduler.Schedule(job);
169-
handle.Parent = parentHandle.Index;
170-
handle.SetDependsOn(parentHandle);
167+
handle.SetParent(parentHandle);
171168
jobScheduler.Flush(handle);
172169
}
173170

@@ -184,7 +181,7 @@ private static void BenchC()
184181
var timer = new JobTimer();
185182
for (var sindex = 0; sindex < loopCount; sindex++)
186183
{
187-
var job = new ParallelJobProducer<HeavyCalculationJob>(jobCount, new(), jobScheduler);
184+
var job = new ParallelJobProducer<HeavyCalculationJob>(0, jobCount, new());
188185
jobScheduler.Wait(job.GetHandle());
189186
}
190187

@@ -206,22 +203,29 @@ private static void BenchD()
206203
timer.End(jobCount * loopCount, "Just Parallel.For");
207204
}
208205

209-
private static long BenchVector(bool dontUseVector)
206+
private static long BenchVector(bool useVector)
210207
{
211-
using var jobScheduler = new JobScheduler();
212208
var timer = new JobTimer();
213209
var data = new VectorCalculationJob { a = new float[jobCount], b = new float[jobCount], result = new float[jobCount], Repetitions = 500 };
210+
var parentJob = ParallelForJobCommon.GlobalScheduler.Schedule();
214211
for (var sindex = 0; sindex < loopCount; sindex++)
215212
{
216-
var job = new ParallelJobProducer<VectorCalculationJob>(jobCount, data, jobScheduler, 16, !dontUseVector);
217-
jobScheduler.Wait(job.GetHandle());
213+
var job = new ParallelJobProducer<VectorCalculationJob>(0, jobCount, data, loopSize: 16, onlySingle: !useVector);
214+
job.CheckAndSplit();
215+
job.GetHandle().SetParent(parentJob);
216+
ParallelForJobCommon.GlobalScheduler.Flush(job.GetHandle());
218217
}
219-
220-
return timer.End(jobCount * loopCount, $"Use vector: {!dontUseVector}");
218+
ParallelForJobCommon.GlobalScheduler.Flush(parentJob);
219+
ParallelForJobCommon.GlobalScheduler.Wait(parentJob);
220+
return timer.End(jobCount * loopCount, $"Use vector: {useVector}");
221221
}
222222

223223
private static void Main(string[] args)
224224
{
225+
ParallelForJobCommon.SetScheduler(new());
226+
// new JobHierarchyTest();
227+
// return;
228+
225229
// var config = DefaultConfig.Instance.AddJob(Job.Default
226230
// .WithWarmupCount(2)
227231
// .WithMinIterationCount(10)
@@ -232,7 +236,7 @@ private static void Main(string[] args)
232236
// config = config.WithOptions(ConfigOptions.DisableOptimizationsValidator);
233237
// BenchmarkRunner.Run<JobSchedulerBenchmark>(config);
234238
// return;
235-
for (var i = 0;; i++)
239+
for (var i = 0; i < 20; i++)
236240
{
237241
// CorrectnessTestJob();
238242
// BenchB();
@@ -242,6 +246,8 @@ private static void Main(string[] args)
242246
var nonVectorized = BenchVector(false);
243247
Console.WriteLine($"Ratio {(double)nonVectorized / vectorized}");
244248
}
249+
ParallelForJobCommon.DisposeScheduler();
250+
245251
//using var jobScheduler = new JobScheduler();
246252

247253
// Spawn massive jobs and wait for finish

‎JobScheduler.Benchmarks/JobHierarchyTest.cs‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
using Arch.Benchmarks;
2+
3+
namespace Schedulers.Benchmarks;
4+
5+
public class JobHierarchyTest
6+
{
7+
private class JobThing : IJob
8+
{
9+
public void Execute()
10+
{
11+
Interlocked.Increment(ref totalPassedJobs);
12+
}
13+
}
14+
15+
private int layers = 6;
16+
private int growSizePerLayer = 10;
17+
private int totalScheduledJobs = 0;
18+
private static int totalPassedJobs = 0;
19+
private JobScheduler jobScheduler = new();
20+
private JobThing jobThing = new();
21+
22+
public JobHierarchyTest()
23+
{
24+
// InvertedPyramidJobs();
25+
// InvertedPyramidJobs();
26+
// InvertedPyramidJobs();
27+
// InvertedPyramidJobs();
28+
for (int i = 0; i < 10; i++)
29+
{
30+
PyramidJobs();
31+
}
32+
33+
jobScheduler.Dispose();
34+
}
35+
36+
private void PyramidJobs()
37+
{
38+
totalScheduledJobs = 0;
39+
totalPassedJobs = 0;
40+
var timer = new JobTimer();
41+
var topJob = jobScheduler.Schedule();
42+
AddPyramidJobs(topJob, 0);
43+
jobScheduler.Flush(topJob);
44+
jobScheduler.Wait(topJob);
45+
timer.End(totalScheduledJobs, "PyramidJobs test");
46+
var passedExpected = (int)Math.Pow(growSizePerLayer, layers);
47+
Console.WriteLine($"Total jobs scheduled: {totalScheduledJobs}, total passed jobs: {totalPassedJobs}");
48+
if (totalPassedJobs != passedExpected)
49+
{
50+
throw new($"Total passed jobs {totalPassedJobs} does not match expected {passedExpected}.");
51+
}
52+
}
53+
54+
private void InvertedPyramidJobs()
55+
{
56+
totalScheduledJobs = 0;
57+
totalPassedJobs = 0;
58+
var timer = new JobTimer();
59+
var bottomJob = jobScheduler.Schedule();
60+
var topJob = jobScheduler.Schedule();
61+
AddInvertedPyramidJobs(bottomJob, 1, topJob);
62+
jobScheduler.Flush(bottomJob);
63+
jobScheduler.Flush(topJob);
64+
jobScheduler.Wait(topJob);
65+
timer.End(totalScheduledJobs, "InvertedPyramidJobs test");
66+
var passedExpected = (int)Math.Pow(growSizePerLayer, layers - 1);
67+
Console.WriteLine($"Total jobs scheduled: {totalScheduledJobs}, total passed jobs: {totalPassedJobs}");
68+
if (totalPassedJobs != passedExpected)
69+
{
70+
throw new($"Total passed jobs {totalPassedJobs} does not match expected {passedExpected}.");
71+
}
72+
}
73+
74+
private void AddPyramidJobs(JobHandle parent, int layer)
75+
{
76+
if (layer >= layers)
77+
{
78+
return;
79+
}
80+
81+
for (var i = 0; i < growSizePerLayer; i++)
82+
{
83+
var isNotLastLayer = layer + 1 < layers;
84+
var handle = jobScheduler.Schedule(isNotLastLayer ? null : jobThing, parent);
85+
AddPyramidJobs(handle, layer + 1);
86+
jobScheduler.Flush(handle);
87+
totalScheduledJobs++;
88+
}
89+
}
90+
91+
private void AddInvertedPyramidJobs(JobHandle source, int layer, JobHandle topJob)
92+
{
93+
if (layer >= layers)
94+
{
95+
return;
96+
}
97+
98+
for (var i = 0; i < growSizePerLayer; i++)
99+
{
100+
var isLastLayer = layer + 1 == layers;
101+
var target = jobScheduler.Schedule(isLastLayer ? jobThing : null);
102+
target.SetDependsOn(source);
103+
target.SetParent(topJob);
104+
AddInvertedPyramidJobs(target, layer + 1, topJob);
105+
jobScheduler.Flush(target);
106+
totalScheduledJobs++;
107+
}
108+
}
109+
}

‎JobScheduler.Benchmarks/JobSchedulerBenchmark.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,4 @@
1-
using Arch.Benchmarks;
2-
using CommunityToolkit.HighPerformance;
1+
using CommunityToolkit.HighPerformance;
32
using Schedulers.Utils;
43

54
namespace Schedulers.Benchmarks;
@@ -9,6 +8,7 @@ public struct CalculationJob : IJob
98
private int _first;
109
private int _second;
1110
public static int _result;
11+
1212
public CalculationJob(int first, int second)
1313
{
1414
_first = first;
@@ -35,8 +35,8 @@ public class JobSchedulerBenchmark
3535
[IterationSetup]
3636
public void Setup()
3737
{
38-
_jobScheduler = new JobScheduler();
39-
_jobHandles = new List<JobHandle>(Jobs);
38+
_jobScheduler = new();
39+
_jobHandles = new(Jobs);
4040
}
4141

4242
[IterationCleanup]

‎JobScheduler.Test/BenchmarkTests/GraphGeneratorTests.cs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,4 @@
1-
using System.Diagnostics;
2-
3-
namespace Schedulers.Test;
1+
namespace Schedulers.Test;
42

53
/*
64
[TestFixture]

0 commit comments

Comments
 (0)