From eeef43ed85cc1ab47bca700459304d552d0ee3fe Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 10 Jul 2026 12:49:40 -0500 Subject: [PATCH] Artery: inbound lanes for parallel per-connection deserialization and dispatch Fans ordinary-stream messages across N bounded per-lane channels within one accepted connection, keyed by recipient path hash so per-recipient ordering is preserved. Ships with inbound-lanes = 1, which keeps the existing fused single-lane pipeline and materializes no lane machinery. --- BREAKING_CHANGES_V1.6.md | 1 + .../Artery/ArteryConfigSpec.cs | 7 +- .../Artery/ArteryInboundLanesSpec.cs | 582 +++++++++++++++++ .../Artery/ArteryInboundProcessingStage.cs | 593 +++++++++++++++++- src/core/Akka.Remote/Artery/ArteryRemoting.cs | 71 ++- src/core/Akka.Remote/Artery/ArterySettings.cs | 33 +- .../Artery/ArteryTransportSetup.cs | 22 +- .../Akka.Remote/Configuration/Remote.conf | 27 +- 8 files changed, 1313 insertions(+), 23 deletions(-) create mode 100644 src/core/Akka.Remote.Tests/Artery/ArteryInboundLanesSpec.cs diff --git a/BREAKING_CHANGES_V1.6.md b/BREAKING_CHANGES_V1.6.md index 71b2bfc36f1..9652702e6e4 100644 --- a/BREAKING_CHANGES_V1.6.md +++ b/BREAKING_CHANGES_V1.6.md @@ -40,6 +40,7 @@ of `Behavior`, `Wire`, `API` (combine with `+`). | Status | PR / Branch | Component | Type | Change | Migration | |--------|-------------|-----------|------|--------|-----------| +| Planned | `feature/artery-inbound-lanes` | `Akka.Remote` (Artery) | Behavior | `akka.remote.artery.advanced.inbound-lanes` is now actually implemented (previously parsed but ignored -- every value behaved as 1). Its shipping default is changed from `4` to `1`, so out-of-the-box behavior is unaffected. An accepted Ordinary-stream connection now fans ordinary-message deserialization/dispatch out across N per-lane bounded queues + consumer loops, keyed by a stable hash of the recipient path (or, for `ActorSelectionMessage`, the selection's own target path elements), only when this value is explicitly raised above `1`; a new `akka.remote.artery.advanced.inbound-lane-buffer-size` (default `4096`) sizes each lane's queue. Per-recipient ordering is preserved. | No action required at the default. If you had previously set `inbound-lanes` > 1 in your own config expecting it to do something, it now genuinely enables parallel inbound processing -- revert to `1` if you relied on its prior no-op behavior. | | Planned | `fix/artery-pipe-watermarks` | `Akka.Remote` (Artery) | Behavior | Artery TCP connections now raise the pause/resume watermarks of their internal read pipe to 1 MiB (pause 2 MiB), via a new opt-in `Akka.IO.Inet.SO.PipeBufferSize` socket option consulted by `TcpIncomingConnection`/`TcpOutgoingConnection`, configurable via `akka.remote.artery.advanced.tcp.pipe-buffer-size` (default `1m`). Akka.IO's own default (~16 KiB, derived from `akka.io.tcp.receive-buffer-size`) is unchanged for every other TCP connection. | No action required. Artery connections use somewhat more memory per connection and see higher throughput under high-in-flight/one-way traffic; tune via `akka.remote.artery.advanced.tcp.pipe-buffer-size` if needed. | | Merged | [#8317](https://github.com/akkadotnet/akka.net/pull/8317) | `Akka.Remote` / `Akka.Cluster` | API | System/address UID widened `int` → `long` across `AddressUid.Uid`, `AddressUidExtension.Uid()`, `UniqueAddress(Address, int)`/`.Uid`, `QuarantinedEvent(Address, int)`/`.Uid`, `RemoteWatcher.HeartbeatRsp(int)`/`.AddressUid`, and all `Quarantine(Address, int? uid)` members (`IRemoteActorRefProvider` / `RemoteTransport` / `RemoteWatcher`). Hard re-type (v1.6). Prerequisite for Artery. | Recompile against the `long` members; replace `int` uid locals with `long`. Default UID generation stays in 32-bit range; full 64-bit generation is opt-in via `akka.remote.use-64bit-system-uids = on`. | | Merged | [#8317](https://github.com/akkadotnet/akka.net/pull/8317) | `Akka.Cluster` | Wire | Cluster gossip `UniqueAddress.uid` widened `uint32` → `uint64` (`ClusterMessages.proto`). Same varint wire type → binary-compatible for uids ≤ `uint32.MaxValue`. | Rolling upgrade safe provided uid generation stays in 32-bit range (the default); enable `akka.remote.use-64bit-system-uids` only after the whole cluster + all remote peers are on v1.6. | diff --git a/src/core/Akka.Remote.Tests/Artery/ArteryConfigSpec.cs b/src/core/Akka.Remote.Tests/Artery/ArteryConfigSpec.cs index 1d77239ebf7..1b6f8d1aa17 100644 --- a/src/core/Akka.Remote.Tests/Artery/ArteryConfigSpec.cs +++ b/src/core/Akka.Remote.Tests/Artery/ArteryConfigSpec.cs @@ -77,7 +77,11 @@ public void Should_ParseArterySettingsDefaults_When_ConfigNotOverridden() settings.CanonicalHostname.Should().Be("localhost"); settings.CanonicalPort.Should().Be(25520); settings.MaximumFrameSize.Should().Be(256 * 1024); - settings.InboundLanes.Should().Be(4); + + // Ships at 1 (single-lane, fused pipeline -- byte-identical to pre-lanes processing) + // until the scaling gates pass -- see ArterySettings.InboundLanes's remarks. + settings.InboundLanes.Should().Be(1); + settings.InboundLaneBufferSize.Should().Be(4096); settings.OutboundLanes.Should().Be(1); settings.HandshakeTimeout.Should().Be(20.Seconds()); settings.HandshakeRetryInterval.Should().Be(1.Seconds()); @@ -108,6 +112,7 @@ public void Should_ParseArterySettingsDefaults_When_ConfigNotOverridden() [InlineData("akka.remote.artery.advanced.maximum-frame-size = 0b")] [InlineData("akka.remote.artery.advanced.maximum-frame-size = 20000000b")] // > 0x00FFFFFF (16 MiB - 1) [InlineData("akka.remote.artery.advanced.inbound-lanes = 0")] + [InlineData("akka.remote.artery.advanced.inbound-lane-buffer-size = 0")] [InlineData("akka.remote.artery.advanced.outbound-lanes = 0")] [InlineData("akka.remote.artery.transport = aeron")] [InlineData("akka.remote.artery.advanced.system-message-buffer-size = 0")] diff --git a/src/core/Akka.Remote.Tests/Artery/ArteryInboundLanesSpec.cs b/src/core/Akka.Remote.Tests/Artery/ArteryInboundLanesSpec.cs new file mode 100644 index 00000000000..f86a97ea87c --- /dev/null +++ b/src/core/Akka.Remote.Tests/Artery/ArteryInboundLanesSpec.cs @@ -0,0 +1,582 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2022 Lightbend Inc. +// Copyright (C) 2013-2026 .NET Foundation +// +//----------------------------------------------------------------------- + +#nullable enable + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Actor.Setup; +using Akka.Configuration; +using Akka.Remote.Artery; +using Akka.Remote.Configuration; +using Akka.Serialization; +using Akka.TestKit; +using Akka.TestKit.Extensions; +using FluentAssertions; +using FluentAssertions.Extensions; +using Xunit; + +namespace Akka.Remote.Tests.Artery +{ + /// + /// End-to-end tests for inbound lanes (akka.remote.artery.advanced.inbound-lanes > 1): + /// 's per-connection fan-out of ordinary-stream + /// messages across N bounded-channel-fed consumer loops. Every test here goes through the REAL + /// TCP transport (never the low-level GraphStage + /// TestSource/TestSink harness ArteryHandshakeSpec uses) -- lanes only exist inside + /// ArteryRemoting.HandleIncomingConnection's real per-connection composition, and + /// / themselves are + /// completely unmodified by this feature (see 's + /// type-level remarks for why), so re-running THEIR isolated unit tests under a lanes=4 config + /// variant would exercise nothing new; a real handshake-then-dispatch round trip does. + /// + /// + /// Maintainer policy, honored throughout this file: no wall-clock thresholds. Every + /// assertion is progress/order/completion/bounded-state, or a liveness await ("this must + /// eventually happen"), never a measurement of how long something took. + /// + /// + public class ArteryInboundLanesSpec : AkkaSpec + { + public ArteryInboundLanesSpec(ITestOutputHelper output) : base(output) + { + } + + // Generous outbound-message-queue-size headroom (well above the production default of + // 3072): several tests below fire a concurrent multi-recipient burst through ONE + // association's SINGLE shared outbound queue purely to prove per-recipient ORDER on the + // receiving (lanes) side -- an incidental overflow drop on the SENDING side would fail + // those tests for a reason that has nothing to do with inbound lanes. + private static Config ArteryConfig(int inboundLanes = 1) => ConfigurationFactory.ParseString($$""" + akka.actor.provider = "Akka.Remote.RemoteActorRefProvider, Akka.Remote" + akka.remote.artery.enabled = on + akka.remote.artery.canonical.hostname = "127.0.0.1" + akka.remote.artery.canonical.port = 0 + akka.remote.artery.advanced.inbound-lanes = {{inboundLanes}} + akka.remote.artery.advanced.outbound-message-queue-size = 20000 + akka.loggers = ["Akka.TestKit.TestEventListener, Akka.TestKit"] + """); + + private static int BoundPort(ActorSystem system) => RARP.For(system).Provider.DefaultAddress.Port!.Value; + + private static string RemotePath(ActorSystem system, string localName) => + $"akka://{system.Name}@127.0.0.1:{BoundPort(system)}/user/{localName}"; + + /// + /// plus wired in via + /// serialization-bindings (same idiom every other custom-serializer spec in this + /// repo uses -- see e.g. AbstractTransientSerializationErrorSpec -- rather than any + /// lanes-specific mechanism) for the backpressure-parking spec below. Applied to BOTH ends + /// of the association: the SENDER needs the binding to encode a at + /// all (Serialization.FindSerializerFor throws otherwise), even though only the + /// RECEIVING side's call ever actually + /// blocks. + /// + private static Config GatedConfig(int inboundLanes, int inboundLaneBufferSize) => ConfigurationFactory.ParseString($$""" + akka.remote.artery.advanced.inbound-lane-buffer-size = {{inboundLaneBufferSize}} + akka.actor.serializers.gate-blocking = "Akka.Remote.Tests.Artery.ArteryInboundLanesSpec+GateBlockingSerializer, Akka.Remote.Tests" + akka.actor.serialization-bindings { + "Akka.Remote.Tests.Artery.ArteryInboundLanesSpec+GatedInt, Akka.Remote.Tests" = gate-blocking + } + """).WithFallback(ArteryConfig(inboundLanes)); + + /// + /// Payload for the backpressure-parking spec below. Carries the per-test gate key and the + /// recorded value entirely through the wire MANIFEST ( + /// itself is trivial) -- there is nothing to block on the SENDING side, only + /// , on the receiving lane consumer. + /// + private sealed class GatedInt + { + public GatedInt(string gateKey, int value) + { + GateKey = gateKey; + Value = value; + } + + public string GateKey { get; } + public int Value { get; } + } + + /// + /// Test-only serializer (registered via HOCON serialization-bindings -- see + /// -- same idiom as AbstractTransientSerializationErrorSpec.TestSerializer) + /// whose blocks on a per- gate until + /// is called. This is the ONLY way this port can deterministically + /// stall an inbound lane's consumer loop: a slow RECIPIENT cannot do it, because the lane's + /// final dispatch is a fire-and-forget Tell into an unbounded mailbox (see + /// ArteryInboundProcessingStage.RunLaneConsumer's remarks) -- deserialization is the + /// only synchronous, blockable step in a lane consumer's loop. + /// + private sealed class GateBlockingSerializer : SerializerWithStringManifest + { + // Keyed by an arbitrary per-test gate key (a fresh Guid per test method) rather than + // one shared field, so a stray un-opened gate from one run/spec can never bleed into a + // different one's key. + private static readonly ConcurrentDictionary Gates = new(); + + // Observability companion to Gates: lets the test confirm (via AwaitAssertAsync -- a + // progress wait, not a wall-clock one) that the FIRST gated message has actually + // reached FromBinary and is blocked, before it goes on to release the gate. + private static readonly ConcurrentDictionary Entered = new(); + + public GateBlockingSerializer(ExtendedActorSystem system) : base(system) + { + } + + // Arbitrary, hardcoded (never auto-assigned) so both ends of the wire agree on it + // regardless of HOCON parse order -- mirrors AbstractTransientSerializationErrorSpec.TestSerializer's + // Identifier=666 idiom (a different number purely so the two can never collide if ever + // loaded together). + public override int Identifier => 918273; + + public override string Manifest(object o) + { + var gated = (GatedInt)o; + return $"{gated.GateKey}|{gated.Value}"; + } + + public override byte[] ToBinary(object obj) => Array.Empty(); + + public override object FromBinary(byte[] bytes, string manifest) + { + var separator = manifest.IndexOf('|'); + var gateKey = manifest.Substring(0, separator); + var value = int.Parse(manifest.Substring(separator + 1)); + + Entered[gateKey] = true; + + var gate = Gates.GetOrAdd(gateKey, _ => new ManualResetEventSlim(false)); + + // Max-block timeout: if the test aborts/asserts before ever calling OpenGate, this + // throws (loudly, caught and logged by RunLaneConsumer's own try/catch) instead of + // hanging the lane consumer -- and therefore the actor system's shutdown -- forever. + if (!gate.Wait(TimeSpan.FromSeconds(60))) + throw new TimeoutException($"GateBlockingSerializer: gate [{gateKey}] was never opened within 60s."); + + return new GatedInt(gateKey, value); + } + + /// once at least one call for has reached (and is blocked on) its gate. + public static bool HasEntered(string gateKey) => Entered.ContainsKey(gateKey); + + /// Opens (permanently -- never Reset) the gate for , releasing every call blocked on it, past and future. + public static void OpenGate(string gateKey) => Gates.GetOrAdd(gateKey, _ => new ManualResetEventSlim(false)).Set(); + } + + private sealed class Echo : ReceiveActor + { + public Echo() + { + ReceiveAny(msg => Sender.Tell(msg)); + } + } + + /// Watches at construction; forwards to (if supplied). + private sealed class PlainWatcher : ReceiveActor + { + public PlainWatcher(IActorRef target, IActorRef? notify = null) + { + Context.Watch(target); + Receive(_ => notify?.Tell("terminated")); + } + } + + private sealed record GetHistory; + private sealed record History(int[] Values); + + /// Records every it receives, in receipt order; answers with the recorded sequence so far. + private sealed class OrderRecorder : ReceiveActor + { + private readonly List _received = new(); + + public OrderRecorder() + { + Receive(i => _received.Add(i)); + // GatedInt (see the backpressure-parking spec below) unwraps to the same recorded + // int history -- so the SAME OrderRecorder/GetHistory assertions this file already + // uses for order-preservation work unchanged for that spec too. + Receive(g => _received.Add(g.Value)); + Receive(_ => Sender.Tell(new History(_received.ToArray()))); + } + } + + [Fact(DisplayName = "Inbound lanes: settings parse the configured lane count, and a real accepted Ordinary connection actually materializes exactly that many lane consumer loops")] + public async Task Should_Parse_And_Materialize_The_Configured_Lane_Count() + { + const int lanes = 4; + + var arteryConfig = ConfigurationFactory.ParseString($"akka.remote.artery.advanced.inbound-lanes = {lanes}") + .WithFallback(RemoteConfigFactory.Default()) + .GetConfig("akka.remote.artery"); + new ArterySettings(arteryConfig).InboundLanes.Should().Be(lanes); + + var observedLaneCounts = new ConcurrentBag(); + var setup = BootstrapSetup.Create().WithConfig(ArteryConfig(lanes)) + .And(new ArteryTransportSetup(onInboundLanesInitialized: observedLaneCounts.Add)); + + var systemB = ActorSystem.Create("ArteryInboundLanesCountB", setup); + var systemA = ActorSystem.Create("ArteryInboundLanesCountA", ArteryConfig()); + try + { + systemB.ActorOf(Props.Create(() => new Echo()), "echo"); + + var echoRef = await systemA.ActorSelection(RemotePath(systemB, "echo")).ResolveOne(TimeSpan.FromSeconds(10)); + var probe = CreateTestProbe(systemA); + echoRef.Tell("ping", probe.Ref); + await probe.ExpectMsgAsync("ping", TimeSpan.FromSeconds(10)); + + await AwaitAssertAsync(() => + { + observedLaneCounts.Should().NotBeEmpty("the accepted Ordinary connection must have materialized its lane machinery"); + observedLaneCounts.Should().OnlyContain(n => n == lanes, "every materialized lane set must contain exactly the configured lane count"); + }, TimeSpan.FromSeconds(10)); + } + finally + { + await systemA.Terminate().AwaitWithTimeout(10.Seconds()); + await systemB.Terminate().AwaitWithTimeout(10.Seconds()); + } + } + + [Fact(DisplayName = "Inbound lanes: concurrent senders to ~16 distinct recipients at inbound-lanes=4 each preserve their own send order, despite fan-out across lanes")] + public async Task Should_Preserve_PerRecipient_Order_Across_Concurrent_Senders_With_Lanes() + { + const int recipientCount = 16; + const int messagesPerRecipient = 300; + + var systemB = ActorSystem.Create("ArteryInboundLanesOrderB", ArteryConfig(inboundLanes: 4)); + var systemA = ActorSystem.Create("ArteryInboundLanesOrderA", ArteryConfig()); + try + { + var recorders = new IActorRef[recipientCount]; + for (var r = 0; r < recipientCount; r++) + recorders[r] = systemB.ActorOf(Props.Create(() => new OrderRecorder()), $"recorder-{r}"); + + // Resolve every recipient's RemoteActorRef from A up front (outside the concurrent + // send loop below) -- purely local address-based resolution, no extra round trips. + var remoteRefs = new IActorRef[recipientCount]; + for (var r = 0; r < recipientCount; r++) + remoteRefs[r] = await systemA.ActorSelection(RemotePath(systemB, $"recorder-{r}")).ResolveOne(TimeSpan.FromSeconds(10)); + + // Concurrent senders: one Task per recipient, all racing onto the SAME association's + // single outbound queue/connection at once -- exactly the interleaving-across-lanes + // scenario this test exists to prove is still per-recipient-ordered on the receiving + // (lanes=4) side. + var sendTasks = Enumerable.Range(0, recipientCount).Select(r => Task.Run(() => + { + for (var i = 0; i < messagesPerRecipient; i++) + remoteRefs[r].Tell(i, ActorRefs.NoSender); + })).ToArray(); + + await Task.WhenAll(sendTasks).WaitAsync(TimeSpan.FromSeconds(30)); + + var expected = Enumerable.Range(0, messagesPerRecipient).ToArray(); + + for (var r = 0; r < recipientCount; r++) + { + var probe = CreateTestProbe(systemA); + await AwaitAssertAsync(async () => + { + recorders[r].Tell(new GetHistory(), probe.Ref); + var history = await probe.ExpectMsgAsync(TimeSpan.FromSeconds(5)); + history.Values.Should().Equal(expected, + $"recorder-{r} must observe its own {messagesPerRecipient} sends in EXACT send order, regardless of concurrent traffic to other recipients on other lanes"); + }, TimeSpan.FromSeconds(30)); + } + } + finally + { + await systemA.Terminate().AwaitWithTimeout(15.Seconds()); + await systemB.Terminate().AwaitWithTimeout(15.Seconds()); + } + } + + [Fact(DisplayName = "Inbound lanes: handshake completion and the control-stream DeathWatch path (Watch/Terminated) are unaffected at inbound-lanes=4, alongside a concurrent ordinary round trip")] + public async Task Should_Leave_Handshake_And_Control_Stream_Unaffected_At_Lanes4() + { + var systemB = ActorSystem.Create("ArteryInboundLanesHandshakeB", ArteryConfig(inboundLanes: 4)); + var systemA = ActorSystem.Create("ArteryInboundLanesHandshakeA", ArteryConfig(inboundLanes: 4)); + try + { + systemB.ActorOf(Props.Create(() => new Echo()), "echo"); + var watchTarget = systemB.ActorOf(Props.Create(() => new Echo()), "watch-target"); + + // Handshake completion, proven the same way every other Artery spec proves it: an + // ordinary round trip actually arrives. + var echoRef = await systemA.ActorSelection(RemotePath(systemB, "echo")).ResolveOne(TimeSpan.FromSeconds(10)); + var echoProbe = CreateTestProbe(systemA); + echoRef.Tell("ping", echoProbe.Ref); + await echoProbe.ExpectMsgAsync("ping", TimeSpan.FromSeconds(10)); + + // Control-stream DeathWatch path: Watch (outbound control from A) / Terminated + // (delivered back to A once B's watch-target stops) -- entirely independent of the + // Ordinary connection's lane machinery (see ArteryInboundProcessingStage's "Why + // control/large connections and lanes=1 never touch this machinery" remarks: the + // CONTROL connection is never lane-routed regardless of inbound-lanes). + var watchTargetRemote = RARP.For(systemA).Provider.ResolveActorRef(RemotePath(systemB, "watch-target")); + var terminatedProbe = CreateTestProbe(systemA); + systemA.ActorOf(Props.Create(() => new PlainWatcher(watchTargetRemote, terminatedProbe.Ref))); + + // A concurrent burst of ordinary traffic to the SAME echo actor while the watch + // above is in flight -- proves the lane path and the control path do not interfere + // with each other. + var burstProbe = CreateTestProbe(systemA); + for (var i = 0; i < 50; i++) + echoRef.Tell(i, burstProbe.Ref); + for (var i = 0; i < 50; i++) + await burstProbe.ExpectMsgAsync(i, TimeSpan.FromSeconds(10)); + + watchTarget.Tell(PoisonPill.Instance); + await terminatedProbe.ExpectMsgAsync("terminated", TimeSpan.FromSeconds(10)); + } + finally + { + await systemA.Terminate().AwaitWithTimeout(15.Seconds()); + await systemB.Terminate().AwaitWithTimeout(15.Seconds()); + } + } + + /// + /// Adapts ArteryUnwatchShutdownRaceSpec's scenario (a mass-Unwatch termination burst + /// racing graceful ArteryRemoting.Shutdown()) with inbound-lanes > 1 on BOTH + /// sides -- proving 's OWN teardown (lane channels + /// completed in Logic.PostStop, consumer loops draining and exiting on their own, no + /// blocking join) does not hang or error, even while a burst of real inbound Unwatch control + /// traffic AND a burst of in-flight ordinary lane traffic are both landing as the receiving + /// side terminates. + /// + [Fact(DisplayName = "Inbound lanes: clean teardown at inbound-lanes=4 -- graceful shutdown under a mass-Unwatch burst plus concurrent in-flight ordinary lane traffic does not hang or error")] + public async Task Should_Teardown_Cleanly_At_Lanes4_Under_Concurrent_Unwatch_And_Ordinary_Bursts() + { + const int watcheesPerIteration = 150; + const int recipientCount = 8; + + for (var iteration = 0; iteration < 3; iteration++) + { + var systemA = ActorSystem.Create($"ArteryInboundLanesTeardownA{iteration}", ArteryConfig(inboundLanes: 4)); + var systemB = ActorSystem.Create($"ArteryInboundLanesTeardownB{iteration}", ArteryConfig(inboundLanes: 4)); + try + { + for (var i = 0; i < watcheesPerIteration; i++) + systemB.ActorOf(Props.Create(() => new Echo()), $"target-{i}"); + + var recorders = new IActorRef[recipientCount]; + for (var r = 0; r < recipientCount; r++) + recorders[r] = systemA.ActorOf(Props.Create(() => new OrderRecorder()), $"recorder-{r}"); + + // Warm up the association first (one real request/reply round trip), same idiom + // as ArteryUnwatchShutdownRaceSpec. + var warmupTarget = await systemA.ActorSelection(RemotePath(systemB, "target-0")).ResolveOne(TimeSpan.FromSeconds(10)); + var warmupProbe = CreateTestProbe(systemA); + warmupTarget.Tell("warmup", warmupProbe.Ref); + await warmupProbe.ExpectMsgAsync("warmup", TimeSpan.FromSeconds(10)); + + var targets = new IActorRef[watcheesPerIteration]; + for (var i = 0; i < watcheesPerIteration; i++) + targets[i] = RARP.For(systemA).Provider.ResolveActorRef(RemotePath(systemB, $"target-{i}")); + + for (var i = 0; i < watcheesPerIteration; i++) + systemA.ActorOf(Props.Create(() => new PlainWatcher(targets[i], ActorRefs.NoSender))); + + var remoteWatcher = RARP.For(systemA).Provider.RemoteWatcher; + var statsProbe = CreateTestProbe(systemA); + await AwaitAssertAsync(async () => + { + remoteWatcher.Tell(new RemoteWatcher.Stats(0, 0), statsProbe.Ref); + var stats = await statsProbe.ExpectMsgAsync(TimeSpan.FromSeconds(3)); + stats.WatchingRefs.Count.Should().BeGreaterOrEqualTo(watcheesPerIteration); + }, TimeSpan.FromSeconds(10)); + + // Concurrent in-flight ORDINARY lane traffic from B -> A's recorders, still being + // sent as A begins tearing down -- exercises this port's inbound lane teardown + // (Logic.PostStop) on systemA's side WHILE systemA's own Shutdown() is racing the + // RemoteWatcher Unwatch drain above. + var remoteRecorders = new IActorRef[recipientCount]; + for (var r = 0; r < recipientCount; r++) + remoteRecorders[r] = RARP.For(systemB).Provider.ResolveActorRef(RemotePath(systemA, $"recorder-{r}")); + + var burstCts = new System.Threading.CancellationTokenSource(); + var burstTasks = Enumerable.Range(0, recipientCount).Select(r => Task.Run(async () => + { + var i = 0; + while (!burstCts.IsCancellationRequested) + { + remoteRecorders[r].Tell(i++, ActorRefs.NoSender); + await Task.Delay(5); + } + })).ToArray(); + + // THE regression assertion: no control-queue-full ERROR anywhere during graceful + // termination (same wording as ArteryUnwatchShutdownRaceSpec), AND -- new for + // this port's inbound lanes -- Terminate() itself must actually complete (a + // liveness await, not a timing measurement): a lane consumer loop or a parked + // backpressured write that never unblocks would hang shutdown instead. + await CreateEventFilter(systemA).Error(contains: "is full (capacity").ExpectAsync(0, async () => + { + await systemA.Terminate().AwaitWithTimeout(20.Seconds()); + }); + + burstCts.Cancel(); + await Task.WhenAll(burstTasks).WaitAsync(TimeSpan.FromSeconds(10)); + } + finally + { + await systemA.Terminate().AwaitWithTimeout(15.Seconds()); + await systemB.Terminate().AwaitWithTimeout(15.Seconds()); + } + } + } + + /// + /// Covers the ONE path none of the specs above reach: ArteryInboundProcessingStage.Logic.TryRouteToLane + /// returning (a lane's channel is genuinely full) -> the item is + /// PARKED (_writeInFlight = true) -> DeliverOrPull/OnPull suppress + /// further Pull on the connection (connection-level backpressure) -> + /// OnLaneWriteAvailable (the GetAsyncCallback-wrapped continuation of + /// ChannelWriter.WaitToWriteAsync) resumes draining once the lane frees up. + /// + /// + /// Why deserialization, not the recipient, is what has to block. A slow ACTOR cannot + /// fill a lane channel: the lane consumer's final step is a fire-and-forget Tell into + /// an unbounded mailbox (ArteryRemoting.DispatchOrdinaryMessage), so the consumer loop + /// never stalls waiting on a recipient, no matter how slow that recipient is. The only + /// synchronous, blockable step in a lane consumer's loop is + /// + /// itself -- so (registered via + /// serialization-bindings, exactly like every other custom-serializer spec in this + /// repo -- never a lanes-specific mechanism) blocks THAT, deterministically, until the test + /// releases it. + /// + /// + /// + /// Why this reliably parks (not merely "usually"). inbound-lane-buffer-size is + /// set to 4 and -many (32) gated messages are sent to + /// ONE recipient, ALL hashing to the SAME lane (recipient hashing is stable per recipient -- + /// see ArteryInboundProcessingStage.Logic.LaneFor) -- so with that lane's consumer + /// permanently blocked (on message 0) until the gate opens, capacity fills after 4 further + /// messages queue up behind it and the 6th TryRouteToLane call is GUARANTEED to park, + /// purely as a matter of the source ( + /// capacity vs. message count), never a timing race. The 8 PLAIN messages to a SECOND + /// recipient (whichever lane it happens to hash onto -- the wire recipient path includes the + /// resolved actor's uid, so the lane is not something this test can predict or force ahead of + /// creating it, and it does not need to) are sent strictly AFTER all 32 gated ones in this + /// association's one shared outbound queue/connection, so they are guaranteed to be + /// UNREACHABLE by the single frame parser until the park clears too: DrainReadyFramesLaneMode + /// stops draining the WHOLE connection the instant anything parks (see + /// 's "single-parked-item" design), so neither + /// recipient can have received anything before the gate opens -- not a probabilistic + /// argument, a structural one that holds regardless of which lane either recipient lands on. + /// + /// + /// + /// Assertions (maintainer policy: no wall-clock thresholds). (a) no message lost -- + /// all 40 (32 gated + 8 plain) arrive once the gate releases; (b) per-recipient send order is + /// preserved for both; (c) both s Terminate() cleanly + /// afterward (teardown liveness -- a parked write or blocked lane consumer that never + /// resumed would hang this instead). The brief below is + /// pure PACING (same idiom as the burst loop's Task.Delay(5) in + /// + /// above) giving the fused stage's own interpreter thread -- which decodes/routes frames + /// concurrently with, and without waiting on, the blocked lane consumer -- a generous margin + /// to reach and park at message 5 before the gate opens; it is never used to assert a + /// pass/fail condition. + /// + /// + [Fact(DisplayName = "Inbound lanes: a slow DESERIALIZATION parks the fused stage on lane backpressure (TryRouteToLane parks -> Pull suppressed -> OnLaneWriteAvailable resumes) -- releasing the gate afterward delivers every message, on both lanes, in order, and both systems terminate cleanly")] + public async Task Should_Recover_All_Messages_After_A_Deserialization_Blocked_Lane_Parks_On_Backpressure() + { + const int inboundLanes = 2; + const int laneBufferSize = 4; // small on purpose -- comfortably exceeded by gatedMessageCount below (see the type-level remarks for why this guarantees a park, not merely risks one). + const int gatedMessageCount = 32; + const int plainMessageCount = 8; + var gateKey = Guid.NewGuid().ToString("N"); + + var systemB = ActorSystem.Create("ArteryInboundLanesParkB", GatedConfig(inboundLanes, laneBufferSize)); + var systemA = ActorSystem.Create("ArteryInboundLanesParkA", GatedConfig(inboundLanes: 1, inboundLaneBufferSize: laneBufferSize)); + try + { + // "gated" is bound to GateBlockingSerializer; "plain" uses ordinary int messages + // through the default serializer. Whichever lane each hashes onto is incidental -- + // not load-bearing for correctness (see the type-level remarks: EVERY lane recovers + // the same way once the gate opens, and the send order below guarantees neither + // recipient sees anything before then regardless of lane assignment). + var gatedRecorder = systemB.ActorOf(Props.Create(() => new OrderRecorder()), "gated"); + var plainRecorder = systemB.ActorOf(Props.Create(() => new OrderRecorder()), "plain"); + + // ResolveOne is itself a request/reply round trip -- it doubles as this association's + // warm-up (handshake completion) before the real burst below, same as every other + // spec in this file. + var gatedRemote = await systemA.ActorSelection(RemotePath(systemB, "gated")).ResolveOne(TimeSpan.FromSeconds(10)); + var plainRemote = await systemA.ActorSelection(RemotePath(systemB, "plain")).ResolveOne(TimeSpan.FromSeconds(10)); + + try + { + // All 32 gated messages, then all 8 plain ones -- deliberately NOT interleaved, + // so every plain frame sits strictly BEHIND every gated frame in this + // association's one shared outbound queue/connection (see the type-level + // remarks: this is what makes "neither recipient receives anything before the + // gate opens" a structural guarantee rather than a timing race). + for (var i = 0; i < gatedMessageCount; i++) + gatedRemote.Tell(new GatedInt(gateKey, i), ActorRefs.NoSender); + + for (var i = 0; i < plainMessageCount; i++) + plainRemote.Tell(i, ActorRefs.NoSender); + + // Progress wait (not a wall-clock one): confirms message 0 has actually reached + // FromBinary and is blocked, i.e. the parking machinery is now live. + await AwaitAssertAsync( + () => GateBlockingSerializer.HasEntered(gateKey).Should().BeTrue( + "the gated recipient's first message must have reached FromBinary and be blocked on the gate"), + TimeSpan.FromSeconds(10)); + + // Pacing only -- see the type-level remarks' last paragraph. + await Task.Delay(TimeSpan.FromMilliseconds(500)); + } + finally + { + // Always release the gate, even if the wait above itself threw -- otherwise a + // failed assertion here would leave the blocked lane consumer parked for its own + // internal 60s max-block timeout, needlessly extending a failing run (and, + // without this, risking a hang in the Terminate() calls below). + GateBlockingSerializer.OpenGate(gateKey); + } + + // (a) no message lost, (b) exact per-recipient send order, for BOTH lanes. + var gatedProbe = CreateTestProbe(systemA); + await AwaitAssertAsync(async () => + { + gatedRecorder.Tell(new GetHistory(), gatedProbe.Ref); + var history = await gatedProbe.ExpectMsgAsync(TimeSpan.FromSeconds(5)); + history.Values.Should().Equal(Enumerable.Range(0, gatedMessageCount), + "every gated message must arrive, in exact send order, once the gate releases"); + }, TimeSpan.FromSeconds(30)); + + var plainProbe = CreateTestProbe(systemA); + await AwaitAssertAsync(async () => + { + plainRecorder.Tell(new GetHistory(), plainProbe.Ref); + var history = await plainProbe.ExpectMsgAsync(TimeSpan.FromSeconds(5)); + history.Values.Should().Equal(Enumerable.Range(0, plainMessageCount), + "the OTHER lane's traffic -- queued up behind the SAME connection-wide park -- must also fully recover, in order, once the gate releases"); + }, TimeSpan.FromSeconds(30)); + } + finally + { + // (c) teardown liveness: Terminate() must actually complete -- a still-parked write + // or a lane consumer stuck on an un-opened gate would hang this instead. + await systemA.Terminate().AwaitWithTimeout(20.Seconds()); + await systemB.Terminate().AwaitWithTimeout(20.Seconds()); + } + } + } +} diff --git a/src/core/Akka.Remote/Artery/ArteryInboundProcessingStage.cs b/src/core/Akka.Remote/Artery/ArteryInboundProcessingStage.cs index 2409df530a9..09168905388 100644 --- a/src/core/Akka.Remote/Artery/ArteryInboundProcessingStage.cs +++ b/src/core/Akka.Remote/Artery/ArteryInboundProcessingStage.cs @@ -10,6 +10,10 @@ using System; using System.Buffers; using System.Collections.Generic; +using System.Text; +using System.Threading.Channels; +using System.Threading.Tasks; +using Akka.Actor; using Akka.Event; using Akka.Streams; using Akka.Streams.Stage; @@ -62,6 +66,53 @@ namespace Akka.Remote.Artery /// caught here and is left to fail the stage/connection, since it indicates the peer is not /// speaking the protocol correctly. /// + /// + /// + /// Inbound lanes (only when > 1 AND the connection's own + /// preamble declares ). Lanes parallelize WITHIN one + /// accepted connection: framing + header decode ALWAYS stay fused here (cheap, and required to + /// even know which lane a frame belongs to), but the EXPENSIVE part of processing an ordinary + /// message -- , + /// RemoteActorRefProvider.ResolveActorRefWithLocalAddress, and the final + /// Tell -- moves onto one of dedicated background consumer + /// loops, each fed by its own bounded (capacity + /// ). A message's lane is a stable hash of its recipient path + /// (Logic.LaneFor) -- so all ordinary traffic to the SAME recipient stays on the SAME + /// lane and is delivered in send order, exactly like today's single-lane pipeline. + /// + /// + /// + /// Why control/large connections and lanes=1 never touch this machinery. An + /// connection carries handshake/heartbeat/quarantine-notice/ + /// reliable-system-message traffic (see ArteryRemoting.MaterializeOutboundStream's + /// remarks: every control-classified message is enqueued via EnqueueControl/ + /// EnqueueSystemMessage, which ALWAYS materializes the CONTROL outbound stream -- never + /// the ordinary one), so an connection never carries a + /// control-classified envelope in practice. Lane routing is therefore gated on the connection's + /// OWN preamble (known the moment parses it), not on a + /// per-frame runtime check -- Control and Large connections always use the exact pre-lanes + /// decode-and-deserialize-inline path, UNCHANGED, regardless of . A + /// defensive SerializerId check () still catches a + /// control-classified frame arriving on an Ordinary connection (should never happen) and routes + /// it through the SAME inline control path instead of a lane, so a protocol violation degrades + /// safely rather than being silently misrouted. + /// + /// + /// + /// ActorSelectionMessage lane key (design investigation). ArteryRemoting.Send + /// resolves an ActorSelectionMessage's wire RecipientPath from the SELECTION'S + /// ANCHOR ref, not its final target -- and ActorRefFactoryShared.ActorSelection(ActorPath) + /// anchors every selection into a given remote system at that system's ROOT GUARDIAN + /// (RemoteActorRefProvider.RootGuardianAt). So EVERY ActorSelectionMessage sent to + /// ANY actor on a given remote system carries the exact SAME RecipientPath string -- + /// hashing it directly would collapse every inbound selection from that peer onto ONE lane, + /// regardless of . instead + /// parses just the OUTER SelectionEnvelope wire wrapper (never the wrapped application + /// message it carries -- that stays fully deferred to the lane, like every other ordinary + /// message) to recover the selection's OWN target path elements and hashes THOSE, so distinct + /// selection targets land on distinct lanes. Falls back to RecipientPath (same-lane, but + /// never WRONG) if the wrapper cannot be parsed. + /// /// internal sealed class ArteryInboundProcessingStage : GraphStage, IInboundEnvelope>> { @@ -75,11 +126,93 @@ internal sealed class ArteryInboundProcessingStage : GraphStagemaximum-large-frame-size. /// /// The receiving actor system's extension. - public ArteryInboundProcessingStage(int maxFrameLength, int maxLargeFrameLength, Akka.Serialization.Serialization serialization) + /// + /// Number of inbound lanes ordinary-stream messages are fanned out across (see the + /// type-level "Inbound lanes" remarks). (the default -- and the + /// production default at akka.remote.artery.advanced.inbound-lanes) means the lane + /// machinery below is NEVER materialized: every connection uses the exact pre-lanes + /// fused decode-and-deserialize-inline path, regardless of which stream it turns out to + /// carry. GATE: passing anything > 1 REQUIRES and + /// to be supplied. + /// + /// + /// Capacity of each lane's bounded , when + /// > 1. Ignored (and may be 0) at the default of 1. + /// + /// + /// Needed ONLY when > 1: an Ordinary-stream connection's + /// lane path bypasses for its own ordinary traffic (see + /// the type-level remarks on why that stage is a structural no-op for such a connection) + /// but still needs its gate, reused as-is here. + /// + /// + /// Needed ONLY when > 1: invoked by each lane's consumer + /// loop, once per deserialized ordinary message, as (message, senderPath, recipientPath) + /// -- mirrors ArteryRemoting.DispatchOrdinaryMessage exactly (the same resolve-and-Tell + /// logic ArteryRemoting.DispatchInbound uses for the lanes=1/non-lane path). + /// + /// + /// Test-observability hook (see ): + /// invoked once, with the actual lane count, the moment this connection's lane machinery is + /// materialized. (the default, and the only production value) does + /// nothing. Never invoked when is 1 (lanes are never + /// materialized then) or for a Control/Large-stream connection (lanes are Ordinary-only). + /// + public ArteryInboundProcessingStage( + int maxFrameLength, + int maxLargeFrameLength, + Akka.Serialization.Serialization serialization, + int inboundLanes = 1, + int inboundLaneBufferSize = 0, + IInboundContext? inboundContext = null, + Action? dispatchOrdinary = null, + Action? onLanesInitialized = null) { + if (inboundLanes < 1) + throw new ArgumentOutOfRangeException(nameof(inboundLanes), inboundLanes, "Must be >= 1."); + if (inboundLanes > 1 && (inboundContext is null || dispatchOrdinary is null)) + throw new ArgumentException( + $"{nameof(inboundContext)} and {nameof(dispatchOrdinary)} are required when {nameof(inboundLanes)} > 1."); + MaxFrameLength = maxFrameLength; MaxLargeFrameLength = maxLargeFrameLength; Serialization = serialization; + InboundLanes = inboundLanes; + InboundLaneBufferSize = inboundLaneBufferSize; + InboundContext = inboundContext; + DispatchOrdinary = dispatchOrdinary; + OnLanesInitialized = onLanesInitialized; + + // Resolved ONCE per accepted connection (this stage is materialized per-connection), + // never on the per-frame hot path -- and only when lanes are actually enabled, since + // this is pure overhead at the InboundLanes=1 default. A miss (no binding found, which + // should never happen for these built-in types) resolves to a sentinel that can never + // match a real wire SerializerId, so the defensive checks below simply never trigger + // rather than throwing. + // + // LOAD-BEARING INVARIANT: lane mode classifies a frame as "control" by comparing its + // wire SerializerId to ControlMessageSerializerId (see ProcessFrameLaneMode) -- this is + // equivalent to the non-lane path's `payload is IArteryControlMessage` check (see + // DecodeFrame) ONLY as long as (a) every IArteryControlMessage type is serialized by + // ArteryControlMessageSerializer and (b) nothing else is ever bound to that same + // SerializerId. Both hold today (ArteryControlMessageSerializer is the sole serializer + // registered for all nine control message types). If that ever changes -- e.g. a new + // IArteryControlMessage type gets its own dedicated serializer, or + // ArteryControlMessageSerializer starts handling a non-control type too -- this + // SerializerId-based check and ProcessFrameLaneMode's classification must be updated + // together, or lane mode will silently stop treating IsControl the same way the + // non-lane path does for the affected type. + if (inboundLanes > 1) + { + ControlMessageSerializerId = serialization.FindSerializerForType(typeof(HandshakeReq))?.Identifier ?? int.MinValue; + SelectionMessageSerializerId = serialization.FindSerializerForType(typeof(ActorSelectionMessage))?.Identifier ?? int.MinValue; + } + else + { + ControlMessageSerializerId = int.MinValue; + SelectionMessageSerializerId = int.MinValue; + } + Shape = new FlowShape, IInboundEnvelope>(In, Out); } @@ -90,6 +223,39 @@ public ArteryInboundProcessingStage(int maxFrameLength, int maxLargeFrameLength, public Akka.Serialization.Serialization Serialization { get; } + /// See the constructor parameter of the same name. + public int InboundLanes { get; } + + /// See the constructor parameter of the same name. + public int InboundLaneBufferSize { get; } + + /// See the constructor parameter of the same name. when is 1. + public IInboundContext? InboundContext { get; } + + /// See the constructor parameter of the same name. when is 1. + public Action? DispatchOrdinary { get; } + + /// See the constructor parameter of the same name. + public Action? OnLanesInitialized { get; } + + /// + /// The shared 's SerializerId (probed via any + /// one of the nine message types it handles) -- used ONLY as a defensive per-frame + /// classification check when > 1 (see the type-level "Why + /// control/large connections and lanes=1 never touch this machinery" remarks). Never + /// resolved (stays a sentinel that matches nothing) at the =1 + /// default. + /// + public int ControlMessageSerializerId { get; } + + /// + /// 's SerializerId -- used ONLY to decide + /// whether a frame needs the special case (see + /// the type-level "ActorSelectionMessage lane key" remarks). Never resolved at the + /// =1 default. + /// + public int SelectionMessageSerializerId { get; } + public Inlet> In { get; } = new("ArteryInboundProcessing.in"); public Outlet Out { get; } = new("ArteryInboundProcessing.out"); @@ -97,6 +263,36 @@ public ArteryInboundProcessingStage(int maxFrameLength, int maxLargeFrameLength, protected override GraphStageLogic CreateLogic(Attributes inheritedAttributes) => new Logic(this); + /// + /// One decoded-but-not-yet-deserialized ordinary message, handed from the fused + /// per-connection stage to a lane's consumer loop. Carries the RAW payload bytes (a + /// slice into a freshly-allocated, never-pooled, + /// never-mutated byte[] owned by that one TCP read -- see + /// Akka.IO.TcpConnection.ReadPipeChunkAsync, which copies OUT of the pipe's own + /// pooled segments before AdvanceTo for exactly this reason -- so it is always safe + /// to retain and deserialize later, on a different thread, with no risk of the underlying + /// bytes being reused/overwritten out from under the lane consumer) plus everything else + /// the lane needs to finish the job without going back to the connection: the serializer + /// id/manifest to deserialize with, and the already-resolved sender/recipient wire paths. + /// + private readonly struct LaneWorkItem + { + public LaneWorkItem(ReadOnlySequence payload, int serializerId, string manifest, string? senderPath, string recipientPath) + { + Payload = payload; + SerializerId = serializerId; + Manifest = manifest; + SenderPath = senderPath; + RecipientPath = recipientPath; + } + + public ReadOnlySequence Payload { get; } + public int SerializerId { get; } + public string Manifest { get; } + public string? SenderPath { get; } + public string RecipientPath { get; } + } + private sealed class Logic : GraphStageLogic, IInHandler, IOutHandler { private readonly ArteryInboundProcessingStage _stage; @@ -117,11 +313,54 @@ private sealed class Logic : GraphStageLogic, IInHandler, IOutHandler /// private ArteryFrameParser? _frameParser; + /// + /// only for a connection whose OWN preamble declared + /// AND + /// > 1 -- set (once) at the end of , never afterwards. + /// Every other connection (Control, Large, or ANY connection when InboundLanes is 1) + /// leaves this for its entire lifetime, so + /// always takes the untouched pre-lanes + /// path for it -- see the type-level "Why control/large connections and lanes=1 never + /// touch this machinery" remarks. + /// + private bool _laneModeActive; + + /// One bounded channel per lane; created once, lazily, by . + private Channel[]? _lanes; + + /// Captured once (interpreter thread, at lane init) so lane consumer loops never touch the stage's lazily-initialized Log from a background thread. + private ILoggingAdapter? _laneLog; + + // Backpressure-parking state (standard async-GraphStage pattern -- mirrors + // Akka.Streams.Implementation.ChannelSinkLogic's TryWrite/WaitToWriteAsync idiom). At + // most ONE item can ever be parked at a time: DrainReadyFramesLaneMode is a plain + // synchronous loop on the interpreter thread that stops the INSTANT a lane write would + // block, so there is never more than one in-flight async write per connection. + private LaneWorkItem _parkedItem; + private int _parkedLaneIndex; + private bool _writeInFlight; + private Action? _onLaneWriteAvailable; + private Action>? _onLaneWriteReady; + public Logic(ArteryInboundProcessingStage stage) : base(stage.Shape) { _stage = stage; SetHandler(stage.In, this); SetHandler(stage.Out, this); + + if (_stage.InboundLanes > 1) + { + _onLaneWriteAvailable = GetAsyncCallback(OnLaneWriteAvailable); + _onLaneWriteReady = t => + { + if (t.IsFaulted) + _onLaneWriteAvailable!(false); + else if (t.IsCanceled) + _onLaneWriteAvailable!(false); + else + _onLaneWriteAvailable!(t.Result); + }; + } } public void OnPush() @@ -135,7 +374,12 @@ public void OnPush() } AppendToParser(chunk); - DrainReadyFrames(); + + if (_laneModeActive) + DrainReadyFramesLaneMode(); + else + DrainReadyFrames(); + DeliverOrPull(); } @@ -145,11 +389,17 @@ public void OnPull() { Push(_stage.Out, _pending.Dequeue()); } - else if (IsClosed(_stage.In)) + else if (IsClosed(_stage.In) && !_writeInFlight) { CompleteStage(); } - else if (!HasBeenPulled(_stage.In)) + // !_writeInFlight guards this exactly like DeliverOrPull's matching guard: while + // parked on lane backpressure, do NOT request more input from the connection, even + // if downstream re-signals demand (e.g. after consuming an earlier, defensively + // routed control envelope) -- otherwise the frame parser's own unbounded internal + // buffer would grow without limit while we wait for lane capacity, defeating the + // whole point of backpressure. Always false (no-op) at InboundLanes=1. + else if (!IsClosed(_stage.In) && !HasBeenPulled(_stage.In) && !_writeInFlight) { Pull(_stage.In); } @@ -157,16 +407,37 @@ public void OnPull() public void OnUpstreamFinish() { - if (_pending.Count == 0) + if (_pending.Count == 0 && !_writeInFlight) CompleteStage(); - // else: swallow the termination and let OnPull drain `_pending`, completing once empty. + // else: swallow the termination and let OnPull (non-lane path) / OnLaneWriteAvailable + // (lane path) drain the remainder, completing once both are empty/clear. } public void OnUpstreamFailure(Exception e) => FailStage(e); public void OnDownstreamFinish(Exception cause) => InternalOnDownstreamFinish(cause); + /// + /// Lane channels are completed on EVERY termination path (graceful completion, + /// failure, or downstream cancellation all funnel through -- + /// the universal teardown hook), so a lane's consumer + /// loop always eventually observes + /// returning once it has drained whatever was already + /// buffered, and exits on its own -- no orphaned tasks, and no message that had + /// already been handed to a lane is ever silently dropped by teardown itself + /// (only a message still sitting unread upstream in the TCP pipe, which was never + /// going to be delivered anyway once the connection is gone). + /// + public override void PostStop() + { + if (_lanes is not null) + { + foreach (var lane in _lanes) + lane.Writer.TryComplete(); + } + } + /// /// Consumes as much of the connection preamble as can supply. /// Returns if the caller should stop processing this push (either @@ -211,6 +482,16 @@ private bool TryConsumePreamble(ref ReadOnlySequence chunk) ? new ArteryFrameParser(_stage.MaxLargeFrameLength) : new ArteryFrameParser(_stage.MaxFrameLength); + // GATE A: only an Ordinary-stream connection, and only when lanes are actually + // configured, ever enters lane mode -- every Control/Large connection, and EVERY + // connection at the InboundLanes=1 default, is untouched from here on (see the + // type-level remarks). + if (streamId == ArteryStreamId.Ordinary && _stage.InboundLanes > 1) + { + _laneModeActive = true; + InitLanes(); + } + _preambleParsed = true; return true; } @@ -299,9 +580,309 @@ private void DeliverOrPull() return; } + // Parked on lane backpressure -- do NOT request more input from the connection + // until the lane frees up room (this IS the connection-level backpressure signal: + // stop reading more off the TCP pipe while a decoded item is stuck waiting for its + // lane). + if (_writeInFlight) + return; + if (!IsClosed(_stage.In) && !HasBeenPulled(_stage.In)) Pull(_stage.In); } + + #region Lane mode (InboundLanes > 1, Ordinary-stream connections only) + + private void InitLanes() + { + var laneCount = _stage.InboundLanes; + _laneLog = Log; + var lanes = new Channel[laneCount]; + + for (var i = 0; i < laneCount; i++) + { + var channel = Channel.CreateBounded(new BoundedChannelOptions(_stage.InboundLaneBufferSize) + { + SingleWriter = true, + SingleReader = true, + AllowSynchronousContinuations = false, + FullMode = BoundedChannelFullMode.Wait + }); + lanes[i] = channel; + + var reader = channel.Reader; + var log = _laneLog; + var serialization = _stage.Serialization; + var dispatch = _stage.DispatchOrdinary!; + _ = Task.Run(() => RunLaneConsumer(reader, serialization, dispatch, log)); + } + + _lanes = lanes; + + // Test-observability only (see ArteryTransportSetup.OnInboundLanesInitialized) -- + // null (the production default) does nothing. + _stage.OnLanesInitialized?.Invoke(laneCount); + } + + /// + /// One lane's long-running consumer loop (design's benchmark-locked mechanism: one + /// bounded per lane, one dedicated consumer loop per lane -- + /// runs on the default .NET ThreadPool via , no dedicated + /// dispatcher/thread). Awaits only BETWEEN messages + /// (/) + /// -- + /// is a plain synchronous call with NO internal , so it always + /// runs start-to-finish on whichever thread invokes it, which is exactly what keeps + /// (a + /// [ThreadStatic]) correct here even though different loop iterations may + /// legitimately resume on different pool threads. + /// + private static async Task RunLaneConsumer( + ChannelReader reader, + Akka.Serialization.Serialization serialization, + Action dispatch, + ILoggingAdapter log) + { + try + { + while (await reader.WaitToReadAsync().ConfigureAwait(false)) + { + while (reader.TryRead(out var item)) + { + try + { + var message = serialization.Deserialize(item.Payload, item.SerializerId, item.Manifest); + dispatch(message, item.SenderPath, item.RecipientPath); + } + catch (Exception ex) + { + log.Warning( + ex, + "Transient error deserializing/dispatching an inbound Artery lane message " + + "(serializer id [{0}], manifest [{1}]); connection remains live.", + item.SerializerId, item.Manifest); + } + } + } + } + catch (Exception ex) + { + log.Warning(ex, "Inbound Artery lane consumer loop terminated unexpectedly."); + } + } + + private void DrainReadyFramesLaneMode() + { + if (_writeInFlight) + return; + + while (_frameParser!.TryReadFrame(out var frameBody)) + { + bool handled; + try + { + handled = ProcessFrameLaneMode(frameBody); + } + catch (Exception ex) + { + Log.Warning(ex, "Transient error decoding an inbound Artery frame; connection remains live."); + continue; + } + + if (!handled) + return; // parked -- stop draining further frames until lane capacity frees up. + } + } + + /// + /// Decodes (never deserializes the application payload) one frame on an Ordinary-stream, + /// lane-enabled connection, then either (a) routes a genuine control-classified frame + /// (should never happen here -- defensive only, see the type-level remarks) through the + /// SAME inline path uses, unchanged, or (b) hashes its + /// recipient (or, for a selection, its target path elements) to a lane and hands it off. + /// Returns only when (b) had to park on a full lane channel -- + /// the caller must stop draining further frames until that clears. + /// + private bool ProcessFrameLaneMode(ReadOnlySequence frameBody) + { + var decoded = ArteryEnvelopeCodec.Decode(frameBody); + + if (!decoded.TryGetManifest(out var manifest)) + { + Log.Warning("Dropping inbound Artery frame: COMPRESSED manifest tag (ref/manifest compression is not implemented at G2)."); + return true; + } + + if (decoded.Header.SerializerId == _stage.ControlMessageSerializerId) + { + // Defensive only: a control-classified frame is not expected on an Ordinary + // connection (see the type-level remarks) -- deserialize INLINE exactly as the + // non-lane path does and push it through the ordinary Out pipeline so + // InboundHandshakeStage/SystemMessageAckerStage/DispatchInbound still see it. + var controlPayload = _stage.Serialization.Deserialize(decoded.Payload, decoded.Header.SerializerId, manifest); + _pending.Enqueue(new InboundEnvelope(controlPayload, null, null, decoded.Header.OriginUid, decoded.Header.SerializerId, manifest)); + return true; + } + + // Mirrors InboundHandshakeStage.IsKnownOrigin gating exactly (same shared-registry + // method, reused rather than reimplemented) -- "handshake completes before + // user-message dispatch per connection" holds for the lane path too. + if (!_stage.InboundContext!.IsKnownOrigin(decoded.Header.OriginUid)) + { + Log.Debug( + "Dropping inbound lane-routed Artery message from unknown origin uid [{0}] (no completed handshake for this uid yet).", + decoded.Header.OriginUid); + return true; + } + + if (!decoded.TryGetRecipientPath(out var recipientPath)) + { + Log.Warning( + "Dropping inbound ordinary-stream message (serializer id [{0}]): COMPRESSED recipient tag " + + "(ref compression is not implemented at G2).", decoded.Header.SerializerId); + return true; + } + + if (recipientPath is null) + { + Log.Warning("Dropping inbound ordinary-stream message (serializer id [{0}]) with no recipient.", decoded.Header.SerializerId); + return true; + } + + var senderPath = decoded.TryGetSenderPath(out var s) ? s : null; + + // ActorSelectionMessage special case -- see the type-level "ActorSelectionMessage + // lane key" remarks. + var laneKey = decoded.Header.SerializerId == _stage.SelectionMessageSerializerId + ? BuildSelectionLaneKey(decoded.Payload) ?? recipientPath + : recipientPath; + + var laneIndex = LaneFor(laneKey, _stage.InboundLanes); + + return TryRouteToLane(laneIndex, new LaneWorkItem(decoded.Payload, decoded.Header.SerializerId, manifest, senderPath, recipientPath)); + } + + /// + /// Parses ONLY the outer SelectionEnvelope wire wrapper (MessageContainerSerializer's + /// format) to recover an ActorSelectionMessage's OWN target path elements, WITHOUT + /// deserializing the wrapped application message it carries (that stays fully deferred to + /// the lane, exactly like any other ordinary message) -- see the type-level + /// "ActorSelectionMessage lane key" remarks for why RecipientPath alone cannot be + /// used here. Returns (safe fallback to RecipientPath-based + /// hashing -- same-lane, never WRONG) if the wrapper has no pattern elements or fails to + /// parse for any reason. + /// + private static string? BuildSelectionLaneKey(in ReadOnlySequence payload) + { + try + { + var bytes = payload.IsSingleSegment ? payload.FirstSpan.ToArray() : payload.ToArray(); + var envelope = Akka.Remote.Serialization.Proto.Msg.SelectionEnvelope.Parser.ParseFrom(bytes); + if (envelope.Pattern.Count == 0) + return null; + + var sb = new StringBuilder(); + foreach (var element in envelope.Pattern) + sb.Append('/').Append(element.Matcher); + + return sb.ToString(); + } + catch + { + return null; + } + } + + /// + /// Deterministic FNV-1a hash of , unsigned-mod'd into + /// [0, lanes). Unsigned mod (not on a signed hash) + /// deliberately avoids the classic Math.Abs(int.MinValue) overflow trap (which + /// returns unchanged, still negative). + /// + private static int LaneFor(string key, int lanes) + { + const uint offsetBasis = 2166136261; + const uint prime = 16777619; + + var hash = offsetBasis; + foreach (var c in key) + { + hash ^= c; + hash *= prime; + } + + return (int)(hash % (uint)lanes); + } + + /// + /// Attempts to hand to lane 's + /// channel. Returns if it was written (or safely dropped because + /// the lane's channel is already completing -- a teardown race, not a capacity problem). + /// Returns if the channel was full: the item is PARKED (never + /// dropped) and re-attempted once 's + /// continuation fires () -- the caller must stop + /// draining further frames from this connection until then, which is exactly the + /// backpressure signal design.md calls for (never block a stage thread; never drop). + /// + private bool TryRouteToLane(int laneIndex, LaneWorkItem item) + { + var writer = _lanes![laneIndex].Writer; + if (writer.TryWrite(item)) + return true; + + var continuation = writer.WaitToWriteAsync(); + if (continuation.IsCompletedSuccessfully) + { + var available = continuation.GetAwaiter().GetResult(); + if (available && writer.TryWrite(item)) + return true; + + // Either the channel was completed underneath us (available == false -- a + // teardown race) or another writer already took the freed slot -- SingleWriter + // means the latter cannot happen here, so this is the teardown race. Drop with + // a debug log rather than parking on a channel that will never accept another + // write. + Log.Debug("Dropping inbound Artery ordinary message: lane [{0}]'s channel is already completing.", laneIndex); + return true; + } + + _parkedLaneIndex = laneIndex; + _parkedItem = item; + _writeInFlight = true; + continuation.AsTask().ContinueWith(_onLaneWriteReady!, TaskContinuationOptions.ExecuteSynchronously); + return false; + } + + private void OnLaneWriteAvailable(bool available) + { + _writeInFlight = false; + var writer = _lanes![_parkedLaneIndex].Writer; + + if (available && writer.TryWrite(_parkedItem)) + { + _parkedItem = default; + } + else + { + // Channel completed underneath us (connection tearing down) -- drop; teardown + // is already in progress, there is nothing further to do for this element. + if (!available) + Log.Debug("Dropping inbound Artery ordinary message: lane [{0}]'s channel completed while parked.", _parkedLaneIndex); + _parkedItem = default; + } + + // Resume: drain whatever further frames are already buffered, then either push a + // pending control envelope, pull more input, or -- if upstream had ALREADY + // finished while we were parked -- complete the stage now (nothing else will + // re-trigger that check for a connection carrying only ordinary/lane traffic). + DrainReadyFramesLaneMode(); + DeliverOrPull(); + + if (_pending.Count == 0 && !_writeInFlight && IsClosed(_stage.In)) + CompleteStage(); + } + + #endregion } } } diff --git a/src/core/Akka.Remote/Artery/ArteryRemoting.cs b/src/core/Akka.Remote/Artery/ArteryRemoting.cs index 7139dc00a36..18ef43c9070 100644 --- a/src/core/Akka.Remote/Artery/ArteryRemoting.cs +++ b/src/core/Akka.Remote/Artery/ArteryRemoting.cs @@ -157,6 +157,13 @@ internal sealed class ArteryRemoting : RemoteTransport, IControlMessageSubscribe /// private Func? _dropOutboundControlMessage; + /// + /// Test-observability hook (inbound lanes) -- see . + /// Read once from in ; + /// (production default) disables it entirely. + /// + private Action? _onInboundLanesInitialized; + /// /// Applied to EVERY Artery socket: the accepting Tcp.Bind and both outbound /// Tcp.OutgoingConnection call sites in . @@ -232,6 +239,7 @@ public override void Start() _encodeBufferPool = arteryTransportSetup.Select(s => s.EncodeBufferPool).GetOrElse(null) ?? ArrayPool.Create(); _dropOutboundControlMessage = arteryTransportSetup.Select(s => s.DropOutboundControlMessage).GetOrElse(null); + _onInboundLanesInitialized = arteryTransportSetup.Select(s => s.OnInboundLanesInitialized).GetOrElse(null); // Large-message stream (task 10.2): a dedicated pool, sized from the large-specific // settings -- see _largeEncodeBufferPool's remarks for why this is kept separate from @@ -456,9 +464,29 @@ private void HandleIncomingConnection(Tcp.IncomingConnection connection) // [control only: SystemMessageAcker]" pipeline -- it is a no-op pass-through for every // element that is not a SystemMessageEnvelope, so composing it unconditionally here // (rather than only for control-preamble connections) is correct and simpler. + // + // GATE A / conditional topology (inbound lanes): at the InboundLanes=1 default, this + // constructs the stage EXACTLY as before -- the 3-arg overload, no lane machinery ever + // materialized. Only when InboundLanes > 1 does construction take the lanes-aware + // overload, handing it the shared IInboundContext (reused for its own IsKnownOrigin + // gate on an Ordinary connection's lane path -- see that stage's remarks) and + // DispatchOrdinaryMessage (the SAME resolve-and-Tell logic DispatchInbound's ordinary + // branch uses, invoked directly by each lane's consumer loop instead of via this + // Sink.ForEach). InboundHandshakeStage/SystemMessageAckerStage/DispatchInbound below are + // NEVER modified for lanes -- a lane-enabled Ordinary connection's own ordinary traffic + // never reaches them at all (dispatched directly from the lane), while any (should + // never happen) control-classified frame on that same connection, and everything on + // every Control/Large connection, still flows through them completely unchanged. + var processingStage = _settings.InboundLanes > 1 + ? new ArteryInboundProcessingStage( + _settings.MaximumFrameSize, _settings.MaximumLargeFrameSize, System.Serialization, + _settings.InboundLanes, _settings.InboundLaneBufferSize, _inboundContext!, DispatchOrdinaryMessage, + _onInboundLanesInitialized) + : new ArteryInboundProcessingStage(_settings.MaximumFrameSize, _settings.MaximumLargeFrameSize, System.Serialization); + var inboundSink = Flow.Create>() .Via(_killSwitch.Flow>()) - .Via(new ArteryInboundProcessingStage(_settings.MaximumFrameSize, _settings.MaximumLargeFrameSize, System.Serialization)) + .Via(processingStage) .Via(Flow.FromGraph(new InboundHandshakeStage(_inboundContext!))) .Via(Flow.FromGraph(new SystemMessageAckerStage(_inboundContext!))) .To(Sink.ForEach(DispatchInbound)); @@ -487,8 +515,6 @@ private void DispatchInbound(IInboundEnvelope env) return; } - var recipient = Provider.ResolveActorRefWithLocalAddress(env.RecipientPath, DefaultAddress); - if (env.Message is ISystemMessage systemMessage) { // Reliable system-message delivery (design.md gate G3): SystemMessageAckerStage has @@ -496,18 +522,51 @@ private void DispatchInbound(IInboundEnvelope env) // classic's DefaultMessageDispatcher system-message path, NOT Tell. System messages // never carry a sender in practice (RemoteActorRef.SendSystemMessage always sends // with sender: null) -- see SystemMessageEnvelope's type-level remarks. + var recipient = Provider.ResolveActorRefWithLocalAddress(env.RecipientPath, DefaultAddress); + recipient.SendSystemMessage(systemMessage); + return; + } + + DispatchOrdinaryMessage(env.Message, env.SenderPath, env.RecipientPath); + } + + /// + /// The ordinary-message half of (resolve recipient, resolve + /// sender or fall back to dead letters, Tell) -- factored out so it can ALSO be + /// invoked directly by an inbound lane's consumer loop (, + /// when > 1) once it has deserialized the + /// payload, bypassing this entirely for that traffic. Identical logic, + /// identical semantics, regardless of which caller invokes it -- lanes change WHERE this + /// runs (a dedicated lane thread vs. this connection's own stage thread), never WHAT it does. + /// + private void DispatchOrdinaryMessage(object message, string? senderPath, string recipientPath) + { + var recipient = Provider.ResolveActorRefWithLocalAddress(recipientPath, DefaultAddress); + + // Defensive-only safety branch mirroring DispatchInbound's ISystemMessage handling + // above -- in practice UNREACHABLE for either of this method's callers: DispatchInbound + // itself already special-cases ISystemMessage before ever reaching here (the lanes=1 + // path), and lane-routed traffic is Ordinary-stream-only by construction (every system + // message rides the CONTROL stream, wrapped in a SystemMessageEnvelope, via + // EnqueueSystemMessage -- never the ordinary queue lanes fan out from; see + // ArteryInboundProcessingStage's "Why control/large connections and lanes=1 never touch + // this machinery" remarks). Kept anyway so this method's dispatch semantics stay EXACTLY + // identical to DispatchInbound's regardless of which caller -- the lanes=1 Sink.ForEach + // above, or an inbound lane's consumer loop -- ends up invoking it. + if (message is ISystemMessage systemMessage) + { recipient.SendSystemMessage(systemMessage); return; } - var sender = env.SenderPath is { } senderPath - ? Provider.ResolveActorRefWithLocalAddress(senderPath, DefaultAddress) + var sender = senderPath is { } sp + ? Provider.ResolveActorRefWithLocalAddress(sp, DefaultAddress) : (IActorRef)System.DeadLetters; // Mirrors classic's DefaultMessageDispatcher semantics without depending on classic types: // an unresolvable recipient resolves to an EmptyLocalActorRef, whose Tell publishes a // DeadLetter automatically. - recipient.Tell(env.Message, sender); + recipient.Tell(message, sender); } /// diff --git a/src/core/Akka.Remote/Artery/ArterySettings.cs b/src/core/Akka.Remote/Artery/ArterySettings.cs index 4c0878d52cc..7c3227817ef 100644 --- a/src/core/Akka.Remote/Artery/ArterySettings.cs +++ b/src/core/Akka.Remote/Artery/ArterySettings.cs @@ -62,15 +62,33 @@ internal sealed class ArterySettings public int MaximumFrameSize { get; } /// - /// Number of inbound lanes ordinary-stream messages are fanned out across for parallel - /// deserialization/dispatch. Parsed and validated now but NOT used until the lanes work - /// lands at gate G5 -- see design.md's milestone table. + /// Number of inbound lanes ordinary-stream messages are fanned out across, WITHIN one + /// accepted connection, for parallel payload deserialization + actor-ref resolution + + /// Tell -- see 's remarks for the full + /// mechanism (bounded per-lane s + + /// dedicated consumer loops) and the lane-key derivation (recipient path hash; special + /// cased for , whose RecipientPath is + /// always the SAME remote root-guardian anchor). Ships at 1 (see + /// for the per-lane queue capacity used when this is + /// > 1) -- AT 1, the lane machinery is not materialized at all; the inbound pipeline is + /// byte-identical to the pre-lanes single-threaded-per-connection processing. /// public int InboundLanes { get; } + /// + /// Capacity of the bounded feeding each + /// inbound lane's consumer loop, when > 1 -- see that + /// property's remarks. A full lane channel backpressures the connection (never drops, + /// never grows unbounded): the accepting + /// parks the connection's read side via + /// until room frees up. Irrelevant (never read) when is 1. + /// + public int InboundLaneBufferSize { get; } + /// /// Number of outbound connections (lanes) used to send ordinary-stream messages. Parsed - /// and validated now but NOT used until gate G5; see . + /// and validated now but NOT YET USED -- outbound lane fan-out is separate, still-pending + /// work (this port's inbound side is what now drives). /// public int OutboundLanes { get; } @@ -267,11 +285,16 @@ public ArterySettings(Config arteryConfig) $"offsets stay within their 24-bit tag space, but was [{maximumFrameSize}]."); MaximumFrameSize = (int)maximumFrameSize; - InboundLanes = arteryConfig.GetInt("advanced.inbound-lanes", 4); + InboundLanes = arteryConfig.GetInt("advanced.inbound-lanes", 1); if (InboundLanes < 1) throw new ConfigurationException( $"akka.remote.artery.advanced.inbound-lanes must be >= 1, but was [{InboundLanes}]."); + InboundLaneBufferSize = arteryConfig.GetInt("advanced.inbound-lane-buffer-size", 4096); + if (InboundLaneBufferSize <= 0) + throw new ConfigurationException( + $"akka.remote.artery.advanced.inbound-lane-buffer-size must be greater than 0, but was [{InboundLaneBufferSize}]."); + OutboundLanes = arteryConfig.GetInt("advanced.outbound-lanes", 1); if (OutboundLanes < 1) throw new ConfigurationException( diff --git a/src/core/Akka.Remote/Artery/ArteryTransportSetup.cs b/src/core/Akka.Remote/Artery/ArteryTransportSetup.cs index 25498631795..8cc2aab4900 100644 --- a/src/core/Akka.Remote/Artery/ArteryTransportSetup.cs +++ b/src/core/Akka.Remote/Artery/ArteryTransportSetup.cs @@ -67,10 +67,17 @@ internal sealed class ArteryTransportSetup : Setup /// it. (the default, and the only production value) disables this /// entirely -- every control message is enqueued normally. /// - public ArteryTransportSetup(ArrayPool? encodeBufferPool = null, Func? dropOutboundControlMessage = null) + /// + /// Test-observability hook -- see . + /// + public ArteryTransportSetup( + ArrayPool? encodeBufferPool = null, + Func? dropOutboundControlMessage = null, + Action? onInboundLanesInitialized = null) { EncodeBufferPool = encodeBufferPool; DropOutboundControlMessage = dropOutboundControlMessage; + OnInboundLanesInitialized = onInboundLanesInitialized; } /// @@ -86,5 +93,18 @@ public ArteryTransportSetup(ArrayPool? encodeBufferPool = null, Func (the default) disables this -- see the constructor parameter docs. /// public Func? DropOutboundControlMessage { get; } + + /// + /// Test-observability hook (inbound lanes): invoked once, with the actual configured lane + /// count, every time an accepted Ordinary-stream connection's + /// actually materializes its per-lane + /// s + consumer loops (i.e. lets a test + /// observe "lanes > 1 really did create N lanes" without reaching into the stage's own + /// private instance, which is otherwise + /// unreachable from outside the materialized stream). (the default, + /// and the only production value) disables this entirely -- it is never invoked at the + /// InboundLanes=1 default in any case, since lane machinery is never materialized then. + /// + public Action? OnInboundLanesInitialized { get; } } } diff --git a/src/core/Akka.Remote/Configuration/Remote.conf b/src/core/Akka.Remote/Configuration/Remote.conf index 49f38d54e91..24948d54017 100644 --- a/src/core/Akka.Remote/Configuration/Remote.conf +++ b/src/core/Akka.Remote/Configuration/Remote.conf @@ -694,15 +694,34 @@ akka { # default. large-buffer-pool-size = 32 - # Number of inbound lanes user (ordinary-stream) messages are fanned out across for - # parallel deserialization/dispatch. Parsed now but NOT used until the lanes work lands - # (gate G5) -- G2/G3/G4 run effectively single-lane regardless of this value. - inbound-lanes = 4 + # Number of inbound lanes ordinary-stream (user) messages are fanned out across, WITHIN + # one accepted connection, for parallel payload deserialization + actor-ref resolution + + # Tell. A message's lane is a stable hash of its recipient path (or, for an + # ActorSelectionMessage, the selection's own target path elements -- see + # ArteryInboundProcessingStage's remarks), so all traffic to the SAME recipient stays on + # the SAME lane and is delivered in send order. Handshake/heartbeat/quarantine-notice/ + # reliable-system-message traffic (the CONTROL connection) is completely unaffected by + # this setting regardless of its value -- only the ORDINARY connection's user messages are + # ever lane-routed. + # + # Ships at 1 (single-lane, fused per-connection processing -- byte-identical to the + # pre-lanes pipeline; the lane machinery is not even materialized at this value) until the + # scaling gates pass. Raise this only after validating the improvement for your workload; + # values > 1 need N dedicated lane-consumer background tasks per accepted ordinary + # connection, each with its own bounded queue (see `inbound-lane-buffer-size` below). + inbound-lanes = 1 # Number of outbound connections (lanes) used to send ordinary-stream messages. # Parsed now but NOT used until gate G5; see `inbound-lanes` above. outbound-lanes = 1 + # Capacity of the bounded, in-process queue (a System.Threading.Channels.Channel) feeding + # EACH inbound lane's consumer, when `inbound-lanes` > 1. A full lane queue backpressures + # the connection (the accepting stage stops reading/parking until room frees up) -- it + # never drops a message or grows unbounded. Irrelevant at the `inbound-lanes = 1` default, + # since no lane queues are created at that value. + inbound-lane-buffer-size = 4096 + # How long an association will wait for a handshake to complete before failing and # retrying. See the `handshake-retry-interval` / `inject-handshake-interval` below. handshake-timeout = 20 s