Dispose requests on JobPool Dispose

.github/workflows/dotnet.yml

@@ -33,10 +33,10 @@ jobs:
     - name: Install GitVersion
       uses: gittools/actions/gitversion/setup@v0
       with:
-        versionSpec: '6.x'
+        versionSpec: '5.x'
         includePrerelease: true
         preferLatestVersion: true
-    
+
     - name: Determine Version
       id:   gitversion
       uses: gittools/actions/gitversion/execute@v0

Sally7.Tests/RequestExecutor/JobPoolTests.cs

@@ -0,0 +1,48 @@
+using System.Threading.Channels;
+using Sally7.RequestExecutor;
+
+namespace Sally7.Tests.RequestExecutor;
+
+public class JobPoolTests
+{
+    [Fact]
+    public async Task RentJobIdAsync_Throws_If_Disposed_And_Depleted()
+    {
+        // Arrange
+        var sut = new JobPool(1);
+        sut.Dispose();
+        _ = await sut.RentJobIdAsync(CancellationToken.None); // Empty the pool
+
+        // Act
+        // Assert
+        await Should.ThrowAsync<ChannelClosedException>(() => sut.RentJobIdAsync(CancellationToken.None).AsTask());
+    }
+
+    [Fact]
+    public void ReturnJobId_Does_Not_Throw_If_Disposed()
+    {
+        // Arrange
+        var sut = new JobPool(1);
+        var jobId = sut.RentJobIdAsync(CancellationToken.None).Result;
+        sut.Dispose();
+
+        // Act
+        // Assert
+        sut.ReturnJobId(jobId);
+    }
+
+    [Fact]
+    public void Dispose_Calls_Dispose_On_Requests()
+    {
+        // Arrange
+        var sut = new JobPool(1);
+        var jobId = sut.RentJobIdAsync(CancellationToken.None).Result;
+        var request = sut.GetRequest(jobId);
+
+        // Act
+        sut.Dispose();
+
+        // Assert
+        Should.Throw<ObjectDisposedException>(() => request.GetResult());
+    }
+}

Sally7.Tests/RequestExecutor/RequestTests.cs

@@ -0,0 +1,35 @@
+using FakeItEasy;
+using Sally7.RequestExecutor;
+
+namespace Sally7.Tests.RequestExecutor;
+
+public class RequestTests
+{
+    [Fact]
+    public void Completes_On_Dispose()
+    {
+        // Arrange
+        var sut = new Request();
+        var callback = A.Fake<Action>();
+        sut.OnCompleted(callback);
+
+        // Act
+        sut.Dispose();
+
+        // Assert
+        A.CallTo(() => callback.Invoke()).MustHaveHappenedOnceExactly();
+    }
+
+    [Fact]
+    public async Task Throws_When_Awaited_After_Dispose()
+    {
+        // Arrange
+        var sut = new Request();
+
+        // Act
+        sut.Dispose();
+
+        // Assert
+        await Should.ThrowAsync<ObjectDisposedException>(async () => await sut);
+    }
+}

Sally7.Tests/RequestExecutor/SignalTests.cs

@@ -0,0 +1,19 @@
+using System.Threading.Channels;
+using Sally7.RequestExecutor;
+
+namespace Sally7.Tests.RequestExecutor;
+
+public class SignalTests
+{
+    [Fact]
+    public async Task WaitAsync_Throws_If_Disposed()
+    {
+        // Arrange
+        var sut = new Signal();
+        sut.Dispose();
+
+        // Act
+        // Assert
+        await Should.ThrowAsync<ChannelClosedException>(() => sut.WaitAsync(CancellationToken.None).AsTask());
+    }
+}

Sally7.Tests/Sally7.Tests.csproj

@@ -4,11 +4,19 @@
     <TargetFramework>net6</TargetFramework>
     <LangVersion>latest</LangVersion>
     <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
 
   <ItemGroup>
+    <Using Include="Shouldly" />
+    <Using Include="Xunit" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="FakeItEasy" Version="8.1.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.5.0" />
+    <PackageReference Include="Shouldly" Version="4.2.1" />
     <PackageReference Include="xunit" Version="2.4.2" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.4.5">
       <PrivateAssets>all</PrivateAssets>

Sally7/Internal/DisposableHelper.cs

@@ -0,0 +1,13 @@
+using System;
+
+namespace Sally7.Internal;
+
+internal static class DisposableHelper
+{
+    public static void ThrowIf(bool condition, object instance)
+    {
+        void Throw() => throw new ObjectDisposedException(instance.ToString());
+
+        if (condition) Throw();
+    }
+}

Sally7/RequestExecutor/ConcurrentRequestExecutor.cs

@@ -2,10 +2,8 @@
 using System.Buffers;
 using System.Diagnostics;
 using System.Net.Sockets;
-using System.Runtime.CompilerServices;
 using System.Runtime.InteropServices;
 using System.Threading;
-using System.Threading.Channels;
 using System.Threading.Tasks;
 using Sally7.Internal;
 
@@ -187,120 +185,5 @@ public async ValueTask<Memory<byte>> PerformRequest(ReadOnlyMemory<byte> request
                 jobPool.ReturnJobId(jobId);
             }
         }
