Akka.TestKit: deleted `IAsyncQueue`; replaced with `System.Threading.Channel<T>` #7157

Aaronontheweb · 2024-04-11T12:55:24Z

Changes

Fixes #7145 - this should resolve some of the "async continuation problems" reported there. Going to add a reproduction for that case specifically, but moreover this change allows us to delete 6 files worth of old shared state concurrent programming code, which is always a good thing.

Checklist

For significant changes, please ensure that the following have been completed (delete if not relevant):

This change follows the Akka.NET API Compatibility Guidelines.
This change follows the Akka.NET Wire Compatibility Guidelines.
I have reviewed my own pull request.
Design discussion issue ExpectMsgAsync deadlocks with actor that has async receive handler and testprobe.Tell #7145
Changes in public API reviewed, if any.

Aaronontheweb · 2024-04-11T12:56:26Z

Running locally and seeing a bunch of test failures from InternalTryReceiveOneAsync - which makes me think my CancellationToken usage is wrong there.

Aaronontheweb · 2024-04-11T12:58:45Z

Looks like it's the ExpectNoMsgAsync implementation that's blowing up.

Aaronontheweb · 2024-04-11T17:37:55Z

src/core/Akka.TestKit.Tests/Bugfix7145Spec.cs

+
+namespace Akka.TestKit.Tests;
+
+public class Bugfix7145Spec : AkkaSpec


This spec was able to reproduce the deadlock reported in #7145 on dev - changes so far allows these specs to pass.

…nontheweb/akka.net into fix-7145-testkit-AsyncQueue

Aaronontheweb

Detailed my changes

Aaronontheweb · 2024-04-11T21:59:07Z

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.DotNet.verified.txt

@@ -622,84 +622,6 @@ namespace Akka.TestKit.Extensions
 }
 namespace Akka.TestKit.Internal
 {
-    public class AsyncQueue<T> : Akka.TestKit.Internal.ITestQueue<T>


Deleted this type

Aaronontheweb · 2024-04-11T21:59:32Z

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.DotNet.verified.txt

@@ -831,11 +709,6 @@ namespace Akka.TestKit.Internal
            public virtual void HandleEvent(Akka.TestKit.Internal.EventFilterBase eventFilter, Akka.Event.LogEvent logEvent) { }
        }
    }
-    public class InternalTestActor : Akka.Actor.UntypedActor


Made the InternalTestActor actually internal, instead of public for some forsaken reason.

Aaronontheweb · 2024-04-11T21:59:51Z

src/core/Akka.Streams.Tests/Dsl/FlowSelectAsyncSpec.cs

