diff --git a/AZURESERVICEBUS-PERF-DEEP-DIVE-PLAN.md b/AZURESERVICEBUS-PERF-DEEP-DIVE-PLAN.md index 964c85ffc..6eab35979 100644 --- a/AZURESERVICEBUS-PERF-DEEP-DIVE-PLAN.md +++ b/AZURESERVICEBUS-PERF-DEEP-DIVE-PLAN.md @@ -160,7 +160,19 @@ feed the #3488/GH-3471 release notes directly** — that's the first ledger entr | Optimization | PR | Scenario (AE-cell) | Metric | Before | After | Release-note one-liner | |---|---|---|---|---|---|---| -| _(PrefetchCount validation (#3488), AO2 session de-quadratic, AO4 settlement concurrency, ... — rows added as measured)_ | | | | | | | +| AO2 session de-quadratic (accept-loop listener) | this wave | AE5 | concurrent accept loops at `RequireSessions(5)` | 25 | 5 | `RequireSessions(n)` opens n session accept loops instead of n-squared | +| AO2 session de-quadratic (`ServiceBusSessionProcessor` path, GH-3533) | this wave | AE5 | concurrent sessions at `RequireSessions(8)` | 64 | 8 | Same n-squared on the session-processor path: `MaxConcurrentSessions` was set per processor while ListeningAgent was already building n of them | +| AO3 `MaximumConcurrentCalls` | this wave | AE1 | inline handler concurrency | 1 (SDK default, unreachable) | configurable | Inline Azure Service Bus listeners can now process messages concurrently without dropping to the raw `ConfigureProcessor` hook | +| AO8 batched defer settles the original | this wave | AE8 | duplicates per deferral | 2 deliveries | 1 | Deferring on a buffered/durable Azure Service Bus listener no longer leaves the original message locked to be redelivered | + +**Not measured on a real namespace.** AO2's 25→5 is structural and unit-tested; AO3 and AO8 are +correctness/ergonomics fixes verified by tests. The throughput numbers this plan asks for -- and +the PrefetchCount validation in AE2 -- still need the real Standard/Premium namespace runs +described in §2 and §8. The emulator is explicitly not publishable. + +**Deferred, needs a JasperFx change:** AO4 (settlement concurrency). `RetryBlock` hard-codes a +parallel count of 1 (`Block(1, Unbounded, executeAsync)`), so widening ASB's per-message +`CompleteMessageAsync` concurrency is not a Wolverine-side change. Track upstream before revisiting. ## 7. Sequencing & exit criteria diff --git a/docs/guide/messaging/transports/azureservicebus/performance.md b/docs/guide/messaging/transports/azureservicebus/performance.md index 7ea3ba151..46ceb27fa 100644 --- a/docs/guide/messaging/transports/azureservicebus/performance.md +++ b/docs/guide/messaging/transports/azureservicebus/performance.md @@ -36,15 +36,23 @@ means silent redelivery and a rising delivery count. ### Inline endpoints process one message at a time by default Inline ASB endpoints use a `ServiceBusProcessor`, whose `MaxConcurrentCalls` defaults to **1**. -Wolverine does not change that default, so an inline listener is single-threaded unless you -raise it: +Wolverine leaves that default alone, so an inline listener is single-threaded unless you raise it: ```cs opts.ListenToAzureServiceBusQueue("orders") .ProcessInline() - .ConfigureProcessor(o => o.MaxConcurrentCalls = 10); + .MaximumConcurrentCalls(10); ``` +`MaximumParallelMessages` has no effect on an inline endpoint — that knob sizes Wolverine's own +in-process worker queue, which inline listeners bypass. The raw +`ConfigureProcessor(o => o.MaxConcurrentCalls = 10)` hook still works and takes precedence. + +On a **session** listener driven by a `ServiceBusSessionProcessor`, `MaximumConcurrentCalls` maps +to `MaxConcurrentCallsPerSession` instead, which trades away the per-session FIFO ordering that is +usually the reason for using sessions at all. Leave it alone there unless you mean it — to process +more sessions at once, use `RequireSessions(n)`. + ## Lock duration vs. processing window For Buffered/Durable endpoints, Wolverine does not renew message locks while messages wait in @@ -86,6 +94,7 @@ outgoing batches are additionally grouped by session id so each batch shares a p Sessions give broker-enforced ordering per `SessionId` (mapped automatically from Wolverine's `Envelope.GroupId`) with cluster-wide exclusivity — but session processing is inherently more expensive than plain consumption: each session must be accepted, locked, drained, and released. +`RequireSessions(n)` opens exactly `n` concurrent session accept loops — one per listener. Keep `RequireSessions(n)` counts modest, and note that strict per-session *processing* order on Buffered/Durable endpoints also needs `PartitionProcessingByGroupId(...)` (or inline execution), since the local worker queue otherwise executes a session's batch in parallel — diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs index e91f34855..9abe38151 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs @@ -77,7 +77,7 @@ public void build_session_processor_options_reasserts_peek_lock_and_disables_aut } [Fact] - public void build_session_processor_options_maps_listener_count_to_max_concurrent_sessions() + public void build_session_processor_options_accepts_one_session_per_processor() { var transport = new AzureServiceBusTransport(); var queue = transport.Queues["incoming"]; @@ -88,12 +88,30 @@ public void build_session_processor_options_maps_listener_count_to_max_concurren ((IDelayedEndpointConfiguration)configuration).Apply(); var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue); - options.MaxConcurrentSessions.ShouldBe(8); + + // GH-3494 (AO2): ListeningAgent builds ListenerCount of these listeners, so mapping + // RequireSessions(8) onto each processor's MaxConcurrentSessions gave 8 x 8 = 64 + // concurrent sessions. One per processor keeps the total at the 8 that was asked for. + options.MaxConcurrentSessions.ShouldBe(1); // FIFO ordering per session is preserved options.MaxConcurrentCallsPerSession.ShouldBe(1); } + [Fact] + public void a_user_can_still_override_max_concurrent_sessions() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + configuration.RequireSessions(8).ConfigureSessionProcessor(o => o.MaxConcurrentSessions = 3); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + AzureServiceBusTransport.BuildSessionProcessorOptions(queue).MaxConcurrentSessions.ShouldBe(3); + } + [Fact] public void build_session_processor_options_composes_multiple_actions() { diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/maximum_concurrent_calls_3494.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/maximum_concurrent_calls_3494.cs new file mode 100644 index 000000000..f5762eb92 --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/maximum_concurrent_calls_3494.cs @@ -0,0 +1,98 @@ +using Shouldly; +using Wolverine.AzureServiceBus.Internal; +using Wolverine.Configuration; +using Xunit; + +namespace Wolverine.AzureServiceBus.Tests; + +/// +/// GH-3494 (AO3). Wolverine never set MaxConcurrentCalls, so an inline Azure Service Bus listener +/// ran on the SDK default of 1 -- one message at a time per endpoint -- reachable only through the +/// raw ConfigureProcessor hook. +/// +public class maximum_concurrent_calls_3494 +{ + [Fact] + public void defaults_to_null_so_the_sdk_default_still_applies() + { + var transport = new AzureServiceBusTransport(); + transport.Queues["incoming"].MaximumConcurrentCalls.ShouldBeNull(); + } + + [Fact] + public void must_be_at_least_one() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + Should.Throw(() => queue.MaximumConcurrentCalls = 0); + } + + [Fact] + public void queue_listener_configuration_sets_it() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + configuration.MaximumConcurrentCalls(12); + ((IDelayedEndpointConfiguration)configuration).Apply(); + + queue.MaximumConcurrentCalls.ShouldBe(12); + } + + [Fact] + public void subscription_listener_configuration_sets_it() + { + var transport = new AzureServiceBusTransport(); + var subscription = transport.Topics["topic1"].FindOrCreateSubscription("sub1"); + + var configuration = new AzureServiceBusSubscriptionListenerConfiguration(subscription); + configuration.MaximumConcurrentCalls(4); + ((IDelayedEndpointConfiguration)configuration).Apply(); + + subscription.MaximumConcurrentCalls.ShouldBe(4); + } + + [Fact] + public void flows_onto_the_processor_options() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + queue.MaximumConcurrentCalls = 8; + + AzureServiceBusTransport.BuildProcessorOptions(queue).MaxConcurrentCalls.ShouldBe(8); + } + + [Fact] + public void configure_processor_still_wins_over_the_endpoint_value() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + queue.MaximumConcurrentCalls = 8; + queue.ConfigureProcessor = o => o.MaxConcurrentCalls = 3; + + AzureServiceBusTransport.BuildProcessorOptions(queue).MaxConcurrentCalls.ShouldBe(3); + } + + [Fact] + public void session_processor_keeps_per_session_fifo_unless_asked_otherwise() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + queue.ConfigureSessionProcessor = _ => { }; + + AzureServiceBusTransport.BuildSessionProcessorOptions(queue).MaxConcurrentCallsPerSession.ShouldBe(1); + } + + [Fact] + public void session_processor_honors_the_opt_in() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + queue.MaximumConcurrentCalls = 5; + queue.ConfigureSessionProcessor = _ => { }; + + AzureServiceBusTransport.BuildSessionProcessorOptions(queue).MaxConcurrentCallsPerSession.ShouldBe(5); + } +} diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_listener_accept_loops_3494.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_listener_accept_loops_3494.cs new file mode 100644 index 000000000..3497be4eb --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_listener_accept_loops_3494.cs @@ -0,0 +1,44 @@ +using Azure.Messaging.ServiceBus; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Shouldly; +using Wolverine.AzureServiceBus.Internal; +using Wolverine.Transports; +using Wolverine.Transports.Sending; +using Xunit; + +namespace Wolverine.AzureServiceBus.Tests; + +/// +/// GH-3494 (AO2). ListeningAgent already builds Endpoint.ListenerCount listeners, so the session +/// listener spawning ListenerCount accept loops of its own made RequireSessions(n) open n-squared +/// concurrent AcceptNextSessionAsync loops -- 25 of them for RequireSessions(5). +/// +public class session_listener_accept_loops_3494 +{ + private static AzureServiceBusSessionListener buildListener(int listenerCount) + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + queue.Options.RequiresSession = true; + queue.ListenerCount = listenerCount; + + // The accept loop itself immediately fails against a transport with no live connection and + // backs off inside its own catch, which is all this test needs -- it only asserts on how + // many loops were started. + return new AzureServiceBusSessionListener(transport, queue, Substitute.For(), + Substitute.For>(), NullLogger.Instance, + Substitute.For()); + } + + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(5)] + [InlineData(10)] + public async Task always_runs_exactly_one_accept_loop_per_listener(int listenerCount) + { + await using var listener = buildListener(listenerCount); + listener.AcceptLoopCount.ShouldBe(1); + } +} diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs index dae8dad56..bf137d00d 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs @@ -170,6 +170,22 @@ public AzureServiceBusQueueListenerConfiguration PrefetchCount(int prefetchCount return this; } + /// + /// How many messages an Inline listener for this queue processes concurrently. This + /// is the Azure Service Bus SDK's MaxConcurrentCalls, which Wolverine left at the SDK + /// default of 1 -- so an inline Azure Service Bus listener consumed strictly one message at + /// a time per endpoint. On a session listener driven by a ServiceBusSessionProcessor + /// this sets MaxConcurrentCallsPerSession instead, which trades away the per-session + /// FIFO ordering, so leave it alone on session endpoints unless you mean it. Has no effect + /// on the default Buffered/Durable batch receive loop. See GH-3494. + /// + /// Concurrent handler invocations. Must be at least 1 + public AzureServiceBusQueueListenerConfiguration MaximumConcurrentCalls(int concurrency) + { + add(e => e.MaximumConcurrentCalls = concurrency); + return this; + } + /// /// Completely disable all SQS dead letter queueing for just this queue /// diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs index 8df1afb08..c4d53c160 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs @@ -193,6 +193,21 @@ public AzureServiceBusSubscriptionListenerConfiguration PrefetchCount(int prefet return this; } + /// + /// How many messages an Inline listener for this subscription processes concurrently. + /// This is the Azure Service Bus SDK's MaxConcurrentCalls, which Wolverine left at + /// the SDK default of 1 -- so an inline listener consumed strictly one message at a time per + /// endpoint. On a session listener driven by a ServiceBusSessionProcessor this sets + /// MaxConcurrentCallsPerSession instead, which trades away the per-session FIFO + /// ordering. Has no effect on the default Buffered/Durable batch receive loop. See GH-3494. + /// + /// Concurrent handler invocations. Must be at least 1 + public AzureServiceBusSubscriptionListenerConfiguration MaximumConcurrentCalls(int concurrency) + { + add(e => e.MaximumConcurrentCalls = concurrency); + return this; + } + /// /// Force this subscription listener to require session identifiers. Use this for FIFO semantics /// diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs index 8ff52423c..538ea9e86 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs @@ -190,6 +190,14 @@ internal static ServiceBusProcessorOptions BuildProcessorOptions(AzureServiceBus PrefetchCount = endpoint.PrefetchCount }; + // GH-3494 (AO3): Wolverine never set MaxConcurrentCalls, so an inline listener ran on the + // SDK default of 1 -- single threaded per endpoint, reachable only through the raw + // ConfigureProcessor hook. Applied before ConfigureProcessor so that hook still wins. + if (endpoint.MaximumConcurrentCalls.HasValue) + { + options.MaxConcurrentCalls = endpoint.MaximumConcurrentCalls.Value; + } + endpoint.ConfigureProcessor?.Invoke(options); // Reserved by Wolverine: the inline listener relies on the peek-lock model to complete, @@ -229,14 +237,25 @@ internal static ServiceBusSessionProcessorOptions BuildSessionProcessorOptions(A { PrefetchCount = endpoint.PrefetchCount, - // Map the existing "parallel sessions" knob (ListenerCount, set via RequireSessions(count)) - // onto the processor's concurrency. A user may override this in ConfigureSessionProcessor. - MaxConcurrentSessions = endpoint.ListenerCount > 0 ? endpoint.ListenerCount : 1, + // GH-3494 (AO2): ONE session per processor. ListeningAgent already builds + // Endpoint.ListenerCount of these listeners, so mapping RequireSessions(n) onto each + // processor's MaxConcurrentSessions meant n listeners x n sessions = n-squared + // concurrent sessions -- 64 of them for RequireSessions(8). One per listener keeps the + // documented meaning of RequireSessions(n): n parallel sessions in total. + // A user may still override this in ConfigureSessionProcessor. + MaxConcurrentSessions = 1, // Preserve the in-session FIFO ordering the hand-rolled loop provided MaxConcurrentCallsPerSession = 1 }; + // GH-3494 (AO3): opt-in concurrency WITHIN a session. Left at 1 unless asked for, because + // raising it gives up the per-session FIFO ordering that is the whole point of sessions. + if (endpoint.MaximumConcurrentCalls.HasValue) + { + options.MaxConcurrentCallsPerSession = endpoint.MaximumConcurrentCalls.Value; + } + endpoint.ConfigureSessionProcessor?.Invoke(options); // Reserved by Wolverine: the listener relies on the peek-lock model to explicitly complete, diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs index b3448ceb1..74540c23d 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs @@ -84,6 +84,32 @@ public int PrefetchCount } } + private int? _maximumConcurrentCalls; + + /// + /// How many messages an Inline or session-processor listener for this endpoint hands + /// to the handler pipeline at once. This maps to the Azure Service Bus SDK's + /// MaxConcurrentCalls, which Wolverine never set -- so inline Azure Service Bus + /// listeners ran strictly one message at a time on the SDK default, and the only way to + /// change that was the raw hook. Null keeps the SDK + /// default of 1. Does not apply to the default Buffered/Durable batch receive loop, which + /// scales through MaximumParallelMessages instead. See GH-3494. + /// + public int? MaximumConcurrentCalls + { + get => _maximumConcurrentCalls; + set + { + if (value is < 1) + { + throw new ArgumentOutOfRangeException(nameof(value), value, + "MaximumConcurrentCalls must be at least 1"); + } + + _maximumConcurrentCalls = value; + } + } + /// /// Optional customization of the Azure Service Bus used /// by inline listeners for this endpoint. Wolverine reserves control of the properties it depends diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs index deedd0283..aa2124993 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs @@ -44,8 +44,20 @@ public BatchedAzureServiceBusListener(AzureServiceBusEndpoint endpoint, ILogger _complete = new RetryBlock((e, _) => { return e.CompleteAsync(_cancellation.Token); }, _logger, _cancellation.Token); - _defer = new RetryBlock(async (envelope, _) => { await _requeue.SendAsync(envelope); }, logger, - _cancellation.Token); + _defer = new RetryBlock(async (envelope, _) => + { + // GH-3494 (AO8): settle the original before re-sending the copy, exactly like the + // inline listener already does. Leaving it unsettled meant the message stayed locked + // until the lock expired and Azure Service Bus redelivered it -- so every deferral + // produced a duplicate on top of the copy this block sends. + if (envelope is AzureServiceBusEnvelope e && !e.IsCompleted) + { + await e.CompleteAsync(_cancellation.Token); + e.IsCompleted = true; + } + + await _requeue.SendAsync(envelope); + }, logger, _cancellation.Token); _deadLetter = new RetryBlock( diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/SessionSpecificListener.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/SessionSpecificListener.cs index be8ef8490..75088ffc2 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/SessionSpecificListener.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/SessionSpecificListener.cs @@ -33,19 +33,18 @@ public AzureServiceBusSessionListener(AzureServiceBusTransport transport, AzureS _logger = logger; _requeue = requeue; - var listenerCount = _endpoint.ListenerCount; - if (listenerCount == 0) - { - listenerCount = 1; - } - - for (var i = 0; i < listenerCount; i++) - { - var task = Task.Run(listenForMessages, _cancellation.Token); - _tasks.Add(task); - } + // GH-3494 (AO2): exactly ONE accept loop per listener instance. ListeningAgent already + // builds Endpoint.ListenerCount of these listeners, so spawning ListenerCount loops inside + // each one made RequireSessions(n) open n-squared concurrent AcceptNextSessionAsync loops -- + // 25 for RequireSessions(5) -- all competing for the same sessions. + _tasks.Add(Task.Run(listenForMessages, _cancellation.Token)); } + /// + /// The number of concurrent AcceptNextSessionAsync loops this listener runs. Exposed for testing. + /// + internal int AcceptLoopCount => _tasks.Count; + public IHandlerPipeline? Pipeline => _receiver.Pipeline; public ValueTask CompleteAsync(Envelope envelope)