Misaki/Misaki.HighPerformance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor job API: add JobExecutionContext, update tests

Browse Source 
Major breaking change: job interfaces now use JobExecutionContext
instead of threadIndex, enabling thread-aware and dynamic job
dispatching. Updated all job system, SPMD, and test code to match.
Collections improved with new methods and clearer enumerators.
Renamed IJobScheduler.WaitComplete to Wait. Incremented project
versions. Includes bug fixes, documentation, and style updates.
This commit is contained in:
Misaki
2026-03-04 11:43:39 +09:00
parent b9ca71834f
commit 37d548085e
31 changed files with 652 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
Misaki.HighPerformance.Jobs/contentFiles/cs/any/AssemblyInfo.cs.cs
View File

@@ -1,2 +1,2 @@
global using unsafe JobExecutionFunc = delegate*<void*, ref Misaki.HighPerformance.Jobs.JobRanges, ref int, int, bool>;
global using unsafe JobExecutionFunc = delegate*<void*, ref Misaki.HighPerformance.Jobs.JobRanges, ref int, ref readonly Misaki.HighPerformance.Jobs.JobExecutionContext, bool>;

30
Misaki.HighPerformance.Jobs/contentFiles/cs/any/IJob.cs
View File

@@ -8,8 +8,8 @@ public interface IJob
/// <summary>
/// Executes the job logic.
/// </summary>
/// <param name="threadIndex">The index of the thread executing the job, useful for thread-specific operations.</param>
void Execute(int threadIndex);
/// <param name="ctx">The context of the job execution, providing access to thread-specific information and job scheduling capabilities.</param>
void Execute(ref readonly JobExecutionContext ctx);
}
/// <summary>
@@ -21,8 +21,8 @@ public interface IJobParallelFor
/// Executes the job for a single item at the given index.
/// </summary>
/// <param name="loopIndex">The index of the item to process.</param>
/// <param name="threadIndex">The index of the thread executing the job, useful for thread-specific operations.</param>
void Execute(int loopIndex, int threadIndex);
/// <param name="ctx">The context of the job execution, providing access to thread-specific information and job scheduling capabilities.</param>
void Execute(int loopIndex, ref readonly JobExecutionContext ctx);
}
/// <summary>
@@ -35,36 +35,36 @@ public interface IJobParallel
/// </summary>
/// <param name="startIndex">The zero-based index at which to begin the operation.</param>
/// <param name="endIndex">The zero-based index at which to end the operation.</param>
/// <param name="threadIndex">The index of the thread executing the job, useful for thread-specific operations.</param>
void Execute(int startIndex, int endIndex, int threadIndex);
/// <param name="ctx">The context of the job execution, providing access to thread-specific information and job scheduling capabilities.</param>
void Execute(int startIndex, int endIndex, ref readonly JobExecutionContext ctx);
}
public static class IJobExtensions
{
public static void Run<T>(this ref T job, int threadIndex)
public static void Run<T>(this ref T job, ref readonly JobExecutionContext ctx)
where T : struct, IJob
{
job.Execute(threadIndex);
job.Execute(in ctx);
}
}
public static class IJobParallelForExtensions
{
public static void Run<T>(this ref T job, int totalIterations, int threadIndex)
where T : struct, IJobParallelFor
public static void Run<T>(this ref T job, int totalIterations, ref readonly JobExecutionContext ctx)
where T : struct, IJobParallelFor
{
for (int i = 0; i < totalIterations; i++)
for (var i = 0; i < totalIterations; i++)
{
job.Execute(i, threadIndex);
job.Execute(i, in ctx);
}
}
}
public static class IJobParallelExtensions
{
public static void Run<T>(this ref T job, int totalIterations, int threadIndex)
where T : struct, IJobParallel
public static void Run<T>(this ref T job, int totalIterations, ref readonly JobExecutionContext ctx)
where T : struct, IJobParallel
{
job.Execute(0, totalIterations, threadIndex);
job.Execute(0, totalIterations, in ctx);
}
}

20
Misaki.HighPerformance.Jobs/contentFiles/cs/any/JobExecutionContext.cs Normal file
View File

@@ -0,0 +1,20 @@
namespace Misaki.HighPerformance.Jobs;
public readonly ref struct JobExecutionContext
{
public int ThreadIndex
{
get;
}
public IJobScheduler JobScheduler
{
get;
}
public JobExecutionContext(int threadIndex, IJobScheduler jobScheduler)
{
ThreadIndex = threadIndex;
JobScheduler = jobScheduler;
}
}

12
Misaki.HighPerformance.Jobs/contentFiles/cs/any/JobExecutor.cs
View File