@@ -406,10 +407,10 @@ public async Task A_Flow_with_SelectAsync_must_not_run_more_futures_than_configu
            {
                const int parallelism = 8;
                var counter = new AtomicCounter();
-                var queue = new BlockingQueue<(TaskCompletionSource<int>, long)>();


Had to replace call to non-longer-existing-API

Aaronontheweb · 2024-04-11T22:00:04Z

src/core/Akka.Streams.Tests/Dsl/FlowSelectAsyncUnorderedSpec.cs

@@ -335,10 +336,10 @@ public async Task A_Flow_with_SelectAsyncUnordered_must_not_run_more_futures_tha
            await this.AssertAllStagesStoppedAsync(() => {
                const int parallelism = 8;
                var counter = new AtomicCounter();
-                var queue = new BlockingQueue<(TaskCompletionSource<int>, long)>();


Had to replace call to no-longer-existing-API

Aaronontheweb · 2024-04-11T22:00:33Z

src/core/Akka.TestKit/Internal/InternalTestActor.cs

    {
-        private readonly ITestActorQueue<MessageEnvelope> _queue;
-        private TestKit.TestActor.Ignore _ignore;
+        private readonly ChannelWriter<MessageEnvelope> _queue;


Replaced ITestActorQueue<T> with ChannelWriter<T>

Aaronontheweb · 2024-04-11T22:01:11Z

src/core/Akka.TestKit/TestKitBase.cs

@@ -36,7 +37,7 @@ public TestState()

            public ActorSystem System { get; set; }
            public TestKitSettings TestKitSettings { get; set; }
-            public ITestQueue<MessageEnvelope> Queue { get; set; }
+            public Channel<MessageEnvelope> Queue { get; set; }


Replace TestState.Queue to use a Channel<MessageEnvelope> instead of the old ITestQueue

Aaronontheweb · 2024-04-11T22:16:31Z

src/core/Akka.TestKit/TestKitBase_Receive.cs

-                    ;
+                var delayTask = Task.Delay(maxDuration, cancellationToken);
+                var readTask = _testState.Queue.Reader.WaitToReadAsync(cancellationToken).AsTask();
+                var completedTask = await Task.WhenAny(readTask, delayTask);


Have to do this instead of playing around with CancellationTokens expiring, because otherwise those work via throwing exceptions. This allows us to gracefully determine if we've hit the next point because stuff was ready or because time expired.

Aaronontheweb · 2024-04-11T22:16:39Z

src/core/Akka.TestKit/TestKitBase_Receive.cs

-                peek = await _testState.Queue.TryPeekAsync((int)maxDuration.TotalMilliseconds, cancellationToken)
-                    ;
+                var delayTask = Task.Delay(maxDuration, cancellationToken);
+                var readTask = _testState.Queue.Reader.WaitToReadAsync(cancellationToken).AsTask();


Have to do this instead of playing around with CancellationTokens expiring, because otherwise those work via throwing exceptions. This allows us to gracefully determine if we've hit the next point because stuff was ready or because time expired.

Arkatufus

Looks good overall, left a few possible improvement notes

Arkatufus · 2024-04-12T13:05:57Z

src/core/Akka.TestKit/TestKitBase_Receive.cs

                take = (didTake, item);
            }
            else if (maxDuration.IsPositiveFinite())
            {
                ConditionalLog(shouldLog, "Trying to receive message from TestActor queue within {0}", maxDuration);
-                take = await _testState.Queue.TryTakeAsync((int)maxDuration.TotalMilliseconds, cancellationToken)
-                    ;
+                var delayTask = Task.Delay(maxDuration, cancellationToken);


since maxDuration is infinite, it is possible for the Delay Task to survive until the end of the test process, causing a memory leak. For this to be leak safe, we'd need to use a linked CancellationTokenSource, use that token, and Cancel and Dispose of it with a try..catch

maxDuration cannot be infinite here - that is explicitly checked by IsPositiveFinite. The TimeSpan.Ticks value is -1 when it's infinite.

src/core/Akka.TestKit/TestKitBase_Receive.cs

deleted IAsyncQueue; replaced with System.Threading.Channel

630bda7

Aaronontheweb added api-change akka-testkit Akka.NET Testkit issues labels Apr 11, 2024

added reproduction for akkadotnet#7145

69d5b92

Aaronontheweb commented Apr 11, 2024

View reviewed changes

Aaronontheweb mentioned this pull request Apr 11, 2024

ExpectMsgAsync deadlocks with actor that has async receive handler and testprobe.Tell #7145

Closed

Aaronontheweb added 6 commits April 11, 2024 12:58

fixed issues with ExpectNoMsgAsync and ExpectNoMsg

3cee51b

fixed issues with TryPeekAsync

3fd7512

Merge branch 'dev' into fix-7145-testkit-AsyncQueue

3ba0a17

added API approvals

53ff0aa

Merge branch 'fix-7145-testkit-AsyncQueue' of https://github.com/Aaro…

1c48759

…nontheweb/akka.net into fix-7145-testkit-AsyncQueue

removed contention spec

0e2be2b

Aaronontheweb commented Apr 11, 2024

View reviewed changes

Aaronontheweb marked this pull request as ready for review April 11, 2024 22:16

Aaronontheweb added 2 commits April 11, 2024 17:18

harden spec

0a3b997

fix warning

6553639

Arkatufus requested changes Apr 12, 2024

View reviewed changes

Arkatufus approved these changes Apr 12, 2024

View reviewed changes

Merge branch 'dev' into fix-7145-testkit-AsyncQueue

166600f

Arkatufus enabled auto-merge (squash) April 12, 2024 14:08

Aaronontheweb disabled auto-merge April 12, 2024 14:09

Aaronontheweb merged commit 2057bae into akkadotnet:dev Apr 12, 2024
2 of 11 checks passed

Aaronontheweb deleted the fix-7145-testkit-AsyncQueue branch April 12, 2024 14:09

Arkatufus mentioned this pull request Apr 15, 2024

Update RELEASE_NOTES.md for 1.5.19 release #7159

Merged

SchiessMax mentioned this pull request Apr 18, 2024

FishForMessageAsync deadlocks with actor that has async receive handler and testprobe.Tell #7120

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Akka.TestKit: deleted `IAsyncQueue`; replaced with `System.Threading.Channel<T>` #7157

Akka.TestKit: deleted `IAsyncQueue`; replaced with `System.Threading.Channel<T>` #7157

Aaronontheweb commented Apr 11, 2024 •

edited

Aaronontheweb commented Apr 11, 2024

Aaronontheweb commented Apr 11, 2024

Aaronontheweb Apr 11, 2024

Aaronontheweb left a comment

Aaronontheweb Apr 11, 2024

Aaronontheweb Apr 11, 2024

Aaronontheweb Apr 11, 2024

Aaronontheweb Apr 11, 2024

Aaronontheweb Apr 11, 2024

Aaronontheweb Apr 11, 2024

Aaronontheweb Apr 11, 2024

Aaronontheweb Apr 11, 2024

Arkatufus left a comment

Arkatufus Apr 12, 2024

Aaronontheweb Apr 12, 2024


		namespace Akka.TestKit.Tests;

		public class Bugfix7145Spec : AkkaSpec

Akka.TestKit: deleted IAsyncQueue; replaced with System.Threading.Channel<T> #7157

Akka.TestKit: deleted IAsyncQueue; replaced with System.Threading.Channel<T> #7157

Conversation

Aaronontheweb commented Apr 11, 2024 • edited

Changes

Checklist

Aaronontheweb commented Apr 11, 2024

Aaronontheweb commented Apr 11, 2024

Choose a reason for hiding this comment

Aaronontheweb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Arkatufus left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Akka.TestKit: deleted `IAsyncQueue`; replaced with `System.Threading.Channel<T>` #7157

Akka.TestKit: deleted `IAsyncQueue`; replaced with `System.Threading.Channel<T>` #7157

Aaronontheweb commented Apr 11, 2024 •

edited