Skip to content

Commit 6270a65

Browse files
mgravellmgravell
committed
refactor API so that MemoryPool<T> can be used throughout, and shared between lease and cycle-buffers
1 parent 329abde commit 6270a65

12 files changed

Lines changed: 113 additions & 142 deletions

‎src/RESPite/Buffers/CycleBuffer.cs‎

Lines changed: 7 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -30,26 +30,20 @@ namespace RESPite.Buffers;
3030
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
3131
public partial struct CycleBuffer
3232
{
33-
#if TRACK_MEMORY
34-
private static MemoryPool<byte> DefaultPool => MemoryTrackedPool<byte>.Shared;
35-
#else
36-
private static MemoryPool<byte> DefaultPool => MemoryPool<byte>.Shared;
37-
#endif
38-
3933
// note: if someone uses an uninitialized CycleBuffer (via default): that's a skills issue; git gud
4034
public static CycleBuffer Create(
41-
CycleBufferPool? pool = null,
35+
MemoryPool<byte>? pool = null,
4236
ICycleBufferCallback? callback = null) => new(pool, callback);
4337

44-
private CycleBuffer(CycleBufferPool? pool, ICycleBufferCallback? callback = null)
38+
private CycleBuffer(MemoryPool<byte>? pool, ICycleBufferCallback? callback = null)
4539
{
46-
_pool = pool ?? CycleBufferPool.Default;
40+
_pool = pool ?? CycleBufferPool<byte>.Default;
4741
_callback = callback;
4842
leasedStart = -1;
4943
}
5044

51-
public CycleBufferPool Pool => _pool;
52-
private readonly CycleBufferPool _pool;
45+
public MemoryPool<byte> Pool => _pool;
46+
private readonly MemoryPool<byte> _pool;
5347
private readonly ICycleBufferCallback? _callback;
5448

5549
private Segment? startSegment, endSegment;
@@ -402,7 +396,8 @@ private Segment GetNextSegment()
402396
}
403397
}
404398