@@ -2,11 +2,11 @@ namespace Misaki.HighPerformance.Jobs;
internal static unsafe class JobExecutor
{
public static bool Execute<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, int threadIndex)
public static bool Execute<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, ref readonly JobExecutionContext ctx)
where T : unmanaged, IJob
{
var pJob = (T*)pJobData;
pJob->Execute(threadIndex);
pJob->Execute(in ctx);
return Interlocked.Decrement(ref remainingBatches) == 0;
}
@@ -25,7 +25,7 @@ internal static unsafe class JobExecutor
return true;
}
public static bool ExecuteParallelFor<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, int threadIndex)
public static bool ExecuteParallelFor<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, ref readonly JobExecutionContext ctx)
where T : unmanaged, IJobParallelFor
{
var pJob = (T*)pJobData;
@@ -40,7 +40,7 @@ internal static unsafe class JobExecutor
for (var i = start; i < end; i++)
{
pJob->Execute(i, threadIndex);
pJob->Execute(i, in ctx);
}
if (Interlocked.Decrement(ref remainingBatches) == 0)
@@ -52,7 +52,7 @@ internal static unsafe class JobExecutor
return wasTheLastBatch;
}
public static bool ExecuteParallel<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, int threadIndex)
public static bool ExecuteParallel<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, ref readonly JobExecutionContext ctx)
where T : unmanaged, IJobParallel
{
var pJob = (T*)pJobData;
@@ -65,7 +65,7 @@ internal static unsafe class JobExecutor
break;
}
pJob->Execute(start, end, threadIndex);
pJob->Execute(start, end, in ctx);
if (Interlocked.Decrement(ref remainingBatches) == 0)
{
wasTheLastBatch = true;

37
Misaki.HighPerformance.Jobs/contentFiles/cs/any/JobScheduler.cs
View File

@@ -1,7 +1,9 @@
using Misaki.HighPerformance.Collections;
using Misaki.HighPerformance.LowLevel.Buffer;
using Misaki.HighPerformance.LowLevel.Collections;
using Misaki.HighPerformance.LowLevel.Utilities;
using System.Collections.Concurrent;
using System.Reflection.Metadata;
using System.Runtime.CompilerServices;
namespace Misaki.HighPerformance.Jobs;
@@ -189,27 +191,18 @@ public interface IJobScheduler
/// Blocks the calling thread until the specified job is completed.
/// </summary>
/// <param name="handle">The handle of the job to wait for.</param>
void WaitComplete(JobHandle handle);
void Wait(JobHandle handle);
/// <summary>
/// Blocks the calling thread until all specified job handles have completed.
/// </summary>
/// <remarks>This method waits for all jobs referenced by the provided handles to complete before
/// returning. The calling thread will be blocked until every job has finished. If any handle is invalid or does not
/// correspond to an active job, it is considered completed. This method is not thread-safe and should not be called
/// concurrently from multiple threads.</remarks>
/// <param name="handles">A collection of job handles to wait for. Each handle represents an asynchronous job whose completion is awaited.
/// The collection must not be empty.</param>
/// <param name="handles">A collection of job handles to wait for.</param>
void WaitAll(params ReadOnlySpan<JobHandle> handles);
/// <summary>
/// Waits until any of the specified job handles has completed and returns the first completed handle.
/// </summary>
/// <remarks>This method blocks the calling thread until at least one of the specified jobs has finished.
/// The returned handle corresponds to the job that completed first among those provided. The order of handles in
/// the span may affect which handle is returned if multiple jobs complete simultaneously.</remarks>
/// <param name="handles">A read-only span containing the job handles to monitor for completion. Each handle represents a job whose
/// completion status will be checked.</param>
/// <param name="handles">A read-only span containing the job handles to monitor for completion.</param>
/// <returns>The first job handle from the provided collection that has completed.</returns>
JobHandle WaitAny(params ReadOnlySpan<JobHandle> handles);
}
@@ -730,7 +723,7 @@ public sealed unsafe partial class JobScheduler : IJobScheduler, IDisposable
return (JobState)(Volatile.Read(ref Unsafe.As<JobState, int>(ref jobInfo.state)) & _STATE_MASK);
}
public void WaitComplete(JobHandle handle)
public void Wait(JobHandle handle)
{
if (!handle.IsValid)
{
@@ -756,15 +749,29 @@ public sealed unsafe partial class JobScheduler : IJobScheduler, IDisposable
public void WaitAll(params ReadOnlySpan<JobHandle> handles)
{
if (handles.Length == 0)
{
return;
}
using var orderedHandles = new UnsafeArray<JobHandle>(handles.Length, Allocator.Temp);
var spin = new SpinWait();
var completedCount = 0;
orderedHandles.CopyFrom(handles);
while (true)
{
var completedCount = 0;
foreach (var handle in handles)
for (int i = completedCount; i < orderedHandles.Length; i++)
{
var handle = orderedHandles[i];
if (!_jobInfoPool.Contains(handle.ID, handle.Generation))
{
// Move completed handle to the front (completedCount index) to avoid checking it again.
var temp = orderedHandles[completedCount];
orderedHandles[completedCount] = handle;
orderedHandles[i] = temp;
completedCount++;
}
}

12
Misaki.HighPerformance.Jobs/contentFiles/cs/any/WorkerThread.cs
View File

@@ -103,11 +103,17 @@ internal class WorkerThread : IDisposable
ref var jobInfo = ref _scheduler.GetJobInfoReference(handle, out var exist);
if (exist && Interlocked.CompareExchange(ref jobInfo.state, JobState.Running, JobState.Scheduled) == JobState.Scheduled)
{
if (jobInfo.pExecutionFunc == null
|| jobInfo.pExecutionFunc(jobInfo.pJobData, ref jobInfo.jobRanges, ref jobInfo.remainingBatches, _index))
if (jobInfo.pExecutionFunc != null)
{
_scheduler.MarkJobComplete(handle);
var ctx = new JobExecutionContext(_index, _scheduler);
if (!jobInfo.pExecutionFunc(jobInfo.pJobData, ref jobInfo.jobRanges, ref jobInfo.remainingBatches, in ctx))
{
// If the job returns false, it means it we are not the last worker to process this job, so we should not mark it as complete yet.
continue;
}
}
_scheduler.MarkJobComplete(handle);
}
}
}