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

Refactor job system and update project configuration

Browse Source 
Added:
- Added `JobExecutor.cs` for job execution management.
- Added `JobInfo.cs` to hold job execution information.
- Added `TestJobSystem.cs` for unit tests of the job system.
- Added `TestJobs.cs` for additional job implementation tests.
- Added `WorkerThread.cs` to manage worker threads for jobs.

Changed:
- Changed `AssemblyInfo.cs.cs` to include a global using directive for `unsafe JobExecuteFunc`.
- Changed `IJob.cs` to include an overload of the `Execute` method with a `threadIndex` parameter.
- Changed `JobHandle.cs` to include an `IsValid` property and updated internal structure.
- Changed `JobScheduler.cs` to improve job scheduling and management.
- Changed `JobsUtility.cs` to enhance job management functions.
- Changed `MemoryBlock.cs` to reference the heap from which memory was allocated.
- Changed `ParallelNoiseBenchmark.cs` to include benchmarks for the job system.
- Changed `Program.cs` to execute benchmarks instead of previous test code.

Removed:
- Removed `.gitignore` entries for default ignored files.
- Removed `JobBase.cs` to shift from structs to classes for jobs.
- Removed `JobExtensions.cs` indicating a change in job scheduling.
- Removed `JobStruct.cs` indicating a change in job structure.
- Removed `encodings.xml`, `indexLayout.xml`, and `vcs.xml` files to simplify project configuration.
- Removed fields from `JobData.cs` to simplify the job data structure.
- Removed `TestJobSystem.csproj` entries related to old project structure.
This commit is contained in:
Misaki
2025-09-08 23:17:22 +09:00
parent a2a760594e
commit 07c99b8a5a
31 changed files with 1392 additions and 1204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
.idea/.idea.Misaki.HighPerformance/.idea/.gitignore generated vendored
View File

@@ -1,13 +0,0 @@
# Default ignored files
/shelf/
/workspace.xml
# Rider ignored files
/contentModel.xml
/modules.xml
/projectSettingsUpdater.xml
/.idea.Misaki.HighPerformance.iml
# Editor-based HTTP Client requests
/httpRequests/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml

4
.idea/.idea.Misaki.HighPerformance/.idea/encodings.xml generated
View File

@@ -1,4 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Encoding" addBOMForNewFiles="with BOM under Windows, with no BOM otherwise" />
</project>

8
.idea/.idea.Misaki.HighPerformance/.idea/indexLayout.xml generated
View File

@@ -1,8 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="UserContentModel">
<attachedFolders />
<explicitIncludes />
<explicitExcludes />
</component>
</project>

6
.idea/.idea.Misaki.HighPerformance/.idea/vcs.xml generated
View File

@@ -1,6 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>

2
Misaki.HighPerformance.Jobs/AssemblyInfo.cs.cs Normal file
View File

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

10
Misaki.HighPerformance.Jobs/IJob.cs
View File

@@ -2,25 +2,25 @@
/// <summary> /// <summary>
/// Represents a job that performs a single unit of work. /// Represents a job that performs a single unit of work.
/// Jobs are structs to avoid allocations and enable high-performance execution.
/// </summary> /// </summary>
public interface IJob public interface IJob
{ {
/// <summary> /// <summary>
/// Executes the job logic. /// Executes the job logic.
/// </summary> /// </summary>
void Execute(); /// <param name="threadIndex">The index of the thread executing the job, useful for thread-specific operations.</param>
void Execute(int threadIndex);
} }
/// <summary> /// <summary>
/// Represents a job that performs the same operation for a set of items, executed in parallel. /// Represents a job that performs the same operation for a set of items, executed in parallel.
/// Each job instance processes a range of indices, enabling data parallelism.
/// </summary> /// </summary>
public interface IJobParallelFor public interface IJobParallelFor
{ {
/// <summary> /// <summary>
/// Executes the job for a single item at the given index. /// Executes the job for a single item at the given index.
/// </summary> /// </summary>
/// <param name="index">The index of the item to process.</param> /// <param name="loopIndex">The index of the item to process.</param>
void Execute(int index); /// <param name="threadIndex">The index of the thread executing the job, useful for thread-specific operations.</param>
void Execute(int loopIndex, int threadIndex);
} }

24
Misaki.HighPerformance.Jobs/JobBase.cs
View File

@@ -1,24 +0,0 @@
namespace Misaki.HighPerformance.Jobs;
/// <summary>
/// Base class for all jobs. Jobs are now classes to avoid heap allocation complexities.
/// </summary>
public abstract class JobBase
{
/// <summary>
/// Called when the job should be executed.
/// </summary>
public abstract void Execute();
}
/// <summary>
/// Base class for parallel jobs.
/// </summary>
public abstract class ParallelJobBase
{
/// <summary>
/// Called for each item in the parallel job.
/// </summary>
/// <param name="index">The index of the current item.</param>
public abstract void Execute(int index);
}

124
Misaki.HighPerformance.Jobs/JobData.cs
View File

@@ -1,124 +0,0 @@
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
namespace Misaki.HighPerformance.Jobs;
/// <summary>
/// Internal data structure representing job ranges for parallel execution.
/// This matches Unity's JobRanges structure for work stealing.
/// </summary>
[StructLayout(LayoutKind.Sequential)]
internal unsafe struct JobRanges
{
public int JobIndex;
public int BeginIndex;
public int EndIndex;
public int TotalLength;
public int BatchSize;
/// <summary>
/// Pointer to atomic counter for work stealing.
/// </summary>
public int* CurrentIndex;
}
/// <summary>
/// Internal job data structure that holds job execution information.
/// </summary>
[StructLayout(LayoutKind.Sequential)]
internal unsafe struct JobData
{
/// <summary>
/// Unique identifier for this job.
/// </summary>
public ulong Id;
/// <summary>
/// Version counter to detect reused job slots.
/// </summary>
public int Version;
/// <summary>
/// Job state using atomic operations.
/// 0 = Scheduled, 1 = Running, 2 = Completed
/// </summary>
public int State;
/// <summary>
/// Number of dependencies this job has.
/// </summary>
public int DependencyCount;
/// <summary>
/// Number of completed dependencies.
/// </summary>
public int CompletedDependencies;
/// <summary>
/// Type of job (0 = IJob, 1 = IJobParallelFor).
/// </summary>
public JobType JobType;
/// <summary>
/// Function pointer to the job execution method.
/// </summary>
public ExecuteJobDelegate? ExecuteJobFunction;
/// <summary>
/// Function pointer to the parallel job execution method.
/// </summary>
public ExecuteParallelJobDelegate? ExecuteParallelJobFunction;
/// <summary>
/// Reference to the job data object.
/// </summary>
public object? JobDataObject; /// <summary>
/// For parallel jobs, the total number of iterations.
/// </summary>
public int TotalIterations;
/// <summary>
/// For parallel jobs, the batch size per worker.
/// </summary>
public int BatchSize;
/// <summary>
/// Array of dependency job IDs (inline for small counts).
/// </summary>
public fixed ulong Dependencies[8]; // Inline dependencies for performance
/// <summary>
/// Pointer to additional dependencies if more than 8.
/// </summary>
public ulong* AdditionalDependencies;
/// <summary>
/// Size of additional dependencies array.
/// </summary>
public int AdditionalDependencyCount;
public readonly bool IsCompleted => Volatile.Read(ref Unsafe.AsRef<int>(in State)) == 2;
public readonly bool CanExecute =>
Volatile.Read(ref Unsafe.AsRef<int>(in State)) == 0 &&
Volatile.Read(ref Unsafe.AsRef<int>(in CompletedDependencies)) >= DependencyCount;
}
/// <summary>
/// Type of job being executed.
/// </summary>
internal enum JobType : byte
{
Job = 0,
ParallelFor = 1
}
/// <summary>
/// Function pointer delegate for IJob execution.
/// </summary>
internal delegate void ExecuteJobDelegate(object jobData);
/// <summary>
/// Function pointer delegate for IJobParallelFor execution.
/// </summary>
internal unsafe delegate void ExecuteParallelJobDelegate(object jobData, ref JobRanges ranges, int jobIndex);

54
Misaki.HighPerformance.Jobs/JobExecutor.cs Normal file
View File

@@ -0,0 +1,54 @@
namespace Misaki.HighPerformance.Jobs;
internal unsafe static class JobExecutor
{
public static bool Execute<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, int threadIndex)
where T : unmanaged, IJob
{
var pJob = (T*)pJobData;
pJob->Execute(threadIndex);
return Interlocked.Decrement(ref remainingBatches) == 0;
}
private static bool GetWorkerStealingRange(ref JobRanges jobRanges, out int start, out int end)
{
start = Interlocked.Add(ref jobRanges.currentIndex, jobRanges.batchSize) - jobRanges.batchSize;
if (start >= jobRanges.totalIteration)
{
end = start;
return false;
}
end = Math.Min(start + jobRanges.batchSize, jobRanges.totalIteration);
return true;
}
public static bool ExecuteParallel<T>(void* pJobData, ref JobRanges jobRanges, ref int remainingBatches, int threadIndex)
where T : unmanaged, IJobParallelFor
{
var pJob = (T*)pJobData;
var wasTheLastBatch = false;
while (true)
{
if (!GetWorkerStealingRange(ref jobRanges, out var start, out var end))
{
break;
}
for (var i = start; i < end; i++)
{
pJob->Execute(i, threadIndex);
}
if (Interlocked.Decrement(ref remainingBatches) == 0)
{
wasTheLastBatch = true;
}
}
return wasTheLastBatch;
}
}

50
Misaki.HighPerformance.Jobs/JobExtensions.cs
View File

@@ -1,50 +0,0 @@
namespace Misaki.HighPerformance.Jobs;
/// <summary>
/// Extension methods for scheduling jobs in a more user-friendly way.
/// These methods provide the public API for the job system.
/// </summary>
public static class JobExtensions
{
/// <summary>
/// Schedules an IJob for execution.
/// </summary>
/// <typeparam name="T">The job type implementing IJob.</typeparam>
/// <param name="jobData">The job data to execute.</param>
/// <param name="dependsOn">Optional job handle this job depends on.</param>
/// <returns>A job handle that can be used to wait for completion or create dependencies.</returns>
public static JobHandle Schedule<T>(this T jobData, JobHandle dependsOn = default)
where T : class, IJob
{
return JobStruct<T>.Schedule(jobData, dependsOn);
}
/// <summary>
/// Schedules an IJobParallelFor for parallel execution.
/// </summary>
/// <typeparam name="T">The job type implementing IJobParallelFor.</typeparam>
/// <param name="jobData">The job data to execute.</param>
/// <param name="arrayLength">The total number of iterations to execute.</param>
/// <param name="innerLoopBatchCount">The batch size for each worker thread. If 0 or negative, an optimal batch size will be calculated.</param>
/// <param name="dependsOn">Optional job handle this job depends on.</param>
/// <returns>A job handle that can be used to wait for completion or create dependencies.</returns>
public static JobHandle ScheduleParallel<T>(this T jobData, int arrayLength, int innerLoopBatchCount = 0, JobHandle dependsOn = default)
where T : class, IJobParallelFor
{
return ParallelForJobStruct<T>.ScheduleParallel(jobData, arrayLength, innerLoopBatchCount, dependsOn);
}
/// <summary>
/// Schedules an IJobParallelFor for parallel execution with automatic batch size calculation.
/// </summary>
/// <typeparam name="T">The job type implementing IJobParallelFor.</typeparam>
/// <param name="jobData">The job data to execute.</param>
/// <param name="arrayLength">The total number of iterations to execute.</param>
/// <param name="dependsOn">Optional job handle this job depends on.</param>
/// <returns>A job handle that can be used to wait for completion or create dependencies.</returns>
public static JobHandle ScheduleParallel<T>(this T jobData, int arrayLength, JobHandle dependsOn)
where T : class, IJobParallelFor
{
return ParallelForJobStruct<T>.ScheduleParallel(jobData, arrayLength, 0, dependsOn);
}
}

83
Misaki.HighPerformance.Jobs/JobHandle.cs
View File