405-
Segment newSegment = Segment.Create(endSegment is null ? _pool.Rent() : _pool.Rent(GetAllCommitted()));
399+
// note that here we're using our extended MemoryPool<T> API that allows sizing based on the existing buffers
400+
Segment newSegment = Segment.Create(_pool.Rent(GetAllCommitted()));
406401
if (endSegment is null)
407402
{
408403
// tabula rasa

‎src/RESPite/Buffers/CycleBufferPool.cs‎

Lines changed: 62 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -1,96 +1,87 @@
11
using System.Buffers;
22
using System.Diagnostics.CodeAnalysis;
3+
using System.Runtime.CompilerServices;
34

45
namespace RESPite.Buffers;
56

6-
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
7-
public abstract class CycleBufferPool
7+
internal static class MemoryPoolExtensions
88
{
9-
/// <summary>
10-
/// Create an initial buffer.
11-
/// </summary>
12-
public virtual IMemoryOwner<byte> Rent() => Rent(default);
9+
internal static IMemoryOwner<T> Rent<T>(this MemoryPool<T> pool, in ReadOnlySequence<T> existing)
10+
{
11+
return pool is CycleBufferPool<T> typed
12+
? typed.Rent(existing)
13+
: pool.Rent(Math.Min(pool.MaxBufferSize, NextSize(existing)));
14+
}
15+
16+
private const int DefaultPageSizeBytes = 8 * 1024;
17+
18+
internal static int NextSize<T>(in ReadOnlySequence<T> existing)
19+
{
20+
if (existing.IsEmpty) return DefaultPageSizeBytes / Unsafe.SizeOf<T>();
21+
22+
// use a growth strategy looking at the size of the last segment
23+
int lastChunk;
24+
if (existing.IsSingleSegment)
25+
{
26+
lastChunk = existing.First.Length;
27+
}
28+
else if (existing.End.GetObject() is ReadOnlySequenceSegment<byte> segment)
29+
{
30+
lastChunk = segment.Memory.Length; // note we ignore GetInteger() intentionally
31+
}
32+
else
33+
{
34+
// do it the hard way; note we'll only observe the reserved size, rather
35+
// than the actual buffer size, but that's the best we can do
36+
lastChunk = 0;
37+
foreach (var chunk in existing)
38+
{
39+
if (!chunk.IsEmpty) lastChunk = chunk.Length;
40+
}
41+
42+
if (lastChunk is 0) lastChunk = existing.First.Length; // paranoia
43+
}
44+
45+
// apply a fixed lower bound - don't start with trivial growth
46+
lastChunk = Math.Max(lastChunk, 16);
47+
48+
// apply doubling; "max" here is to account for overflow - i.e. stop growing if that happens (unlikely)
49+
return Math.Max(lastChunk, lastChunk << 1);
50+
}
51+
}
1352

53+
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
54+
public abstract class CycleBufferPool<T> : MemoryPool<T>
55+
{
1456
/// <summary>
1557
/// Create a buffer with knowledge of the existing leased data.
1658
/// </summary>
17-
public abstract IMemoryOwner<byte> Rent(in ReadOnlySequence<byte> existing);
59+
public abstract IMemoryOwner<T> Rent(in ReadOnlySequence<T> existing);
1860

1961
// new MemoryPool(...) would be a non-growing buffer pool.
20-
public static CycleBufferPool Default { get; } = new GrowingMemoryPool(minBytes: 8 * 1024);
62+
public static CycleBufferPool<T> Default { get; } = new DefaultPool();
2163

22-
private class MemoryPool : CycleBufferPool
64+
private class DefaultPool : CycleBufferPool<T>
2365
{
66+
private static readonly MemoryPool<T> Tail =
2467
#if TRACK_MEMORY
25-
private static MemoryPool<byte> DefaultPool => MemoryTrackedPool<byte>.Shared;
68+
MemoryTrackedPool<T>.Shared;
2669
#else
27-
private static MemoryPool<byte> DefaultPool => MemoryPool<byte>.Shared;
70+
MemoryPool<T>.Shared;
2871
#endif
29-
private readonly MemoryPool<byte> _pool;
30-
private readonly int _minBytes, _maxBytes;
3172

32-
public MemoryPool(int minBytes, MemoryPool<byte>? pool = null, int maxBytes = int.MaxValue)
33-
{
34-
_pool = pool ?? DefaultPool;
35-
// capture the max bytes, without exceeding the pool's max size
36-
_maxBytes = Math.Min(maxBytes, _pool.MaxBufferSize);
37-
// capture the min bytes, applying a rigid lower bound, and not overlapping the max bytes
38-
_minBytes = Math.Min(Math.Max(minBytes, 16), _maxBytes);
39-
}
73+
// ReSharper disable once StaticMemberInGenericType - intentional to memoize, but can vary per T
74+
private static readonly int MaxSize = Tail.MaxBufferSize;
4075

41-
/// <summary>
42-
/// Rent a chunk using the specified size as a hint.
43-
/// </summary>
44-
protected IMemoryOwner<byte> Rent(int bytes)
45-
{
46-
#if NET
47-
bytes = Math.Clamp(bytes, _minBytes, _maxBytes);
48-
#else
49-
bytes = Math.Min(Math.Max(bytes, _minBytes), _maxBytes);
50-
#endif
51-
return _pool.Rent(bytes);
52-
}
53-
54-
// by default, use fixed size without reference to the existing data; subclasses can tweak
55-
56-
/// <inheritdoc/>
57-
public override IMemoryOwner<byte> Rent() => Rent(_minBytes);
76+
protected override void Dispose(bool disposing) { }
5877

5978
/// <inheritdoc/>
60-
public override IMemoryOwner<byte> Rent(in ReadOnlySequence<byte> existing) => Rent(_minBytes);
61-
}
79+
public override IMemoryOwner<T> Rent(int minBufferSize = -1) => Tail.Rent(minBufferSize);
6280

63-
private sealed class GrowingMemoryPool(int minBytes, MemoryPool<byte>? pool = null, int maxBytes = int.MaxValue)
64-
: MemoryPool(minBytes, pool, maxBytes)
65-
{
66-
public override IMemoryOwner<byte> Rent(in ReadOnlySequence<byte> existing)
67-
{
68-
if (existing.IsEmpty) return base.Rent(existing);
69-
// use a growth strategy looking at the size of the last segment
70-
int lastChunk;
71-
if (existing.IsSingleSegment)
72-
{
73-
lastChunk = existing.First.Length;
74-
}
75-
else if (existing.End.GetObject() is ReadOnlySequenceSegment<byte> segment)
76-
{
77-
lastChunk = segment.Memory.Length; // note we ignore GetInteger() intentionally
78-
}
79-
else
80-
{
81-
// do it the hard way; note we'll only observe the reserved size, rather
82-
// than the actual buffer size, but that's the best we can do
83-
lastChunk = 0;
84-
foreach (var chunk in existing)
85-
{
86-
if (!chunk.IsEmpty) lastChunk = chunk.Length;
87-
}
88-
89-
if (lastChunk is 0) lastChunk = existing.First.Length; // paranoia
90-
}
81+
public override int MaxBufferSize => Tail.MaxBufferSize;
9182

92-
// "max" here is to account for overflow - i.e. stop growing if that happens (unlikely)
93-
return Rent(Math.Max(lastChunk, lastChunk << 1));
94-
}
83+
/// <inheritdoc/>
84+
public override IMemoryOwner<T> Rent(in ReadOnlySequence<T> existing)
85+
=> Tail.Rent(Math.Min(MemoryPoolExtensions.NextSize(existing), MaxSize));
9586
}
9687
}

‎src/RESPite/PublicAPI/PublicAPI.Unshipped.txt‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,9 @@
44
[SER004]static RESPite.AsciiHash.EqualsCI(System.ReadOnlySpan<char> first, System.ReadOnlySpan<byte> second) -> bool
55
[SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan<byte> first, System.ReadOnlySpan<char> second) -> bool
66
[SER004]static RESPite.AsciiHash.SequenceEqualsCI(System.ReadOnlySpan<char> first, System.ReadOnlySpan<byte> second) -> bool
7-
[SER004]abstract RESPite.Buffers.CycleBufferPool.Rent(in System.Buffers.ReadOnlySequence<byte> existing) -> System.Buffers.IMemoryOwner<byte>!
8-
[SER004]RESPite.Buffers.CycleBufferPool
9-
[SER004]RESPite.Buffers.CycleBufferPool.CycleBufferPool() -> void
10-
[SER004]static RESPite.Buffers.CycleBufferPool.Default.get -> RESPite.Buffers.CycleBufferPool!
11-
[SER004]virtual RESPite.Buffers.CycleBufferPool.Rent() -> System.Buffers.IMemoryOwner<byte>!
12-
[SER004]static RESPite.Buffers.CycleBuffer.Create(RESPite.Buffers.CycleBufferPool? pool = null, RESPite.Buffers.ICycleBufferCallback? callback = null) -> RESPite.Buffers.CycleBuffer
13-
[SER004]RESPite.Buffers.CycleBuffer.Pool.get -> RESPite.Buffers.CycleBufferPool!
7+
[SER004]abstract RESPite.Buffers.CycleBufferPool<T>.Rent(in System.Buffers.ReadOnlySequence<T> existing) -> System.Buffers.IMemoryOwner<T>!
8+
[SER004]RESPite.Buffers.CycleBuffer.Pool.get -> System.Buffers.MemoryPool<byte>!
9+
[SER004]RESPite.Buffers.CycleBufferPool<T>
10+
[SER004]RESPite.Buffers.CycleBufferPool<T>.CycleBufferPool() -> void
11+
[SER004]static RESPite.Buffers.CycleBuffer.Create(System.Buffers.MemoryPool<byte>? pool = null, RESPite.Buffers.ICycleBufferCallback? callback = null) -> RESPite.Buffers.CycleBuffer
12+
[SER004]static RESPite.Buffers.CycleBufferPool<T>.Default.get -> RESPite.Buffers.CycleBufferPool<T>!

‎src/StackExchange.Redis/BufferedStreamWriter.Switchable.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using System;
2+
using System.Buffers;
23
using System.Diagnostics;
34
using System.IO;
45
using System.Threading;
@@ -14,7 +15,7 @@ internal sealed class SwitchableBufferedStreamWriter : CycleBufferStreamWriter,
1415
private ManualResetValueTaskSourceCore<bool> _readerTask;
1516
private bool _syncSignalled;
1617

17-
public SwitchableBufferedStreamWriter(CycleBufferPool? pool, Stream target, CancellationToken cancellationToken, bool initiallySync)
18+
public SwitchableBufferedStreamWriter(MemoryPool<byte>? pool, Stream target, CancellationToken cancellationToken, bool initiallySync)
1819
: base(pool, target, cancellationToken, initiallySync ? StateFlags.None : StateFlags.AsyncMode)
1920
{
2021
_readerTask.RunContinuationsAsynchronously = true; // we never want the flusher to take over the copying

‎src/StackExchange.Redis/BufferedStreamWriter.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -67,8 +67,8 @@ public static BufferedStreamWriter Create(WriteMode mode, ConnectionType connect
6767
}
6868
return mode switch
6969
{
70-
WriteMode.Sync => new SwitchableBufferedStreamWriter(options?.RequestCycleBufferPool, target, cancellationToken, initiallySync: true),
71-
WriteMode.Async => new SwitchableBufferedStreamWriter(options?.RequestCycleBufferPool, target, cancellationToken, initiallySync: false),
70+
WriteMode.Sync => new SwitchableBufferedStreamWriter(options?.RequestBufferPool, target, cancellationToken, initiallySync: true),
71+
WriteMode.Async => new SwitchableBufferedStreamWriter(options?.RequestBufferPool, target, cancellationToken, initiallySync: false),
7272
WriteMode.Pipe => new PipeStreamWriter(target, cancellationToken),
7373
_ => throw new ArgumentOutOfRangeException(nameof(mode)),
7474
};
@@ -115,7 +115,7 @@ public virtual void DebugSetLog(Action<string> log) { }
115115

116116
internal abstract class CycleBufferStreamWriter : BufferedStreamWriter, ICycleBufferCallback
117117
{
118-
protected CycleBufferStreamWriter(CycleBufferPool? pool, Stream target, CancellationToken cancellationToken, StateFlags flags = StateFlags.None)
118+
protected CycleBufferStreamWriter(MemoryPool<byte>? pool, Stream target, CancellationToken cancellationToken, StateFlags flags = StateFlags.None)
119119
: base(target, cancellationToken)
120120
{
121121
_buffer = CycleBuffer.Create(pool: pool, callback: this);

‎src/StackExchange.Redis/ConfigurationOptions.cs‎

Lines changed: 5 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -974,6 +974,8 @@ public static ConfigurationOptions Parse(string configuration, bool ignoreUnknow
974974
#if DEBUG
975975
OutputLog = OutputLog,
976976
#endif
977+
RequestBufferPool = RequestBufferPool,
978+
ResponseBufferPool = ResponseBufferPool,
977979
};
978980

979981
/// <summary>
@@ -1419,36 +1421,13 @@ internal static bool TryParseRedisProtocol(string? value, out RedisProtocol prot
14191421
}
14201422

14211423
/// <summary>
1422-
/// The buffer pool to use when buffering responses.
1424+
/// The buffer pool to use when buffering responses, and for allocating <see cref="Lease{byte}"/> results.
14231425
/// </summary>
1424-
public CycleBufferPool? ResponseCycleBufferPool
1425-
{
1426-
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
1427-
get;
1428-
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
1429-
set;
1430-
}
1426+
public MemoryPool<byte>? ResponseBufferPool { get; set; }
14311427

14321428
/// <summary>
14331429
/// The buffer pool to use when buffering requests.
14341430
/// </summary>
1435-
public CycleBufferPool? RequestCycleBufferPool
1436-
{
1437-
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
1438-
get;
1439-
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
1440-
set;
1441-
}
1442-
1443-
/// <summary>
1444-
/// The memory pool to use when buffering responses.
1445-
/// </summary>
1446-
public MemoryPool<byte>? ResponseMemoryPool
1447-
{
1448-
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
1449-
get;
1450-
[Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
1451-
set;
1452-
}
1431+
public MemoryPool<byte>? RequestBufferPool { get; set; }
14531432
}
14541433
}

‎src/StackExchange.Redis/Lease.cs‎

Lines changed: 16 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -34,26 +34,33 @@ public sealed class Lease<T> : IMemoryOwner<T>
3434
/// </summary>
3535
/// <param name="length">The size required.</param>
3636
/// <param name="clear">Whether to erase the memory.</param>
37+
#pragma warning disable RS0026 // multiple overloads with optional args; not ambiguous in this case.
3738
public static Lease<T> Create(int length, bool clear = true)
39+
#pragma warning restore RS0026
3840
{
39-
if (length == 0) return Empty;
41+
if (length is 0) return Empty;
4042
var arr = ArrayPool<T>.Shared.Rent(length);
41-
if (clear) System.Array.Clear(arr, 0, length);
42-
return new Lease<T>(arr, length);
43+
var lease = new Lease<T>(arr, length);
44+
if (clear) lease.Span.Clear();
45+
return lease;
4346
}
4447

4548
/// <summary>
4649
/// Create a new lease.
4750
/// </summary>
4851
/// <param name="length">The size required.</param>
49-
/// <param name="memoryOwner">Buffer.</param>
50-
public static Lease<T> Create(int length, IMemoryOwner<T> memoryOwner)
52+
/// <param name="pool">The buffer source to use; if <c>null</c>, <see cref="ArrayPool{T}.Shared"/> is used.</param>
53+
/// <param name="clear">Whether to erase the memory.</param>
54+
#pragma warning disable RS0026 // multiple overloads with optional args; not ambiguous in this case.
55+
public static Lease<T> Create(int length, MemoryPool<T>? pool, bool clear = true)
56+
#pragma warning restore RS0026
5157
{
52-
if (length == 0) return Empty;
53-
if ((uint)length > memoryOwner.Memory.Length)
54-
throw new ArgumentOutOfRangeException(nameof(length));
58+
if (pool is null) return Create(length, clear);
59+
if (length is 0) return Empty;
5560

56-
return new Lease<T>(memoryOwner, length);
61+
var lease = new Lease<T>(pool.Rent(length), length);
62+
if (clear) lease.Span.Clear();
63+
return lease;
5764
}
5865

5966
private Lease(T[] arr, int length)

‎src/StackExchange.Redis/PhysicalConnection.Read.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ static void StartReadingSync(PhysicalConnection conn, CancellationToken cancella
6565
private void StartReadAllAsync(CancellationToken cancellationToken)
6666
=> Task.Run(() => ReadAllAsync(cancellationToken)).RedisFireAndForget();
6767

68-
private CycleBufferPool? ReaderBufferPool => BridgeCouldBeNull?.Multiplexer?.RawConfig?.ResponseCycleBufferPool;
68+
private MemoryPool<byte>? ReaderBufferPool => BridgeCouldBeNull?.Multiplexer?.RawConfig?.ResponseBufferPool;
6969

7070
private async Task ReadAllAsync(CancellationToken cancellationToken)
7171
{
@@ -401,7 +401,7 @@ private void OnResponseFrame(RespPrefix prefix, ReadOnlySequence<byte> payload)
401401
else
402402
{
403403
var len = checked((int)payload.Length);
404-
var memoryPool = BridgeCouldBeNull?.Multiplexer.RawConfig.ResponseMemoryPool ?? MemoryPool<byte>.Shared;
404+
var memoryPool = BridgeCouldBeNull?.Multiplexer.RawConfig.RequestBufferPool ?? MemoryPool<byte>.Shared;
405405
var memoryOwner = memoryPool.Rent(len);
406406
Span<byte> oversized = memoryOwner.Memory.Span.Slice(0, len);
407407

‎src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1734,7 +1734,6 @@ static StackExchange.Redis.HashEntry.operator !=(StackExchange.Redis.HashEntry x
17341734
static StackExchange.Redis.HashEntry.operator ==(StackExchange.Redis.HashEntry x, StackExchange.Redis.HashEntry y) -> bool
17351735
static StackExchange.Redis.KeyspaceIsolation.DatabaseExtensions.WithKeyPrefix(this StackExchange.Redis.IDatabase! database, StackExchange.Redis.RedisKey keyPrefix) -> StackExchange.Redis.IDatabase!
17361736
static StackExchange.Redis.Lease<T>.Create(int length, bool clear = true) -> StackExchange.Redis.Lease<T>!
1737-
static StackExchange.Redis.Lease<T>.Create(int length, System.Buffers.IMemoryOwner<T>! memoryOwner) -> StackExchange.Redis.Lease<T>!
17381737
static StackExchange.Redis.Lease<T>.Empty.get -> StackExchange.Redis.Lease<T>!
17391738
static StackExchange.Redis.ListPopResult.Null.get -> StackExchange.Redis.ListPopResult
17401739
static StackExchange.Redis.LuaScript.GetCachedScriptCount() -> int

‎src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,9 @@
11
#nullable enable
2-
[SER004]StackExchange.Redis.ConfigurationOptions.ResponseMemoryPool.get -> System.Buffers.MemoryPool<byte>?
3-
[SER004]StackExchange.Redis.ConfigurationOptions.ResponseMemoryPool.set -> void
4-
[SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool?
5-
[SER004]StackExchange.Redis.ConfigurationOptions.RequestCycleBufferPool.set -> void
6-
[SER004]StackExchange.Redis.ConfigurationOptions.ResponseCycleBufferPool.get -> RESPite.Buffers.CycleBufferPool?
7-
[SER004]StackExchange.Redis.ConfigurationOptions.ResponseCycleBufferPool.set -> void
2+
StackExchange.Redis.ConfigurationOptions.RequestBufferPool.get -> System.Buffers.MemoryPool<byte>?
3+
StackExchange.Redis.ConfigurationOptions.RequestBufferPool.set -> void
4+
StackExchange.Redis.ConfigurationOptions.ResponseBufferPool.get -> System.Buffers.MemoryPool<byte>?
5+
StackExchange.Redis.ConfigurationOptions.ResponseBufferPool.set -> void
6+
static StackExchange.Redis.Lease<T>.Create(int length, System.Buffers.MemoryPool<T>? pool, bool clear = true) -> StackExchange.Redis.Lease<T>!
87
[SER005]StackExchange.Redis.TestHarness
98
[SER005]StackExchange.Redis.TestHarness.BufferValidator
109
[SER005]StackExchange.Redis.TestHarness.ChannelPrefix.get -> StackExchange.Redis.RedisChannel

0 commit comments

Comments
 (0)