Implement scheduling system with parallel execution support and dummy job systems

This commit is contained in:
max
2026-08-07 12:14:02 +02:00
parent a0f2713fdc
commit 0132c1326d
14 changed files with 844 additions and 5 deletions
+270 -1
View File
@@ -1,7 +1,67 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Runtime.ExceptionServices;
namespace Nerfed.Runtime;
public enum JobDeadlineMode
{
FrameCritical,
Deferred,
}
public enum JobExecutionStatus
{
Invalid,
Queued,
Running,
Completed,
Faulted,
Canceled,
}
public readonly struct JobHandle : IEquatable<JobHandle>
{
internal readonly int id;
internal JobHandle(int id)
{
this.id = id;
}
public bool IsValid => id > 0;
public bool Equals(JobHandle other)
{
return id == other.id;
}
public override bool Equals(object obj)
{
return obj is JobHandle other && Equals(other);
}
public override int GetHashCode()
{
return id;
}
public static bool operator ==(JobHandle left, JobHandle right)
{
return left.Equals(right);
}
public static bool operator !=(JobHandle left, JobHandle right)
{
return !left.Equals(right);
}
public override string ToString()
{
return IsValid ? $"JobHandle({id})" : "JobHandle(Invalid)";
}
}
public sealed class JobSystem : IDisposable
{
public static readonly JobSystem Default = new JobSystem();
@@ -9,6 +69,13 @@ public sealed class JobSystem : IDisposable
private Thread[] workers;
private SemaphoreSlim startSignal;
private CountdownEvent completionEvent;
private Thread[] asyncWorkers;
private SemaphoreSlim asyncSignal;
private ConcurrentQueue<int> asyncQueue;
private ConcurrentDictionary<int, AsyncJobState> asyncJobStates;
private int nextAsyncJobId;
private volatile bool running;
// Shared per-dispatch state written by main thread before workers wake.
@@ -16,10 +83,41 @@ public sealed class JobSystem : IDisposable
private int workCount;
private int nextIndex; // grabbed with Interlocked.Increment for work-stealing
private sealed class AsyncJobState : IDisposable
{
public readonly Action Action;
public readonly JobDeadlineMode Deadline;
public readonly string Name;
public readonly ManualResetEventSlim Completion = new(false);
public ExceptionDispatchInfo CapturedException;
public volatile JobExecutionStatus Status;
public volatile bool CancellationRequested;
public AsyncJobState(Action action, JobDeadlineMode deadline, string name)
{
Action = action;
Deadline = deadline;
Name = name;
Status = JobExecutionStatus.Queued;
}
public bool IsTerminal =>
Status == JobExecutionStatus.Completed ||
Status == JobExecutionStatus.Faulted ||
Status == JobExecutionStatus.Canceled;
public void Dispose()
{
Completion.Dispose();
}
}
public int WorkerCount => workers?.Length ?? 0;
public int AsyncWorkerCount => asyncWorkers?.Length ?? 0;
public bool IsInitialized => workers != null;
public void Initialize(int threadCount = -1)
public void Initialize(int threadCount = -1, int asyncThreadCount = -1)
{
if (IsInitialized)
{
@@ -33,7 +131,16 @@ public sealed class JobSystem : IDisposable
running = true;
startSignal = new SemaphoreSlim(0);
completionEvent = new CountdownEvent(1);
asyncSignal = new SemaphoreSlim(0);
workers = new Thread[threadCount];
asyncQueue = new ConcurrentQueue<int>();
asyncJobStates = new ConcurrentDictionary<int, AsyncJobState>();
asyncThreadCount = asyncThreadCount < 0
? Math.Max(1, Environment.ProcessorCount >= 8 ? 2 : 1)
: Math.Max(0, asyncThreadCount);
asyncWorkers = new Thread[asyncThreadCount];
for (int i = 0; i < threadCount; i++)
{
@@ -44,6 +151,98 @@ public sealed class JobSystem : IDisposable
};
workers[i].Start();
}
for (int i = 0; i < asyncThreadCount; i++)
{
asyncWorkers[i] = new Thread(AsyncWorkerLoop)
{
IsBackground = true,
Name = $"AsyncJob-{i}",
};
asyncWorkers[i].Start();
}
}
public JobHandle Submit(Action action, JobDeadlineMode deadline = JobDeadlineMode.Deferred, string name = "")
{
ArgumentNullException.ThrowIfNull(action);
if (!running)
{
throw new InvalidOperationException("JobSystem is not running. Call Initialize before submitting jobs.");
}
int id = Interlocked.Increment(ref nextAsyncJobId);
var state = new AsyncJobState(action, deadline, name ?? string.Empty);
if (!asyncJobStates.TryAdd(id, state))
{
state.Dispose();
throw new InvalidOperationException($"Failed to register async job {id}.");
}
asyncQueue.Enqueue(id);
asyncSignal.Release();
return new JobHandle(id);
}
public JobExecutionStatus GetStatus(JobHandle handle)
{
if (!handle.IsValid)
{
return JobExecutionStatus.Invalid;
}
return asyncJobStates.TryGetValue(handle.id, out AsyncJobState state)
? state.Status
: JobExecutionStatus.Invalid;
}
public bool IsCompleted(JobHandle handle)
{
return GetStatus(handle) is JobExecutionStatus.Completed or JobExecutionStatus.Faulted or JobExecutionStatus.Canceled;
}
public bool Cancel(JobHandle handle)
{
if (!handle.IsValid || !asyncJobStates.TryGetValue(handle.id, out AsyncJobState state))
{
return false;
}
state.CancellationRequested = true;
return true;
}
public void Wait(JobHandle handle)
{
if (!handle.IsValid || !asyncJobStates.TryGetValue(handle.id, out AsyncJobState state))
{
throw new InvalidOperationException($"Unknown job handle {handle}.");
}
state.Completion.Wait();
if (state.Status == JobExecutionStatus.Faulted)
{
state.CapturedException?.Throw();
}
}
public bool TryForget(JobHandle handle)
{
if (!handle.IsValid || !asyncJobStates.TryGetValue(handle.id, out AsyncJobState state) || !state.IsTerminal)
{
return false;
}
if (asyncJobStates.TryRemove(handle.id, out AsyncJobState removedState))
{
removedState.Dispose();
return true;
}
return false;
}
public void Dispatch(int count, Action<int> action)
@@ -79,8 +278,30 @@ public sealed class JobSystem : IDisposable
foreach (Thread t in workers)
t.Join();
for (int i = 0; i < asyncWorkers.Length; i++)
{
asyncSignal.Release();
}
foreach (Thread t in asyncWorkers)
{
t.Join();
}
completionEvent.Dispose();
startSignal.Dispose();
asyncSignal.Dispose();
foreach (KeyValuePair<int, AsyncJobState> pair in asyncJobStates)
{
pair.Value.Dispose();
}
asyncJobStates.Clear();
asyncWorkers = null;
asyncQueue = null;
asyncJobStates = null;
workers = null;
}
@@ -106,4 +327,52 @@ public sealed class JobSystem : IDisposable
completionEvent.Signal();
}
}
private void AsyncWorkerLoop()
{
while (true)
{
asyncSignal.Wait();
if (!running)
{
return;
}
if (!asyncQueue.TryDequeue(out int jobId))
{
continue;
}
if (!asyncJobStates.TryGetValue(jobId, out AsyncJobState state))
{
continue;
}
if (state.CancellationRequested)
{
state.Status = JobExecutionStatus.Canceled;
state.Completion.Set();
continue;
}
state.Status = JobExecutionStatus.Running;
try
{
state.Action();
state.Status = state.CancellationRequested
? JobExecutionStatus.Canceled
: JobExecutionStatus.Completed;
}
catch (Exception ex)
{
state.CapturedException = ExceptionDispatchInfo.Capture(ex);
state.Status = JobExecutionStatus.Faulted;
}
finally
{
state.Completion.Set();
}
}
}
}
@@ -0,0 +1,7 @@
namespace Nerfed.Runtime.Scheduling;
public interface IParallelSystemMetadata
{
string ScheduleName { get; }
SystemAccessDeclaration AccessDeclaration { get; }
}
@@ -0,0 +1,69 @@
namespace Nerfed.Runtime.Scheduling;
public enum SystemAccessMode
{
Read,
WriteExisting,
StructuralWrite,
}
public readonly struct ComponentAccess
{
public ComponentAccess(Type componentType, SystemAccessMode mode)
{
ComponentType = componentType;
Mode = mode;
}
public Type ComponentType { get; }
public SystemAccessMode Mode { get; }
}
public readonly struct SystemAccessDeclaration
{
private readonly ComponentAccess[] accesses;
public static readonly SystemAccessDeclaration Empty = new(Array.Empty<ComponentAccess>());
public SystemAccessDeclaration(params ComponentAccess[] accesses)
{
this.accesses = accesses ?? Array.Empty<ComponentAccess>();
}
public ReadOnlySpan<ComponentAccess> Accesses => accesses ?? Array.Empty<ComponentAccess>();
public bool ConflictsWith(in SystemAccessDeclaration other)
{
ReadOnlySpan<ComponentAccess> left = Accesses;
ReadOnlySpan<ComponentAccess> right = other.Accesses;
for (int i = 0; i < left.Length; i++)
{
for (int j = 0; j < right.Length; j++)
{
if (left[i].Mode == SystemAccessMode.StructuralWrite || right[j].Mode == SystemAccessMode.StructuralWrite)
{
return true;
}
if (left[i].ComponentType != right[j].ComponentType)
{
continue;
}
if (left[i].Mode == SystemAccessMode.Read && right[j].Mode == SystemAccessMode.Read)
{
continue;
}
return true;
}
}
return false;
}
public static ComponentAccess Read<T>() where T : unmanaged => new(typeof(T), SystemAccessMode.Read);
public static ComponentAccess WriteExisting<T>() where T : unmanaged => new(typeof(T), SystemAccessMode.WriteExisting);
public static ComponentAccess StructuralWrite<T>() where T : unmanaged => new(typeof(T), SystemAccessMode.StructuralWrite);
}
@@ -0,0 +1,37 @@
using MoonTools.ECS;
namespace Nerfed.Runtime.Scheduling;
public sealed class SystemScheduleEntry
{
public SystemScheduleEntry(
MoonTools.ECS.System system,
SystemSchedulePhase phase,
string[] dependsOn,
int registrationIndex)
{
System = system;
Phase = phase;
DependsOn = dependsOn ?? Array.Empty<string>();
RegistrationIndex = registrationIndex;
if (system is IParallelSystemMetadata metadata)
{
Name = metadata.ScheduleName;
Access = metadata.AccessDeclaration;
}
else
{
Name = system.GetType().Name;
// Unknown systems are treated as structural to preserve safety until metadata is declared.
Access = new SystemAccessDeclaration(new ComponentAccess(typeof(object), SystemAccessMode.StructuralWrite));
}
}
public MoonTools.ECS.System System { get; }
public string Name { get; }
public SystemSchedulePhase Phase { get; }
public string[] DependsOn { get; }
public int RegistrationIndex { get; }
public SystemAccessDeclaration Access { get; }
}
@@ -0,0 +1,9 @@
namespace Nerfed.Runtime.Scheduling;
public enum SystemSchedulePhase
{
PreUpdate = 0,
Simulation = 1,
LateSimulation = 2,
PreRender = 3,
}
@@ -0,0 +1,188 @@
using MoonTools.ECS;
namespace Nerfed.Runtime.Scheduling;
public sealed class SystemScheduler
{
private readonly List<SystemScheduleEntry> entries = new List<SystemScheduleEntry>();
private readonly SystemSchedulerOptions options;
public SystemScheduler(SystemSchedulerOptions options)
{
this.options = options ?? new SystemSchedulerOptions();
}
public IReadOnlyList<SystemScheduleEntry> Entries => entries;
public void Register(MoonTools.ECS.System system, SystemSchedulePhase phase, params string[] dependsOn)
{
ArgumentNullException.ThrowIfNull(system);
string[] dependencies = dependsOn?.Where(static value => !string.IsNullOrWhiteSpace(value)).ToArray() ?? Array.Empty<string>();
entries.Add(new SystemScheduleEntry(system, phase, dependencies, entries.Count));
}
public void Execute(TimeSpan delta)
{
if (entries.Count == 0)
{
return;
}
Dictionary<string, SystemScheduleEntry> byName = BuildNameIndex(entries);
ValidateDependencies(byName, entries);
foreach (SystemSchedulePhase phase in Enum.GetValues<SystemSchedulePhase>())
{
ExecutePhase(phase, delta, byName);
}
}
private static Dictionary<string, SystemScheduleEntry> BuildNameIndex(List<SystemScheduleEntry> allEntries)
{
var byName = new Dictionary<string, SystemScheduleEntry>(StringComparer.Ordinal);
foreach (SystemScheduleEntry entry in allEntries)
{
if (byName.ContainsKey(entry.Name))
{
throw new InvalidOperationException($"Duplicate scheduled system name '{entry.Name}'.");
}
byName.Add(entry.Name, entry);
}
return byName;
}
private void ValidateDependencies(Dictionary<string, SystemScheduleEntry> byName, List<SystemScheduleEntry> allEntries)
{
foreach (SystemScheduleEntry entry in allEntries)
{
foreach (string dependency in entry.DependsOn)
{
if (!byName.TryGetValue(dependency, out SystemScheduleEntry dependencyEntry))
{
if (options.StrictDependencyValidation)
{
throw new InvalidOperationException($"Scheduled system '{entry.Name}' depends on missing system '{dependency}'.");
}
continue;
}
if (dependencyEntry.Phase > entry.Phase)
{
throw new InvalidOperationException(
$"Scheduled system '{entry.Name}' in phase {entry.Phase} cannot depend on later phase system '{dependency}' in phase {dependencyEntry.Phase}.");
}
}
}
}
private void ExecutePhase(SystemSchedulePhase phase, TimeSpan delta, Dictionary<string, SystemScheduleEntry> byName)
{
List<SystemScheduleEntry> phaseEntries = entries
.Where(entry => entry.Phase == phase)
.OrderBy(entry => entry.RegistrationIndex)
.ToList();
if (phaseEntries.Count == 0)
{
return;
}
var completed = new HashSet<string>(StringComparer.Ordinal);
while (completed.Count < phaseEntries.Count)
{
List<SystemScheduleEntry> ready = phaseEntries
.Where(entry => !completed.Contains(entry.Name) && DependenciesSatisfied(entry, completed, byName))
.OrderBy(entry => entry.RegistrationIndex)
.ToList();
if (ready.Count == 0)
{
throw new InvalidOperationException($"Cyclic or unsatisfied dependencies detected in phase {phase}.");
}
List<SystemScheduleEntry> batch = BuildNonConflictingBatch(ready);
ExecuteBatch(batch, delta, phase);
foreach (SystemScheduleEntry entry in batch)
{
completed.Add(entry.Name);
}
}
}
private static bool DependenciesSatisfied(SystemScheduleEntry entry, HashSet<string> completed, Dictionary<string, SystemScheduleEntry> byName)
{
foreach (string dependency in entry.DependsOn)
{
if (!byName.TryGetValue(dependency, out SystemScheduleEntry dependencyEntry))
{
continue;
}
if (dependencyEntry.Phase == entry.Phase && !completed.Contains(dependency))
{
return false;
}
}
return true;
}
private static List<SystemScheduleEntry> BuildNonConflictingBatch(List<SystemScheduleEntry> ready)
{
var batch = new List<SystemScheduleEntry>(ready.Count);
for (int i = 0; i < ready.Count; i++)
{
SystemScheduleEntry candidate = ready[i];
bool conflicts = false;
for (int j = 0; j < batch.Count; j++)
{
if (candidate.Access.ConflictsWith(batch[j].Access))
{
conflicts = true;
break;
}
}
if (!conflicts)
{
batch.Add(candidate);
}
}
if (batch.Count == 0)
{
batch.Add(ready[0]);
}
return batch;
}
private void ExecuteBatch(List<SystemScheduleEntry> batch, TimeSpan delta, SystemSchedulePhase phase)
{
using ProfilerScope phaseScope = new($"Schedule.{phase}.Batch[{batch.Count}]");
if (options.EnableParallelSystemExecution && batch.Count > 1)
{
Parallel.ForEach(batch, entry => ExecuteSystem(entry, delta));
return;
}
for (int i = 0; i < batch.Count; i++)
{
ExecuteSystem(batch[i], delta);
}
}
private static void ExecuteSystem(SystemScheduleEntry entry, TimeSpan delta)
{
using ProfilerScope scope = new(entry.Name);
entry.System.Update(delta);
}
}
@@ -0,0 +1,7 @@
namespace Nerfed.Runtime.Scheduling;
public sealed class SystemSchedulerOptions
{
public bool EnableParallelSystemExecution { get; set; }
public bool StrictDependencyValidation { get; set; } = true;
}
@@ -0,0 +1,47 @@
using MoonTools.ECS;
using Nerfed.Runtime.Scheduling;
namespace Nerfed.Runtime.Systems.Synthetic;
public sealed class DummyLongJobConsumerSystem : MoonTools.ECS.System, IParallelSystemMetadata
{
private readonly JobSystem jobs;
private readonly DummyLongJobSharedState state;
private readonly bool requireCompletion;
public DummyLongJobConsumerSystem(
World world,
DummyLongJobSharedState state,
JobSystem jobs,
bool requireCompletion) : base(world)
{
this.state = state;
this.jobs = jobs;
this.requireCompletion = requireCompletion;
}
public string ScheduleName => nameof(DummyLongJobConsumerSystem);
// Consumer/apply stage is where structural world work should happen in real systems.
public SystemAccessDeclaration AccessDeclaration => SystemAccessDeclaration.Empty;
public override void Update(TimeSpan delta)
{
JobHandle handle = state.InFlight;
if (!handle.IsValid)
{
return;
}
if (!requireCompletion && !jobs.IsCompleted(handle))
{
return;
}
jobs.Wait(handle);
jobs.TryForget(handle);
state.InFlight = default;
state.ConsumeCount++;
}
}
@@ -0,0 +1,53 @@
using MoonTools.ECS;
using Nerfed.Runtime.Scheduling;
namespace Nerfed.Runtime.Systems.Synthetic;
public sealed class DummyLongJobProducerSystem : MoonTools.ECS.System, IParallelSystemMetadata
{
private readonly JobSystem jobs;
private readonly DummyLongJobSharedState state;
private readonly TimeSpan duration;
private readonly DummyWorkloadMode mode;
private readonly JobDeadlineMode deadline;
private int seed;
public DummyLongJobProducerSystem(
World world,
DummyLongJobSharedState state,
JobSystem jobs,
TimeSpan duration,
DummyWorkloadMode mode,
JobDeadlineMode deadline,
int initialSeed = 1) : base(world)
{
this.state = state;
this.jobs = jobs;
this.duration = duration;
this.mode = mode;
this.deadline = deadline;
seed = initialSeed;
}
public string ScheduleName => nameof(DummyLongJobProducerSystem);
// Producer pattern: reads current world state and emits immutable job payload.
public SystemAccessDeclaration AccessDeclaration => SystemAccessDeclaration.Empty;
public override void Update(TimeSpan delta)
{
if (state.InFlight.IsValid)
{
return;
}
int capturedSeed = seed;
state.InFlight = jobs.Submit(
() => state.LastResult = DummyWorkloadRunner.Run(duration, mode, capturedSeed),
deadline,
nameof(DummyLongJobProducerSystem));
seed = unchecked(capturedSeed + 1);
state.SubmitCount++;
}
}
@@ -0,0 +1,9 @@
namespace Nerfed.Runtime.Systems.Synthetic;
public sealed class DummyLongJobSharedState
{
public JobHandle InFlight;
public int LastResult;
public int SubmitCount;
public int ConsumeCount;
}
@@ -0,0 +1,34 @@
using MoonTools.ECS;
using Nerfed.Runtime.Scheduling;
namespace Nerfed.Runtime.Systems.Synthetic;
public sealed class DummyMainThreadWorkSystem : MoonTools.ECS.System, IParallelSystemMetadata
{
private readonly TimeSpan duration;
private readonly DummyWorkloadMode mode;
private readonly string scheduleName;
private int seed;
public DummyMainThreadWorkSystem(
World world,
TimeSpan duration,
DummyWorkloadMode mode,
string scheduleName,
int initialSeed = 1) : base(world)
{
this.duration = duration;
this.mode = mode;
this.scheduleName = scheduleName;
seed = initialSeed;
}
public string ScheduleName => scheduleName;
public SystemAccessDeclaration AccessDeclaration => SystemAccessDeclaration.Empty;
public override void Update(TimeSpan delta)
{
seed = DummyWorkloadRunner.Run(duration, mode, seed);
}
}
@@ -0,0 +1,8 @@
namespace Nerfed.Runtime.Systems.Synthetic;
public enum DummyWorkloadMode
{
Spin,
Sleep,
Compute,
}
@@ -0,0 +1,54 @@
using System.Diagnostics;
namespace Nerfed.Runtime.Systems.Synthetic;
internal static class DummyWorkloadRunner
{
public static int Run(TimeSpan duration, DummyWorkloadMode mode, int seed)
{
return mode switch
{
DummyWorkloadMode.Sleep => Sleep(duration, seed),
DummyWorkloadMode.Compute => Compute(duration, seed),
_ => Spin(duration, seed),
};
}
private static int Sleep(TimeSpan duration, int seed)
{
if (duration > TimeSpan.Zero)
{
Thread.Sleep(duration);
}
return seed;
}
private static int Spin(TimeSpan duration, int seed)
{
Stopwatch stopwatch = Stopwatch.StartNew();
int value = seed;
while (stopwatch.Elapsed < duration)
{
value = unchecked((value * 1664525) + 1013904223);
Thread.SpinWait(128);
}
return value;
}
private static int Compute(TimeSpan duration, int seed)
{
Stopwatch stopwatch = Stopwatch.StartNew();
int value = seed;
while (stopwatch.Elapsed < duration)
{
for (int i = 0; i < 2048; i++)
{
value = unchecked((value << 5) - value + i);
}
}
return value;
}
}