Skip to content

Fix #7578, Cancelled sinks are blocking other BroadcastHub consumers #7579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 25 commits into from
Closed

Fix #7578, Cancelled sinks are blocking other BroadcastHub consumers #7579

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
0d7a04e
Add reproduction unit test
Arkatufus Mar 24, 2025
428195e
Port https://github.com/akka/akka/pull/23970
Arkatufus Mar 25, 2025
aaefe69
Update API Approval list
Arkatufus Mar 25, 2025
3772f15
Update API Approval list
Arkatufus Mar 25, 2025
59070d0
Add async callback unit tests
Arkatufus Mar 25, 2025
59ab4ad
skip non-implemented feature
Arkatufus Mar 26, 2025
7a9e8f6
Fix TaskCompletionSource allocation problem
Arkatufus Mar 26, 2025
647a481
Code cleanup
Arkatufus Mar 26, 2025
2a1ded1
Implement early callback from https://github.com/akka/akka/pull/23185
Arkatufus Mar 27, 2025
5712140
Make sure early callbacks also get cancelled on stop
Arkatufus Mar 27, 2025
6d70d58
Fix naming and copyright header
Arkatufus Mar 28, 2025
6ccadf1
Update API Approval list
Arkatufus Mar 28, 2025
adb41f7
Fix missing feedback callback
Arkatufus Mar 31, 2025
f779ce3
rerun tests
Arkatufus Mar 31, 2025
a6499c0
Implement locks
Arkatufus Mar 31, 2025
2a7ffe0
cleanup lock code
Arkatufus Mar 31, 2025
f680ea7
extend locking to `_callbackWaitingForInterpreter`
Arkatufus Mar 31, 2025
0a1d84f
Fix unit test
Arkatufus Mar 31, 2025
bf53eeb
rerun tests
Arkatufus Mar 31, 2025
1dd0fe8
Merge branch 'dev' into #7578-Fix-BroadcastHub-block-bug
Aaronontheweb Apr 3, 2025
aa91f10
Merge branch 'dev' into #7578-Fix-BroadcastHub-block-bug
Arkatufus Apr 4, 2025
2f08067
Merge branch 'dev' into #7578-Fix-BroadcastHub-block-bug
Aaronontheweb Apr 18, 2025
ffc1c8e
Merge branch 'dev' into #7578-Fix-BroadcastHub-block-bug
Aaronontheweb Apr 18, 2025
543df2f
Merge branch 'dev' into #7578-Fix-BroadcastHub-block-bug
Aaronontheweb Apr 25, 2025
9a54eac
Merge branch 'dev' into #7578-Fix-BroadcastHub-block-bug
Aaronontheweb May 1, 2025
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
12 changes: 8 additions & 4 deletions src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveStreams.DotNet.verified.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3967,7 +3967,8 @@ namespace Akka.Streams.Implementation.Fusing
public readonly object Event;
public readonly System.Action<object> Handler;
public readonly Akka.Streams.Stage.GraphStageLogic Logic;
public AsyncInput(Akka.Streams.Implementation.Fusing.GraphInterpreterShell shell, Akka.Streams.Stage.GraphStageLogic logic, object @event, System.Action<object> handler) { }
public readonly System.Threading.Tasks.TaskCompletionSource<Akka.Done> Promise;
public AsyncInput(Akka.Streams.Implementation.Fusing.GraphInterpreterShell shell, Akka.Streams.Stage.GraphStageLogic logic, object @event, System.Threading.Tasks.TaskCompletionSource<Akka.Done> promise, System.Action<object> handler) { }
public Akka.Streams.Implementation.Fusing.GraphInterpreterShell Shell { get; }
}
public class BatchingActorInputBoundary : Akka.Streams.Implementation.Fusing.GraphInterpreter.UpstreamBoundaryStageLogic
Expand Down Expand Up @@ -4202,7 +4203,7 @@ namespace Akka.Streams.Implementation.Fusing
public readonly Akka.Streams.Stage.GraphStageLogic[] Logics;
public readonly Akka.Streams.IMaterializer Materializer;
public const Akka.Streams.Implementation.Fusing.GraphInterpreter.Connection NoEvent = null;
public readonly System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Action<object>> OnAsyncInput;
public readonly System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Threading.Tasks.TaskCompletionSource<Akka.Done>, System.Action<object>> OnAsyncInput;
public const int OutClosed = 32;
public const int OutReady = 8;
public const int PullEndFlip = 10;
Expand All @@ -4213,7 +4214,7 @@ namespace Akka.Streams.Implementation.Fusing
public const int Pushing = 4;
public int RunningStagesCount;
public static readonly Akka.Streams.Attributes[] SingleNoAttribute;
public GraphInterpreter(Akka.Streams.Implementation.Fusing.GraphAssembly assembly, Akka.Streams.IMaterializer materializer, Akka.Event.ILoggingAdapter log, Akka.Streams.Stage.GraphStageLogic[] logics, Connection[] connections, System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Action<object>> onAsyncInput, bool fuzzingMode, Akka.Actor.IActorRef context) { }
public GraphInterpreter(Akka.Streams.Implementation.Fusing.GraphAssembly assembly, Akka.Streams.IMaterializer materializer, Akka.Event.ILoggingAdapter log, Akka.Streams.Stage.GraphStageLogic[] logics, Connection[] connections, System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Threading.Tasks.TaskCompletionSource<Akka.Done>, System.Action<object>> onAsyncInput, bool fuzzingMode, Akka.Actor.IActorRef context) { }
public Akka.Actor.IActorRef Context { get; }
public static Akka.Streams.Implementation.Fusing.GraphInterpreter Current { get; }
public static Akka.Streams.Implementation.Fusing.GraphInterpreter CurrentInterpreterOrNull { get; }
Expand All @@ -4228,7 +4229,7 @@ namespace Akka.Streams.Implementation.Fusing
public int Execute(int eventLimit) { }
public void Finish() { }
public void Init(Akka.Streams.IMaterializer subMaterializer) { }
public void RunAsyncInput(Akka.Streams.Stage.GraphStageLogic logic, object evt, System.Action<object> handler) { }
public void RunAsyncInput(Akka.Streams.Stage.GraphStageLogic logic, object evt, System.Threading.Tasks.TaskCompletionSource<Akka.Done> promise, System.Action<object> handler) { }
public void SetHandler(Akka.Streams.Implementation.Fusing.GraphInterpreter.Connection connection, Akka.Streams.Stage.IInHandler handler) { }
public void SetHandler(Akka.Streams.Implementation.Fusing.GraphInterpreter.Connection connection, Akka.Streams.Stage.IOutHandler handler) { }
public override string ToString() { }
Expand Down Expand Up @@ -4884,6 +4885,7 @@ namespace Akka.Streams.Stage
public static readonly Akka.Streams.Stage.InHandler IgnoreTerminateInput;
public static readonly Akka.Streams.Stage.OutHandler IgnoreTerminateOutput;
public readonly int InCount;
public static readonly System.Threading.Tasks.TaskCompletionSource<Akka.Done> NoPromise;
public readonly int OutCount;
public static readonly Akka.Streams.Stage.InHandler TotallyIgnorantInput;
protected GraphStageLogic(int inCount, int outCount) { }
Expand Down Expand Up @@ -4918,6 +4920,8 @@ namespace Akka.Streams.Stage
public void FailStage(System.Exception reason) { }
protected System.Action<T> GetAsyncCallback<T>(System.Action<T> handler) { }
protected System.Action GetAsyncCallback(System.Action handler) { }
protected System.Func<T, System.Threading.Tasks.Task<Akka.Done>> GetAsyncCallbackWithTask<T>(System.Action<T> handler) { }
protected System.Func<System.Threading.Tasks.Task> GetAsyncCallbackWithTask(System.Action handler) { }
protected Akka.Streams.Stage.IInHandler GetHandler<T>(Akka.Streams.Inlet<T> inlet) { }
protected Akka.Streams.Stage.IOutHandler GetHandler<T>(Akka.Streams.Outlet<T> outlet) { }
[Akka.Annotations.ApiMayChangeAttribute()]
Expand Down
12 changes: 8 additions & 4 deletions src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveStreams.Net.verified.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3949,7 +3949,8 @@ namespace Akka.Streams.Implementation.Fusing
public readonly object Event;
public readonly System.Action<object> Handler;
public readonly Akka.Streams.Stage.GraphStageLogic Logic;
public AsyncInput(Akka.Streams.Implementation.Fusing.GraphInterpreterShell shell, Akka.Streams.Stage.GraphStageLogic logic, object @event, System.Action<object> handler) { }
public readonly System.Threading.Tasks.TaskCompletionSource<Akka.Done> Promise;
public AsyncInput(Akka.Streams.Implementation.Fusing.GraphInterpreterShell shell, Akka.Streams.Stage.GraphStageLogic logic, object @event, System.Threading.Tasks.TaskCompletionSource<Akka.Done> promise, System.Action<object> handler) { }
public Akka.Streams.Implementation.Fusing.GraphInterpreterShell Shell { get; }
}
public class BatchingActorInputBoundary : Akka.Streams.Implementation.Fusing.GraphInterpreter.UpstreamBoundaryStageLogic
Expand Down Expand Up @@ -4175,7 +4176,7 @@ namespace Akka.Streams.Implementation.Fusing
public readonly Akka.Streams.Stage.GraphStageLogic[] Logics;
public readonly Akka.Streams.IMaterializer Materializer;
public const Akka.Streams.Implementation.Fusing.GraphInterpreter.Connection NoEvent = null;
public readonly System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Action<object>> OnAsyncInput;
public readonly System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Threading.Tasks.TaskCompletionSource<Akka.Done>, System.Action<object>> OnAsyncInput;
public const int OutClosed = 32;
public const int OutReady = 8;
public const int PullEndFlip = 10;
Expand All @@ -4186,7 +4187,7 @@ namespace Akka.Streams.Implementation.Fusing
public const int Pushing = 4;
public int RunningStagesCount;
public static readonly Akka.Streams.Attributes[] SingleNoAttribute;
public GraphInterpreter(Akka.Streams.Implementation.Fusing.GraphAssembly assembly, Akka.Streams.IMaterializer materializer, Akka.Event.ILoggingAdapter log, Akka.Streams.Stage.GraphStageLogic[] logics, Connection[] connections, System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Action<object>> onAsyncInput, bool fuzzingMode, Akka.Actor.IActorRef context) { }
public GraphInterpreter(Akka.Streams.Implementation.Fusing.GraphAssembly assembly, Akka.Streams.IMaterializer materializer, Akka.Event.ILoggingAdapter log, Akka.Streams.Stage.GraphStageLogic[] logics, Connection[] connections, System.Action<Akka.Streams.Stage.GraphStageLogic, object, System.Threading.Tasks.TaskCompletionSource<Akka.Done>, System.Action<object>> onAsyncInput, bool fuzzingMode, Akka.Actor.IActorRef context) { }
public Akka.Actor.IActorRef Context { get; }
public static Akka.Streams.Implementation.Fusing.GraphInterpreter Current { get; }
public static Akka.Streams.Implementation.Fusing.GraphInterpreter CurrentInterpreterOrNull { get; }
Expand All @@ -4201,7 +4202,7 @@ namespace Akka.Streams.Implementation.Fusing
public int Execute(int eventLimit) { }
public void Finish() { }
public void Init(Akka.Streams.IMaterializer subMaterializer) { }
public void RunAsyncInput(Akka.Streams.Stage.GraphStageLogic logic, object evt, System.Action<object> handler) { }
public void RunAsyncInput(Akka.Streams.Stage.GraphStageLogic logic, object evt, System.Threading.Tasks.TaskCompletionSource<Akka.Done> promise, System.Action<object> handler) { }
public void SetHandler(Akka.Streams.Implementation.Fusing.GraphInterpreter.Connection connection, Akka.Streams.Stage.IInHandler handler) { }
public void SetHandler(Akka.Streams.Implementation.Fusing.GraphInterpreter.Connection connection, Akka.Streams.Stage.IOutHandler handler) { }
public override string ToString() { }
Expand Down Expand Up @@ -4857,6 +4858,7 @@ namespace Akka.Streams.Stage
public static readonly Akka.Streams.Stage.InHandler IgnoreTerminateInput;
public static readonly Akka.Streams.Stage.OutHandler IgnoreTerminateOutput;
public readonly int InCount;
public static readonly System.Threading.Tasks.TaskCompletionSource<Akka.Done> NoPromise;
public readonly int OutCount;
public static readonly Akka.Streams.Stage.InHandler TotallyIgnorantInput;
protected GraphStageLogic(int inCount, int outCount) { }
Expand Down Expand Up @@ -4891,6 +4893,8 @@ namespace Akka.Streams.Stage
public void FailStage(System.Exception reason) { }
protected System.Action<T> GetAsyncCallback<T>(System.Action<T> handler) { }
protected System.Action GetAsyncCallback(System.Action handler) { }
protected System.Func<T, System.Threading.Tasks.Task<Akka.Done>> GetAsyncCallbackWithTask<T>(System.Action<T> handler) { }
protected System.Func<System.Threading.Tasks.Task> GetAsyncCallbackWithTask(System.Action handler) { }
protected Akka.Streams.Stage.IInHandler GetHandler<T>(Akka.Streams.Inlet<T> inlet) { }
protected Akka.Streams.Stage.IOutHandler GetHandler<T>(Akka.Streams.Outlet<T> outlet) { }
[Akka.Annotations.ApiMayChangeAttribute()]
Expand Down
32 changes: 32 additions & 0 deletions src/core/Akka.Streams.Tests/Dsl/HubSpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -615,6 +615,38 @@ await Awaiting(async () =>
}, Materializer);
}

