Streams ensure cts cancel fix statecheck #6935
Conversation
…confirm failure and provide future regression checks
| _enumerable = enumerable; | ||
| _outlet = shape.Outlet; | ||
| _onSuccess = GetAsyncCallback<T>(OnSuccess); | ||
| _onFailure = GetAsyncCallback<Exception>(OnFailure); | ||
| _onComplete = GetAsyncCallback(OnComplete); | ||
|
|
||
| _completionCts = new CancellationTokenSource(); |
There was a problem hiding this comment.
Controversial possibly, but I felt it safer to move this to ctor.
| @@ -3859,26 +3880,12 @@ public override void OnPull() | |||
| // if result is false, it means enumerator was closed. Complete stage in that case. | |||
| CompleteStage(); | |||
| } | |||
| } | |||
| else if (vtask.IsCompleted) // IsCompleted covers Faulted, Cancelled, and RanToCompletion async state | |||
There was a problem hiding this comment.
This is racy, so removed. Exception case is not as big a deal anyway.
| #pragma warning disable CS4014 | ||
| ProcessTask(); | ||
| #pragma warning restore CS4014 | ||
| _ = ProcessTask(); |
There was a problem hiding this comment.
Discard = no pragma disable needed
| } | ||
| } | ||
|
|
||
| public override void OnDownstreamFinish(Exception cause) | ||
| { | ||
| _completionCts.Cancel(); | ||
| _completionCts.Dispose(); |
There was a problem hiding this comment.
Moved to PostStop(). AFAIK we will call it after CompleteStage() in interpreter, plz correct if I'm wrong.
There was a problem hiding this comment.
I believe that's the case.
| _completionCts.Cancel(); | ||
| _completionCts.Dispose(); |
There was a problem hiding this comment.
We should ALWAYS be disposing the CTS.
What is more interesting to ask, is whether it's worth us having some sort of flag to know whether we should bother canceling or not. I went this route because in theory, under happy-completion path nobody cares about this CTS anyway.
That said there -may- be value in some cases of not calling cancel, but can't think of any.
| } | ||
| try | ||
| { | ||
| _enumerator.DisposeAsync().ConfigureAwait(false); |
There was a problem hiding this comment.
Just feels like the right thing to do.
There was a problem hiding this comment.
I think that this is the actual fix, the enumerator needs to be disposed for the resources to be released.
| [Fact] | ||
| public async Task AsyncEnumerableSource_Disposes_OnCancel() | ||
| { | ||
| var resource = new Resource(); | ||
| var tcs = new System.Threading.Tasks.TaskCompletionSource<NotUsed>(TaskCreationOptions | ||
| .RunContinuationsAsynchronously); | ||
| var src = Source.From(() => | ||
| CancelTestGenerator(tcs, resource, default)); | ||
| src.To(Sink.Ignore<int>()).Run(Materializer); | ||
| await tcs.Task; | ||
| Materializer.Shutdown(); | ||
| await Task.Delay(500); | ||
| Assert.False(resource.IsActive); | ||
| } | ||
|
|
There was a problem hiding this comment.
Adapted from #6903 report.
Calling Materializer.Shutdown() is, as far as I know, going to happen when the actorsystem shuts down.
OTOH this test was added in a red->green refactoring, so I know without the changes in AsyncEnumerable it fails.
| _enumerable = enumerable; | ||
| _outlet = shape.Outlet; | ||
| _onSuccess = GetAsyncCallback<T>(OnSuccess); | ||
| _onFailure = GetAsyncCallback<Exception>(OnFailure); | ||
| _onComplete = GetAsyncCallback(OnComplete); | ||
|
|
||
| _completionCts = new CancellationTokenSource(); |
| #pragma warning disable CS4014 | ||
| ProcessTask(); | ||
| #pragma warning restore CS4014 | ||
| _ = ProcessTask(); |
| } | ||
| } | ||
|
|
||
| public override void OnDownstreamFinish(Exception cause) | ||
| { | ||
| _completionCts.Cancel(); | ||
| _completionCts.Dispose(); |
There was a problem hiding this comment.
I believe that's the case.
Fixes #6903
Changes
Changed cleanup pattern to fix AsyncEnumerable stage, also fixed race condition in ValueTask state check.
Checklist
For significant changes, please ensure that the following have been completed (delete if not relevant):