@@ -1,80 +1,38 @@
using System.Runtime.CompilerServices; namespace Misaki.HighPerformance.Jobs;
using System.Runtime.InteropServices;
namespace Misaki.HighPerformance.Jobs;
/// <summary>
/// A handle that represents a scheduled job and can be used to manage dependencies and wait for completion.
/// JobHandle is designed to be a lightweight value type to avoid allocations.
/// </summary>
[StructLayout(LayoutKind.Sequential)]
public readonly struct JobHandle : IEquatable<JobHandle> public readonly struct JobHandle : IEquatable<JobHandle>
{ {
internal readonly ulong _id; internal readonly int _id;
internal readonly int _version; internal readonly int _generation;
internal JobHandle(ulong id, int version) public static JobHandle Invalid => new(-1, -1);
public bool IsValid => this != Invalid;
internal JobHandle(int id, int generation)
{ {
_id = id; _id = id;
_version = version; _generation = generation;
} }
/// <summary>
/// A completed job handle that can be used as a dependency that is already satisfied.
/// </summary>
public static JobHandle Completed => new(0, 0);
/// <summary>
/// Gets whether this job handle represents a completed job.
/// </summary>
public bool IsCompleted => _id == 0 || JobScheduler.IsCompleted(this);
/// <summary>
/// Blocks the calling thread until the job completes.
/// </summary>
public void Complete()
{
if (_id != 0)
{
JobScheduler.Complete(this);
}
}
/// <summary>
/// Combines multiple job handles into a single dependency.
/// The resulting handle will be complete when all input handles are complete.
/// </summary>
/// <param name="dependencies">The job handles to combine.</param>
/// <returns>A new job handle that depends on all input handles.</returns>
public static JobHandle CombineDependencies(params ReadOnlySpan<JobHandle> dependencies)
{
if (dependencies.Length == 0)
{
return Completed;
}
if (dependencies.Length == 1)
{
return dependencies[0];
}
return JobScheduler.CombineDependencies(dependencies);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool Equals(JobHandle other) public bool Equals(JobHandle other)
{ {
return _id == other._id && _version == other._version; return _id == other._id && _generation == other._generation;
} }
public override bool Equals(object? obj) public override bool Equals(object? obj)
{ {
return obj is JobHandle other && Equals(other); return obj is JobHandle handle && Equals(handle);
} }
public override int GetHashCode() public override int GetHashCode()
{ {
return HashCode.Combine(_id, _version); return HashCode.Combine(_id, _generation);
}
public override string ToString()
{
return IsValid ? $"JobHandle({_id}, {_generation})" : "JobHandle(Invalid)";
} }
public static bool operator ==(JobHandle left, JobHandle right) public static bool operator ==(JobHandle left, JobHandle right)
@@ -84,11 +42,6 @@ public readonly struct JobHandle : IEquatable<JobHandle>
public static bool operator !=(JobHandle left, JobHandle right) public static bool operator !=(JobHandle left, JobHandle right)
{ {
return !left.Equals(right); return !(left == right);
}
public override string ToString()
{
return _id == 0 ? "JobHandle(Completed)" : $"JobHandle(ID:{_id}, Version:{_version})";
} }
} }

45
Misaki.HighPerformance.Jobs/JobInfo.cs Normal file
View File

@@ -0,0 +1,45 @@
namespace Misaki.HighPerformance.Jobs;
public enum JobStatus
{
Invalid = -1,
Created = 0,
Scheduled = 1,
Running = 2,
Completed = 3
}
internal unsafe struct JobInfo
{
public const int MAX_DEPENDENTS = 8;
// The list of jobs that are waiting for THIS job to complete.
public fixed int dependentsID[MAX_DEPENDENTS]; // The actual list of IDs
public fixed int dependentsGeneration[MAX_DEPENDENTS]; // The actual list of generations
public int dependentCount;
public JobRanges jobRanges;
public void* pJobData;
public JobExecuteFunc executeDelegate;
public JobStatus status;
public int remainingBatches;
public int threadIndex; // The preferred thread index to run this job on, -1 means any thread
public int dependencyCount; // Numbers of jobs that this job depends on, when it reaches 0, the job can be executed
}
internal struct JobRanges
{
public int batchSize;
public int totalIteration;
public int currentIndex;
public static JobRanges Single => new()
{
batchSize = 1,
totalIteration = 1,
currentIndex = 0,
};
}

707
Misaki.HighPerformance.Jobs/JobScheduler.cs
View File

@@ -1,455 +1,400 @@
using Misaki.HighPerformance.Collections;
using Misaki.HighPerformance.LowLevel.Buffer; using Misaki.HighPerformance.LowLevel.Buffer;
using Misaki.HighPerformance.LowLevel.Collections; using Misaki.HighPerformance.LowLevel.Helpers;
using System.Collections.Concurrent; using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
namespace Misaki.HighPerformance.Jobs; namespace Misaki.HighPerformance.Jobs;
/// <summary> public unsafe sealed class JobScheduler : IDisposable
/// High-performance job scheduler that manages job execution and dependencies.
/// Designed to minimize allocations and provide efficient work distribution.
/// </summary>
public static unsafe class JobScheduler
{ {
private const int _Init_INLINE_JOBS = 64; private FreeList _jobDataAllocator;
private const int _MAX_WORKER_THREADS = 64; private readonly ConcurrentSlotMap<JobInfo> _jobInfoPool;
private readonly ConcurrentQueue<JobHandle> _jobQueue;
private readonly WorkerThread[] _workerThreads;
private static readonly Lock _lock = new(); private readonly Lock _lock;
private static SlotMap<JobData>? _jobPool; private readonly SemaphoreSlim _workSignal;
private static int _jobVersion; private readonly CancellationTokenSource _cts;
private static volatile bool _isInitialized;
private static volatile bool _isShuttingDown;
// Worker thread management private bool _disposed = false;
private static Thread[]? _workerThreads;
private static int _workerThreadCount;
private static readonly ManualResetEventSlim _workAvailableEvent = new ManualResetEventSlim(false);
private static readonly ConcurrentQueue<int> _readyJobs = new ConcurrentQueue<int>();
// Fast lookup for active jobs public int WorkerCount => _workerThreads.Length;
private static readonly ConcurrentDictionary<ulong, int> _activeJobs = new ConcurrentDictionary<ulong, int>();
static JobScheduler() internal bool IsCancellationRequested => _cts.IsCancellationRequested;
public JobScheduler(int threadCount)
{ {
Initialize(); _jobDataAllocator = new(8);
_jobInfoPool = new();
_jobQueue = new();
_lock = new();
_workSignal = new(0);
_cts = new();
var workerCount = Math.Max(1, threadCount);
_workerThreads = new WorkerThread[workerCount];
for (var i = 0; i < workerCount; i++)
{
_workerThreads[i] = new WorkerThread(i, this);
}
foreach (var worker in _workerThreads)
{
worker.Start();
}
} }
/// <summary> ~JobScheduler()
/// Initializes the job scheduler with default settings.
/// </summary>
public static void Initialize(int InitialJobsSize = _Init_INLINE_JOBS, int workerThreadCount = -1)
{ {
if (_isInitialized) Dispose();
return; }
lock (_lock) private void EnqueueJobIfReady(JobHandle handle)
{
ref var jobInfo = ref _jobInfoPool.GetElementReferenceAt(handle._id, handle._generation, out var exist);
if (exist && Volatile.Read(ref jobInfo.dependencyCount) == 0)
{ {
if (_isInitialized) if (Interlocked.CompareExchange(ref jobInfo.status, JobStatus.Scheduled, JobStatus.Created) != JobStatus.Created)
return;
_jobPool = new SlotMap<JobData>(InitialJobsSize);
// Initialize worker threads
if (workerThreadCount <= 0)
workerThreadCount = Math.Min(Environment.ProcessorCount, _MAX_WORKER_THREADS);
_workerThreadCount = workerThreadCount;
_workerThreads = new Thread[workerThreadCount];
for (var i = 0; i < workerThreadCount; i++)
{ {
var thread = new Thread(WorkerThreadLoop) return;
{
Name = $"JobWorker-{i}",
IsBackground = true
};
_workerThreads[i] = thread;
thread.Start(i);
} }
_isInitialized = true; ConcurrentQueue<JobHandle> jobQueue;
} if (jobInfo.threadIndex >= 0 && jobInfo.threadIndex < _workerThreads.Length)
}
/// <summary>
/// Shuts down the job scheduler and cleans up resources.
/// </summary>
public static void Shutdown()
{
if (!_isInitialized || _isShuttingDown)
return;
lock (_lock)
{
if (_isShuttingDown)
return;
_isShuttingDown = true;
// Signal all worker threads to exit
_workAvailableEvent.Set();
// Wait for all worker threads to complete
if (_workerThreads != null)
{ {
for (var i = 0; i < _workerThreadCount; i++) jobQueue = _workerThreads[jobInfo.threadIndex].LocalQueue;
{ }
_workerThreads[i]?.Join(5000); // 5 second timeout else
} {
jobQueue = _jobQueue;
} }
_jobPool?.Clear(); // Ensure the count of this job handle won't exceed the number of worker threads.
_jobPool = null; // Worker threads will steal parallel iteration ranges from each other.
var handleCount = Math.Min(jobInfo.remainingBatches, _workerThreads.Length);
_isInitialized = false; for (var i = 0; i < handleCount; i++)
_isShuttingDown = false; {
jobQueue.Enqueue(handle);
}
_workSignal.Release(handleCount);
} }
} }
/// <summary> private JobHandle CreateJobHandle(ref JobInfo jobInfo, params ReadOnlySpan<JobHandle> dependencies)
/// Schedules a job for execution.
/// </summary>
internal static JobHandle ScheduleJob(object jobData, ExecuteJobDelegate executeFunction, JobType jobType,
int totalIterations, int batchSize, JobHandle dependsOn)
{ {
if (!_isInitialized) var id = _jobInfoPool.Add(jobInfo, out var generation);
throw new InvalidOperationException("JobScheduler is not initialized"); ref var infoInPool = ref _jobInfoPool.GetElementReferenceAt(id, generation, out _);
var jobSlot = AllocateJobSlot(); var handle = new JobHandle(id, generation);
var jobId = JobsUtility.GetNextJobId();
var version = Interlocked.Increment(ref _jobVersion);
ref var job = ref _jobPool![jobSlot];
job.Id = jobId;
job.Version = version;
job.State = 0; // Scheduled
job.JobType = jobType;
job.ExecuteJobFunction = executeFunction;
job.ExecuteParallelJobFunction = null;
job.JobDataObject = jobData;
job.TotalIterations = totalIterations;
job.BatchSize = batchSize;
job.DependencyCount = dependsOn._id == 0 ? 0 : 1;
job.CompletedDependencies = 0;
job.AdditionalDependencies = null;
job.AdditionalDependencyCount = 0;
// Set up dependencies
if (dependsOn._id != 0)
{
job.Dependencies[0] = dependsOn._id;
}
_activeJobs.TryAdd(jobId, jobSlot);
// Check if job can be executed immediately
if (job.CanExecute)
{
_readyJobs.Enqueue(jobSlot);
_workAvailableEvent.Set();
}
return new JobHandle(jobId, version);
}
/// <summary>
/// Schedules a parallel job for execution.
/// </summary>
internal static JobHandle ScheduleParallelJob(object jobData, ExecuteParallelJobDelegate executeFunction,
int totalIterations, int batchSize, JobHandle dependsOn)
{
if (!_isInitialized)
throw new InvalidOperationException("JobScheduler is not initialized");
var jobSlot = AllocateJobSlot();
var jobId = JobsUtility.GetNextJobId();
var version = Interlocked.Increment(ref _jobVersion);
ref var job = ref _jobPool![jobSlot];
job.Id = jobId;
job.Version = version;
job.State = 0; // Scheduled
job.JobType = JobType.ParallelFor;
job.ExecuteJobFunction = null;
job.ExecuteParallelJobFunction = executeFunction;
job.JobDataObject = jobData;
job.TotalIterations = totalIterations;
job.BatchSize = batchSize;
job.DependencyCount = dependsOn._id == 0 ? 0 : 1;
job.CompletedDependencies = 0;
job.AdditionalDependencies = null;
job.AdditionalDependencyCount = 0;
// Set up dependencies
if (dependsOn._id != 0)
{
job.Dependencies[0] = dependsOn._id;
}
_activeJobs.TryAdd(jobId, jobSlot);
// Check if job can be executed immediately
if (job.CanExecute)
{
_readyJobs.Enqueue(jobSlot);
_workAvailableEvent.Set();
}
return new JobHandle(jobId, version);
}
/// <summary>
/// Combines multiple job dependencies into a single handle.
/// </summary>
internal static JobHandle CombineDependencies(ReadOnlySpan<JobHandle> dependencies)
{
if (dependencies.Length == 0)
{
return JobHandle.Completed;
}
if (dependencies.Length == 1)
{
return dependencies[0];
}
// Filter out completed dependencies
var activeDeps = stackalloc JobHandle[dependencies.Length];
var activeCount = 0;
for (var i = 0; i < dependencies.Length; i++) for (var i = 0; i < dependencies.Length; i++)
{ {
if (dependencies[i]._id != 0 && !IsCompleted(dependencies[i])) var dependency = dependencies[i];
if (!dependency.IsValid)
{ {
activeDeps[activeCount++] = dependencies[i]; continue;
} }
}
if (activeCount == 0) lock (_lock)
return JobHandle.Completed;
if (activeCount == 1)
return activeDeps[0];
// Create a combined dependency job
var jobSlot = AllocateJobSlot();
var jobId = JobsUtility.GetNextJobId();
var version = Interlocked.Increment(ref _jobVersion);
ref var job = ref _jobPool![jobSlot];
job.Id = jobId;
job.Version = version;
job.State = 0; // Scheduled
job.JobType = JobType.Job; // Dependency-only job
job.ExecuteJobFunction = null; // No execution needed
job.ExecuteParallelJobFunction = null;
job.JobDataObject = null;
job.TotalIterations = 0;
job.BatchSize = 0;
job.DependencyCount = activeCount;
job.CompletedDependencies = 0;
// Set up dependencies
for (var i = 0; i < Math.Min(activeCount, 8); i++)
{
job.Dependencies[i] = activeDeps[i]._id;
}
// Handle additional dependencies if more than 8
if (activeCount > 8)
{
var additionalSize = activeCount - 8;
var handle = AllocationManager.GetAllocationHandle(Allocator.Temp);
job.AdditionalDependencies = (ulong*)handle.Alloc(handle.Allocator, (nuint)(sizeof(ulong) * additionalSize), sizeof(ulong), AllocationOption.None);
job.AdditionalDependencyCount = additionalSize;
for (var i = 0; i < additionalSize; i++)
{ {
job.AdditionalDependencies[i] = activeDeps[i + 8]._id; ref var depJobInfo = ref _jobInfoPool.GetElementReferenceAt(dependency._id, dependency._generation, out var exist);
if (!exist || Volatile.Read(ref Unsafe.As<JobStatus, int>(ref depJobInfo.status)) == (int)JobStatus.Completed)
{
continue;
}
if (depJobInfo.dependentCount >= JobInfo.MAX_DEPENDENTS)
{
// Too many dependents
// TODO: Handle this case properly
_jobDataAllocator.Free(jobInfo.pJobData);
return JobHandle.Invalid;
}
depJobInfo.dependentsID[depJobInfo.dependentCount] = id;
depJobInfo.dependentsGeneration[depJobInfo.dependentCount] = generation;
depJobInfo.dependentCount++;
} }
Interlocked.Increment(ref infoInPool.dependencyCount);
} }
_activeJobs.TryAdd(jobId, jobSlot); EnqueueJobIfReady(handle);
return new JobHandle(jobId, version); return handle;
} }
/// <summary> [MethodImpl(MethodImplOptions.AggressiveInlining)]
/// Checks if a job is completed. internal bool HasWork()
/// </summary>
internal static bool IsCompleted(JobHandle handle)
{ {
if (handle._id == 0) return !_jobQueue.IsEmpty || _workerThreads.Any(w => !w.LocalQueue.IsEmpty);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void WaitForWork()
{
_workSignal.Wait(_cts.Token);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal bool TryStealJob(int threadIndex, out JobHandle outHandle)
{
if (threadIndex >= 0 && threadIndex < _workerThreads.Length
&& _workerThreads[threadIndex].LocalQueue.TryDequeue(out outHandle))
{
return true; return true;
}
if (_activeJobs.TryGetValue(handle._id, out var jobSlot)) else if (_jobQueue.TryDequeue(out outHandle))
{ {
return _jobPool![jobSlot].IsCompleted; return true;
} }
return true; // Job not found, assume completed outHandle = JobHandle.Invalid;
return false;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal ref JobInfo GetJobInfoReference(JobHandle handle, out bool exist)
{
if (!handle.IsValid)
{
exist = false;
return ref Unsafe.NullRef<JobInfo>();
}
return ref _jobInfoPool.GetElementReferenceAt(handle._id, handle._generation, out exist);
}
internal void MarkJobComplete(JobHandle handle)
{
if (!handle.IsValid)
{
return;
}
ref var info = ref _jobInfoPool.GetElementReferenceAt(handle._id, handle._generation, out var exist);
if (!exist)
{
return;
}
if (Interlocked.CompareExchange(ref info.status, JobStatus.Completed, JobStatus.Running) != JobStatus.Running)
{
return;
}
var dependentsToNotify = stackalloc JobHandle[JobInfo.MAX_DEPENDENTS];
var dependentCount = 0;
lock (_lock)
{
dependentCount = info.dependentCount;
for (var i = 0; i < dependentCount; i++)
{
dependentsToNotify[i] = new JobHandle(info.dependentsID[i], info.dependentsGeneration[i]);
}
}
_jobDataAllocator.Free(info.pJobData);
_jobInfoPool.Remove(handle._id, handle._generation);
for (var i = 0; i < dependentCount; i++)
{
var depHandle = dependentsToNotify[i];
ref var depJobInfo = ref _jobInfoPool.GetElementReferenceAt(depHandle._id, depHandle._generation, out var depExist);
if (depExist && Interlocked.Decrement(ref depJobInfo.dependencyCount) == 0)
{
EnqueueJobIfReady(depHandle);
}
}
} }
/// <summary> /// <summary>
/// Blocks until the specified job completes. /// Schedules a single job for execution on a specified thread, with an optional dependency on another job.
/// </summary> /// </summary>
internal static void Complete(JobHandle handle) /// <typeparam name="T">The type of the job to execute. Must implement <see cref="IJob"/> and be unmanaged.</typeparam>
/// <param name="job">The job instance to be executed. The job data will be copied internally.</param>
/// <param name="threadIndex">The index of the thread that will execute the job. This is used to assign thread-specific data. Use -1 to allow any thread to execute the job.</param>
/// <param name="dependency">A <see cref="JobHandle"/> representing the dependencies that must be completed before this job can begin.
/// Use <see cref="JobHandle.Invalid"/> if there are no dependencies.</param>
/// <returns>A <see cref="JobHandle"/> that can be used to track the completion of the scheduled job.
/// Returns <see cref="JobHandle.Invalid"/> if the job data allocation fails.</returns>
public JobHandle Schedule<T>(ref T job, int threadIndex, JobHandle dependency)
where T : unmanaged, IJob
{ {
if (handle._id == 0) var jobData = _jobDataAllocator.Allocate(MemoryUtilities.SizeOf<T>(), MemoryUtilities.AlignOf<T>());
return; if (jobData == null)
while (!IsCompleted(handle))
{ {
// Try to help with work while waiting return JobHandle.Invalid;
if (_readyJobs.TryDequeue(out var jobSlot))
{
ExecuteJob(jobSlot);
}
else
{
Thread.Yield();
}
} }
}
private static int AllocateJobSlot() fixed (T* pJob = &job)
{
// Create a new JobData and add it to the SlotMap
var jobData = new JobData();
return _jobPool!.Add(jobData);
}
private static void WorkerThreadLoop(object? threadIndexObj)
{
var threadIndex = (int)threadIndexObj!;
while (!_isShuttingDown)
{ {
if (_readyJobs.TryDequeue(out var jobSlot)) MemoryUtilities.MemCpy(pJob, jobData, MemoryUtilities.SizeOf<T>());
{
ExecuteJob(jobSlot);
}
else
{
_workAvailableEvent.Wait(100); // Wait with timeout
_workAvailableEvent.Reset();
}
} }
}
private static void ExecuteJob(int jobSlot) var jobInfo = new JobInfo
{
ref var job = ref _jobPool![jobSlot];
// Mark as running
if (Interlocked.CompareExchange(ref job.State, 1, 0) != 0)
return; // Job already taken by another thread
try
{ {
if (job.ExecuteJobFunction != null) pJobData = jobData,
{ executeDelegate = &JobExecutor.Execute<T>,
if (job.JobType == JobType.Job)
{
// Execute IJob
job.ExecuteJobFunction(job.JobDataObject!);
}
}
else if (job.ExecuteParallelJobFunction != null)
{
if (job.JobType == JobType.ParallelFor)
{
// Execute IJobParallelFor
ExecuteParallelJob(ref job);
}
}
}
finally
{
// Mark as completed
Volatile.Write(ref job.State, 2);
// Clean up additional dependencies remainingBatches = 1,
if (job.AdditionalDependencies != null) threadIndex = threadIndex,
{
var handle = AllocationManager.GetAllocationHandle(Allocator.Temp);
handle.Free(handle.Allocator, job.AdditionalDependencies);
job.AdditionalDependencies = null;
}
// Remove from active jobs and notify dependent jobs jobRanges = JobRanges.Single,
_activeJobs.TryRemove(job.Id, out _);
NotifyDependentJobs(job.Id);
}
}
private static void ExecuteParallelJob(ref JobData job)
{
var batchSize = job.BatchSize > 0 ? job.BatchSize : Math.Max(1, job.TotalIterations / (_workerThreadCount * 4));
var currentIndex = 0;
var ranges = new JobRanges
{
JobIndex = 0,
BeginIndex = 0,
EndIndex = job.TotalIterations,
TotalLength = job.TotalIterations,
BatchSize = batchSize,
CurrentIndex = &currentIndex
}; };
var executeDelegate = job.ExecuteParallelJobFunction!; return CreateJobHandle(ref jobInfo, dependency);
var jobDataObject = job.JobDataObject!;
// Execute in parallel using available threads
Parallel.For(0, _workerThreadCount, threadIndex =>
{
executeDelegate(jobDataObject, ref ranges, threadIndex);
});
} }
// TODO: Optimize by maintaining a reverse dependency graph /// <summary>
private static void NotifyDependentJobs(ulong completedJobId) /// Schedules a single job for execution on a specified thread, with an optional dependency on another job.
/// </summary>
/// <typeparam name="T">The type of the job to execute. Must implement <see cref="IJob"/> and be unmanaged.</typeparam>
/// <param name="job">The job instance to be executed. The job data will be copied internally.</param>
/// <param name="threadIndex">The index of the thread that will execute the job. This is used to assign thread-specific data. Use -1 to allow any thread to execute the job.</param>
/// <returns>A <see cref="JobHandle"/> that can be used to track the completion of the scheduled job.
/// Returns <see cref="JobHandle.Invalid"/> if the job data allocation fails.</returns>
public JobHandle Schedule<T>(ref T job, int threadIndex)
where T : unmanaged, IJob
=> Schedule(ref job, threadIndex, JobHandle.Invalid);
/// <summary>
/// Schedules a parallel job for execution, dividing the workload into batches and distributing it across threads.
/// </summary>
/// <typeparam name="T">The type of the job to execute. Must implement <see cref="IJobParallelFor"/> and be unmanaged.</typeparam>
/// <param name="job">The job instance to be executed. The job data will be copied internally.</param>
/// <param name="totalIteration">The total number of iterations to be processed by the job.</param>
/// <param name="batchSize">The number of iterations to include in each batch.</param>
/// <param name="threadIndex">The index of the thread that will execute the job. This is used to assign thread-specific data. Use -1 to allow any thread to execute the job.</param>
/// <param name="dependency">A <see cref="JobHandle"/> representing the dependencies that must be completed before this job can begin.
/// Use <see cref="JobHandle.Invalid"/> if there are no dependencies.</param>
/// <returns>A <see cref="JobHandle"/> that can be used to track the completion of the scheduled job.
/// Returns <see cref="JobHandle.Invalid"/> if the job data allocation fails.</returns>
public JobHandle ScheduleParallel<T>(ref T job, int totalIteration, int batchSize, int threadIndex, JobHandle dependency)
where T : unmanaged, IJobParallelFor
{ {
// Scan for jobs that depend on this completed job var jobData = _jobDataAllocator.Allocate(MemoryUtilities.SizeOf<T>(), MemoryUtilities.AlignOf<T>());
for (var i = 0; i < _jobPool!.Count; i++) if (jobData == null)
{ {
ref var job = ref _jobPool[i]; return JobHandle.Invalid;
if (job.State == 0 && job.DependencyCount > 0) // Scheduled and has dependencies }
fixed (T* pJob = &job)
{
MemoryUtilities.MemCpy(pJob, jobData, MemoryUtilities.SizeOf<T>());
}
var optimalBatchSize = Math.Max(1, batchSize);
var totalBatches = (totalIteration + optimalBatchSize - 1) / optimalBatchSize;
var jobInfo = new JobInfo
{
pJobData = jobData,
executeDelegate = &JobExecutor.ExecuteParallel<T>,
remainingBatches = totalBatches,
threadIndex = threadIndex,
jobRanges = new()
{ {
var isDependent = false; currentIndex = 0,
batchSize = optimalBatchSize,
totalIteration = totalIteration,
},
};
// Check inline dependencies return CreateJobHandle(ref jobInfo, dependency);
for (var j = 0; j < Math.Min(job.DependencyCount, 8); j++) }
{
if (job.Dependencies[j] == completedJobId)
{
isDependent = true;
break;
}
}
// Check additional dependencies /// <summary>
if (!isDependent && job.AdditionalDependencies != null) /// Schedules a parallel job for execution, dividing the workload into batches and distributing it across threads.
{ /// </summary>
for (var j = 0; j < job.AdditionalDependencyCount; j++) /// <typeparam name="T">The type of the job to execute. Must implement <see cref="IJobParallelFor"/> and be unmanaged.</typeparam>
{ /// <param name="job">The job instance to be executed. The job data will be copied internally.</param>
if (job.AdditionalDependencies[j] == completedJobId) /// <param name="totalIteration">The total number of iterations to be processed by the job.</param>
{ /// <param name="batchSize">The number of iterations to include in each batch.</param>
isDependent = true; /// <param name="threadIndex">The index of the thread that will execute the job. This is used to assign thread-specific data. Use -1 to allow any thread to execute the job.</param>
break; /// <returns>A <see cref="JobHandle"/> that can be used to track the completion of the scheduled job.
} /// Returns <see cref="JobHandle.Invalid"/> if the job data allocation fails.</returns>
} public JobHandle ScheduleParallel<T>(ref T job, int totalIteration, int batchSize, int threadIndex)
} where T : unmanaged, IJobParallelFor
=> ScheduleParallel(ref job, totalIteration, batchSize, threadIndex, JobHandle.Invalid);
if (isDependent) /// <summary>
{ /// Combines multiple job dependencies into a single <see cref="JobHandle"/>.
var completedCount = Interlocked.Increment(ref job.CompletedDependencies); /// </summary>
if (completedCount >= job.DependencyCount) /// <param name="dependencies">A collection of <see cref="JobHandle"/> instances representing the dependencies to combine.</param>
{ /// <returns>A <see cref="JobHandle"/> that represents the combined dependencies. The returned handle can be used to ensure
_readyJobs.Enqueue(i); /// that all specified dependencies are completed before proceeding.</returns>
_workAvailableEvent.Set(); public JobHandle CombineDependencies(params ReadOnlySpan<JobHandle> dependencies)
} {
} var jobInfo = new JobInfo
{
pJobData = null,
executeDelegate = null,
remainingBatches = 1,
threadIndex = -1,
jobRanges = JobRanges.Single,
};
return CreateJobHandle(ref jobInfo, dependencies);
}
/// <summary>
/// Blocks the calling thread until the specified job is completed.
/// </summary>
/// <param name="handle">The handle of the job to wait for.</param>
public void WaitComplete(JobHandle handle)
{
if (!handle.IsValid)
{
return;
}
var spin = new SpinWait();
while (_jobInfoPool.TryGetElement(handle._id, handle._generation, out var jobInfo))
{
if (jobInfo.status == JobStatus.Completed)
{
return;
} }
spin.SpinOnce(-1);
} }
} }
public void Dispose()
{
if (_disposed)
{
return;
}
_cts.Cancel();
foreach (var worker in _workerThreads)
{
worker.Dispose();
}
_jobInfoPool.Clear();
_jobQueue.Clear();
_jobDataAllocator.Dispose();
_cts.Dispose();
_disposed = true;
GC.SuppressFinalize(this);
}
} }

108
Misaki.HighPerformance.Jobs/JobStruct.cs
View File

@@ -1,108 +0,0 @@
namespace Misaki.HighPerformance.Jobs;
/// <summary>
/// Internal job struct for IJob execution, similar to Unity's JobStruct pattern.
/// This provides the bridge between the job scheduler and user job implementations.
/// </summary>
/// <typeparam name="T">The job type implementing IJob.</typeparam>
internal struct JobStruct<T> where T : class, IJob
{
/// <summary>
/// Cached function delegate for this job type.
/// This avoids allocations during job scheduling.
/// </summary>
internal static readonly ExecuteJobDelegate ExecuteDelegate;
static JobStruct()
{
// Create and cache the function delegate
ExecuteDelegate = Execute;
}
/// <summary>
/// Executes the job. This method matches the ExecuteJobDelegate signature.
/// </summary>
/// <param name="jobData">The job data object.</param>
public static void Execute(object jobData)
{
var typedJobData = (T)jobData;
typedJobData.Execute();
}
/// <summary>
/// Schedules this job type for execution.
/// </summary>
/// <param name="jobData">The job data.</param>
/// <param name="dependsOn">Job handle this job depends on.</param>
/// <returns>A job handle for the scheduled job.</returns>
public static JobHandle Schedule(T jobData, JobHandle dependsOn = default)
{
return JobScheduler.ScheduleJob(jobData, ExecuteDelegate, JobType.Job, 0, 0, dependsOn);
}
}
/// <summary>
/// Internal job struct for IJobParallelFor execution, similar to Unity's ParallelForJobStruct.
/// This provides efficient parallel execution with work stealing.
/// </summary>
/// <typeparam name="T">The job type implementing IJobParallelFor.</typeparam>
internal struct ParallelForJobStruct<T> where T : class, IJobParallelFor
{
/// <summary>
/// Cached function delegate for this job type.
/// </summary>
internal static readonly ExecuteParallelJobDelegate ExecuteDelegate;
static ParallelForJobStruct()
{
// Create and cache the function delegate
ExecuteDelegate = Execute;
}
/// <summary>
/// Executes the parallel job using work stealing. This method matches the ExecuteParallelJobDelegate signature.
/// </summary>
/// <param name="jobData">The job data object.</param>
/// <param name="ranges">Job ranges for work distribution.</param>
/// <param name="jobIndex">Index of the current worker thread.</param>
public static unsafe void Execute(object jobData, ref JobRanges ranges, int jobIndex)
{
var typedJobData = (T)jobData;
while (true)
{
if (!JobsUtility.GetWorkStealingRange(ref ranges, jobIndex, out var begin, out var end))
break;
// Execute the batch
var endThatCompilerCanSeeWillNeverChange = end;
for (var i = begin; i < endThatCompilerCanSeeWillNeverChange; ++i)
{
typedJobData.Execute(i);
}
}
}
/// <summary>
/// Schedules this parallel job type for execution.
/// </summary>
/// <param name="jobData">The job data.</param>
/// <param name="arrayLength">Total number of iterations.</param>
/// <param name="innerLoopBatchCount">Batch size for each worker. If <= 0, an optimal batch size will be calculated.</param>
/// <param name="dependsOn">Job handle this job depends on.</param>
/// <returns>A job handle for the scheduled job.</returns>
public static JobHandle ScheduleParallel(T jobData, int arrayLength, int innerLoopBatchCount = 0, JobHandle dependsOn = default)
{
if (arrayLength <= 0)
throw new ArgumentException("Array length must be greater than 0", nameof(arrayLength));
// Calculate optimal batch size if not specified
if (innerLoopBatchCount <= 0)
{
var workerCount = Environment.ProcessorCount;
innerLoopBatchCount = Math.Max(1, arrayLength / (workerCount * 4));
}
return JobScheduler.ScheduleParallelJob(jobData, ExecuteDelegate, arrayLength, innerLoopBatchCount, dependsOn);
}
}

41
Misaki.HighPerformance.Jobs/JobsUtility.cs
View File

@@ -1,41 +0,0 @@
using System.Runtime.CompilerServices;
namespace Misaki.HighPerformance.Jobs;
/// <summary>
/// Utilities for job execution, similar to Unity's JobsUtility.
/// Provides low-level job management functions.
/// </summary>
internal static unsafe class JobsUtility
{
private static ulong s_nextJobId = 1;
/// <summary>
/// Gets the next unique job ID.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static ulong GetNextJobId()
{
return (ulong)Interlocked.Increment(ref Unsafe.As<ulong, long>(ref s_nextJobId));
}
/// <summary>
/// Implements work stealing for parallel jobs.
/// Returns false when no more work is available.
/// </summary>
/// <param name="ranges">The job ranges containing work distribution information.</param>
/// <param name="jobIndex">The index of the current worker thread.</param>
/// <param name="beginIndex">Output: The starting index for this work batch.</param>
/// <param name="endIndex">Output: The ending index for this work batch.</param>
/// <returns>True if work was acquired, false if no more work is available.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static bool GetWorkStealingRange(ref JobRanges ranges, int jobIndex, out int beginIndex, out int endIndex)
{
var currentIndex = Interlocked.Add(ref *ranges.CurrentIndex, ranges.BatchSize);
beginIndex = currentIndex - ranges.BatchSize;
endIndex = Math.Min(currentIndex, ranges.TotalLength);
return beginIndex < ranges.TotalLength;
}
}

95
Misaki.HighPerformance.Jobs/WorkerThread.cs Normal file
View File

@@ -0,0 +1,95 @@
using System.Collections.Concurrent;
namespace Misaki.HighPerformance.Jobs;
internal class WorkerThread : IDisposable
{
private readonly int _index;
private readonly Thread _thread;
private readonly ConcurrentQueue<JobHandle> _localQueue;
private readonly JobScheduler _scheduler;
private readonly Random _random;
internal ConcurrentQueue<JobHandle> LocalQueue => _localQueue;
public WorkerThread(int index, JobScheduler scheduler)
{
_index = index;
_localQueue = new();
_scheduler = scheduler;
_random = new Random(index * 9973 + Environment.TickCount);
_thread = new Thread(WorkLoop)
{
IsBackground = true,
Name = $"WorkerThread-{index}"
};
}
public void Start() => _thread.Start();
private unsafe void WorkLoop()
{
while (!_scheduler.IsCancellationRequested)
{
var handle = JobHandle.Invalid;
// Always try the local thread and main thread queue first.
if (!_localQueue.TryDequeue(out handle)
&& !_scheduler.TryStealJob(-1, out handle))
{
var randomIndex = _random.Next(0, _scheduler.WorkerCount);
if (_scheduler.TryStealJob(randomIndex, out var tempHandle))
{
handle = tempHandle;
}
}
ref var jobInfo = ref _scheduler.GetJobInfoReference(handle, out var exist);
if (exist)
{
Interlocked.CompareExchange(ref jobInfo.status, JobStatus.Running, JobStatus.Scheduled);
var executeDelegate = jobInfo.executeDelegate;
if (executeDelegate == null
|| executeDelegate(jobInfo.pJobData, ref jobInfo.jobRanges, ref jobInfo.remainingBatches, _index))
{
_scheduler.MarkJobComplete(handle);
}
}
else
{
var spinner = new SpinWait();
for (var i = 0; i < 25; i++)
{
spinner.SpinOnce(-1);
if (_scheduler.HasWork())
{
goto FoundWork;
}
}
try
{
_scheduler.WaitForWork();
}
catch (OperationCanceledException)
{
continue;
}
}
FoundWork:
;
}
}
public void Dispose()
{
_thread.Join();
_localQueue.Clear();
}
}

6
Misaki.HighPerformance.LowLevel/AssemblyInfo.cs
View File

@@ -1,6 +1,6 @@
global using static Misaki.HighPerformance.LowLevel.Helpers.MemoryUtilities; global using static Misaki.HighPerformance.LowLevel.Helpers.MemoryUtilities;
global using unsafe AllocFunc = delegate* unmanaged<void*, nuint, nuint, Misaki.HighPerformance.LowLevel.Buffer.AllocationOption, void*>; global using unsafe AllocFunc = delegate*<void*, nuint, nuint, Misaki.HighPerformance.LowLevel.Buffer.AllocationOption, void*>;
global using unsafe FreeFunc = delegate* unmanaged<void*, void*, void>; global using unsafe FreeFunc = delegate*<void*, void*, void>;
global using unsafe ReallocFunc = delegate* unmanaged<void*, void*, nuint, nuint, void*>; global using unsafe ReallocFunc = delegate*<void*, void*, nuint, nuint, void*>;

38
Misaki.HighPerformance.LowLevel/Buffer/AllocationManager.cs
View File

@@ -61,11 +61,10 @@ public static unsafe class AllocationManager
public void Init(uint initialSize) public void Init(uint initialSize)
{ {
_arena = new DynamicArena(initialSize); _arena = new(initialSize);
_handle = new AllocationHandle(Unsafe.AsPointer(ref this), &Allocate, &Reallocate, &FreeBlock); _handle = new(Unsafe.AsPointer(ref this), &Allocate, &Reallocate, &FreeBlock);
} }
[UnmanagedCallersOnly]
private static void* Allocate(void* instance, nuint size, nuint alignment, AllocationOption allocationOption) private static void* Allocate(void* instance, nuint size, nuint alignment, AllocationOption allocationOption)
{ {
var selfPtr = (ArenaAllocator*)instance; var selfPtr = (ArenaAllocator*)instance;
@@ -74,7 +73,6 @@ public static unsafe class AllocationManager
return ptr; return ptr;
} }
[UnmanagedCallersOnly]
private static void* Reallocate(void* instance, void* ptr, nuint size, nuint alignment) private static void* Reallocate(void* instance, void* ptr, nuint size, nuint alignment)
{ {
var selfPtr = (ArenaAllocator*)instance; var selfPtr = (ArenaAllocator*)instance;
@@ -84,7 +82,6 @@ public static unsafe class AllocationManager
return newPtr; return newPtr;
} }
[UnmanagedCallersOnly]
private static void FreeBlock(void* instance, void* ptr) private static void FreeBlock(void* instance, void* ptr)
{ {
// The arena allocator does not free individual blocks, as it manages memory in chunks. // The arena allocator does not free individual blocks, as it manages memory in chunks.
@@ -109,10 +106,9 @@ public static unsafe class AllocationManager
public void Init() public void Init()
{ {
_handle = new AllocationHandle(Unsafe.AsPointer(ref this), &Allocate, &Reallocate, &FreeBlock); _handle = new(Unsafe.AsPointer(ref this), &Allocate, &Reallocate, &FreeBlock);
} }
[UnmanagedCallersOnly]
private static void* Allocate(void* instance, nuint size, nuint alignment, AllocationOption allocationOption) private static void* Allocate(void* instance, nuint size, nuint alignment, AllocationOption allocationOption)
{ {
var ptr = AlignedAlloc(size, alignment); var ptr = AlignedAlloc(size, alignment);
@@ -130,7 +126,6 @@ public static unsafe class AllocationManager
return ptr; return ptr;
} }
[UnmanagedCallersOnly]
private static void* Reallocate(void* instance, void* ptr, nuint size, nuint alignment) private static void* Reallocate(void* instance, void* ptr, nuint size, nuint alignment)
{ {
var newPtr = AlignedRealloc(ptr, size, alignment); var newPtr = AlignedRealloc(ptr, size, alignment);
@@ -138,7 +133,6 @@ public static unsafe class AllocationManager
return newPtr; return newPtr;
} }
[UnmanagedCallersOnly]
private static void FreeBlock(void* instance, void* ptr) private static void FreeBlock(void* instance, void* ptr)
{ {
AlignedFree(ptr); AlignedFree(ptr);
@@ -154,16 +148,6 @@ public static unsafe class AllocationManager
private static bool s_debugLayer; private static bool s_debugLayer;
private static ConcurrentDictionary<nint, AllocationInfo>? s_allocated; private static ConcurrentDictionary<nint, AllocationInfo>? s_allocated;
/// <summary>
/// Gets a reference to the allocation handle for temporary allocations.
/// </summary>
public static ref AllocationHandle TempHandle => ref s_arenaAllocator->Handle;
/// <summary>
/// Gets a reference to the persistent allocation handle.
/// </summary>
public static ref AllocationHandle PersistentHandle => ref s_persistentAllocator->Handle;
static AllocationManager() static AllocationManager()
{ {
s_arenaAllocator = (ArenaAllocator*)NativeMemory.Alloc((nuint)sizeof(ArenaAllocator)); s_arenaAllocator = (ArenaAllocator*)NativeMemory.Alloc((nuint)sizeof(ArenaAllocator));
@@ -176,6 +160,7 @@ public static unsafe class AllocationManager
/// <summary> /// <summary>
/// Enables the debug layer, allowing additional diagnostic information to be collected. /// Enables the debug layer, allowing additional diagnostic information to be collected.
/// </summary> /// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void EnableDebugLayer() public static void EnableDebugLayer()
{ {
s_debugLayer = true; s_debugLayer = true;
@@ -188,14 +173,15 @@ public static unsafe class AllocationManager
/// <param name="allocator">The allocator type for which to retrieve the allocation handle.</param> /// <param name="allocator">The allocator type for which to retrieve the allocation handle.</param>
/// <returns>A reference to the allocation handle associated with the specified allocator type.</returns> /// <returns>A reference to the allocation handle associated with the specified allocator type.</returns>
/// <exception cref="ArgumentException"></exception> /// <exception cref="ArgumentException"></exception>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static ref AllocationHandle GetAllocationHandle(Allocator allocator) public static ref AllocationHandle GetAllocationHandle(Allocator allocator)
{ {
switch (allocator) switch (allocator)
{ {
case Allocator.Temp: case Allocator.Temp:
return ref TempHandle; return ref s_arenaAllocator->Handle;
case Allocator.Persistent: case Allocator.Persistent:
return ref PersistentHandle; return ref s_persistentAllocator->Handle;
default: default:
throw new ArgumentException("Target allocator type does not support custom allocation.", nameof(allocator)); throw new ArgumentException("Target allocator type does not support custom allocation.", nameof(allocator));
} }
@@ -258,6 +244,7 @@ public static unsafe class AllocationManager
/// Removes the specified memory allocation from the tracking system. /// Removes the specified memory allocation from the tracking system.
/// </summary> /// </summary>
/// <param name="ptr">A pointer to the memory allocation to untrack.</param> /// <param name="ptr">A pointer to the memory allocation to untrack.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void UntrackAllocation(void* ptr) public static void UntrackAllocation(void* ptr)
{ {
if (s_allocated == null) if (s_allocated == null)
@@ -268,6 +255,15 @@ public static unsafe class AllocationManager
s_allocated.Remove((nint)ptr, out _); s_allocated.Remove((nint)ptr, out _);
} }
/// <summary>
/// Resets the temporary memory allocator, clearing all allocated memory.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void ResetTempAllocator()
{
s_arenaAllocator->Reset();
}
/// <summary> /// <summary>
/// Disposes of the AllocationManager, freeing all allocated memory and resources. /// Disposes of the AllocationManager, freeing all allocated memory and resources.
/// </summary> /// </summary>

204
Misaki.HighPerformance.LowLevel/Buffer/FreeList.cs
View File

@@ -1,5 +1,4 @@
using Misaki.HighPerformance.LowLevel.Helpers; using System.Runtime.CompilerServices;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
namespace Misaki.HighPerformance.LowLevel.Buffer; namespace Misaki.HighPerformance.LowLevel.Buffer;
@@ -30,9 +29,6 @@ namespace Misaki.HighPerformance.LowLevel.Buffer;
[StructLayout(LayoutKind.Explicit, Size = 256)] // Cache line aligned to prevent false sharing [StructLayout(LayoutKind.Explicit, Size = 256)] // Cache line aligned to prevent false sharing
public unsafe struct FreeList : IDisposable public unsafe struct FreeList : IDisposable
{ {
/// <summary>
/// Node structure for the lock-free free list with size information.
/// </summary>
[StructLayout(LayoutKind.Sequential)] [StructLayout(LayoutKind.Sequential)]
private struct FreeNode private struct FreeNode
{ {
@@ -40,9 +36,6 @@ public unsafe struct FreeList : IDisposable
public nuint size; public nuint size;
} }
/// <summary>
/// Memory chunk that contains variable-size blocks.
/// </summary>
[StructLayout(LayoutKind.Sequential)] [StructLayout(LayoutKind.Sequential)]
private struct MemoryChunk private struct MemoryChunk
{ {
@@ -52,20 +45,35 @@ public unsafe struct FreeList : IDisposable
public nuint used; // Amount of memory used in this chunk public nuint used; // Amount of memory used in this chunk
} }
/// <summary> [StructLayout(LayoutKind.Explicit, Size = 32)]
/// Size bucket for different allocation sizes.
/// </summary>
[StructLayout(LayoutKind.Sequential)]
private struct SizeBucket private struct SizeBucket
{ {
public nint freeHead; // Free list head for this size [FieldOffset(0)]
public nuint blockSize; // Fixed size for this bucket
public long freeCount; // Number of free blocks public long freeCount; // Number of free blocks
[FieldOffset(8)]
public nint freeHead; // Free list head for this size
[FieldOffset(16)]
public nuint blockSize; // Fixed size for this bucket
[FieldOffset(24)]
public int creationLock;
}
[StructLayout(LayoutKind.Explicit, Size = 24)]
private struct BlockHeader
{
// Ensure the size is fixed across x86 and x64
[FieldOffset(0)]
public MemoryChunk* ownerChunk;
[FieldOffset(8)]
public nuint blockSize;
[FieldOffset(16)]
public ulong magicNumber;
} }
private const int _MAX_BUCKETS = 16; // Number of size buckets private const int _MAX_BUCKETS = 16; // Number of size buckets
private const nuint _MIN_BLOCK_SIZE = 16; // Minimum block size private const nuint _MIN_BLOCK_SIZE = 16; // Minimum block size
private const nuint _DEFAULT_CHUNK_SIZE = 64 * 1024; // 64KB chunks private const nuint _DEFAULT_CHUNK_SIZE = 64 * 1024; // 64KB chunks
private const ulong _MAGIC_NUMBER = 0xDEADBEEFDEADBEEF; // For validating blocks
[FieldOffset(0)] [FieldOffset(0)]
private fixed byte _buckets[_MAX_BUCKETS * 32]; // SizeBucket array (32 bytes per bucket) private fixed byte _buckets[_MAX_BUCKETS * 32]; // SizeBucket array (32 bytes per bucket)
@@ -77,10 +85,10 @@ public unsafe struct FreeList : IDisposable
private MemoryChunk* _chunks; // 8 private MemoryChunk* _chunks; // 8
[FieldOffset(648)] [FieldOffset(648)]
private nuint _chunkSize; // 8 private readonly nuint _chunkSize; // 8
[FieldOffset(656)] [FieldOffset(656)]
private nuint _alignment; // 8 private readonly nuint _alignment; // 8
[FieldOffset(664)] [FieldOffset(664)]
private long _totalAllocatedBytes; // 8 private long _totalAllocatedBytes; // 8
@@ -124,13 +132,17 @@ public unsafe struct FreeList : IDisposable
/// </summary> /// </summary>
/// <param name="alignment">Alignment requirement for blocks (must be power of 2).</param> /// <param name="alignment">Alignment requirement for blocks (must be power of 2).</param>
/// <param name="chunkSize">Size of memory chunks to allocate (default: 64KB).</param> /// <param name="chunkSize">Size of memory chunks to allocate (default: 64KB).</param>
public FreeList(nuint alignment = 8, nuint chunkSize = _DEFAULT_CHUNK_SIZE) public FreeList(nuint alignment, nuint chunkSize = _DEFAULT_CHUNK_SIZE)
{ {
if (alignment == 0 || (alignment & (alignment - 1)) != 0) if (alignment == 0 || (alignment & (alignment - 1)) != 0)
{
throw new ArgumentException("Alignment must be a power of 2", nameof(alignment)); throw new ArgumentException("Alignment must be a power of 2", nameof(alignment));
}
if (chunkSize < 1024) if (chunkSize < 1024)
{
throw new ArgumentException("Chunk size must be at least 1KB", nameof(chunkSize)); throw new ArgumentException("Chunk size must be at least 1KB", nameof(chunkSize));
}
_alignment = alignment; _alignment = alignment;
_chunkSize = chunkSize; _chunkSize = chunkSize;
@@ -144,9 +156,6 @@ public unsafe struct FreeList : IDisposable
InitializeBuckets(); InitializeBuckets();
} }
/// <summary>
/// Initializes the size buckets with exponential sizes.
/// </summary>
private readonly void InitializeBuckets() private readonly void InitializeBuckets()
{ {
var buckets = GetBuckets(); var buckets = GetBuckets();
@@ -161,9 +170,6 @@ public unsafe struct FreeList : IDisposable
} }
} }
/// <summary>
/// Gets a pointer to the size buckets array.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
private readonly SizeBucket* GetBuckets() private readonly SizeBucket* GetBuckets()
{ {
@@ -173,11 +179,6 @@ public unsafe struct FreeList : IDisposable
} }
} }
/// <summary>
/// Finds the appropriate bucket for the given size.
/// </summary>
/// <param name="size">Size to find bucket for.</param>
/// <returns>Bucket index, or -1 if too large for buckets.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
private readonly int FindBucket(nuint size) private readonly int FindBucket(nuint size)
{ {
@@ -186,7 +187,9 @@ public unsafe struct FreeList : IDisposable
for (var i = 0; i < _MAX_BUCKETS; i++) for (var i = 0; i < _MAX_BUCKETS; i++)
{ {
if (size <= buckets[i].blockSize) if (size <= buckets[i].blockSize)
{
return i; return i;
}
} }
return -1; // Size too large for buckets return -1; // Size too large for buckets
@@ -200,19 +203,25 @@ public unsafe struct FreeList : IDisposable
/// <param name="allocationOption">Options for allocation (e.g., clear memory).</param> /// <param name="allocationOption">Options for allocation (e.g., clear memory).</param>
/// <returns>MemoryBlock containing allocated memory, or Invalid if allocation fails.</returns> /// <returns>MemoryBlock containing allocated memory, or Invalid if allocation fails.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
public MemoryBlock Allocate(nuint size, nuint alignment = 0, AllocationOption allocationOption = AllocationOption.None) public void* Allocate(nuint size, nuint alignment, AllocationOption allocationOption = AllocationOption.None)
{ {
if (_disposed != 0 || size == 0) if (_disposed != 0 || size == 0)
return MemoryBlock.Invalid; {
return null;
}
if (alignment == 0) if (alignment == 0)
{
alignment = _alignment; alignment = _alignment;
}
// Align size to alignment boundary // Align size to alignment boundary
var alignedSize = (size + alignment - 1) & ~(alignment - 1); var alignedSize = (size + alignment - 1) & ~(alignment - 1);
alignedSize = Math.Max(alignedSize, _MIN_BLOCK_SIZE); alignedSize = Math.Max(alignedSize, _MIN_BLOCK_SIZE);
var bucketIndex = FindBucket(alignedSize); var totalSize = alignedSize + (nuint)sizeof(BlockHeader);
var bucketIndex = FindBucket(totalSize);
void* ptr = null; void* ptr = null;
if (bucketIndex >= 0) if (bucketIndex >= 0)
@@ -233,22 +242,24 @@ public unsafe struct FreeList : IDisposable
if (ptr == null) if (ptr == null)
{ {
// Fallback to direct allocation from chunk // Fallback to direct allocation from chunk
ptr = AllocateFromChunk(alignedSize, alignment); ptr = AllocateFromChunk(totalSize, alignment);
} }
if (ptr != null) if (ptr != null)
{ {
Interlocked.Add(ref _totalAllocatedBytes, (long)alignedSize); var header = (BlockHeader*)ptr;
Interlocked.Add(ref _totalAllocatedBytes, (long)header->blockSize);
var pUserData = (byte*)ptr + sizeof(BlockHeader);
if (allocationOption.HasFlag(AllocationOption.Clear)) if (allocationOption.HasFlag(AllocationOption.Clear))
{ {
MemClear(ptr, alignedSize); MemClear(pUserData, alignedSize);
} }
return new MemoryBlock(ptr, alignedSize, alignment); return pUserData;
} }
return MemoryBlock.Invalid; return null;
} }
/// <summary> /// <summary>
@@ -256,26 +267,39 @@ public unsafe struct FreeList : IDisposable
/// </summary> /// </summary>
/// <param name="block">MemoryBlock to free.</param> /// <param name="block">MemoryBlock to free.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Free(MemoryBlock block) public void Free(void* ptr)
{ {
if (!block.IsValid || _disposed != 0) if (_disposed != 0 || ptr == null)
return;
if (!IsValidBlock(block.Ptr))
return; // Invalid pointer, ignore
var bucketIndex = FindBucket(block.Size);
if (bucketIndex >= 0)
{ {
PushToBucket(bucketIndex, block.Ptr, block.Size); return;
} }
Interlocked.Add(ref _totalAllocatedBytes, -(long)block.Size); var blockStartPtr = (byte*)ptr - sizeof(BlockHeader);
var header = (BlockHeader*)blockStartPtr;
if (header->magicNumber != _MAGIC_NUMBER)
{
return;
}
var chuck = header->ownerChunk;
if (chuck == null)
{
return;
}
var bucketIndex = FindBucket(header->blockSize);
if (bucketIndex >= 0)
{
PushToBucket(bucketIndex, blockStartPtr, header->blockSize);
}
Interlocked.Add(ref _totalAllocatedBytes, -(long)header->blockSize);
header->ownerChunk = null;
header->blockSize = 0;
header->magicNumber = 0;
} }
/// <summary>
/// Tries to pop a free block from the specified bucket.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
private readonly void* TryPopFromBucket(int bucketIndex) private readonly void* TryPopFromBucket(int bucketIndex)
{ {
@@ -289,7 +313,9 @@ public unsafe struct FreeList : IDisposable
{ {
head = bucket->freeHead; head = bucket->freeHead;
if (head == 0) if (head == 0)
{
return null; return null;
}
headPtr = (FreeNode*)head; headPtr = (FreeNode*)head;
newHead = (nint)headPtr->next; newHead = (nint)headPtr->next;
@@ -300,9 +326,6 @@ public unsafe struct FreeList : IDisposable
return (void*)head; return (void*)head;
} }
/// <summary>
/// Pushes a block to the specified bucket's free list.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
private readonly void PushToBucket(int bucketIndex, void* ptr, nuint size) private readonly void PushToBucket(int bucketIndex, void* ptr, nuint size)
{ {
@@ -323,29 +346,45 @@ public unsafe struct FreeList : IDisposable
Interlocked.Increment(ref bucket->freeCount); Interlocked.Increment(ref bucket->freeCount);
} }
/// <summary> [MethodImpl(MethodImplOptions.AggressiveInlining)]
/// Creates new blocks for the specified bucket. private static void AssignBlockHeader(BlockHeader* header, MemoryChunk* ownerChunk, nuint blockSize)
/// </summary> {
header->ownerChunk = ownerChunk;
header->blockSize = blockSize;
header->magicNumber = _MAGIC_NUMBER;
}
private bool TryCreateBlocksForBucket(int bucketIndex) private bool TryCreateBlocksForBucket(int bucketIndex)
{ {
while (Interlocked.CompareExchange(ref _chunkCreationLock, 1, 0) != 0) var buckets = GetBuckets();
var bucket = &buckets[bucketIndex];
while (Interlocked.CompareExchange(ref bucket->creationLock, 1, 0) != 0)
{ {
Thread.SpinWait(1); Thread.SpinWait(1);
} }
try try
{ {
var buckets = GetBuckets(); if (bucket->freeHead != 0)
var blockSize = buckets[bucketIndex].blockSize; {
return true; // Another thread did the work for us!
}
var blockSize = bucket->blockSize;
var blocksToCreate = Math.Min(_chunkSize / blockSize, 256); // Limit number of blocks var blocksToCreate = Math.Min(_chunkSize / blockSize, 256); // Limit number of blocks
if (blocksToCreate == 0) if (blocksToCreate == 0)
{
return false; return false;
}
var totalSize = blocksToCreate * blockSize; var totalSize = blocksToCreate * blockSize;
var memory = (byte*)AlignedAlloc(totalSize, _alignment); var memory = (byte*)AlignedAlloc(totalSize, _alignment);
if (memory == null) if (memory == null)
{
return false; return false;
}
var chunk = (MemoryChunk*)_chunkArena.Allocate(SizeOf<MemoryChunk>(), AlignOf<MemoryChunk>(), AllocationOption.None); var chunk = (MemoryChunk*)_chunkArena.Allocate(SizeOf<MemoryChunk>(), AlignOf<MemoryChunk>(), AllocationOption.None);
if (chunk == null) if (chunk == null)
@@ -363,21 +402,20 @@ public unsafe struct FreeList : IDisposable
// Add all blocks to the bucket's free list // Add all blocks to the bucket's free list
for (nuint i = 0; i < blocksToCreate; i++) for (nuint i = 0; i < blocksToCreate; i++)
{ {
var blockPtr = memory + (i * blockSize); var blockStartPtr = memory + (i * blockSize);
PushToBucket(bucketIndex, blockPtr, blockSize);
AssignBlockHeader((BlockHeader*)blockStartPtr, chunk, blockSize);
PushToBucket(bucketIndex, blockStartPtr, blockSize);
} }
return true; return true;
} }
finally finally
{ {
Interlocked.Exchange(ref _chunkCreationLock, 0); Interlocked.Exchange(ref bucket->creationLock, 0);
} }
} }
/// <summary>
/// Allocates memory directly from a chunk (for large allocations).
/// </summary>
private void* AllocateFromChunk(nuint size, nuint alignment) private void* AllocateFromChunk(nuint size, nuint alignment)
{ {
while (Interlocked.CompareExchange(ref _chunkCreationLock, 1, 0) != 0) while (Interlocked.CompareExchange(ref _chunkCreationLock, 1, 0) != 0)
@@ -397,9 +435,11 @@ public unsafe struct FreeList : IDisposable
if (totalNeeded <= available) if (totalNeeded <= available)
{ {
var ptr = chunk->memory + alignedOffset; var blockStartPtr = chunk->memory + alignedOffset;
chunk->used += totalNeeded;
return ptr; // Write the header and return the pointer WITH the header
AssignBlockHeader((BlockHeader*)blockStartPtr, chunk, size);
return blockStartPtr;
} }
chunk = chunk->next; chunk = chunk->next;
@@ -409,7 +449,9 @@ public unsafe struct FreeList : IDisposable
var newChunkSize = Math.Max(_chunkSize, size + alignment); var newChunkSize = Math.Max(_chunkSize, size + alignment);
var newMemory = (byte*)AlignedAlloc(newChunkSize, alignment); var newMemory = (byte*)AlignedAlloc(newChunkSize, alignment);
if (newMemory == null) if (newMemory == null)
{
return null; return null;
}
var newChunk = (MemoryChunk*)_chunkArena.Allocate(SizeOf<MemoryChunk>(), AlignOf<MemoryChunk>(), AllocationOption.None); var newChunk = (MemoryChunk*)_chunkArena.Allocate(SizeOf<MemoryChunk>(), AlignOf<MemoryChunk>(), AllocationOption.None);
if (newChunk == null) if (newChunk == null)
@@ -424,6 +466,8 @@ public unsafe struct FreeList : IDisposable
newChunk->next = _chunks; newChunk->next = _chunks;
_chunks = newChunk; _chunks = newChunk;
// Write the header and return the pointer WITH the header
AssignBlockHeader((BlockHeader*)newMemory, newChunk, size);
return newMemory; return newMemory;
} }
finally finally
@@ -432,27 +476,6 @@ public unsafe struct FreeList : IDisposable
} }
} }
/// <summary>
/// Validates that a pointer belongs to one of our memory chunks.
/// </summary>
private readonly bool IsValidBlock(void* ptr)
{
var chunk = _chunks;
while (chunk != null)
{
var chunkStart = (nuint)chunk->memory;
var chunkEnd = chunkStart + chunk->size;
var ptrValue = (nuint)ptr;
if (ptrValue >= chunkStart && ptrValue < chunkEnd)
return true;
chunk = chunk->next;
}
return false;
}
/// <summary> /// <summary>
/// Disposes the free list and frees all allocated memory. /// Disposes the free list and frees all allocated memory.
/// Note: This method is NOT thread-safe by design as requested. /// Note: This method is NOT thread-safe by design as requested.
@@ -467,10 +490,11 @@ public unsafe struct FreeList : IDisposable
{ {
var next = chunk->next; var next = chunk->next;
AlignedFree(chunk->memory); AlignedFree(chunk->memory);
MemoryUtilities.Free(chunk);
chunk = next; chunk = next;
} }
_chunkArena.Dispose();
_chunks = null; _chunks = null;
_totalAllocatedBytes = 0; _totalAllocatedBytes = 0;
_totalFreeBytes = 0; _totalFreeBytes = 0;

13
Misaki.HighPerformance.LowLevel/Buffer/MemoryBlock.cs
View File

@@ -16,6 +16,14 @@ public unsafe readonly struct MemoryBlock
get; get;
} }
/// <summary>
/// The heap from which the memory was allocated.
/// </summary>
public void* Heap
{
get;
}
/// <summary> /// <summary>
/// Size of the allocated memory in bytes. /// Size of the allocated memory in bytes.
/// </summary> /// </summary>
@@ -43,9 +51,10 @@ public unsafe readonly struct MemoryBlock
/// <param name="ptr">Pointer to the allocated memory.</param> /// <param name="ptr">Pointer to the allocated memory.</param>
/// <param name="size">Size of the allocated memory.</param> /// <param name="size">Size of the allocated memory.</param>
/// <param name="alignment">Alignment of the allocated memory.</param> /// <param name="alignment">Alignment of the allocated memory.</param>
public MemoryBlock(void* ptr, nuint size, nuint alignment) public MemoryBlock(void* ptr, void* heap, nuint size, nuint alignment)
{ {
Ptr = ptr; Ptr = ptr;
Heap = heap;
Size = size; Size = size;
Alignment = alignment; Alignment = alignment;
} }
@@ -53,7 +62,7 @@ public unsafe readonly struct MemoryBlock
/// <summary> /// <summary>
/// Creates an invalid MemoryBlock. /// Creates an invalid MemoryBlock.
/// </summary> /// </summary>
public static MemoryBlock Invalid => new(null, 0, 0); public static MemoryBlock Invalid => new(null, null, 0, 0);
public Span<T> AsSpan<T>() public Span<T> AsSpan<T>()
where T : unmanaged where T : unmanaged

2
Misaki.HighPerformance.Test/Benchmark/CollectionBenchmark.cs
View File

@@ -34,7 +34,7 @@ public unsafe class CollectionBenchmark
array[i] = i; array[i] = i;
} }
((ArenaAllocator*)AllocationManager.TempHandle.Allocator)->Reset(); AllocationManager.ResetTempAllocator();
} }
[Benchmark] [Benchmark]

62
Misaki.HighPerformance.Test/Benchmark/ParallelNoiseBenchmark.cs
View File

@@ -1,4 +1,5 @@
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using Misaki.HighPerformance.Jobs;
using Misaki.HighPerformance.LowLevel.Buffer; using Misaki.HighPerformance.LowLevel.Buffer;
using Misaki.HighPerformance.LowLevel.Collections; using Misaki.HighPerformance.LowLevel.Collections;
using Misaki.HighPerformance.Test.Jobs; using Misaki.HighPerformance.Test.Jobs;
@@ -9,61 +10,62 @@ namespace Misaki.HighPerformance.Test.Benchmark;
[MemoryDiagnoser] [MemoryDiagnoser]
public class ParallelNoiseBenchmark public class ParallelNoiseBenchmark
{ {
private const int _WIDTH = 512; private const int _WIDTH = 64;
private const int _HEIGHT = 512; private const int _HEIGHT = 64;
private const int _LENGTH = _WIDTH * _HEIGHT; private const int _LENGTH = _WIDTH * _HEIGHT;
//[GlobalSetup] internal JobScheduler _jobScheduler = null!;
//public void Setup() private UnsafeArray<float> _buffers;
//{
// JobScheduler.Initialize();
//}
//[GlobalCleanup] [GlobalSetup]
//public void Cleanup() public void Setup()
//{ {
// JobScheduler.Shutdown(); _jobScheduler = new JobScheduler(Environment.ProcessorCount - 1);
//} _buffers = new UnsafeArray<float>(_LENGTH, Allocator.Persistent);
}
//[Benchmark] [GlobalCleanup]
//public void JobSystem() public void Cleanup()
//{ {
// using var buffers = new UnsafeArray<float>(_LENGTH, Allocator.Persistent, AllocationOption.None); _jobScheduler.Dispose();
// var job = new NoiseJob() _buffers.Dispose();
// { }
// buffers = buffers,
// width = _WIDTH,
// height = _HEIGHT
// };
// var handle = job.Schedule(_LENGTH, 64); [Benchmark]
// handle.Complete(); public void JobSystem()
//} {
var job = new NoiseJob()
{
buffers = _buffers,
width = _WIDTH,
height = _HEIGHT
};
var handle = _jobScheduler.ScheduleParallel(ref job, _LENGTH, 64, -1);
_jobScheduler.WaitComplete(handle);
}
[Benchmark] [Benchmark]
public void ParallelFor() public void ParallelFor()
{ {
using var buffers = new UnsafeArray<float>(_LENGTH, Allocator.Persistent, AllocationOption.None);
Parallel.For(0, _LENGTH, i => Parallel.For(0, _LENGTH, i =>
{ {
var x = i % _WIDTH; var x = i % _WIDTH;
var y = i / _HEIGHT; var y = i / _HEIGHT;
var uv = new Vector2(x, y); var uv = new Vector2(x, y);
buffers[i] = NoiseJob.GradientNoise(uv); _buffers[i] = NoiseJob.GradientNoise(uv);
}); });
} }
[Benchmark(Baseline = true)] [Benchmark(Baseline = true)]
public void For() public void For()
{ {
using var buffers = new UnsafeArray<float>(_LENGTH, Allocator.Persistent, AllocationOption.None);
for (var i = 0; i < _LENGTH; i++) for (var i = 0; i < _LENGTH; i++)
{ {
var x = i % _WIDTH; var x = i % _WIDTH;
var y = i / _HEIGHT; var y = i / _HEIGHT;
var uv = new Vector2(x, y); var uv = new Vector2(x, y);
buffers[i] = NoiseJob.GradientNoise(uv); _buffers[i] = NoiseJob.GradientNoise(uv);
} }
} }
} }

184
Misaki.HighPerformance.Test/Jobs/JobSystemExample.cs
View File

@@ -1,184 +0,0 @@
using Misaki.HighPerformance.Jobs;
using Misaki.HighPerformance.LowLevel.Buffer;
using Misaki.HighPerformance.LowLevel.Collections;
namespace Misaki.HighPerformance.Test.Jobs;
/// <summary>
/// Simple job that adds a value to each element in an array.
/// </summary>
public unsafe class AddValueJob : IJobParallelFor
{
public float* Data;
public float Value;
public void Execute(int index)
{
Data[index] += Value;
}
}
/// <summary>
/// Simple job that multiplies each element in an array by a value.
/// </summary>
public unsafe class MultiplyJob : IJobParallelFor
{
public float* Data;
public float Multiplier;
public void Execute(int index)
{
Data[index] *= Multiplier;
}
}
/// <summary>
/// Simple job that computes the sum of an array (single-threaded).
/// </summary>
/// <remarks>
/// This job uses the Kahan summation algorithm to reduce numerical error.
/// </remarks>
public unsafe class KahanSumJob : IJob
{
public float* Data;
public int Length;
public float* Result;
public void Execute()
{
var sum = 0f;
var c = 0f; // Compensation for lost low-order bits
for (var i = 0; i < Length; i++)
{
var y = Data[i] - c; // So far, so good: c is zero
var t = sum + y; // Alas, sum is big, y small, so low-order digits of y are lost
c = (t - sum) - y; // (t - sum) cancels the high-order part of y; subtracting y recovers negative (low part of y)
sum = t; // Algebraically, c should always be zero. Beware overly-clever compilers!
}
*Result = sum;
}
}
/// <summary>
/// Example program demonstrating the job system with dependencies.
/// </summary>
public static class JobSystemExample
{
public static unsafe void RunExample()
{
Console.WriteLine("=== Job System Example ===");
const int arraySize = 10000;
// Create test data
using var array = new UnsafeArray<float>(arraySize, Allocator.Persistent);
// Initialize with values 1, 2, 3, ...
for (var i = 0; i < arraySize; i++)
{
array[i] = i + 1;
}
Console.WriteLine($"Initial sum: {ComputeSum((float*)array.GetUnsafePtr(), arraySize)}");
// Job 1: Add 10 to each element
var addJob = new AddValueJob
{
Data = (float*)array.GetUnsafePtr(),
Value = 10f
};
// Job 2: Multiply each element by 2 (depends on addJob)
var multiplyJob = new MultiplyJob
{
Data = (float*)array.GetUnsafePtr(),
Multiplier = 2f
};
// Job 3: Compute final sum (depends on multiplyJob)
var result = stackalloc float[1];
var sumJob = new KahanSumJob
{
Data = (float*)array.GetUnsafePtr(),
Length = arraySize,
Result = result
};
Console.WriteLine("Scheduling jobs with dependencies...");
// Schedule jobs with dependencies
var addHandle = addJob.ScheduleParallel(arraySize, 64);
var multiplyHandle = multiplyJob.ScheduleParallel(arraySize, 64, addHandle);
var sumHandle = sumJob.Schedule(multiplyHandle);
// Wait for all jobs to complete
sumHandle.Complete();
Console.WriteLine($"Final sum: {*result}");
Console.WriteLine($"Expected sum: {ComputeExpectedSum(arraySize)}");
Console.WriteLine("Jobs completed successfully!");
// Test dependency combination
Console.WriteLine("\n=== Testing Dependency Combination ===");
// Reset array
for (var i = 0; i < arraySize; i++)
{
array[i] = 1f;
}
// Create multiple independent jobs
var basePtr = (float*)array.GetUnsafePtr();
var job1 = new AddValueJob { Data = basePtr, Value = 1f };
var job2 = new AddValueJob { Data = basePtr + arraySize / 2, Value = 2f };
var job3 = new AddValueJob { Data = basePtr + arraySize / 4, Value = 3f };
var handle1 = job1.ScheduleParallel(arraySize / 2, 32);
var handle2 = job2.ScheduleParallel(arraySize / 2, 32);
var handle3 = job3.ScheduleParallel(arraySize / 4, 32);
// Combine dependencies
var combinedHandle = JobHandle.CombineDependencies(handle1, handle2, handle3);
// Final job that depends on all previous jobs
var finalSum = stackalloc float[1];
var finalSumJob = new KahanSumJob
{
Data = (float*)array.GetUnsafePtr(),
Length = arraySize,
Result = finalSum
};
var finalHandle = finalSumJob.Schedule(combinedHandle);
finalHandle.Complete();
Console.WriteLine($"Final sum after combined dependencies: {*finalSum}");
Console.WriteLine("Dependency combination test completed!");
}
private static unsafe float ComputeSum(float* data, int length)
{
var sum = 0f;
for (var i = 0; i < length; i++)
{
sum += data[i];
}
return sum;
}
private static float ComputeExpectedSum(int arraySize)
{
// Original sum: 1 + 2 + 3 + ... + n = n(n+1)/2
var originalSum = arraySize * (arraySize + 1) / 2f;
// After adding 10: each element increases by 10, so total increases by 10 * n
var afterAdd = originalSum + (10f * arraySize);
// After multiplying by 2: everything doubles
var afterMultiply = afterAdd * 2f;
return afterMultiply;
}
}

12
Misaki.HighPerformance.Test/Jobs/NoiseJob.cs
View File

@@ -5,7 +5,7 @@ using System.Runtime.CompilerServices;
namespace Misaki.HighPerformance.Test.Jobs; namespace Misaki.HighPerformance.Test.Jobs;
internal unsafe struct NoiseJob : IJobParallelFor internal struct NoiseJob : IJobParallelFor
{ {
public UnsafeArray<float> buffers; public UnsafeArray<float> buffers;
public int width; public int width;
@@ -41,11 +41,11 @@ internal unsafe struct NoiseJob : IJobParallelFor
return float.Lerp(float.Lerp(d00, d10, fp.Y), float.Lerp(d01, d11, fp.Y), fp.X); return float.Lerp(float.Lerp(d00, d10, fp.Y), float.Lerp(d01, d11, fp.Y), fp.X);
} }
public void Execute(int index) public void Execute(int loopIndex, int threadIndex)
{ {
var x = index % width; var x = loopIndex % width;
var y = index / height; var y = loopIndex / height;
var uv = new Vector2(x, y); var uv = new Vector2(x, y) / new Vector2(width, height);
buffers[index] = float.Clamp(GradientNoise(uv), 0.0f, 1.0f); buffers[loopIndex] = GradientNoise(uv);
} }
} }

4
Misaki.HighPerformance.Test/Misaki.HighPerformance.Test.csproj
View File

@@ -22,8 +22,4 @@
<ProjectReference Include="..\Misaki.HighPerformance\Misaki.HighPerformance.csproj" /> <ProjectReference Include="..\Misaki.HighPerformance\Misaki.HighPerformance.csproj" />
</ItemGroup> </ItemGroup>
<ItemGroup>
<Folder Include="UnitTest\" />
</ItemGroup>
</Project> </Project>

52
Misaki.HighPerformance.Test/Program.cs
View File

@@ -1,12 +1,46 @@
using Misaki.HighPerformance.Test.Benchmark; //var threadCount = 8;
using Misaki.HighPerformance.Test.Jobs; //var map = new ConcurrentSlotMap<int>();
// Test the job system //var barrier = new Barrier(threadCount);
JobSystemExample.RunExample();
Console.WriteLine("\nPress any key to run benchmarks..."); //Parallel.For(0, threadCount, threadIndex =>
Console.ReadKey(); //{
// barrier.SignalAndWait();
// for (var i = 0; i < 1000; i++)
// {
// var id = map.Add(i + threadIndex * 1000, out var gen);
// if (i % 100 == 0)
// {
// map.Remove(id, gen);
// }
// }
//});
BenchmarkDotNet.Running.BenchmarkRunner.Run<MathematicsBenchmark>(); //Console.WriteLine($"Count should be {threadCount * 990}, actual: {map.Count}");
//var b = new MathematicsBenchmark();
//b.Vector2Add(); using Misaki.HighPerformance.Test.Benchmark;
//BenchmarkDotNet.Running.BenchmarkRunner.Run<ParallelNoiseBenchmark>();
var benchmark = new ParallelNoiseBenchmark();
var sw = new System.Diagnostics.Stopwatch();
benchmark.Setup();
for (var i = 0; i < 1024; i++)
{
benchmark.JobSystem();
}
sw.Start();
for (var i = 0; i < 1024; i++)
{
benchmark.JobSystem();
}
sw.Stop();
benchmark.Cleanup();
Console.WriteLine($"JobSystem: {sw.Elapsed.TotalMilliseconds / 1024.0} ms");

201
Misaki.HighPerformance.Test/UnitTest/Jobs/TestJobSystem.cs Normal file
View File

@@ -0,0 +1,201 @@
using Misaki.HighPerformance.Jobs;
using Misaki.HighPerformance.LowLevel.Buffer;
using Misaki.HighPerformance.LowLevel.Collections;
using Misaki.HighPerformance.LowLevel.Helpers;
namespace Misaki.HighPerformance.Test.UnitTest.Jobs;
[TestClass]
public unsafe class TestJobSystem
{
private JobScheduler _jobScheduler = null!;
[TestInitialize]
public void Initialize()
{
_jobScheduler = new JobScheduler(Environment.ProcessorCount);
}
[TestCleanup]
public void Cleanup()
{
_jobScheduler.Dispose();
}
[TestMethod]
public void SingleJob()
{
var result = stackalloc float[1];
var job = new TwoSumJob
{
value1 = 1.5f,
value2 = 2.5f,
result = result
};
var handle = _jobScheduler.Schedule(ref job, -1);
_jobScheduler.WaitComplete(handle);
Assert.AreEqual(4.0f, *result);
}
[TestMethod]
public void JobDependency()
{
var result = stackalloc float[1];
var job1 = new TwoSumJob
{
value1 = 1.5f,
value2 = 2.5f,
result = result
};
var handle1 = _jobScheduler.Schedule(ref job1, -1);
var job2 = new AddJob
{
value = 4.0f,
result = result
};
var handle2 = _jobScheduler.Schedule(ref job2, -1, handle1);
_jobScheduler.WaitComplete(handle2);
Assert.AreEqual(8.0f, *result);
}
[TestMethod]
public void CompletedDependency()
{
var result = stackalloc float[1];
var job1 = new TwoSumJob
{
value1 = 1.5f,
value2 = 2.5f,
result = result
};
var handle1 = _jobScheduler.Schedule(ref job1, -1);
_jobScheduler.WaitComplete(handle1);
var job2 = new AddJob
{
value = 4.0f,
result = result
};
var handle2 = _jobScheduler.Schedule(ref job2, -1, handle1);
_jobScheduler.WaitComplete(handle2);
Assert.AreEqual(8.0f, *result);
}
[TestMethod]
public void CombineDependencies()
{
var result = stackalloc float[1];
var job1 = new TwoSumJob
{
value1 = 2.5f,
value2 = 2.5f,
result = result
};
var handle1 = _jobScheduler.Schedule(ref job1, -1);
var job2 = new AddJob
{
value = 4.0f,
result = result
};
var handle2 = _jobScheduler.Schedule(ref job2, -1, handle1);
var job3 = new AddJob
{
value = 10.0f,
result = result
};
var combinedHandle = _jobScheduler.CombineDependencies(handle1, handle2);
var handle3 = _jobScheduler.Schedule(ref job3, -1, combinedHandle);
_jobScheduler.WaitComplete(handle3);
Assert.AreEqual(19.0f, *result);
}
[TestMethod]
public void SingleParallelJob()
{
const int size = 1000;
var result = stackalloc float[size];
MemoryUtilities.MemSet(result, 0, sizeof(float) * size);
var job = new ParallelAddJob
{
value = 1.0f,
inout = result
};
var handle = _jobScheduler.ScheduleParallel(ref job, size, 64, -1, JobHandle.Invalid);
_jobScheduler.WaitComplete(handle);
Assert.AreEqual(1.0f, result[500]);
}
private static float ComputeExpectedSum(int arraySize)
{
// Original sum: 1 + 2 + 3 + ... + n = n(n+1)/2
var originalSum = arraySize * (arraySize + 1) / 2f;
// After adding 10: each element increases by 10, so total increases by 10 * n
var afterAdd = originalSum + (10f * arraySize);
// After multiplying by 2: everything doubles
var afterMultiply = afterAdd * 2f;
return afterMultiply;
}
[TestMethod]
public void ChainJob()
{
const int arraySize = 10000;
using var array = new UnsafeArray<float>(arraySize, Allocator.Persistent);
for (var i = 0; i < arraySize; i++)
{
array[i] = i + 1;
}
var addJob = new ParallelAddJob
{
value = 10f,
inout = (float*)array.GetUnsafePtr()
};
var multiplyJob = new ParallelMultiplyJob
{
multiplier = 2f,
inout = (float*)array.GetUnsafePtr()
};
var result = stackalloc float[1];
var sumJob = new KahanSumJob
{
input = (float*)array.GetUnsafePtr(),
length = arraySize,
output = result
};
var handle1 = _jobScheduler.ScheduleParallel(ref addJob, arraySize, 64, -1, JobHandle.Invalid);
var handle2 = _jobScheduler.ScheduleParallel(ref multiplyJob, arraySize, 64, -1, handle1);
var handle3 = _jobScheduler.Schedule(ref sumJob, -1, handle2);
_jobScheduler.WaitComplete(handle3);
var expected = ComputeExpectedSum(arraySize);
Assert.AreEqual(expected, *result, 0.01f);
}
}

73
Misaki.HighPerformance.Test/UnitTest/Jobs/TestJobs.cs Normal file
View File

@@ -0,0 +1,73 @@
using Misaki.HighPerformance.Jobs;
namespace Misaki.HighPerformance.Test.UnitTest.Jobs;
internal unsafe struct TwoSumJob : IJob
{
public float value1;
public float value2;
public float* result;
public void Execute(int threadIndex)
{
*result = value1 + value2;
}
}
internal unsafe struct AddJob : IJob
{
public float value;
public float* result;
public void Execute(int threadIndex)
{
*result += value;
}
}
internal unsafe struct KahanSumJob : IJob
{
public float* input;
public int length;
public float* output;
public void Execute(int threadIndex)
{
var sum = 0f;
var c = 0f; // Compensation for lost low-order bits
for (var i = 0; i < length; i++)
{
var y = input[i] - c; // So far, so good: c is zero
var t = sum + y; // Alas, sum is big, y small, so low-order digits of y are lost
c = (t - sum) - y; // (t - sum) cancels the high-order part of y; subtracting y recovers negative (low part of y)
sum = t; // Algebraically, c should always be zero. Beware overly-clever compilers!
}
*output = sum;
}
}
internal unsafe struct ParallelAddJob : IJobParallelFor
{
public float value;
public float* inout;
public void Execute(int loopIndex, int threadIndex)
{
inout[loopIndex] += value;
}
}
internal unsafe struct ParallelMultiplyJob : IJobParallelFor
{
public float multiplier;
public float* inout;
public void Execute(int loopIndex, int threadIndex)
{
inout[loopIndex] *= multiplier;
}
}

0
Misaki.HighPerformance/Collections/AtomicCounter32.cs → Misaki.HighPerformance/Collections/AtomicCounter.cs
View File

321
Misaki.HighPerformance/Collections/ConcureentSlotMap.cs Normal file
View File

@@ -0,0 +1,321 @@
using System.Collections;
using System.Collections.Concurrent;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
namespace Misaki.HighPerformance.Collections;
public class ConcurrentSlotMap<T> : IEnumerable<T>
{
public struct Enumerator : IEnumerator<T>
{
private readonly ConcurrentSlotMap<T> _slotMap;
private int _currentIndex;
public Enumerator(ConcurrentSlotMap<T> slotMap)
{
_slotMap = slotMap;
_currentIndex = -1;
}
public readonly T Current => _slotMap._data[_currentIndex].value!;
readonly object? IEnumerator.Current => Current;
public bool MoveNext()
{
var capacity = Volatile.Read(ref _slotMap._capacity);
while (++_currentIndex < capacity)
{
if (Volatile.Read(ref _slotMap._data[_currentIndex].isValid) == 1)
{
return true;
}
}
return false;
}
public void Reset() => _currentIndex = -1;
public void Dispose()
{
}
}
// Lock-free slot using separate fields for atomic operations
private struct SlotEntry
{
public T? value;
public int generation;
public int isValid;
public SlotEntry()
{
value = default;
generation = 0;
isValid = 0;
}
}
private volatile SlotEntry[] _data;
private readonly ConcurrentQueue<int> _freeSlots;
private int _count;
private int _capacity;
private int _nextSlotIndex;
// For lock-free resizing
private int _isResizing;
public int Count => Volatile.Read(ref _count);
public int Capacity => Volatile.Read(ref _capacity);
public IEnumerator<T> GetEnumerator() => new Enumerator(this);
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
public ConcurrentSlotMap(int initialCapacity = 16)
{
_capacity = initialCapacity;
_count = 0;
_nextSlotIndex = 0;
_isResizing = 0;
_data = new SlotEntry[initialCapacity];
_freeSlots = new();
}
[MethodImpl(MethodImplOptions.NoInlining)]
private void TryResize(int requiredCapacity)
{
// Use CAS to ensure only one thread does the resize
if (Interlocked.CompareExchange(ref _isResizing, 1, 0) != 0)
{
// Another thread is resizing, wait for it to complete
var spinWait = new SpinWait();
while (Volatile.Read(ref _isResizing) == 1)
{
spinWait.SpinOnce();
}
return;
}
try
{
var currentCapacity = Volatile.Read(ref _capacity);
if (currentCapacity >= requiredCapacity)
{
return; // Another thread already resized
}
var newCapacity = currentCapacity;
while (newCapacity < requiredCapacity)
{
newCapacity *= 2;
}
var newData = new SlotEntry[newCapacity];
var oldData = _data;
// Copy existing data
Array.Copy(oldData, newData, currentCapacity);
// Initialize new slots
for (var i = currentCapacity; i < newCapacity; i++)
{
newData[i] = new SlotEntry();
}
// Atomically update the array reference and capacity
_data = newData;
Volatile.Write(ref _capacity, newCapacity);
}
finally
{
// Release the resize lock
Volatile.Write(ref _isResizing, 0);
}
}
public int Add(T item, out int generation)
{
// Try to get a free slot first
if (_freeSlots.TryDequeue(out var slotIndex))
{
ref var slot = ref _data[slotIndex];
// Atomically mark as valid and get the current generation
var currentGeneration = Volatile.Read(ref slot.generation);
slot.value = item;
// Use CAS to mark as valid atomically
if (Interlocked.CompareExchange(ref slot.isValid, 1, 0) == 0)
{
generation = currentGeneration;
Interlocked.Increment(ref _count);
return slotIndex;
}
else
{
// Slot was somehow already valid, put it back and try again
_freeSlots.Enqueue(slotIndex);
return Add(item, out generation);
}
}
// Need a new slot
slotIndex = Interlocked.Increment(ref _nextSlotIndex) - 1;
// Check if we need to resize
var currentCapacity = Volatile.Read(ref _capacity);
if (slotIndex >= currentCapacity)
{
TryResize(slotIndex + 1);
}
// Initialize the new slot
ref var newSlot = ref _data[slotIndex];
newSlot.value = item;
newSlot.generation = 0;
Volatile.Write(ref newSlot.isValid, 1);
generation = 0;
Interlocked.Increment(ref _count);
return slotIndex;
}
public bool Remove(int slotIndex, int generation)
{
var capacity = Volatile.Read(ref _capacity);
if (slotIndex < 0 || slotIndex >= capacity)
{
return false;
}
ref var slot = ref _data[slotIndex];
// Check if slot is valid and generation matches
if (Volatile.Read(ref slot.isValid) == 0 || Volatile.Read(ref slot.generation) != generation)
{
return false;
}
// Atomically mark as invalid
if (Interlocked.CompareExchange(ref slot.isValid, 0, 1) == 1)
{
Interlocked.Increment(ref slot.generation);
slot.value = default;
_freeSlots.Enqueue(slotIndex);
Interlocked.Decrement(ref _count);
return true;
}
return false; // Another thread already removed it
}
public bool TryGetElement(int slotIndex, int generation, [MaybeNullWhen(false)] out T value)
{
value = default;
if (slotIndex < 0 || slotIndex >= Volatile.Read(ref _capacity))
{
return false;
}
ref var slot = ref _data[slotIndex];
// Read generation first, then validity, then value for consistency
var currentGeneration = Volatile.Read(ref slot.generation);
var isValid = Volatile.Read(ref slot.isValid) == 1;
if (isValid && currentGeneration == generation)
{
// Double-check that the slot is still valid with same generation
// to avoid race condition where slot gets removed between reads
if (Volatile.Read(ref slot.isValid) == 1 && Volatile.Read(ref slot.generation) == generation)
{
value = slot.value!;
return true;
}
}
return false;
}
public T GetElementAt(int slotIndex, int generation)
{
if (slotIndex < 0 || slotIndex >= Volatile.Read(ref _capacity))
{
throw new ArgumentOutOfRangeException(nameof(slotIndex), "Slot index is out of range.");
}
ref var slot = ref _data[slotIndex];
if (Volatile.Read(ref slot.isValid) == 0 || Volatile.Read(ref slot.generation) != generation)
{
throw new InvalidOperationException($"Slot {slotIndex} is not occupied or generation mismatch.");
}
return slot.value!;
}
public ref T GetElementReferenceAt(int slotIndex, int generation, out bool exist)
{
if (slotIndex < 0 || slotIndex >= Volatile.Read(ref _capacity))
{
exist = false;
return ref Unsafe.NullRef<T>();
}
ref var slot = ref _data[slotIndex];
if (Volatile.Read(ref slot.isValid) == 0 || Volatile.Read(ref slot.generation) != generation)
{
exist = false;
return ref Unsafe.NullRef<T>();
}
exist = true;
return ref slot.value!;
}
public void UpdateElement(int slotIndex, int generation, T newValue)
{
if (slotIndex < 0 || slotIndex >= Volatile.Read(ref _capacity))
{
throw new ArgumentOutOfRangeException(nameof(slotIndex), "Slot index is out of range.");
}
ref var slot = ref _data[slotIndex];
if (Volatile.Read(ref slot.isValid) == 0 || Volatile.Read(ref slot.generation) != generation)
{
throw new InvalidOperationException($"Slot {slotIndex} is not occupied or generation mismatch.");
}
slot.value = newValue;
}
public void Clear()
{
// Reset counters
Volatile.Write(ref _count, 0);
Volatile.Write(ref _nextSlotIndex, 0);
// Clear all slots
var capacity = Volatile.Read(ref _capacity);
for (var i = 0; i < capacity; i++)
{
ref var slot = ref _data[i];
Volatile.Write(ref slot.isValid, 0);
slot.generation = 0;
slot.value = default;
}
// Clear free slots queue
while (_freeSlots.TryDequeue(out _))
{
}
}
}

46
Misaki.HighPerformance.LowLevel/Collections/SlotMap.cs → Misaki.HighPerformance/Collections/SlotMap.cs
View File

@@ -42,6 +42,7 @@ public class SlotMap<T> : IEnumerable<T>
private struct SlotData private struct SlotData
{ {
public T value; public T value;
public int generation;
public bool isValid; public bool isValid;
} }
@@ -54,25 +55,6 @@ public class SlotMap<T> : IEnumerable<T>
public int Count => _count; public int Count => _count;
public int Capacity => _capacity; public int Capacity => _capacity;
public ref T this[int slotIndex]
{
get
{
if (slotIndex < 0 || slotIndex >= _capacity)
{
throw new ArgumentOutOfRangeException(nameof(slotIndex), "Slot index is out of range.");
}
ref var slot = ref _data[slotIndex];
if (!slot.isValid)
{
throw new InvalidOperationException($"Slot {slotIndex} is not occupied.");
}
return ref slot.value;
}
}
public IEnumerator<T> GetEnumerator() => new Enumerator(this); public IEnumerator<T> GetEnumerator() => new Enumerator(this);
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator(); IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
@@ -82,7 +64,7 @@ public class SlotMap<T> : IEnumerable<T>
_capacity = initialCapacity; _capacity = initialCapacity;
_data = new SlotData[initialCapacity]; _data = new SlotData[initialCapacity];
_freeSlots = new Queue<int>(initialCapacity); _freeSlots = new(initialCapacity);
} }
private void Resize() private void Resize()
@@ -95,7 +77,7 @@ public class SlotMap<T> : IEnumerable<T>
_capacity = newCapacity; _capacity = newCapacity;
} }
public int Add(T item) public int Add(T item, out int generation)
{ {
if (_count >= _capacity) if (_count >= _capacity)
{ {
@@ -115,13 +97,14 @@ public class SlotMap<T> : IEnumerable<T>
ref var slot = ref _data[slotIndex]; ref var slot = ref _data[slotIndex];
slot.value = item; slot.value = item;
slot.isValid = true; slot.isValid = true;
generation = slot.generation;
_count++; _count++;
return slotIndex; return slotIndex;
} }
public bool Remove(int slotIndex) public bool Remove(int slotIndex, int generation)
{ {
if (slotIndex < 0 || slotIndex >= _capacity) if (slotIndex < 0 || slotIndex >= _capacity)
{ {
@@ -129,11 +112,12 @@ public class SlotMap<T> : IEnumerable<T>
} }
ref var slot = ref _data[slotIndex]; ref var slot = ref _data[slotIndex];
if (!slot.isValid) if (slot.generation != generation)
{ {
return false; return false;
} }
slot.generation++;
slot.isValid = false; slot.isValid = false;
_freeSlots.Enqueue(slotIndex); _freeSlots.Enqueue(slotIndex);
@@ -142,6 +126,22 @@ public class SlotMap<T> : IEnumerable<T>
return true; return true;
} }
public ref T GetElementAt(int slotIndex, int generation)
{
if (slotIndex < 0 || slotIndex >= _capacity)
{
throw new ArgumentOutOfRangeException(nameof(slotIndex), "Slot index is out of range.");
}
ref var slot = ref _data[slotIndex];
if (slot.generation != generation)
{
throw new InvalidOperationException($"Slot {slotIndex} is not occupied.");
}
return ref slot.value;
}
public void Clear() public void Clear()
{ {
_count = 0; _count = 0;