diff --git a/docs/exp/SER009.md b/docs/exp/SER009.md new file mode 100644 index 000000000..a0f921859 --- /dev/null +++ b/docs/exp/SER009.md @@ -0,0 +1,22 @@ +`RESPite.Transports.DuplexTransport` (and `TransportReceiver`) are an experimental transport +abstraction: staged-and-flushed outbound (the transport itself is the `IBufferWriter`), push +inbound with transport-owned memory, and an explicit batch-end notification. They exist to let +alternative IO engines plug in beneath the library (see `Tunnel.ConnectTransportAsync`) without being +constrained to `Stream` or pipe semantics. + +**This API is not intended for external use at this time.** It exists to support internal +experimentation with alternative transports, and is public only because a seam has to be public to be +implemented. Specifically: + +1. **No stability whatsoever is implied.** The shape may gain members, change semantics, be renamed, + move assembly or namespace, or be removed outright — between *any* two versions, including + patches, with no breaking-change ceremony, no deprecation period, and no migration notes. +2. **Do not implement or consume it in code you ship.** If you build on it anyway, you are accepting + that any update may break you without warning, and issues asking for compatibility or support for + it will be closed. +3. The `[Experimental]` diagnostic (`SER009`) is the enforcement mechanism: suppressing it is you + signing up to the above. + +If a transport seam useful beyond this experiment emerges, it will be stabilised deliberately — with +its own announcement and a removed `[Experimental]` marker — rather than by this surface quietly +hardening into a contract. diff --git a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt index 7dc5c5811..3b10f3da4 100644 --- a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt @@ -1 +1,14 @@ #nullable enable +[SER009]RESPite.Transports.DuplexTransport +[SER009]RESPite.Transports.DuplexTransport.DuplexTransport() -> void +[SER009]abstract RESPite.Transports.DuplexTransport.Advance(int count) -> void +[SER009]abstract RESPite.Transports.DuplexTransport.DisposeAsync() -> System.Threading.Tasks.ValueTask +[SER009]abstract RESPite.Transports.DuplexTransport.Flush() -> bool +[SER009]abstract RESPite.Transports.DuplexTransport.GetMemory(int sizeHint = 0) -> System.Memory +[SER009]virtual RESPite.Transports.DuplexTransport.GetSpan(int sizeHint = 0) -> System.Span +[SER009]abstract RESPite.Transports.DuplexTransport.Start(RESPite.Transports.TransportReceiver! receiver) -> void +[SER009]RESPite.Transports.TransportReceiver +[SER009]RESPite.Transports.TransportReceiver.TransportReceiver() -> void +[SER009]abstract RESPite.Transports.TransportReceiver.OnReceived(System.ReadOnlySpan payload) -> bool +[SER009]virtual RESPite.Transports.TransportReceiver.OnBatchEnd() -> void +[SER009]virtual RESPite.Transports.TransportReceiver.OnClosed(System.Exception? fault) -> void diff --git a/src/RESPite/Shared/Experiments.cs b/src/RESPite/Shared/Experiments.cs index f72ece073..39fb2426d 100644 --- a/src/RESPite/Shared/Experiments.cs +++ b/src/RESPite/Shared/Experiments.cs @@ -20,6 +20,7 @@ internal static class Experiments public const string UnitTesting = "SER005"; public const string GeoRedundantFailover = "SER007"; public const string Server_8_10 = "SER008"; + public const string Transport = "SER009"; // ReSharper restore InconsistentNaming diff --git a/src/RESPite/Transports/DuplexTransport.cs b/src/RESPite/Transports/DuplexTransport.cs new file mode 100644 index 000000000..2701d60b6 --- /dev/null +++ b/src/RESPite/Transports/DuplexTransport.cs @@ -0,0 +1,72 @@ +using System; +using System.Buffers; +using System.Diagnostics.CodeAnalysis; +using System.Threading.Tasks; + +namespace RESPite.Transports; + +/// +/// A duplex byte transport, deliberately NOT a and NOT a pipe: outbound +/// is staged-and-flushed (batching is an explicit contract point, not an implementation accident), and +/// inbound is PUSH — the transport delivers bytes to a on the +/// transport's own schedule, in transport-owned memory. +/// +/// The shape is derived from measured transport work rather than taste: any-thread copying writes with +/// an explicit flush (batching at the caller's natural boundaries was the largest single lever +/// measured), push delivery (pull adapters over a push transport measured 24-40% overhead), and a +/// batch-end notification (coalescing responses produced during a delivery burst into one flush +/// eliminated a measured 3x send amplification). +/// +/// The transport IS the outbound — there is no separate output object. +/// Staging is callable from any thread (single logical writer at a time); bytes are owned by the +/// transport once returns; hands the staged bytes to the +/// wire. Passing the transport AS deliberately grants stage-only +/// access: the holder composes, the owner flushes at its batch boundary. +/// +[Experimental(Experiments.Transport, UrlFormat = Experiments.UrlFormat)] +public abstract class DuplexTransport : IBufferWriter, IAsyncDisposable +{ + /// Request writable space to stage outbound bytes (see ). + public abstract Memory GetMemory(int sizeHint = 0); + + /// + /// Defaults to GetMemory(sizeHint).Span; override when the transport has a cheaper + /// span path than a round-trip. + public virtual Span GetSpan(int sizeHint = 0) => GetMemory(sizeHint).Span; + + /// Commit bytes obtained via or + /// ; the transport owns them when this returns, so caller state need not + /// survive it. + public abstract void Advance(int count); + + /// Hand everything staged since the last flush to the wire, as one send where the + /// transport allows. Returns false if the transport is closed (staged bytes are dropped). + public abstract bool Flush(); + + /// Begin inbound delivery. Exactly one receiver, set once, before any data is expected; + /// delivery runs on the transport's schedule and threads. + public abstract void Start(TransportReceiver receiver); + + public abstract ValueTask DisposeAsync(); +} + +/// +/// The consumer half of . Callbacks run on the transport's threads and +/// must be bounded and non-blocking; anything long-running belongs on the consumer's own scheduler. +/// +[Experimental(Experiments.Transport, UrlFormat = Experiments.UrlFormat)] +public abstract class TransportReceiver +{ + /// Bytes arrived. is TRANSPORT-OWNED and valid only for the + /// duration of the call — copy anything retained. Return false to request the transport close. + public abstract bool OnReceived(ReadOnlySpan payload); + + /// A delivery burst has ended (for loop transports: the event batch is drained). Flush + /// anything staged in response to the burst HERE, once, rather than per — + /// per-callback flushing measurably amplifies peer segmentation. + public virtual void OnBatchEnd() { } + + /// The transport closed; fires exactly once. is the failure when + /// the transport can attribute one, else null (a clean or unattributed close). + public virtual void OnClosed(Exception? fault) { } +} diff --git a/src/StackExchange.Redis/Configuration/Tunnel.cs b/src/StackExchange.Redis/Configuration/Tunnel.cs index 9c28cbd2d..72cdbc4d2 100644 --- a/src/StackExchange.Redis/Configuration/Tunnel.cs +++ b/src/StackExchange.Redis/Configuration/Tunnel.cs @@ -1,5 +1,6 @@ using System; using System.Buffers; +using System.Diagnostics.CodeAnalysis; using System.IO; using System.Net; using System.Net.Sockets; @@ -35,6 +36,16 @@ public abstract class Tunnel /// public virtual ValueTask BeforeAuthenticateAsync(EndPoint endpoint, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) => default; + /// + /// Optionally supply the ENTIRE transport for this connection — the same hijack as + /// one level deeper: instead of yielding a + /// over a socket the library owns, yield a + /// the tunnel owns, and no socket is created at + /// all. Return null (the default for every existing tunnel) for the standard socket path. + /// + [Experimental(RESPite.Experiments.Transport, UrlFormat = RESPite.Experiments.UrlFormat)] + public virtual ValueTask ConnectTransportAsync(EndPoint endpoint, ConnectionType connectionType, CancellationToken cancellationToken) => default; + private sealed class HttpProxyTunnel : Tunnel { public EndPoint Proxy { get; } diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 7dc5c5811..ddf237e1a 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1 +1,2 @@ #nullable enable +[SER009]virtual StackExchange.Redis.Configuration.Tunnel.ConnectTransportAsync(System.Net.EndPoint! endpoint, StackExchange.Redis.ConnectionType connectionType, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask