From 3c67d4b35a1b4d2cab46de1626bb697184697724 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 29 Sep 2026 20:37:30 +0000 Subject: [PATCH] Akka.Streams: fix a possible lost wakeup in MergeHub (missing memory fence) MergeHub's consumer stores _needWakeup = true and then re-polls the ConcurrentQueue. The queue's empty path only does volatile reads, so the store could be reordered after the re-poll (store-load reordering, legal on x86 and ARM). A producer could then enqueue, read _needWakeup == false and skip the wakeup while the consumer saw an empty queue. The hub stalled until the next enqueue, which never comes when perProducerBufferSize is 1. JVM Akka marks the flag @volatile, which is sequentially consistent. C# volatile is not, so add a full fence after the store, and the same after _shuttingDown = true in PostStop, which has the same shape. Mark both fields volatile to match upstream. Un-skip MergeHub_must_work_with_long_streams_when_buffer_size_is_1. --- src/core/Akka.Streams.Tests/Dsl/HubSpec.cs | 11 +++++------ src/core/Akka.Streams/Dsl/Hub.cs | 14 ++++++++++++-- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/src/core/Akka.Streams.Tests/Dsl/HubSpec.cs b/src/core/Akka.Streams.Tests/Dsl/HubSpec.cs index 08070b8ffa4..af6e43d81a9 100644 --- a/src/core/Akka.Streams.Tests/Dsl/HubSpec.cs +++ b/src/core/Akka.Streams.Tests/Dsl/HubSpec.cs @@ -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 () => @@ -189,13 +189,12 @@ await this.AssertAllStagesStoppedAsync(async () => .Take(20000) .ToMaterialized(Sink.Seq(), 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] diff --git a/src/core/Akka.Streams/Dsl/Hub.cs b/src/core/Akka.Streams/Dsl/Hub.cs index fe7dd7482d0..eb53e2a8afb 100644 --- a/src/core/Akka.Streams/Dsl/Hub.cs +++ b/src/core/Akka.Streams/Dsl/Hub.cs @@ -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; @@ -177,8 +178,9 @@ private sealed class HubLogic : OutGraphStageLogic private readonly AtomicCounterLong _producerCount; private readonly Dictionary _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 stage, AtomicCounterLong producerCount) : base(stage.Shape) { @@ -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)) @@ -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; } @@ -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) {