Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 17 additions & 2 deletions src/LogicLooper/LogicLooperPool.Shared.cs
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,13 @@ namespace Cysharp.Threading;

public sealed partial class LogicLooperPool
{
private static readonly ILogicLooperPool _notInitializedSharedPool = new NotInitializedLogicLooperPool();
private static ILogicLooperPool _shared = _notInitializedSharedPool;

/// <summary>
/// Gets the shared pool of loopers. Requires to call <see cref="InitializeSharedPool"/> method before use.
/// </summary>
public static ILogicLooperPool Shared { get; private set; } = new NotInitializedLogicLooperPool();
public static ILogicLooperPool Shared => Volatile.Read(ref _shared);

/// <summary>
/// Initializes the shared pool of loopers with specified options.
Expand All @@ -16,18 +19,30 @@ public sealed partial class LogicLooperPool
/// <param name="looperCount"></param>
/// <param name="balancer"></param>
/// <param name="looperFactory"></param>
/// <exception cref="InvalidOperationException">The shared pool has already been initialized.</exception>
public static void InitializeSharedPool(int targetFrameRate, int looperCount = 0, ILogicLooperPoolBalancer? balancer = null, ILogicLooperPoolLooperFactory? looperFactory = null)
{
if (!ReferenceEquals(Volatile.Read(ref _shared), _notInitializedSharedPool))
{
throw new InvalidOperationException("LogicLooperPool.Shared has already been initialized.");
}

if (looperCount == 0)
{
looperCount = Math.Max(1, Environment.ProcessorCount - 1);
}

Shared = new LogicLooperPool(
var sharedPool = new LogicLooperPool(
targetFrameRate,
looperCount,
balancer ?? RoundRobinLogicLooperPoolBalancer.Instance,
looperFactory ?? DefaultLogicLooperPoolLooperFactory.Instance
);

if (!ReferenceEquals(Interlocked.CompareExchange(ref _shared, sharedPool, _notInitializedSharedPool), _notInitializedSharedPool))
{
sharedPool.Dispose();
throw new InvalidOperationException("LogicLooperPool.Shared has already been initialized.");
}
}
}
97 changes: 97 additions & 0 deletions test/LogicLooper.Test/LogicLooperPoolSharedTest.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
using Cysharp.Threading;

namespace LogicLooper.Test;

public class LogicLooperPoolSharedTest
{
[Fact]
public void InitializeSharedPool_CanOnlySucceedOnce()
{
const int concurrentCalls = 8;
using var factory = new TrackingLogicLooperFactory(concurrentCalls);
using var start = new ManualResetEventSlim();
var results = new Exception[concurrentCalls];

var tasks = Enumerable.Range(0, concurrentCalls)
.Select(index => Task.Factory.StartNew(
() =>
{
start.Wait();
try
{
LogicLooperPool.InitializeSharedPool(60, 1, looperFactory: factory);
}
catch (Exception ex)
{
results[index] = ex;
}
},
CancellationToken.None,
TaskCreationOptions.LongRunning,
TaskScheduler.Default))
.ToArray();

start.Set();
Task.WaitAll(tasks);

try
{
Assert.Single(results, static x => x == null);
Assert.Equal(concurrentCalls - 1, results.Count(static x => x is InvalidOperationException));
Assert.Equal(concurrentCalls, factory.CreatedCount);
Assert.Equal(concurrentCalls - 1, factory.DisposedCount);

Assert.Throws<InvalidOperationException>(() =>
LogicLooperPool.InitializeSharedPool(60, 1, looperFactory: factory));
Assert.Equal(concurrentCalls, factory.CreatedCount);
}
finally
{
LogicLooperPool.Shared.Dispose();
}

Assert.Equal(concurrentCalls, factory.DisposedCount);
}

private sealed class TrackingLogicLooperFactory(int expectedCreateCount) : ILogicLooperPoolLooperFactory, IDisposable
{
private readonly Barrier _createBarrier = new(expectedCreateCount);
private int _createdCount;
private int _disposedCount;

public int CreatedCount => Volatile.Read(ref _createdCount);
public int DisposedCount => Volatile.Read(ref _disposedCount);

public ILogicLooper Create(TimeSpan targetFrameTime)
{
Interlocked.Increment(ref _createdCount);
if (!_createBarrier.SignalAndWait(TimeSpan.FromSeconds(5)))
{
throw new TimeoutException("Timed out waiting for concurrent pool creation.");
}
return new TrackingLogicLooper(this);
}

public void Dispose() => _createBarrier.Dispose();

private sealed class TrackingLogicLooper(TrackingLogicLooperFactory owner) : ILogicLooper
{
public int Id => 0;
public int ApproximatelyRunningActions => 0;
public TimeSpan LastProcessingDuration => TimeSpan.Zero;
public double TargetFrameRate => 60;
public long CurrentFrame => 0;

public Task RegisterActionAsync(LogicLooperActionDelegate loopAction) => throw new NotSupportedException();
public Task RegisterActionAsync(LogicLooperActionDelegate loopAction, LooperActionOptions options) => throw new NotSupportedException();
public Task RegisterActionAsync<TState>(LogicLooperActionWithStateDelegate<TState> loopAction, TState state) => throw new NotSupportedException();
public Task RegisterActionAsync<TState>(LogicLooperActionWithStateDelegate<TState> loopAction, TState state, LooperActionOptions options) => throw new NotSupportedException();
public Task RegisterActionAsync(LogicLooperAsyncActionDelegate loopAction) => throw new NotSupportedException();
public Task RegisterActionAsync(LogicLooperAsyncActionDelegate loopAction, LooperActionOptions options) => throw new NotSupportedException();
public Task RegisterActionAsync<TState>(LogicLooperAsyncActionWithStateDelegate<TState> loopAction, TState state) => throw new NotSupportedException();
public Task RegisterActionAsync<TState>(LogicLooperAsyncActionWithStateDelegate<TState> loopAction, TState state, LooperActionOptions options) => throw new NotSupportedException();
public Task ShutdownAsync(TimeSpan shutdownDelay) => Task.CompletedTask;
public void Dispose() => Interlocked.Increment(ref owner._disposedCount);
}
}
}
Loading