Skip to content

Keep a blocking receive loop on its dedicated thread past the first await (GH-4330 follow-up) #5096

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
jeremydmiller merged 1 commit into from
Oct 20, 2026
Merged

Keep a blocking receive loop on its dedicated thread past the first await (GH-4330 follow-up) #5096

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
150 changes: 150 additions & 0 deletions src/Testing/CoreTests/Transports/background_receive_loop_blocking_iteration_4330.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using JasperFx.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Wolverine.Transports;
using Xunit;

namespace CoreTests.Transports;

// GH-4330 gave BackgroundReceiveLoop an opt-in UsesBlockingIteration so that an iteration which blocks its thread
// (the Kafka consumer's Consume(CancellationToken)) runs on a dedicated thread instead of a pool worker. Starting the
// async loop with TaskCreationOptions.LongRunning only covered it up to its first await that did not complete
// synchronously: the loop then resumed on the pool thread that completed it, so from the second iteration on every
// blocking call held a pool worker for as long as it blocked.
public class background_receive_loop_blocking_iteration_4330
{
private static BackgroundReceiveLoop blockingLoop(Func<CancellationToken, Task<bool>> body)
=> new(new Uri("test://blocking-loop"), NullLogger.Instance, body, CancellationToken.None, 20.Milliseconds())
{
UsesBlockingIteration = true
};

[Fact]
public async Task the_blocking_part_of_every_iteration_runs_on_one_dedicated_thread()
{
var threads = new ConcurrentQueue<CallingThread>();
var enough = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);

await using var theLoop = blockingLoop(async token =>
{
// The synchronous part of an iteration -- Consume(token) in Kafka -- runs on whichever thread called it
threads.Enqueue(CallingThread.Current());
if (threads.Count >= 5)
{
enough.TrySetResult();
}

// ...and what it awaits completes on the pool, as the durable inbox insert behind ReceivedAsync does
await Task.Delay(5, token);
return true;
});

theLoop.Start();
await enough.Task.WaitAsync(10.Seconds(), TestContext.Current.CancellationToken);

var recorded = threads.ToArray();
recorded.ShouldAllBe(thread => !thread.IsThreadPoolThread);
recorded.Select(thread => thread.ManagedThreadId).Distinct().Count().ShouldBe(1);
}

[Fact]
public async Task failures_still_back_off_and_recover_on_the_dedicated_thread()
{
var fail = true;
var threads = new ConcurrentQueue<CallingThread>();

await using var theLoop = blockingLoop(async token =>
{
threads.Enqueue(CallingThread.Current());
await Task.Delay(5, token);

if (Volatile.Read(ref fail))
{
throw new InvalidOperationException("boom");
}

return true;
});

theLoop.Start();
await waitUntil(() => theLoop.ConsecutiveFailures >= 2);
theLoop.ReceiveLoopStatus.ShouldBe(ReceiveLoopStatus.Running);

Volatile.Write(ref fail, false);
await waitUntil(() => theLoop.ConsecutiveFailures == 0);

threads.ToArray().ShouldAllBe(thread => !thread.IsThreadPoolThread);
}

[Fact]
public async Task dispose_stops_an_iteration_blocked_the_way_consume_blocks()
{
var iterations = 0;
var blocked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);

var theLoop = blockingLoop(async token =>
{
if (Interlocked.Increment(ref iterations) == 1)
{
await Task.Yield();
return true;
}

blocked.TrySetResult();

// Consume(token) shape: hold the calling thread until the token is cancelled, then throw
token.WaitHandle.WaitOne();
token.ThrowIfCancellationRequested();
return true;
});

theLoop.Start();
await blocked.Task.WaitAsync(5.Seconds(), TestContext.Current.CancellationToken);

var stopwatch = Stopwatch.StartNew();
await theLoop.DisposeAsync().AsTask().WaitAsync(5.Seconds(), TestContext.Current.CancellationToken);
stopwatch.Elapsed.ShouldBeLessThan(2.Seconds());

theLoop.ReceiveLoopStatus.ShouldBe(ReceiveLoopStatus.Stopped);
}

[Fact]
public async Task a_loop_that_does_not_block_stays_on_the_pool()
{
var threads = new ConcurrentQueue<CallingThread>();

await using var theLoop = new BackgroundReceiveLoop(new Uri("test://async-loop"), NullLogger.Instance,
async token =>
{
threads.Enqueue(CallingThread.Current());
await Task.Delay(5, token);
return true;
}, CancellationToken.None, 20.Milliseconds());

theLoop.Start();
await waitUntil(() => threads.Count >= 3);

threads.ToArray().ShouldAllBe(thread => thread.IsThreadPoolThread);
}