[Fact]
public async Task BroadcastHub_must_handle_cancelled_Sink()
Copy link
Contributor Author

@Arkatufus Arkatufus Mar 26, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The actual reproduction

{
await this.AssertAllStagesStoppedAsync(async () =>
{
var upstream = this.CreatePublisherProbe<int>();
var hubSource = Source.FromPublisher(upstream).RunWith(BroadcastHub.Sink<int>(4), Materializer);
var downstream = this.CreateSubscriberProbe<int>();

hubSource.RunWith(Sink.Cancelled<int>(), Materializer);
hubSource.RunWith(Sink.FromSubscriber(downstream), Materializer);

await downstream.EnsureSubscriptionAsync();

await downstream.RequestAsync(10);
await upstream.ExpectRequestAsync();
await upstream.SendNextAsync(1);
await downstream.ExpectNextAsync(1);
await upstream.SendNextAsync(2);
await downstream.ExpectNextAsync(2);
await upstream.SendNextAsync(3);
await downstream.ExpectNextAsync(3);
await upstream.SendNextAsync(4);
await downstream.ExpectNextAsync(4);
await upstream.SendNextAsync(5);
await downstream.ExpectNextAsync(5);

await upstream.SendCompleteAsync();
await downstream.ExpectCompleteAsync();
}, Materializer);
}

[Fact]
public async Task PartitionHub_must_work_in_the_happy_case_with_one_stream()
{
Expand Down
Loading