From 87c1c8097893e5aa738bca7857a224669932bcab Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 17 Jun 2026 13:20:47 +0100 Subject: [PATCH 1/6] - introduce CycleBufferPool as a concept that encapsulates MemoryPool with the growth logic - support `-d` in the benchmark --- src/RESPite.Benchmark/BenchmarkBase.cs | 21 +++- src/RESPite.Benchmark/readme.md | 23 ++--- src/RESPite/Buffers/CycleBuffer.cs | 26 ++--- src/RESPite/Buffers/CycleBufferPool.cs | 96 +++++++++++++++++++ src/RESPite/PublicAPI/PublicAPI.Shipped.txt | 3 - src/RESPite/PublicAPI/PublicAPI.Unshipped.txt | 7 ++ 6 files changed, 142 insertions(+), 34 deletions(-) create mode 100644 src/RESPite/Buffers/CycleBufferPool.cs diff --git a/src/RESPite.Benchmark/BenchmarkBase.cs b/src/RESPite.Benchmark/BenchmarkBase.cs index 50994b23f..6e08b0515 100644 --- a/src/RESPite.Benchmark/BenchmarkBase.cs +++ b/src/RESPite.Benchmark/BenchmarkBase.cs @@ -59,7 +59,7 @@ public enum PipelineStrategy protected BenchmarkBase(string[] args) { - int operations = 100_000; + int operations = 100_000, payloadSize = -1; string tests = ""; for (int i = 0; i < args.Length; i++) @@ -81,6 +81,9 @@ protected BenchmarkBase(string[] args) case "-P" when i != args.Length - 1 && int.TryParse(args[++i], out int tmp) && tmp > 0: PipelineDepth = tmp; break; + case "-d" when i != args.Length - 1 && int.TryParse(args[++i], out int tmp) && tmp > 0: + payloadSize = tmp; + break; case "-w" when i != args.Length - 1 && int.TryParse(args[++i], out int tmp): WriteMode = tmp; break; @@ -125,7 +128,21 @@ protected BenchmarkBase(string[] args) _operationsPerClient = operations / ClientCount; - Payload = "abc"u8.ToArray(); + if (payloadSize < 0) + { + Payload = "abc"u8.ToArray(); + } + else + { + var arr = new byte[payloadSize]; + var rand = new Random(payloadSize); // use the size as the seed, for repeatability + for (int i = 0; i < arr.Length; i++) + { + arr[i] = (byte)rand.Next(32, 127); // space thru ~ (ignore DEL) + } + + Payload = arr; + } } public abstract Task RunAll(); diff --git a/src/RESPite.Benchmark/readme.md b/src/RESPite.Benchmark/readme.md index ea8fe2105..cf8a6b0c0 100644 --- a/src/RESPite.Benchmark/readme.md +++ b/src/RESPite.Benchmark/readme.md @@ -21,35 +21,36 @@ Example usage: Basic options, for parity: -- `-h ` Server hostname (default 127.0.0.1) -- `-p ` Server port (default 6379) +- `-h ` Server hostname (default 127.0.0.1). +- `-p ` Server port (default 6379). - `-c ` Number of parallel connections (default 50). -- `-n ` Total number of requests (default 100000) +- `-n ` Total number of requests (default 100000). +- `-d ` Data size of SET/GET value in bytes (default 3). - `-P ` Pipeline requests. Default 1 (no pipeline). -- `-l` Loop. Run the tests forever -- `-q` Quiet. Just show query/sec values +- `-l` Loop. Run the tests forever. +- `-q` Quiet. Just show query/sec values. - `-t ` Only run the comma separated list of tests. The test names are the same as the ones produced as output. ## Custom options Additional options specific to this tool: -- `+m` / `-m`: enable or disable (default) multiplexing: when enabled clients share a connection, otherwise each client has a separate connection -- `--batch` / `--queue` pipelining should using batching (default) or queueing strategy -- `--basic` : perform basic typical IO operations rather than synthetic benchmarks +- `+m` / `-m`: enable or disable (default) multiplexing: when enabled clients share a connection, otherwise each client has a separate connection. +- `--batch` / `--queue` pipelining should using batching (default) or queueing strategy. +- `--basic` : perform basic typical IO operations rather than synthetic benchmarks. ## Internal options These exist mostly for Marc's benefit: -- `-w ` Specify the internal write-mode -- `+x` / `-x`: enable or disable (default) cancellation support (irrelevant until later v3 tranche) +- `-w ` Specify the internal write-mode. +- `+x` / `-x`: enable or disable (default) cancellation support (irrelevant until later v3 tranche). ## Local example To build and run from source, `dotnet run` can be used with everything after `--` being args to the command: ``` -dotnet run -p:TargetVer=3 -f net10.0 -c Release -- -q +m --batch -n 500000 +dotnet run -p:TargetVer=3 -f net10.0 -c Release -- -q -c 50 -P 100 +m --queue -n 500000 -q -l -t INCR ``` \ No newline at end of file diff --git a/src/RESPite/Buffers/CycleBuffer.cs b/src/RESPite/Buffers/CycleBuffer.cs index 14774b357..334171c12 100644 --- a/src/RESPite/Buffers/CycleBuffer.cs +++ b/src/RESPite/Buffers/CycleBuffer.cs @@ -38,28 +38,18 @@ public partial struct CycleBuffer // note: if someone uses an uninitialized CycleBuffer (via default): that's a skills issue; git gud public static CycleBuffer Create( - MemoryPool? pool = null, - int pageSize = DefaultPageSize, - ICycleBufferCallback? callback = null) - { - pool ??= DefaultPool; - if (pageSize <= 0) pageSize = DefaultPageSize; - if (pageSize > pool.MaxBufferSize) pageSize = pool.MaxBufferSize; - return new CycleBuffer(pool, pageSize, callback); - } + CycleBufferPool? pool = null, + ICycleBufferCallback? callback = null) => new(pool, callback); - private CycleBuffer(MemoryPool pool, int pageSize, ICycleBufferCallback? callback) + private CycleBuffer(CycleBufferPool? pool, ICycleBufferCallback? callback = null) { - Pool = pool; - PageSize = pageSize; + _pool = pool ?? CycleBufferPool.Default; _callback = callback; leasedStart = -1; } - private const int DefaultPageSize = 8 * 1024; - - public int PageSize { get; } - public MemoryPool Pool { get; } + public CycleBufferPool Pool => _pool; + private readonly CycleBufferPool _pool; private readonly ICycleBufferCallback? _callback; private Segment? startSegment, endSegment; @@ -412,7 +402,7 @@ private Segment GetNextSegment() } } - Segment newSegment = Segment.Create(Pool.Rent(PageSize)); + Segment newSegment = Segment.Create(endSegment is null ? _pool.Rent() : _pool.Rent(GetAllCommitted())); if (endSegment is null) { // tabula rasa @@ -452,7 +442,7 @@ public Memory GetUncommittedMemory(int hint = 0) { if (!memory.IsEmpty) return MemoryMarshal.AsMemory(memory); } - else if (memory.Length >= Math.Min(hint, PageSize >> 2)) // respect the hint up to 1/4 of the page size + else if (memory.Length >= Math.Min(hint, 1024)) // respect the hint, or 1k (only relevant when large requests are made) { return MemoryMarshal.AsMemory(memory); } diff --git a/src/RESPite/Buffers/CycleBufferPool.cs b/src/RESPite/Buffers/CycleBufferPool.cs new file mode 100644 index 000000000..76a87e4dc --- /dev/null +++ b/src/RESPite/Buffers/CycleBufferPool.cs @@ -0,0 +1,96 @@ +using System.Buffers; +using System.Diagnostics.CodeAnalysis; + +namespace RESPite.Buffers; + +[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] +public abstract class CycleBufferPool +{ + /// + /// Create an initial buffer. + /// + public virtual IMemoryOwner Rent() => Rent(default); + + /// + /// Create a buffer with knowledge of the existing leased data. + /// + public abstract IMemoryOwner Rent(in ReadOnlySequence existing); + + // new MemoryPool(...) would be a non-growing buffer pool. + public static CycleBufferPool Default { get; } = new GrowingMemoryPool(minBytes: 8 * 1024); + + private class MemoryPool : CycleBufferPool + { +#if TRACK_MEMORY + private static MemoryPool DefaultPool => MemoryTrackedPool.Shared; +#else + private static MemoryPool DefaultPool => MemoryPool.Shared; +#endif + private readonly MemoryPool _pool; + private readonly int _minBytes, _maxBytes; + + public MemoryPool(int minBytes, MemoryPool? pool = null, int maxBytes = int.MaxValue) + { + _pool = pool ?? DefaultPool; + // capture the max bytes, without exceeding the pool's max size + _maxBytes = Math.Min(maxBytes, _pool.MaxBufferSize); + // capture the min bytes, applying a rigid lower bound, and not overlapping the max bytes + _minBytes = Math.Min(Math.Max(minBytes, 16), _maxBytes); + } + + /// + /// Rent a chunk using the specified size as a hint. + /// + protected IMemoryOwner Rent(int bytes) + { +#if NET + bytes = Math.Clamp(bytes, _minBytes, _maxBytes); +#else + bytes = Math.Min(Math.Max(bytes, _minBytes), _maxBytes); +#endif + return _pool.Rent(bytes); + } + + // by default, use fixed size without reference to the existing data; subclasses can tweak + + /// + public override IMemoryOwner Rent() => Rent(_minBytes); + + /// + public override IMemoryOwner Rent(in ReadOnlySequence existing) => Rent(_minBytes); + } + + private sealed class GrowingMemoryPool(int minBytes, MemoryPool? pool = null, int maxBytes = int.MaxValue) + : MemoryPool(minBytes, pool, maxBytes) + { + public override IMemoryOwner Rent(in ReadOnlySequence existing) + { + if (existing.IsEmpty) return base.Rent(existing); + // use a growth strategy looking at the size of the last segment + int lastChunk; + if (existing.IsSingleSegment) + { + lastChunk = existing.First.Length; + } + else if (existing.End.GetObject() is ReadOnlySequenceSegment segment) + { + lastChunk = segment.Memory.Length; // note we ignore GetInteger() intentionally + } + else + { + // do it the hard way; note we'll only observe the reserved size, rather + // than the actual buffer size, but that's the best we can do + lastChunk = 0; + foreach (var chunk in existing) + { + if (!chunk.IsEmpty) lastChunk = chunk.Length; + } + + if (lastChunk is 0) lastChunk = existing.First.Length; // paranoia + } + + // "max" here is to account for overflow - i.e. stop growing if that happens (unlikely) + return Rent(Math.Max(lastChunk, lastChunk << 1)); + } + } +} diff --git a/src/RESPite/PublicAPI/PublicAPI.Shipped.txt b/src/RESPite/PublicAPI/PublicAPI.Shipped.txt index 27160d830..f5ab197bf 100644 --- a/src/RESPite/PublicAPI/PublicAPI.Shipped.txt +++ b/src/RESPite/PublicAPI/PublicAPI.Shipped.txt @@ -32,8 +32,6 @@ [SER004]RESPite.Buffers.CycleBuffer.GetCommittedLength() -> long [SER004]RESPite.Buffers.CycleBuffer.GetUncommittedMemory(int hint = 0) -> System.Memory [SER004]RESPite.Buffers.CycleBuffer.GetUncommittedSpan(int hint = 0) -> System.Span -[SER004]RESPite.Buffers.CycleBuffer.PageSize.get -> int -[SER004]RESPite.Buffers.CycleBuffer.Pool.get -> System.Buffers.MemoryPool! [SER004]RESPite.Buffers.CycleBuffer.Release() -> void [SER004]RESPite.Buffers.CycleBuffer.TryGetCommitted(out System.ReadOnlySpan span) -> bool [SER004]RESPite.Buffers.CycleBuffer.TryGetFirstCommittedMemory(int minBytes, out System.ReadOnlyMemory memory) -> bool @@ -203,7 +201,6 @@ [SER004]RESPite.Messages.RespScanState.TryRead(System.ReadOnlySpan value, out int bytesRead) -> bool [SER004]RESPite.RespException [SER004]RESPite.RespException.RespException(string! message) -> void -[SER004]static RESPite.Buffers.CycleBuffer.Create(System.Buffers.MemoryPool? pool = null, int pageSize = 8192, RESPite.Buffers.ICycleBufferCallback? callback = null) -> RESPite.Buffers.CycleBuffer [SER004]static RESPite.Messages.RespFrameScanner.Default.get -> RESPite.Messages.RespFrameScanner! [SER004]static RESPite.Messages.RespFrameScanner.Subscription.get -> RESPite.Messages.RespFrameScanner! [SER004]virtual RESPite.Messages.RespAttributeReader.Read(ref RESPite.Messages.RespReader reader, ref T value) -> void diff --git a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt index 9acf1fc40..f10c7cc88 100644 --- a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt @@ -4,3 +4,10 @@ [SER004]static RESPite.AsciiHash.EqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool [SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool [SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool +[SER004]abstract RESPite.Buffers.CycleBufferPool.Rent(in System.Buffers.ReadOnlySequence existing) -> System.Buffers.IMemoryOwner! +[SER004]RESPite.Buffers.CycleBufferPool +[SER004]RESPite.Buffers.CycleBufferPool.CycleBufferPool() -> void +[SER004]static RESPite.Buffers.CycleBufferPool.Default.get -> RESPite.Buffers.CycleBufferPool! +[SER004]virtual RESPite.Buffers.CycleBufferPool.Rent() -> System.Buffers.IMemoryOwner! +[SER004]static RESPite.Buffers.CycleBuffer.Create(RESPite.Buffers.CycleBufferPool? pool = null, RESPite.Buffers.ICycleBufferCallback? callback = null) -> RESPite.Buffers.CycleBuffer +[SER004]RESPite.Buffers.CycleBuffer.Pool.get -> RESPite.Buffers.CycleBufferPool! \ No newline at end of file From 6350db9d94ee3e298ada3cd0a3305bde88281d02 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 17 Jun 2026 14:32:41 +0100 Subject: [PATCH 2/6] - add ConfigurationOptions support - add [SkipLocalsInit] to RESPite --- src/RESPite/AssemblyInfo.cs | 1 + src/RESPite/Messages/RespReader.cs | 1 + src/RESPite/Shared/AsciiHash.Instance.cs | 1 + .../Shared}/SkipLocalsInit.cs | 0 src/StackExchange.Redis/AssemblyInfoHack.cs | 7 +++--- .../BufferedStreamWriter.Switchable.cs | 5 ++-- .../BufferedStreamWriter.cs | 10 ++++---- .../ConfigurationOptions.cs | 25 +++++++++++++++++++ .../PhysicalConnection.Read.cs | 6 +++-- .../PhysicalConnection.Write.cs | 5 ++-- .../PublicAPI/PublicAPI.Unshipped.txt | 4 +++ .../BufferedStreamWriterTests.cs | 10 ++++---- .../StackExchange.Redis.Tests/ConfigTests.cs | 2 ++ 13 files changed, 57 insertions(+), 20 deletions(-) create mode 100644 src/RESPite/AssemblyInfo.cs rename src/{StackExchange.Redis => RESPite/Shared}/SkipLocalsInit.cs (100%) diff --git a/src/RESPite/AssemblyInfo.cs b/src/RESPite/AssemblyInfo.cs new file mode 100644 index 000000000..f5476101b --- /dev/null +++ b/src/RESPite/AssemblyInfo.cs @@ -0,0 +1 @@ +[assembly: CLSCompliant(true)] diff --git a/src/RESPite/Messages/RespReader.cs b/src/RESPite/Messages/RespReader.cs index 86ff08f40..6b5dc2228 100644 --- a/src/RESPite/Messages/RespReader.cs +++ b/src/RESPite/Messages/RespReader.cs @@ -714,6 +714,7 @@ internal readonly T ParseBytes(Parser parser, TState } } + [CLSCompliant(false)] public readonly unsafe bool TryParseScalar( delegate* managed, out T, bool> parser, out T value) { diff --git a/src/RESPite/Shared/AsciiHash.Instance.cs b/src/RESPite/Shared/AsciiHash.Instance.cs index 53db4ff27..50ac4d245 100644 --- a/src/RESPite/Shared/AsciiHash.Instance.cs +++ b/src/RESPite/Shared/AsciiHash.Instance.cs @@ -35,6 +35,7 @@ public AsciiHash(string? value) : this(value is null ? [] : Encoding.ASCII.GetBy public override bool Equals(object? other) => other is AsciiHash hash && Equals(hash); /// + [CLSCompliant(false)] public bool Equals(in AsciiHash other) { return (_length == other.Length & _hashCS == other._hashCS) diff --git a/src/StackExchange.Redis/SkipLocalsInit.cs b/src/RESPite/Shared/SkipLocalsInit.cs similarity index 100% rename from src/StackExchange.Redis/SkipLocalsInit.cs rename to src/RESPite/Shared/SkipLocalsInit.cs diff --git a/src/StackExchange.Redis/AssemblyInfoHack.cs b/src/StackExchange.Redis/AssemblyInfoHack.cs index 50cdc2c1c..f2e7744d4 100644 --- a/src/StackExchange.Redis/AssemblyInfoHack.cs +++ b/src/StackExchange.Redis/AssemblyInfoHack.cs @@ -1,6 +1,5 @@ -// Yes, this is embarrassing. However, in .NET Core the including AssemblyInfo (ifdef'd or not) will screw with -// your version numbers. Therefore, we need to move the attribute out into another file...this file. -// When .csproj merges in, this should be able to return to Properties/AssemblyInfo.cs -using System; +using System; +using System.Runtime.CompilerServices; [assembly: CLSCompliant(true)] +[module: SkipLocalsInit] diff --git a/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs b/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs index 153c3962a..4fe9c5f31 100644 --- a/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs +++ b/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs @@ -4,6 +4,7 @@ using System.Threading; using System.Threading.Tasks; using System.Threading.Tasks.Sources; +using RESPite.Buffers; namespace StackExchange.Redis; @@ -13,8 +14,8 @@ internal sealed class SwitchableBufferedStreamWriter : CycleBufferStreamWriter, private ManualResetValueTaskSourceCore _readerTask; private bool _syncSignalled; - public SwitchableBufferedStreamWriter(Stream target, CancellationToken cancellationToken, bool initiallySync) - : base(target, cancellationToken, initiallySync ? StateFlags.None : StateFlags.AsyncMode) + public SwitchableBufferedStreamWriter(CycleBufferPool? pool, Stream target, CancellationToken cancellationToken, bool initiallySync) + : base(pool, target, cancellationToken, initiallySync ? StateFlags.None : StateFlags.AsyncMode) { _readerTask.RunContinuationsAsynchronously = true; // we never want the flusher to take over the copying if (initiallySync) diff --git a/src/StackExchange.Redis/BufferedStreamWriter.cs b/src/StackExchange.Redis/BufferedStreamWriter.cs index 85a993a71..a839a8ad5 100644 --- a/src/StackExchange.Redis/BufferedStreamWriter.cs +++ b/src/StackExchange.Redis/BufferedStreamWriter.cs @@ -57,7 +57,7 @@ public enum WriteMode public virtual bool IsSync => false; - public static BufferedStreamWriter Create(WriteMode mode, ConnectionType connectionType, Stream target, CancellationToken cancellationToken) + public static BufferedStreamWriter Create(WriteMode mode, ConnectionType connectionType, Stream target, ConfigurationOptions? options, CancellationToken cancellationToken) { if (connectionType is ConnectionType.Subscription | mode is WriteMode.Default) { @@ -67,8 +67,8 @@ public static BufferedStreamWriter Create(WriteMode mode, ConnectionType connect } return mode switch { - WriteMode.Sync => new SwitchableBufferedStreamWriter(target, cancellationToken, initiallySync: true), - WriteMode.Async => new SwitchableBufferedStreamWriter(target, cancellationToken, initiallySync: false), + WriteMode.Sync => new SwitchableBufferedStreamWriter(options?.RequestCycleBufferPool, target, cancellationToken, initiallySync: true), + WriteMode.Async => new SwitchableBufferedStreamWriter(options?.RequestCycleBufferPool, target, cancellationToken, initiallySync: false), WriteMode.Pipe => new PipeStreamWriter(target, cancellationToken), _ => throw new ArgumentOutOfRangeException(nameof(mode)), }; @@ -115,10 +115,10 @@ public virtual void DebugSetLog(Action log) { } internal abstract class CycleBufferStreamWriter : BufferedStreamWriter, ICycleBufferCallback { - protected CycleBufferStreamWriter(Stream target, CancellationToken cancellationToken, StateFlags flags = StateFlags.None) + protected CycleBufferStreamWriter(CycleBufferPool? pool, Stream target, CancellationToken cancellationToken, StateFlags flags = StateFlags.None) : base(target, cancellationToken) { - _buffer = CycleBuffer.Create(callback: this); + _buffer = CycleBuffer.Create(pool: pool, callback: this); _stateFlags = flags; } diff --git a/src/StackExchange.Redis/ConfigurationOptions.cs b/src/StackExchange.Redis/ConfigurationOptions.cs index e8114cdb6..91de712f4 100644 --- a/src/StackExchange.Redis/ConfigurationOptions.cs +++ b/src/StackExchange.Redis/ConfigurationOptions.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.ComponentModel; +using System.Diagnostics.CodeAnalysis; using System.Linq; using System.Net; using System.Net.Security; @@ -12,6 +13,8 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; +using RESPite; +using RESPite.Buffers; using StackExchange.Redis.Configuration; namespace StackExchange.Redis @@ -1413,5 +1416,27 @@ internal static bool TryParseRedisProtocol(string? value, out RedisProtocol prot protocol = default; return false; } + + /// + /// The buffer pool to use when buffering responses. + /// + public CycleBufferPool? ResponseCycleBufferPool + { + [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] + get; + [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] + set; + } + + /// + /// The buffer pool to use when buffering requests. + /// + public CycleBufferPool? RequestCycleBufferPool + { + [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] + get; + [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] + set; + } } } diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index af5421301..774db2c6b 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -65,6 +65,8 @@ static void StartReadingSync(PhysicalConnection conn, CancellationToken cancella private void StartReadAllAsync(CancellationToken cancellationToken) => Task.Run(() => ReadAllAsync(cancellationToken)).RedisFireAndForget(); + private CycleBufferPool? ReaderBufferPool => BridgeCouldBeNull?.Multiplexer?.RawConfig?.ResponseCycleBufferPool; + private async Task ReadAllAsync(CancellationToken cancellationToken) { var tail = _ioStream ?? Stream.Null; @@ -73,7 +75,7 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) // preserve existing state if transitioning _readStatus = ReadStatus.Init; _readState = default; - _readBuffer = CycleBuffer.Create(); + _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool); } try { @@ -130,7 +132,7 @@ private void ReadAllSync(CancellationToken cancellationToken) var tail = _ioStream ?? Stream.Null; _readStatus = ReadStatus.Init; _readState = default; - _readBuffer = CycleBuffer.Create(); + _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool); try { int read; diff --git a/src/StackExchange.Redis/PhysicalConnection.Write.cs b/src/StackExchange.Redis/PhysicalConnection.Write.cs index 84d3430e1..0d213927d 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Write.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Write.cs @@ -22,9 +22,10 @@ private void InitOutput(Stream? stream) { if (stream is null) return; _ioStream = stream; - _output = BufferedStreamWriter.Create(WriteMode, connectionType, stream, OutputCancel); + var config = BridgeCouldBeNull?.Multiplexer?.RawConfig; + _output = BufferedStreamWriter.Create(WriteMode, connectionType, stream, config, OutputCancel); #if DEBUG - if (BridgeCouldBeNull?.Multiplexer.RawConfig.OutputLog is { } log) + if (config?.OutputLog is { } log) { _output.DebugSetLog(log); } diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 56c23463e..4a8bf8b68 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1,4 +1,8 @@ #nullable enable +[SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool? +[SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.set -> void +[SER004]StackExchange.Redis.ConfigurationOptions.ResponseCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool? +[SER004]StackExchange.Redis.ConfigurationOptions.ResponseCycleBufferPool.set -> void [SER005]StackExchange.Redis.TestHarness [SER005]StackExchange.Redis.TestHarness.BufferValidator [SER005]StackExchange.Redis.TestHarness.ChannelPrefix.get -> StackExchange.Redis.RedisChannel diff --git a/tests/StackExchange.Redis.Tests/BufferedStreamWriterTests.cs b/tests/StackExchange.Redis.Tests/BufferedStreamWriterTests.cs index 56e1c86e9..932153db0 100644 --- a/tests/StackExchange.Redis.Tests/BufferedStreamWriterTests.cs +++ b/tests/StackExchange.Redis.Tests/BufferedStreamWriterTests.cs @@ -17,7 +17,7 @@ public class BufferedStreamWriterTests public async Task FlushStateDoesNotLeakIntoNextPageActivation(WriteMode mode) { var stream = new ObservedStream(); - var writer = BufferedStreamWriter.Create((BufferedStreamWriter.WriteMode)mode, ConnectionType.Interactive, stream, CancellationToken.None); + var writer = BufferedStreamWriter.Create((BufferedStreamWriter.WriteMode)mode, ConnectionType.Interactive, stream, null, CancellationToken.None); try { Write(writer, 1, 1); @@ -50,7 +50,7 @@ public async Task FlushStateDoesNotLeakIntoNextPageActivation(WriteMode mode) public async Task WriterDoesNotLoseFlushRequestedDuringDrainFlush(WriteMode mode) { var stream = new ObservedStream(); - var writer = BufferedStreamWriter.Create((BufferedStreamWriter.WriteMode)mode, ConnectionType.Interactive, stream, CancellationToken.None); + var writer = BufferedStreamWriter.Create((BufferedStreamWriter.WriteMode)mode, ConnectionType.Interactive, stream, null, CancellationToken.None); try { stream.BlockNextFlush(); @@ -81,7 +81,7 @@ public async Task WriterFaultsWriteCompleteAfterTargetWriteFailure(WriteMode mod { var failure = new IOException("simulated target write failure"); var stream = new ObservedStream { WriteException = failure }; - var writer = BufferedStreamWriter.Create((BufferedStreamWriter.WriteMode)mode, ConnectionType.Interactive, stream, CancellationToken.None); + var writer = BufferedStreamWriter.Create((BufferedStreamWriter.WriteMode)mode, ConnectionType.Interactive, stream, null, CancellationToken.None); Write(writer, 1, 1); writer.Flush(); @@ -101,7 +101,7 @@ public async Task WriterFaultsWriteCompleteAfterTargetWriteFailure(WriteMode mod public async Task SyncWriterTransitionsToAsyncWhileIdleAndPreservesBufferedData() { var stream = new ObservedStream(); - var writer = BufferedStreamWriter.Create(BufferedStreamWriter.WriteMode.Sync, ConnectionType.Interactive, stream, CancellationToken.None); + var writer = BufferedStreamWriter.Create(BufferedStreamWriter.WriteMode.Sync, ConnectionType.Interactive, stream, null, CancellationToken.None); try { Assert.True(writer.IsSync); @@ -135,7 +135,7 @@ public async Task SyncWriterTransitionsToAsyncWhileIdleAndPreservesBufferedData( public async Task SyncWriterTransitionsToAsyncAfterActiveSyncDrain() { var stream = new ObservedStream(); - var writer = BufferedStreamWriter.Create(BufferedStreamWriter.WriteMode.Sync, ConnectionType.Interactive, stream, CancellationToken.None); + var writer = BufferedStreamWriter.Create(BufferedStreamWriter.WriteMode.Sync, ConnectionType.Interactive, stream, null, CancellationToken.None); try { stream.BlockNextWrite(); diff --git a/tests/StackExchange.Redis.Tests/ConfigTests.cs b/tests/StackExchange.Redis.Tests/ConfigTests.cs index 561308d30..a117fc9e5 100644 --- a/tests/StackExchange.Redis.Tests/ConfigTests.cs +++ b/tests/StackExchange.Redis.Tests/ConfigTests.cs @@ -90,6 +90,8 @@ orderby name "password", "proxy", "reconnectRetryPolicy", + "RequestCycleBufferPool", + "ResponseCycleBufferPool", "responseTimeout", "ServiceName", "SocketManager", From 13cc1b7ce738b835b7a7047b3315d3fb6568d7bd Mon Sep 17 00:00:00 2001 From: Ivan Tikhonov Date: Mon, 22 Jun 2026 17:52:22 +0300 Subject: [PATCH 3/6] add ResponseArrayPool (#3104) * add ResponseArrayPool * Lease * IMemoryOwner? --------- Co-authored-by: ITikhonov --- .../ConfigurationOptions.cs | 12 ++++ src/StackExchange.Redis/Lease.cs | 64 +++++++++++++++---- .../PhysicalConnection.Read.cs | 24 +++---- .../PublicAPI/PublicAPI.Shipped.txt | 1 + .../PublicAPI/PublicAPI.Unshipped.txt | 2 + .../RespReaderExtensions.cs | 3 +- 6 files changed, 80 insertions(+), 26 deletions(-) diff --git a/src/StackExchange.Redis/ConfigurationOptions.cs b/src/StackExchange.Redis/ConfigurationOptions.cs index 91de712f4..8ad1662be 100644 --- a/src/StackExchange.Redis/ConfigurationOptions.cs +++ b/src/StackExchange.Redis/ConfigurationOptions.cs @@ -1,4 +1,5 @@ using System; +using System.Buffers; using System.Collections.Generic; using System.ComponentModel; using System.Diagnostics.CodeAnalysis; @@ -1438,5 +1439,16 @@ public CycleBufferPool? RequestCycleBufferPool [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] set; } + + /// + /// The memory pool to use when buffering responses. + /// + public MemoryPool? ResponseMemoryPool + { + [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] + get; + [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] + set; + } } } diff --git a/src/StackExchange.Redis/Lease.cs b/src/StackExchange.Redis/Lease.cs index a5a88e4eb..c77f6d6e9 100644 --- a/src/StackExchange.Redis/Lease.cs +++ b/src/StackExchange.Redis/Lease.cs @@ -1,6 +1,7 @@ using System; using System.Buffers; using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; using System.Threading; namespace StackExchange.Redis @@ -16,7 +17,7 @@ public sealed class Lease : IMemoryOwner /// public static Lease Empty { get; } = new Lease(System.Array.Empty(), 0); - private T[]? _arr; + private object? _buffer; /// /// Gets whether this lease is empty. @@ -41,9 +42,29 @@ public static Lease Create(int length, bool clear = true) return new Lease(arr, length); } + /// + /// Create a new lease. + /// + /// The size required. + /// Buffer. + public static Lease Create(int length, IMemoryOwner memoryOwner) + { + if (length == 0) return Empty; + if ((uint)length > memoryOwner.Memory.Length) + throw new ArgumentOutOfRangeException(nameof(length)); + + return new Lease(memoryOwner, length); + } + private Lease(T[] arr, int length) { - _arr = arr; + _buffer = arr; + Length = length; + } + + private Lease(IMemoryOwner memoryOwner, int length) + { + _buffer = memoryOwner; Length = length; } @@ -54,33 +75,50 @@ public void Dispose() { if (Length != 0) { - var arr = Interlocked.Exchange(ref _arr, null); - if (arr != null) ArrayPool.Shared.Return(arr); + var buffer = Interlocked.Exchange(ref _buffer, null); + if (buffer != null) + { + if (buffer is T[] arr) + ArrayPool.Shared.Return(arr); + else + ((IMemoryOwner)buffer).Dispose(); + } } } [MethodImpl(MethodImplOptions.NoInlining)] private static T[] ThrowDisposed() => throw new ObjectDisposedException(nameof(Lease)); - private T[] Array - { - [MethodImpl(MethodImplOptions.AggressiveInlining)] - get => _arr ?? ThrowDisposed(); - } - /// /// The data as a . /// - public Memory Memory => new Memory(Array, 0, Length); + public Memory Memory => _buffer is IMemoryOwner memoryOwner + ? memoryOwner.Memory.Slice(0, Length) + : new Memory((T[]?)_buffer ?? ThrowDisposed(), 0, Length); /// /// The data as a . /// - public Span Span => new Span(Array, 0, Length); + public Span Span => _buffer is IMemoryOwner memoryOwner + ? memoryOwner.Memory.Span.Slice(0, Length) + : new Span((T[]?)_buffer ?? ThrowDisposed(), 0, Length); /// /// The data as an . /// - public ArraySegment ArraySegment => new ArraySegment(Array, 0, Length); + public ArraySegment ArraySegment + { + get + { + if (_buffer is IMemoryOwner memoryOwner) + { + if (!MemoryMarshal.TryGetArray((ReadOnlyMemory)memoryOwner.Memory, out var segment)) + throw new NotSupportedException("Only array-backed buffers are supported"); + + return new ArraySegment(segment.Array!, segment.Offset, Length); + } + return new ArraySegment((T[]?)_buffer ?? ThrowDisposed(), 0, Length); + } + } } } diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 774db2c6b..7ee1897c4 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Buffers; using System.Buffers.Binary; using System.Diagnostics; @@ -205,7 +205,7 @@ private bool ShouldTransitionToAsync() private bool ForceReconnect => BridgeCouldBeNull?.NeedsReconnect == true; - private static byte[]? SharedNoLease; + private static IMemoryOwner? SharedNoLease; private CycleBuffer _readBuffer; private RespScanState _readState = default; @@ -401,15 +401,15 @@ private void OnResponseFrame(RespPrefix prefix, ReadOnlySequence payload) else { var len = checked((int)payload.Length); - byte[]? oversized = ArrayPool.Shared.Rent(len); + var memoryPool = BridgeCouldBeNull?.Multiplexer.RawConfig.ResponseMemoryPool ?? MemoryPool.Shared; + var memoryOwner = memoryPool.Rent(len); + Span oversized = memoryOwner.Memory.Span.Slice(0, len); + payload.CopyTo(oversized); - OnResponseFrame(prefix, new(oversized, 0, len), ref oversized); - // the lease could have been claimed by the activation code (to prevent another memcpy); otherwise, free - if (oversized is not null) - { - ArrayPool.Shared.Return(oversized); - } + OnResponseFrame(prefix, oversized, ref memoryOwner); + + memoryOwner?.Dispose(); } } @@ -420,7 +420,7 @@ private void UpdateBufferStats(int lastResult, long inBuffer) bytesLastResult = lastResult; } - private void OnResponseFrame(RespPrefix prefix, ReadOnlySpan frame, ref byte[]? lease) + private void OnResponseFrame(RespPrefix prefix, ReadOnlySpan frame, ref IMemoryOwner? memoryOwner) { DebugValidateSingleFrame(frame); _readStatus = ReadStatus.MatchResult; @@ -431,7 +431,7 @@ private void OnResponseFrame(RespPrefix prefix, ReadOnlySpan frame, ref by case RespPrefix.Array when (_protocol is RedisProtocol.Resp2 & connectionType is ConnectionType.Subscription) && !IsArrayPong(frame): // could be a RESP2 pub/sub payload // out-of-band; pub/sub etc - if (OnOutOfBand(frame, ref lease)) + if (OnOutOfBand(frame, ref memoryOwner)) { OnDetailLog($"out-of-band message, not dequeuing: {prefix}"); return; @@ -502,7 +502,7 @@ internal static ReadOnlySpan StackCopyLengthChecked(scoped in RespReader r return buffer.Slice(0, len); } - private bool OnOutOfBand(ReadOnlySpan payload, ref byte[]? lease) + private bool OnOutOfBand(ReadOnlySpan payload, ref IMemoryOwner? memoryOwner) { var muxer = BridgeCouldBeNull?.Multiplexer; if (muxer is null) return true; // consume it blindly diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt index 269cc11d2..47422be39 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt @@ -1734,6 +1734,7 @@ static StackExchange.Redis.HashEntry.operator !=(StackExchange.Redis.HashEntry x static StackExchange.Redis.HashEntry.operator ==(StackExchange.Redis.HashEntry x, StackExchange.Redis.HashEntry y) -> bool static StackExchange.Redis.KeyspaceIsolation.DatabaseExtensions.WithKeyPrefix(this StackExchange.Redis.IDatabase! database, StackExchange.Redis.RedisKey keyPrefix) -> StackExchange.Redis.IDatabase! static StackExchange.Redis.Lease.Create(int length, bool clear = true) -> StackExchange.Redis.Lease! +static StackExchange.Redis.Lease.Create(int length, System.Buffers.IMemoryOwner! memoryOwner) -> StackExchange.Redis.Lease! static StackExchange.Redis.Lease.Empty.get -> StackExchange.Redis.Lease! static StackExchange.Redis.ListPopResult.Null.get -> StackExchange.Redis.ListPopResult static StackExchange.Redis.LuaScript.GetCachedScriptCount() -> int diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 4a8bf8b68..648acbb60 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1,4 +1,6 @@ #nullable enable +[SER004]StackExchange.Redis.ConfigurationOptions.ResponseMemoryPool.get -> System.Buffers.MemoryPool? +[SER004]StackExchange.Redis.ConfigurationOptions.ResponseMemoryPool.set -> void [SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool? [SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.set -> void [SER004]StackExchange.Redis.ConfigurationOptions.ResponseCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool? diff --git a/src/StackExchange.Redis/RespReaderExtensions.cs b/src/StackExchange.Redis/RespReaderExtensions.cs index a5bb77bdd..4884690fe 100644 --- a/src/StackExchange.Redis/RespReaderExtensions.cs +++ b/src/StackExchange.Redis/RespReaderExtensions.cs @@ -1,4 +1,5 @@ -using System; +using System; +using System.Buffers; using System.Diagnostics; using System.Threading.Tasks; using RESPite.Messages; From 89ea7b374b26d280abdd8d2407668f79a99ad8c4 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 22 Jun 2026 15:50:15 +0100 Subject: [PATCH 4/6] refactor API so that MemoryPool can be used throughout, and shared between lease and cycle-buffers --- src/RESPite/Buffers/CycleBuffer.cs | 19 +-- src/RESPite/Buffers/CycleBufferPool.cs | 133 ++++++++---------- src/RESPite/PublicAPI/PublicAPI.Unshipped.txt | 13 +- .../BufferedStreamWriter.Switchable.cs | 3 +- .../BufferedStreamWriter.cs | 6 +- .../ConfigurationOptions.cs | 31 +--- src/StackExchange.Redis/Lease.cs | 25 ++-- .../PhysicalConnection.Read.cs | 4 +- .../PublicAPI/PublicAPI.Shipped.txt | 1 - .../PublicAPI/PublicAPI.Unshipped.txt | 11 +- .../RespReaderExtensions.cs | 5 +- .../ResultProcessor.Lease.cs | 4 +- 12 files changed, 113 insertions(+), 142 deletions(-) diff --git a/src/RESPite/Buffers/CycleBuffer.cs b/src/RESPite/Buffers/CycleBuffer.cs index 334171c12..3434eff4c 100644 --- a/src/RESPite/Buffers/CycleBuffer.cs +++ b/src/RESPite/Buffers/CycleBuffer.cs @@ -30,26 +30,20 @@ namespace RESPite.Buffers; [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] public partial struct CycleBuffer { - #if TRACK_MEMORY - private static MemoryPool DefaultPool => MemoryTrackedPool.Shared; - #else - private static MemoryPool DefaultPool => MemoryPool.Shared; - #endif - // note: if someone uses an uninitialized CycleBuffer (via default): that's a skills issue; git gud public static CycleBuffer Create( - CycleBufferPool? pool = null, + MemoryPool? pool = null, ICycleBufferCallback? callback = null) => new(pool, callback); - private CycleBuffer(CycleBufferPool? pool, ICycleBufferCallback? callback = null) + private CycleBuffer(MemoryPool? pool, ICycleBufferCallback? callback = null) { - _pool = pool ?? CycleBufferPool.Default; + _pool = pool ?? CycleBufferPool.Default; _callback = callback; leasedStart = -1; } - public CycleBufferPool Pool => _pool; - private readonly CycleBufferPool _pool; + public MemoryPool Pool => _pool; + private readonly MemoryPool _pool; private readonly ICycleBufferCallback? _callback; private Segment? startSegment, endSegment; @@ -402,7 +396,8 @@ private Segment GetNextSegment() } } - Segment newSegment = Segment.Create(endSegment is null ? _pool.Rent() : _pool.Rent(GetAllCommitted())); + // note that here we're using our extended MemoryPool API that allows sizing based on the existing buffers + Segment newSegment = Segment.Create(_pool.Rent(GetAllCommitted())); if (endSegment is null) { // tabula rasa diff --git a/src/RESPite/Buffers/CycleBufferPool.cs b/src/RESPite/Buffers/CycleBufferPool.cs index 76a87e4dc..23f53fbcd 100644 --- a/src/RESPite/Buffers/CycleBufferPool.cs +++ b/src/RESPite/Buffers/CycleBufferPool.cs @@ -1,96 +1,87 @@ using System.Buffers; using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; namespace RESPite.Buffers; -[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] -public abstract class CycleBufferPool +internal static class MemoryPoolExtensions { - /// - /// Create an initial buffer. - /// - public virtual IMemoryOwner Rent() => Rent(default); + internal static IMemoryOwner Rent(this MemoryPool pool, in ReadOnlySequence existing) + { + return pool is CycleBufferPool typed + ? typed.Rent(existing) + : pool.Rent(Math.Min(pool.MaxBufferSize, NextSize(existing))); + } + + private const int DefaultPageSizeBytes = 8 * 1024; + + internal static int NextSize(in ReadOnlySequence existing) + { + if (existing.IsEmpty) return DefaultPageSizeBytes / Unsafe.SizeOf(); + + // use a growth strategy looking at the size of the last segment + int lastChunk; + if (existing.IsSingleSegment) + { + lastChunk = existing.First.Length; + } + else if (existing.End.GetObject() is ReadOnlySequenceSegment segment) + { + lastChunk = segment.Memory.Length; // note we ignore GetInteger() intentionally + } + else + { + // do it the hard way; note we'll only observe the reserved size, rather + // than the actual buffer size, but that's the best we can do + lastChunk = 0; + foreach (var chunk in existing) + { + if (!chunk.IsEmpty) lastChunk = chunk.Length; + } + + if (lastChunk is 0) lastChunk = existing.First.Length; // paranoia + } + + // apply a fixed lower bound - don't start with trivial growth + lastChunk = Math.Max(lastChunk, 16); + + // apply doubling; "max" here is to account for overflow - i.e. stop growing if that happens (unlikely) + return Math.Max(lastChunk, lastChunk << 1); + } +} +[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] +public abstract class CycleBufferPool : MemoryPool +{ /// /// Create a buffer with knowledge of the existing leased data. /// - public abstract IMemoryOwner Rent(in ReadOnlySequence existing); + public abstract IMemoryOwner Rent(in ReadOnlySequence existing); // new MemoryPool(...) would be a non-growing buffer pool. - public static CycleBufferPool Default { get; } = new GrowingMemoryPool(minBytes: 8 * 1024); + public static CycleBufferPool Default { get; } = new DefaultPool(); - private class MemoryPool : CycleBufferPool + private class DefaultPool : CycleBufferPool { + private static readonly MemoryPool Tail = #if TRACK_MEMORY - private static MemoryPool DefaultPool => MemoryTrackedPool.Shared; + MemoryTrackedPool.Shared; #else - private static MemoryPool DefaultPool => MemoryPool.Shared; + MemoryPool.Shared; #endif - private readonly MemoryPool _pool; - private readonly int _minBytes, _maxBytes; - public MemoryPool(int minBytes, MemoryPool? pool = null, int maxBytes = int.MaxValue) - { - _pool = pool ?? DefaultPool; - // capture the max bytes, without exceeding the pool's max size - _maxBytes = Math.Min(maxBytes, _pool.MaxBufferSize); - // capture the min bytes, applying a rigid lower bound, and not overlapping the max bytes - _minBytes = Math.Min(Math.Max(minBytes, 16), _maxBytes); - } + // ReSharper disable once StaticMemberInGenericType - intentional to memoize, but can vary per T + private static readonly int MaxSize = Tail.MaxBufferSize; - /// - /// Rent a chunk using the specified size as a hint. - /// - protected IMemoryOwner Rent(int bytes) - { -#if NET - bytes = Math.Clamp(bytes, _minBytes, _maxBytes); -#else - bytes = Math.Min(Math.Max(bytes, _minBytes), _maxBytes); -#endif - return _pool.Rent(bytes); - } - - // by default, use fixed size without reference to the existing data; subclasses can tweak - - /// - public override IMemoryOwner Rent() => Rent(_minBytes); + protected override void Dispose(bool disposing) { } /// - public override IMemoryOwner Rent(in ReadOnlySequence existing) => Rent(_minBytes); - } + public override IMemoryOwner Rent(int minBufferSize = -1) => Tail.Rent(minBufferSize); - private sealed class GrowingMemoryPool(int minBytes, MemoryPool? pool = null, int maxBytes = int.MaxValue) - : MemoryPool(minBytes, pool, maxBytes) - { - public override IMemoryOwner Rent(in ReadOnlySequence existing) - { - if (existing.IsEmpty) return base.Rent(existing); - // use a growth strategy looking at the size of the last segment - int lastChunk; - if (existing.IsSingleSegment) - { - lastChunk = existing.First.Length; - } - else if (existing.End.GetObject() is ReadOnlySequenceSegment segment) - { - lastChunk = segment.Memory.Length; // note we ignore GetInteger() intentionally - } - else - { - // do it the hard way; note we'll only observe the reserved size, rather - // than the actual buffer size, but that's the best we can do - lastChunk = 0; - foreach (var chunk in existing) - { - if (!chunk.IsEmpty) lastChunk = chunk.Length; - } - - if (lastChunk is 0) lastChunk = existing.First.Length; // paranoia - } + public override int MaxBufferSize => Tail.MaxBufferSize; - // "max" here is to account for overflow - i.e. stop growing if that happens (unlikely) - return Rent(Math.Max(lastChunk, lastChunk << 1)); - } + /// + public override IMemoryOwner Rent(in ReadOnlySequence existing) + => Tail.Rent(Math.Min(MemoryPoolExtensions.NextSize(existing), MaxSize)); } } diff --git a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt index f10c7cc88..6e3af51d8 100644 --- a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt @@ -4,10 +4,9 @@ [SER004]static RESPite.AsciiHash.EqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool [SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool [SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool -[SER004]abstract RESPite.Buffers.CycleBufferPool.Rent(in System.Buffers.ReadOnlySequence existing) -> System.Buffers.IMemoryOwner! -[SER004]RESPite.Buffers.CycleBufferPool -[SER004]RESPite.Buffers.CycleBufferPool.CycleBufferPool() -> void -[SER004]static RESPite.Buffers.CycleBufferPool.Default.get -> RESPite.Buffers.CycleBufferPool! -[SER004]virtual RESPite.Buffers.CycleBufferPool.Rent() -> System.Buffers.IMemoryOwner! -[SER004]static RESPite.Buffers.CycleBuffer.Create(RESPite.Buffers.CycleBufferPool? pool = null, RESPite.Buffers.ICycleBufferCallback? callback = null) -> RESPite.Buffers.CycleBuffer -[SER004]RESPite.Buffers.CycleBuffer.Pool.get -> RESPite.Buffers.CycleBufferPool! \ No newline at end of file +[SER004]abstract RESPite.Buffers.CycleBufferPool.Rent(in System.Buffers.ReadOnlySequence existing) -> System.Buffers.IMemoryOwner! +[SER004]RESPite.Buffers.CycleBuffer.Pool.get -> System.Buffers.MemoryPool! +[SER004]RESPite.Buffers.CycleBufferPool +[SER004]RESPite.Buffers.CycleBufferPool.CycleBufferPool() -> void +[SER004]static RESPite.Buffers.CycleBuffer.Create(System.Buffers.MemoryPool? pool = null, RESPite.Buffers.ICycleBufferCallback? callback = null) -> RESPite.Buffers.CycleBuffer +[SER004]static RESPite.Buffers.CycleBufferPool.Default.get -> RESPite.Buffers.CycleBufferPool! diff --git a/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs b/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs index 4fe9c5f31..79f5104d1 100644 --- a/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs +++ b/src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs @@ -1,4 +1,5 @@ using System; +using System.Buffers; using System.Diagnostics; using System.IO; using System.Threading; @@ -14,7 +15,7 @@ internal sealed class SwitchableBufferedStreamWriter : CycleBufferStreamWriter, private ManualResetValueTaskSourceCore _readerTask; private bool _syncSignalled; - public SwitchableBufferedStreamWriter(CycleBufferPool? pool, Stream target, CancellationToken cancellationToken, bool initiallySync) + public SwitchableBufferedStreamWriter(MemoryPool? pool, Stream target, CancellationToken cancellationToken, bool initiallySync) : base(pool, target, cancellationToken, initiallySync ? StateFlags.None : StateFlags.AsyncMode) { _readerTask.RunContinuationsAsynchronously = true; // we never want the flusher to take over the copying diff --git a/src/StackExchange.Redis/BufferedStreamWriter.cs b/src/StackExchange.Redis/BufferedStreamWriter.cs index a839a8ad5..00198f09f 100644 --- a/src/StackExchange.Redis/BufferedStreamWriter.cs +++ b/src/StackExchange.Redis/BufferedStreamWriter.cs @@ -67,8 +67,8 @@ public static BufferedStreamWriter Create(WriteMode mode, ConnectionType connect } return mode switch { - WriteMode.Sync => new SwitchableBufferedStreamWriter(options?.RequestCycleBufferPool, target, cancellationToken, initiallySync: true), - WriteMode.Async => new SwitchableBufferedStreamWriter(options?.RequestCycleBufferPool, target, cancellationToken, initiallySync: false), + WriteMode.Sync => new SwitchableBufferedStreamWriter(options?.RequestBufferPool, target, cancellationToken, initiallySync: true), + WriteMode.Async => new SwitchableBufferedStreamWriter(options?.RequestBufferPool, target, cancellationToken, initiallySync: false), WriteMode.Pipe => new PipeStreamWriter(target, cancellationToken), _ => throw new ArgumentOutOfRangeException(nameof(mode)), }; @@ -115,7 +115,7 @@ public virtual void DebugSetLog(Action log) { } internal abstract class CycleBufferStreamWriter : BufferedStreamWriter, ICycleBufferCallback { - protected CycleBufferStreamWriter(CycleBufferPool? pool, Stream target, CancellationToken cancellationToken, StateFlags flags = StateFlags.None) + protected CycleBufferStreamWriter(MemoryPool? pool, Stream target, CancellationToken cancellationToken, StateFlags flags = StateFlags.None) : base(target, cancellationToken) { _buffer = CycleBuffer.Create(pool: pool, callback: this); diff --git a/src/StackExchange.Redis/ConfigurationOptions.cs b/src/StackExchange.Redis/ConfigurationOptions.cs index 8ad1662be..879fdf298 100644 --- a/src/StackExchange.Redis/ConfigurationOptions.cs +++ b/src/StackExchange.Redis/ConfigurationOptions.cs @@ -974,6 +974,8 @@ public static ConfigurationOptions Parse(string configuration, bool ignoreUnknow #if DEBUG OutputLog = OutputLog, #endif + RequestBufferPool = RequestBufferPool, + ResponseBufferPool = ResponseBufferPool, }; /// @@ -1419,36 +1421,13 @@ internal static bool TryParseRedisProtocol(string? value, out RedisProtocol prot } /// - /// The buffer pool to use when buffering responses. + /// The buffer pool to use when buffering responses, and for allocating results. /// - public CycleBufferPool? ResponseCycleBufferPool - { - [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] - get; - [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] - set; - } + public MemoryPool? ResponseBufferPool { get; set; } /// /// The buffer pool to use when buffering requests. /// - public CycleBufferPool? RequestCycleBufferPool - { - [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] - get; - [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] - set; - } - - /// - /// The memory pool to use when buffering responses. - /// - public MemoryPool? ResponseMemoryPool - { - [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] - get; - [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)] - set; - } + public MemoryPool? RequestBufferPool { get; set; } } } diff --git a/src/StackExchange.Redis/Lease.cs b/src/StackExchange.Redis/Lease.cs index c77f6d6e9..05f742c84 100644 --- a/src/StackExchange.Redis/Lease.cs +++ b/src/StackExchange.Redis/Lease.cs @@ -34,26 +34,33 @@ public sealed class Lease : IMemoryOwner /// /// The size required. /// Whether to erase the memory. +#pragma warning disable RS0026 // multiple overloads with optional args; not ambiguous in this case. public static Lease Create(int length, bool clear = true) +#pragma warning restore RS0026 { - if (length == 0) return Empty; + if (length is 0) return Empty; var arr = ArrayPool.Shared.Rent(length); - if (clear) System.Array.Clear(arr, 0, length); - return new Lease(arr, length); + var lease = new Lease(arr, length); + if (clear) lease.Span.Clear(); + return lease; } /// /// Create a new lease. /// /// The size required. - /// Buffer. - public static Lease Create(int length, IMemoryOwner memoryOwner) + /// The buffer source to use; if null, is used. + /// Whether to erase the memory. +#pragma warning disable RS0026 // multiple overloads with optional args; not ambiguous in this case. + public static Lease Create(int length, MemoryPool? pool, bool clear = true) +#pragma warning restore RS0026 { - if (length == 0) return Empty; - if ((uint)length > memoryOwner.Memory.Length) - throw new ArgumentOutOfRangeException(nameof(length)); + if (pool is null) return Create(length, clear); + if (length is 0) return Empty; - return new Lease(memoryOwner, length); + var lease = new Lease(pool.Rent(length), length); + if (clear) lease.Span.Clear(); + return lease; } private Lease(T[] arr, int length) diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 7ee1897c4..ec3d0c22f 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -65,7 +65,7 @@ static void StartReadingSync(PhysicalConnection conn, CancellationToken cancella private void StartReadAllAsync(CancellationToken cancellationToken) => Task.Run(() => ReadAllAsync(cancellationToken)).RedisFireAndForget(); - private CycleBufferPool? ReaderBufferPool => BridgeCouldBeNull?.Multiplexer?.RawConfig?.ResponseCycleBufferPool; + private MemoryPool? ReaderBufferPool => BridgeCouldBeNull?.Multiplexer?.RawConfig?.ResponseBufferPool; private async Task ReadAllAsync(CancellationToken cancellationToken) { @@ -401,7 +401,7 @@ private void OnResponseFrame(RespPrefix prefix, ReadOnlySequence payload) else { var len = checked((int)payload.Length); - var memoryPool = BridgeCouldBeNull?.Multiplexer.RawConfig.ResponseMemoryPool ?? MemoryPool.Shared; + var memoryPool = BridgeCouldBeNull?.Multiplexer.RawConfig.RequestBufferPool ?? MemoryPool.Shared; var memoryOwner = memoryPool.Rent(len); Span oversized = memoryOwner.Memory.Span.Slice(0, len); diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt index 47422be39..269cc11d2 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt @@ -1734,7 +1734,6 @@ static StackExchange.Redis.HashEntry.operator !=(StackExchange.Redis.HashEntry x static StackExchange.Redis.HashEntry.operator ==(StackExchange.Redis.HashEntry x, StackExchange.Redis.HashEntry y) -> bool static StackExchange.Redis.KeyspaceIsolation.DatabaseExtensions.WithKeyPrefix(this StackExchange.Redis.IDatabase! database, StackExchange.Redis.RedisKey keyPrefix) -> StackExchange.Redis.IDatabase! static StackExchange.Redis.Lease.Create(int length, bool clear = true) -> StackExchange.Redis.Lease! -static StackExchange.Redis.Lease.Create(int length, System.Buffers.IMemoryOwner! memoryOwner) -> StackExchange.Redis.Lease! static StackExchange.Redis.Lease.Empty.get -> StackExchange.Redis.Lease! static StackExchange.Redis.ListPopResult.Null.get -> StackExchange.Redis.ListPopResult static StackExchange.Redis.LuaScript.GetCachedScriptCount() -> int diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 648acbb60..0f75682cb 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1,10 +1,9 @@ #nullable enable -[SER004]StackExchange.Redis.ConfigurationOptions.ResponseMemoryPool.get -> System.Buffers.MemoryPool? -[SER004]StackExchange.Redis.ConfigurationOptions.ResponseMemoryPool.set -> void -[SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool? -[SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.set -> void -[SER004]StackExchange.Redis.ConfigurationOptions.ResponseCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool? -[SER004]StackExchange.Redis.ConfigurationOptions.ResponseCycleBufferPool.set -> void +StackExchange.Redis.ConfigurationOptions.RequestBufferPool.get -> System.Buffers.MemoryPool? +StackExchange.Redis.ConfigurationOptions.RequestBufferPool.set -> void +StackExchange.Redis.ConfigurationOptions.ResponseBufferPool.get -> System.Buffers.MemoryPool? +StackExchange.Redis.ConfigurationOptions.ResponseBufferPool.set -> void +static StackExchange.Redis.Lease.Create(int length, System.Buffers.MemoryPool? pool, bool clear = true) -> StackExchange.Redis.Lease! [SER005]StackExchange.Redis.TestHarness [SER005]StackExchange.Redis.TestHarness.BufferValidator [SER005]StackExchange.Redis.TestHarness.ChannelPrefix.get -> StackExchange.Redis.RedisChannel diff --git a/src/StackExchange.Redis/RespReaderExtensions.cs b/src/StackExchange.Redis/RespReaderExtensions.cs index 4884690fe..6f80c57a4 100644 --- a/src/StackExchange.Redis/RespReaderExtensions.cs +++ b/src/StackExchange.Redis/RespReaderExtensions.cs @@ -136,7 +136,7 @@ public void MovePastBof() public RedisValue[]? ReadPastRedisValues() => reader.ReadPastArray(static (ref r) => r.ReadRedisValue(), scalar: true); - public Lease? AsLease() + public Lease? AsLease(PhysicalConnection? connection) { if (!reader.IsScalar) throw new InvalidCastException("Cannot convert to Lease: " + reader.Prefix); if (reader.IsNull) return null; @@ -144,7 +144,8 @@ public void MovePastBof() var length = reader.ScalarLength(); if (length == 0) return Lease.Empty; - var lease = Lease.Create(length, clear: false); + var pool = connection?.BridgeCouldBeNull?.Multiplexer?.RawConfig?.ResponseBufferPool; + var lease = Lease.Create(length, pool, clear: false); if (reader.TryGetSpan(out var span)) { span.CopyTo(lease.Span); diff --git a/src/StackExchange.Redis/ResultProcessor.Lease.cs b/src/StackExchange.Redis/ResultProcessor.Lease.cs index 91dfcd2ad..4544321fe 100644 --- a/src/StackExchange.Redis/ResultProcessor.Lease.cs +++ b/src/StackExchange.Redis/ResultProcessor.Lease.cs @@ -152,7 +152,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes { if (reader.IsScalar) { - SetResult(message, reader.AsLease()!); + SetResult(message, reader.AsLease(connection)!); return true; } return false; @@ -167,7 +167,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes && reader.TryMoveNext() && reader.IsScalar) { // treat an array of 1 like a single reply - SetResult(message, reader.AsLease()!); + SetResult(message, reader.AsLease(connection)!); return true; } return false; From c1788755d8120d14bc43ef2336611fd35b1c75dd Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 22 Jun 2026 16:03:51 +0100 Subject: [PATCH 5/6] we don't actually need a default CycleBufferPool impl; --- src/RESPite/Buffers/CycleBuffer.cs | 2 +- src/RESPite/Buffers/CycleBufferPool.cs | 30 ++----------------- src/RESPite/PublicAPI/PublicAPI.Unshipped.txt | 3 +- .../ConfigurationOptions.cs | 2 +- .../StackExchange.Redis.Tests/ConfigTests.cs | 4 +-- 5 files changed, 7 insertions(+), 34 deletions(-) diff --git a/src/RESPite/Buffers/CycleBuffer.cs b/src/RESPite/Buffers/CycleBuffer.cs index 3434eff4c..5a4f53c29 100644 --- a/src/RESPite/Buffers/CycleBuffer.cs +++ b/src/RESPite/Buffers/CycleBuffer.cs @@ -37,7 +37,7 @@ public static CycleBuffer Create( private CycleBuffer(MemoryPool? pool, ICycleBufferCallback? callback = null) { - _pool = pool ?? CycleBufferPool.Default; + _pool = pool ?? MemoryPool.Shared; _callback = callback; leasedStart = -1; } diff --git a/src/RESPite/Buffers/CycleBufferPool.cs b/src/RESPite/Buffers/CycleBufferPool.cs index 23f53fbcd..ea207f93c 100644 --- a/src/RESPite/Buffers/CycleBufferPool.cs +++ b/src/RESPite/Buffers/CycleBufferPool.cs @@ -56,32 +56,6 @@ public abstract class CycleBufferPool : MemoryPool /// /// Create a buffer with knowledge of the existing leased data. /// - public abstract IMemoryOwner Rent(in ReadOnlySequence existing); - - // new MemoryPool(...) would be a non-growing buffer pool. - public static CycleBufferPool Default { get; } = new DefaultPool(); - - private class DefaultPool : CycleBufferPool - { - private static readonly MemoryPool Tail = -#if TRACK_MEMORY - MemoryTrackedPool.Shared; -#else - MemoryPool.Shared; -#endif - - // ReSharper disable once StaticMemberInGenericType - intentional to memoize, but can vary per T - private static readonly int MaxSize = Tail.MaxBufferSize; - - protected override void Dispose(bool disposing) { } - - /// - public override IMemoryOwner Rent(int minBufferSize = -1) => Tail.Rent(minBufferSize); - - public override int MaxBufferSize => Tail.MaxBufferSize; - - /// - public override IMemoryOwner Rent(in ReadOnlySequence existing) - => Tail.Rent(Math.Min(MemoryPoolExtensions.NextSize(existing), MaxSize)); - } + public virtual IMemoryOwner Rent(in ReadOnlySequence existing) + => Rent(Math.Min(MemoryPoolExtensions.NextSize(existing), MaxBufferSize)); } diff --git a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt index 6e3af51d8..f91f0fee0 100644 --- a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt @@ -4,9 +4,8 @@ [SER004]static RESPite.AsciiHash.EqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool [SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool [SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan first, System.ReadOnlySpan second) -> bool -[SER004]abstract RESPite.Buffers.CycleBufferPool.Rent(in System.Buffers.ReadOnlySequence existing) -> System.Buffers.IMemoryOwner! +[SER004]virtual RESPite.Buffers.CycleBufferPool.Rent(in System.Buffers.ReadOnlySequence existing) -> System.Buffers.IMemoryOwner! [SER004]RESPite.Buffers.CycleBuffer.Pool.get -> System.Buffers.MemoryPool! [SER004]RESPite.Buffers.CycleBufferPool [SER004]RESPite.Buffers.CycleBufferPool.CycleBufferPool() -> void [SER004]static RESPite.Buffers.CycleBuffer.Create(System.Buffers.MemoryPool? pool = null, RESPite.Buffers.ICycleBufferCallback? callback = null) -> RESPite.Buffers.CycleBuffer -[SER004]static RESPite.Buffers.CycleBufferPool.Default.get -> RESPite.Buffers.CycleBufferPool! diff --git a/src/StackExchange.Redis/ConfigurationOptions.cs b/src/StackExchange.Redis/ConfigurationOptions.cs index 879fdf298..eef0bde81 100644 --- a/src/StackExchange.Redis/ConfigurationOptions.cs +++ b/src/StackExchange.Redis/ConfigurationOptions.cs @@ -1421,7 +1421,7 @@ internal static bool TryParseRedisProtocol(string? value, out RedisProtocol prot } /// - /// The buffer pool to use when buffering responses, and for allocating results. + /// The buffer pool to use when buffering responses, and for allocating results. /// public MemoryPool? ResponseBufferPool { get; set; } diff --git a/tests/StackExchange.Redis.Tests/ConfigTests.cs b/tests/StackExchange.Redis.Tests/ConfigTests.cs index a117fc9e5..89ffd851e 100644 --- a/tests/StackExchange.Redis.Tests/ConfigTests.cs +++ b/tests/StackExchange.Redis.Tests/ConfigTests.cs @@ -90,8 +90,8 @@ orderby name "password", "proxy", "reconnectRetryPolicy", - "RequestCycleBufferPool", - "ResponseCycleBufferPool", + "RequestBufferPool", + "ResponseBufferPool", "responseTimeout", "ServiceName", "SocketManager", From 7ba4a5d524f284112782b522cb34146c7824f1f9 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 23 Jun 2026 10:03:56 +0100 Subject: [PATCH 6/6] fix typo --- src/StackExchange.Redis/PhysicalConnection.Read.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index ec3d0c22f..1f75e3879 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -401,7 +401,7 @@ private void OnResponseFrame(RespPrefix prefix, ReadOnlySequence payload) else { var len = checked((int)payload.Length); - var memoryPool = BridgeCouldBeNull?.Multiplexer.RawConfig.RequestBufferPool ?? MemoryPool.Shared; + var memoryPool = BridgeCouldBeNull?.Multiplexer.RawConfig.ResponseBufferPool ?? MemoryPool.Shared; var memoryOwner = memoryPool.Rent(len); Span oversized = memoryOwner.Memory.Span.Slice(0, len);