// Captured when the iteration is called: a LongRunning thread can have exited by the time the test asserts, and
// reading IsThreadPoolThread from a dead Thread throws.
private record struct CallingThread(int ManagedThreadId, bool IsThreadPoolThread)
{
public static CallingThread Current()
=> new(Environment.CurrentManagedThreadId, Thread.CurrentThread.IsThreadPoolThread);
}

private static async Task waitUntil(Func<bool> condition, int timeoutMs = 5000)
{
var stopwatch = Stopwatch.StartNew();
while (!condition() && stopwatch.ElapsedMilliseconds < timeoutMs)
{
await Task.Delay(10);
}

condition().ShouldBeTrue();
}
}
7 changes: 6 additions & 1 deletion src/Transports/Kafka/Wolverine.Kafka/Internals/KafkaTopicGroupListener.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,12 @@ public KafkaTopicGroupListener(KafkaTopicGroup endpoint, ConsumerConfig config,

// Subscribe to all topics at once — single consumer, multiple topics.
_consumer.Subscribe(endpoint.TopicNames);
_loop = new BackgroundReceiveLoop(Address, logger, consumeOnceAsync, _cancellation.Token);
// GH-4330: the same blocking IConsumer.Consume(CancellationToken) as KafkaListener, so the same
// dedicated thread rather than a pool worker held for the listener's whole lifetime.
_loop = new BackgroundReceiveLoop(Address, logger, consumeOnceAsync, _cancellation.Token)
{
UsesBlockingIteration = true
};
_loop.Start();
}

Expand Down
183 changes: 115 additions & 68 deletions src/Wolverine/Transports/BackgroundReceiveLoop.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,12 @@ public DateTimeOffset? LastReceiveLoopActivityAt
/// from the pool the handler pipeline and every other transport's continuations share.
///
/// <para>
/// When set, the loop runs on a dedicated thread for its whole lifetime: each pass is waited on
/// rather than awaited, so the loop is back on that thread before it calls the iteration again.
/// What a pass awaits, such as the receiver's durable inbox insert, still completes on the pool.
/// </para>
///
/// <para>
/// Leave false for genuinely async iterations (SQS, Redis, the database queues): they release
/// their thread at every await, so a dedicated thread would waste one.
/// </para>
Expand All @@ -133,81 +139,46 @@ public void Start()
_status = ReceiveLoopStatus.Running;

_task = UsesBlockingIteration
? Task.Factory.StartNew(runAsync, _cancellation.Token, TaskCreationOptions.LongRunning,
TaskScheduler.Default).Unwrap()
? Task.Factory.StartNew(runOnDedicatedThread, _cancellation.Token, TaskCreationOptions.LongRunning,
TaskScheduler.Default)
: Task.Run(runAsync, _cancellation.Token);
}

private async Task runAsync()
{
try
{
while (!_cancellation.IsCancellationRequested)
while (!_cancellation.IsCancellationRequested && await runOnceAsync().ConfigureAwait(false))
{
// Heartbeat: bump before each iteration so a hung iteration shows a stale timestamp.
Interlocked.Exchange(ref _lastActivityTicks, DateTimeOffset.UtcNow.Ticks);
}

try
{
var didWork = await _iteration(_cancellation.Token).ConfigureAwait(false);
_consecutiveFailures = 0;

// GH-4215: a successful iteration clears an EntityMissing streak, so a loop that healed
// reports Running again rather than staying stuck on the diagnosis that got it there.
if (_status == ReceiveLoopStatus.EntityMissing)
{
_status = ReceiveLoopStatus.Running;
_logger.LogInformation("The receive loop for {Uri} is consuming again", _uri);
}

if (!didWork)
{
await WaitForWakeupOrTimeoutAsync(_idleDelay, _cancellation.Token).ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (_cancellation.IsCancellationRequested)
{
break;
}
catch (Exception e)
{
var failures = Interlocked.Increment(ref _consecutiveFailures);
var entityMissing = IsEntityMissing?.Invoke(e) ?? false;

if (entityMissing)
{
_status = ReceiveLoopStatus.EntityMissing;

// GH-4215: logged on the first failure of a streak and then every 60th, rather than every
// iteration. The old behaviour drowned out everything else the host logged, which is its
// own outage on top of the one being reported.
if (failures == 1 || failures % 60 == 0)
{
_logger.LogWarning(e,
"The broker entity for {Uri} does not exist (consecutive failures: {Count}). Retrying the receive cannot succeed until it is re-declared.",
_uri, failures);
}

await tryRedeclareAsync().ConfigureAwait(false);
}
else
{
_logger.LogError(e,
"Error in the receive loop for {Uri} (consecutive failures: {Count}); backing off and retrying",
_uri, failures);
}

try
{
await Task.Delay(backoffFor(failures, entityMissing), _cancellation.Token)
.ConfigureAwait(false);
}
catch (OperationCanceledException)
{
break;
}
}
_status = ReceiveLoopStatus.Stopped;
}
catch (OperationCanceledException)
{
_status = ReceiveLoopStatus.Stopped;
}
catch (Exception e)
{
markFaulted(e);
}
}

// GH-4330: starting runAsync with LongRunning only kept the loop on its dedicated thread until the first await
// that did not complete synchronously. The thread then exited, the loop resumed on whichever pool thread
// completed that await, and every later Consume(token) blocked a pool worker after all -- one per Kafka
// listener, held for as long as it waited for a record. Waiting on each pass here instead of awaiting it brings
// the loop back to this thread before every iteration. The pass's own continuations still run on the pool, and
// nothing ever needs this thread to complete them, so the wait cannot deadlock.
private void runOnDedicatedThread()
{
try
{
#pragma warning disable VSTHRD002 // Deliberate: blocking this dedicated thread is what keeps the next iteration on it
while (!_cancellation.IsCancellationRequested && runOnceAsync().GetAwaiter().GetResult())
{
}
#pragma warning restore VSTHRD002

_status = ReceiveLoopStatus.Stopped;
}
Expand All @@ -217,12 +188,88 @@ await Task.Delay(backoffFor(failures, entityMissing), _cancellation.Token)
}
catch (Exception e)
{
// The loop scaffolding itself fell over — this is the silent-death case GH-3236 exists to surface.
_status = ReceiveLoopStatus.Faulted;
_logger.LogError(e, "The receive loop for {Uri} terminated unexpectedly and is no longer consuming", _uri);
markFaulted(e);
}
}

// One pass of the loop: the iteration, then the idle wait or the failure backoff that follows it. Returns false
// once the loop has been cancelled and should stop.
private async Task<bool> runOnceAsync()
{
// Heartbeat: bump before each iteration so a hung iteration shows a stale timestamp.
Interlocked.Exchange(ref _lastActivityTicks, DateTimeOffset.UtcNow.Ticks);

try
{
var didWork = await _iteration(_cancellation.Token).ConfigureAwait(false);
_consecutiveFailures = 0;

// GH-4215: a successful iteration clears an EntityMissing streak, so a loop that healed
// reports Running again rather than staying stuck on the diagnosis that got it there.
if (_status == ReceiveLoopStatus.EntityMissing)
{
_status = ReceiveLoopStatus.Running;
_logger.LogInformation("The receive loop for {Uri} is consuming again", _uri);
}

if (!didWork)
{
await WaitForWakeupOrTimeoutAsync(_idleDelay, _cancellation.Token).ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (_cancellation.IsCancellationRequested)
{
return false;
}
catch (Exception e)
{
var failures = Interlocked.Increment(ref _consecutiveFailures);
var entityMissing = IsEntityMissing?.Invoke(e) ?? false;

if (entityMissing)
{
_status = ReceiveLoopStatus.EntityMissing;

// GH-4215: logged on the first failure of a streak and then every 60th, rather than every
// iteration. The old behaviour drowned out everything else the host logged, which is its
// own outage on top of the one being reported.
if (failures == 1 || failures % 60 == 0)
{
_logger.LogWarning(e,
"The broker entity for {Uri} does not exist (consecutive failures: {Count}). Retrying the receive cannot succeed until it is re-declared.",
_uri, failures);
}

await tryRedeclareAsync().ConfigureAwait(false);
}
else
{
_logger.LogError(e,
"Error in the receive loop for {Uri} (consecutive failures: {Count}); backing off and retrying",
_uri, failures);
}

try
{
await Task.Delay(backoffFor(failures, entityMissing), _cancellation.Token)
.ConfigureAwait(false);
}
catch (OperationCanceledException)
{
return false;
}
}

return true;
}

private void markFaulted(Exception e)
{
// The loop scaffolding itself fell over — this is the silent-death case GH-3236 exists to surface.
_status = ReceiveLoopStatus.Faulted;
_logger.LogError(e, "The receive loop for {Uri} terminated unexpectedly and is no longer consuming", _uri);
}

// Mirrors the historical SQS backoff curve: a gentle ramp, capped at one second.
// GH-4215: a missing entity gets a far longer floor than a transient blip. Retrying the receive cannot
// succeed until something re-declares, so a one-second cadence buys nothing and costs a log storm.
Expand Down
Loading