Skip to content

Akka.Streams: fix a possible lost wakeup in MergeHub (missing memory fence) #8665

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

Merged
Aaronontheweb merged 1 commit into from
Sep 29, 2026
Merged

Akka.Streams: fix a possible lost wakeup in MergeHub (missing memory fence) #8665

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
11 changes: 5 additions & 6 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 @@ -180,7 +180,7 @@ await this.AssertAllStagesStoppedAsync(async () =>
}, Materializer);
}

[Fact(Skip = "Very racy")] // @Aaronontheweb - I believe the issue here is a genuine race condition with the hub itself. Messages start getting read before any subscribers are attached. Very tricky N+1 error.
[Fact]
public async Task MergeHub_must_work_with_long_streams_when_buffer_size_is_1()
{
await this.AssertAllStagesStoppedAsync(async () =>
Expand All @@ -189,13 +189,12 @@ await this.AssertAllStagesStoppedAsync(async () =>
.Take(20000)
.ToMaterialized(Sink.Seq<int>(), Keep.Both)
.Run(Materializer);



Source.From(Enumerable.Range(1, 10000)).RunWith(sink, Materializer);
Source.From(Enumerable.Range(10001, 10000)).RunWith(sink, Materializer);
(await result).OrderBy(x => x).Should().BeEquivalentTo(Enumerable.Range(1, 20000));
}, Materializer, 10.Seconds());

(await result.WaitAsync(10.Seconds())).OrderBy(x => x).Should().BeEquivalentTo(Enumerable.Range(1, 20000));
}, Materializer, 15.Seconds());
}

[Fact]
Expand Down
14 changes: 12 additions & 2 deletions src/core/Akka.Streams/Dsl/Hub.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Akka.Annotations;
using Akka.Streams.Stage;
Expand Down Expand Up @@ -177,8 +178,9 @@ private sealed class HubLogic : OutGraphStageLogic
private readonly AtomicCounterLong _producerCount;
private readonly Dictionary<long, InputState> _demands = new();
private Action _wakeupCallback;
private bool _needWakeup;
private bool _shuttingDown;
// volatile also publishes _wakeupCallback to producers that see _needWakeup == true
private volatile bool _needWakeup;
private volatile bool _shuttingDown;

public HubLogic(MergeHub<T> stage, AtomicCounterLong producerCount) : base(stage.Shape)
{
Expand All @@ -198,6 +200,8 @@ public override void PostStop()
{
// First announce that we are shutting down. This will notify late-comers to not even put anything in the queue
_shuttingDown = true;
// Full fence: see the comment in TryProcessNext. Producers check the flag after they enqueue.
Interlocked.MemoryBarrier();

// Anybody that missed the announcement needs to be notified.
while (_queue.TryDequeue(out var e))
Expand Down Expand Up @@ -258,6 +262,9 @@ private void TryProcessNext(bool firstAttempt)
_needWakeup = true;
if (firstAttempt)
{
// Full fence so the store above can't be reordered after the re-poll. C# volatile
// doesn't prevent that (JVM @volatile does); the proof in Enqueue relies on it.
Interlocked.MemoryBarrier();
firstAttempt = false;
continue;
}
Expand Down Expand Up @@ -298,6 +305,9 @@ public void Enqueue(IEvent ev)
// wakeup (otherwise needWakeup = true). Now, if the consumer is still running (2) is violated,
// if not running then needWakeup = false is violated (which comes from (6)). No matter what,
// contradiction. QED.
//
// The proof assumes sequentially consistent accesses (JVM @volatile). In .NET that takes the full fence
// in TryProcessNext; on this side the CAS inside _queue.Enqueue already acts as one.

if (_needWakeup)
{
Expand Down
Loading