-
-        private class JobPool : IDisposable
-        {
-            private readonly Channel<int> jobIdPool;
-            private readonly Request[] requests;
-            private volatile bool disposed;
-
-            public JobPool(int maxNumberOfConcurrentRequests)
-            {
-                jobIdPool = Channel.CreateBounded<int>(maxNumberOfConcurrentRequests);
-                requests = new Request[maxNumberOfConcurrentRequests];
-
-                for (int i = 0; i < maxNumberOfConcurrentRequests; ++i)
-                {
-                    if (!jobIdPool.Writer.TryWrite(i + 1))
-                    {
-                        Sally7Exception.ThrowFailedToInitJobPool();
-                    }
-
-                    requests[i] = new Request();
-                }
-            }
-
-            public void Dispose()
-            {
-                disposed = true;
-                jobIdPool.Writer.Complete();
-            }
-
-            public ValueTask<int> RentJobIdAsync(CancellationToken cancellationToken) => jobIdPool.Reader.ReadAsync(cancellationToken);
-
-            public void ReturnJobId(int jobId)
-            {
-                if (!jobIdPool.Writer.TryWrite(jobId) && !disposed)
-                {
-                    Sally7Exception.ThrowFailedToReturnJobIDToPool(jobId);
-                }
-            }
-
-            [DebuggerNonUserCode]
-            public Request GetRequest(int jobId) => requests[jobId - 1];
-
-            public void SetBufferForRequest(int jobId, Memory<byte> buffer)
-            {
-                Request req = GetRequest(jobId);
-                req.Reset();
-                req.SetBuffer(buffer);
-            }
-        }
-
-        [DebuggerNonUserCode]
-        [DebuggerDisplay(nameof(NeedToWait) + ": {" + nameof(NeedToWait) + ",nq}")]
-        private class Signal : IDisposable
-        {
-            private readonly Channel<int> channel = Channel.CreateBounded<int>(1);
-
-            public void Dispose() => channel.Writer.Complete();
-
-            public bool TryInit() => channel.Writer.TryWrite(0);
-
-            public ValueTask<int> WaitAsync(CancellationToken cancellationToken) => channel.Reader.ReadAsync(cancellationToken);
-
-            public bool TryRelease() => channel.Writer.TryWrite(0);
-
-            private bool NeedToWait => channel.Reader.Count == 0;
-        }
-
-        private class Request : INotifyCompletion
-        {
-            private static readonly Action Sentinel = () => { };
-
-            private Memory<byte> buffer;
-
-            public bool IsCompleted { get; private set; }
-            private int length;
-            private Action? continuation = Sentinel;
-
-            public Memory<byte> Buffer => buffer;
-
-            public void Complete(int length)
-            {
-                this.length = length;
-                IsCompleted = true;
-
-                var prev = continuation ?? Interlocked.CompareExchange(ref continuation, Sentinel, null);
-                prev?.Invoke();
-            }
-
-            public Memory<byte> GetResult()
-            {
-                return buffer.Slice(0, length);
-            }
-
-            public Request GetAwaiter() => this;
-
-            public void OnCompleted(Action continuation)
-            {
-                if (this.continuation == Sentinel ||
-                    Interlocked.CompareExchange(ref this.continuation, continuation, null) == Sentinel)
-                {
-                    continuation.Invoke();
-                }
-            }
-
-            public void Reset()
-            {
-                continuation = null;
-                IsCompleted = false;
-            }
-
-            public void SetBuffer(Memory<byte> buffer)
-            {
-                this.buffer = buffer;
-            }
-        }
     }
 }

Sally7/RequestExecutor/JobPool.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Sally7.Internal;
+
+namespace Sally7.RequestExecutor;
+
+internal class JobPool : IDisposable
+{
+    private readonly Channel<int> _jobIdPool;
+    private readonly Request[] _requests;
+    private bool _disposed;
+
+    public JobPool(int maxNumberOfConcurrentRequests)
+    {
+        _jobIdPool = Channel.CreateBounded<int>(maxNumberOfConcurrentRequests);
+        _requests = new Request[maxNumberOfConcurrentRequests];
+
+        for (int i = 0; i < maxNumberOfConcurrentRequests; ++i)
+        {
+            if (!_jobIdPool.Writer.TryWrite(i + 1))
+            {
+                Sally7Exception.ThrowFailedToInitJobPool();
+            }
+
+            _requests[i] = new Request();
+        }
+    }
+
+    public void Dispose()
+    {
+        Volatile.Write(ref _disposed, true);
+        _jobIdPool.Writer.Complete();
+
+        foreach (var request in _requests)
+        {
+            request.Dispose();
+        }
+    }
+
+    public ValueTask<int> RentJobIdAsync(CancellationToken cancellationToken) => _jobIdPool.Reader.ReadAsync(cancellationToken);
+
+    public void ReturnJobId(int jobId)
+    {
+        if (!_jobIdPool.Writer.TryWrite(jobId) && !Volatile.Read(ref _disposed))
+        {
+            Sally7Exception.ThrowFailedToReturnJobIDToPool(jobId);
+        }
+    }
+
+    [DebuggerNonUserCode]
+    public Request GetRequest(int jobId)
+    {
+        DisposableHelper.ThrowIf(Volatile.Read(ref _disposed), this);
+
+        return _requests[jobId - 1];
+    }
+
+    public void SetBufferForRequest(int jobId, Memory<byte> buffer)
+    {
+        Request req = GetRequest(jobId);
+        req.Reset();
+        req.SetBuffer(buffer);
+    }
+}