fix toy server output

Author: mgravell

src/RESPite/Buffers/CycleBuffer.cs

@@ -30,13 +30,19 @@ namespace RESPite.Buffers;
 [Experimental(Experiments.Respite, UrlFormat = Experiments.UrlFormat)]
 public partial struct CycleBuffer
 {
+    #if TRACK_MEMORY
+    private static MemoryPool<byte> DefaultPool => MemoryTrackedPool<byte>.Shared;
+    #else
+    private static MemoryPool<byte> DefaultPool => MemoryPool<byte>.Shared;
+    #endif
+
     // note: if someone uses an uninitialized CycleBuffer (via default): that's a skills issue; git gud
     public static CycleBuffer Create(
         MemoryPool<byte>? pool = null,
         int pageSize = DefaultPageSize,
         ICycleBufferCallback? callback = null)
     {
-        pool ??= MemoryPool<byte>.Shared;
+        pool ??= DefaultPool;
         if (pageSize <= 0) pageSize = DefaultPageSize;
         if (pageSize > pool.MaxBufferSize) pageSize = pool.MaxBufferSize;
         return new CycleBuffer(pool, pageSize, callback);

src/RESPite/Buffers/MemoryTrackedPool.cs

@@ -0,0 +1,63 @@
+#if TRACK_MEMORY
+using System.Buffers;
+using System.Diagnostics.CodeAnalysis;
+
+namespace RESPite.Buffers;
+
+internal sealed class MemoryTrackedPool<T> : MemoryPool<T>
+{
+    // like MemoryPool<T>, but tracks and reports double disposal via a custom memory manager, which
+    // allows all future use of a Memory<T> to be tracked; contrast ArrayMemoryPool<T>, which only tracks
+    // the initial fetch of .Memory from the lease
+    public override IMemoryOwner<T> Rent(int minBufferSize = -1) => MemoryManager.Rent(minBufferSize);
+
+    protected override void Dispose(bool disposing)
+    {
+    }
+
+    // ReSharper disable once ArrangeModifiersOrder - you're wrong
+    public static new MemoryTrackedPool<T> Shared { get; } = new();
+
+    public override int MaxBufferSize => MemoryPool<T>.Shared.MaxBufferSize;
+
+    private MemoryTrackedPool()
+    {
+    }
+
+    private sealed class MemoryManager : MemoryManager<T>
+    {
+        public static IMemoryOwner<T> Rent(int minBufferSize = -1) => new MemoryManager(minBufferSize);
+
+        private T[]? array;
+        private MemoryManager(int minBufferSize)
+        {
+            array = ArrayPool<T>.Shared.Rent(Math.Max(64, minBufferSize));
+        }
+
+        private T[] CheckDisposed()
+        {
+            return array ?? Throw();
+            [DoesNotReturn]
+            static T[] Throw() => throw new ObjectDisposedException("Use-after-free of Memory-" + typeof(T).Name);
+        }
+
+        public override MemoryHandle Pin(int elementIndex = 0) => throw new NotSupportedException(nameof(Pin));
+
+        public override void Unpin() => throw new NotSupportedException(nameof(Unpin));
+
+        public override Span<T> GetSpan() => CheckDisposed();
+
+        protected override bool TryGetArray(out ArraySegment<T> segment)
+        {
+            segment = new ArraySegment<T>(CheckDisposed());
+            return true;
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            var arr = Interlocked.Exchange(ref array, null);
+            if (arr is not null) ArrayPool<T>.Shared.Return(arr);
+        }
+    }
+}
+#endif

src/RESPite/Messages/RespReader.Debug.cs

@@ -1,4 +1,6 @@
-using System.Diagnostics;
+using System.Buffers;
+using System.Diagnostics;
+using System.Text;
 
 #pragma warning disable IDE0079 // Remove unnecessary suppression
 #pragma warning disable CS0282 // There is no defined ordering between fields in multiple declarations of partial struct
@@ -30,4 +32,28 @@ internal void DebugReset()
 #if DEBUG
     internal bool VectorizeDisabled { get; set; }
 #endif
+
+    private partial ReadOnlySpan<byte> ActiveBuffer { get; }
+
+    internal readonly string BufferUtf8()
+    {
+        var clone = Clone();
+        var active = clone.ActiveBuffer;
+        var totalLen = checked((int)(active.Length + clone._remainingTailLength));
+        var oversized = ArrayPool<byte>.Shared.Rent(totalLen);
+        Span<byte> target = oversized.AsSpan(0, totalLen);
+
+        while (!target.IsEmpty)
+        {
+            active.CopyTo(target);
+            target = target.Slice(active.Length);
+            if (!clone.TryMoveToNextSegment()) break;
+            active = clone.ActiveBuffer;
+        }
+        if (!target.IsEmpty) throw new EndOfStreamException();
+
+        var s = Encoding.UTF8.GetString(oversized, 0, totalLen);
+        ArrayPool<byte>.Shared.Return(oversized);
+        return s;
+    }
 }

src/RESPite/Messages/RespReader.Span.cs

@@ -52,6 +52,7 @@ private partial void SetCurrent(ReadOnlySpan<byte> value)
         _bufferRoot = ref MemoryMarshal.GetReference(value);
         _bufferLength = value.Length;
     }
+    private partial ReadOnlySpan<byte> ActiveBuffer => MemoryMarshal.CreateReadOnlySpan(ref _bufferRoot, _bufferLength);
 }
 #else
 public ref partial struct RespReader // much more conservative - uses slices etc
@@ -80,5 +81,6 @@ private readonly partial int CurrentLength
     private readonly partial ReadOnlySpan<byte> CurrentSpan() => _buffer.Slice(_bufferIndex);
 
     private partial void SetCurrent(ReadOnlySpan<byte> value) => _buffer = value;
+    private partial ReadOnlySpan<byte> ActiveBuffer => _buffer;
 }
 #endif

src/RESPite/Messages/RespReader.Utils.cs

@@ -90,7 +90,10 @@ private readonly LengthPrefixResult TryReadLengthPrefix(ReadOnlySpan<byte> bytes
         }
     }
 
-    private readonly RespReader Clone() => this; // useful for performing streaming operations without moving the primary
+    /// <summary>
+    /// Create an isolated copy of this reader, which can be advanced independently.
+    /// </summary>
+    public readonly RespReader Clone() => this; // useful for performing streaming operations without moving the primary
 
     [MethodImpl(MethodImplOptions.NoInlining), DoesNotReturn]
     private readonly void ThrowProtocolFailure(string message)

src/RESPite/Messages/RespReader.cs

@@ -110,6 +110,11 @@ public readonly int ScalarLength() =>
     public readonly bool ScalarIsEmpty() =>
         IsInlineScalar ? _length == 0 : (IsNullScalar || !ScalarChunks().MoveNext());
 
+    /// <summary>
+    /// Indicates whether this aggregate value is zero-length.
+    /// </summary>
+    public readonly bool AggregateIsEmpty() => AggregateLengthIs(0);
+
     /// <summary>
     /// The payload length of this scalar element (includes combined length for streaming scalars).
     /// </summary>

src/RESPite/PublicAPI/PublicAPI.Unshipped.txt

@@ -48,7 +48,9 @@
 [SER004]RESPite.Messages.RespReader.AggregateEnumerator.FillAll<TState, TResult>(scoped System.Span<TResult> target, ref TState state, RESPite.Messages.RespReader.Projection<TState, TResult>! projection) -> void
 [SER004]RESPite.Messages.RespReader.AggregateEnumerator.MoveNextRaw() -> bool
 [SER004]RESPite.Messages.RespReader.AggregateEnumerator.MoveNextRaw<T>(RESPite.Messages.RespAttributeReader<T>! respAttributeReader, ref T attributes) -> bool
+[SER004]RESPite.Messages.RespReader.AggregateIsEmpty() -> bool
 [SER004]RESPite.Messages.RespReader.AggregateLengthIs(int count) -> bool
+[SER004]RESPite.Messages.RespReader.Clone() -> RESPite.Messages.RespReader
 [SER004]RESPite.Messages.RespReader.FillAll<TState, TResult>(scoped System.Span<TResult> target, ref TState state, RESPite.Messages.RespReader.Projection<TState, TResult>! projection) -> void
 [SER004]RESPite.Messages.RespReader.Projection<TState, TResult>
 [SER004]RESPite.Messages.RespReader.ReadArray<TState, TResult>(ref TState state, RESPite.Messages.RespReader.Projection<TState, TResult>! projection, bool scalar = false) -> TResult[]?

src/StackExchange.Redis/RespReaderExtensions.cs

@@ -41,8 +41,34 @@ public string DebugReadTruncatedString(int maxChars)
         public RedisChannel ReadRedisChannel(RedisChannel.RedisChannelOptions options)
             => new(reader.ReadByteArray(), options);
 
+        private bool TryGetFirst(out string first)
+        {
+            if (reader.IsNonNullAggregate && !reader.AggregateIsEmpty())
+            {
+                var clone = reader.Clone();
+                if (clone.TryMoveNext())
+                {
+                    unsafe
+                    {
+                        if (clone.IsScalar &&
+                            clone.TryParseScalar(&PhysicalConnection.PushKindMetadata.TryParse, out PhysicalConnection.PushKind kind))
+                        {
+                            first = kind.ToString();
+                            return true;
+                        }
+                    }
+
+                    first = clone.GetOverview();
+                    return true;
+                }
+            }
+            first = "";
+            return false;
+        }
+
         public string GetOverview()
         {
+            // return reader.BufferUtf8(); // <== for when you really can't grok what is happening
             if (reader.Prefix is RespPrefix.None)
             {
                 var copy = reader;
@@ -54,6 +80,7 @@ public string GetOverview()
             return reader.Prefix switch
             {
                 RespPrefix.SimpleString or RespPrefix.Integer or RespPrefix.SimpleError or RespPrefix.Double => $"{reader.Prefix}: {reader.ReadString()}",
+                RespPrefix.Push when reader.TryGetFirst(out var first) => $"{reader.Prefix} ({first}): {reader.AggregateLength()} items",
                 _ when reader.IsScalar => $"{reader.Prefix}: {reader.ScalarLength()} bytes, '{reader.DebugReadTruncatedString(16)}'",
                 _ when reader.IsAggregate => $"{reader.Prefix}: {reader.AggregateLength()} items",
                 _ => $"(unknown: {reader.Prefix})",

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -129,6 +129,21 @@ protected override void OnOutOfBand(RedisClient client, TypedRedisValue message)
         base.OnOutOfBand(client, message);
     }
 
+    /*
+    public override void OnFlush(RedisClient client, int messages, long bytes)
+    {
+        if (bytes >= 0)
+        {
+            _log?.WriteLine($"[{client}] flushed {messages} messages, {bytes} bytes");
+        }
+        else
+        {
+            _log?.WriteLine($"[{client}] flushed {messages} messages"); // bytes not available
+        }
+        base.OnFlush(client, messages, bytes);
+    }
+    */
+
     public override TypedRedisValue OnUnknownCommand(in RedisClient client, in RedisRequest request, ReadOnlySpan<byte> command)
     {
         _log?.WriteLine($"[{client}] unknown command: {Encoding.ASCII.GetString(command)}");

tests/StackExchange.Redis.Tests/PubSubTests.cs

@@ -154,6 +154,16 @@ public async Task TestBasicPubSub(string? channelPrefix, bool wildCard, string b
         Assert.Equal(0, count);
     }
 
+    [Fact]
+    public async Task Ping()
+    {
+        await using var conn = ConnectFactory(shared: false);
+        var pub = GetAnyPrimary(conn.DefaultClient);
+        var sub = conn.GetSubscriber();
+        await sub.SubscribeAsync(RedisChannel.Literal(Me()), (_, __) => { }); // to ensure we're in subscriber mode
+        await PingAsync(pub, sub, 5).ForAwait();
+    }
+
     [Fact]
     public async Task TestBasicPubSubFireAndForget()
     {
@@ -313,6 +323,7 @@ public async Task TestMassivePublishWithWithoutFlush_Local()
     public async Task TestMassivePublishWithWithoutFlush_Remote()
     {
         Skip.UnlessLongRunning();
+        SkipIfWouldUseInProcessServer();
         await using var conn = Create(configuration: TestConfig.Current.RemoteServerAndPort);
 
         var sub = conn.GetSubscriber();
@@ -436,6 +447,7 @@ await sub.SubscribeAsync(channel, (_, val) =>
     [Fact]
     public async Task PubSubGetAllCorrectOrder()
     {
+        SkipIfWouldUseInProcessServer();
         await using (var conn = Create(configuration: TestConfig.Current.RemoteServerAndPort, syncTimeout: 20000, log: Writer))
         {
             var sub = conn.GetSubscriber();
@@ -506,6 +518,7 @@ async Task RunLoop()
     [Fact]
     public async Task PubSubGetAllCorrectOrder_OnMessage_Sync()
     {
+        SkipIfWouldUseInProcessServer();
         await using (var conn = Create(configuration: TestConfig.Current.RemoteServerAndPort, syncTimeout: 20000, log: Writer))
         {
             var sub = conn.GetSubscriber();
@@ -572,6 +585,7 @@ public async Task PubSubGetAllCorrectOrder_OnMessage_Sync()
     [Fact]
     public async Task PubSubGetAllCorrectOrder_OnMessage_Async()
     {
+        SkipIfWouldUseInProcessServer();
         await using (var conn = Create(configuration: TestConfig.Current.RemoteServerAndPort, syncTimeout: 20000, log: Writer))
         {
             var sub = conn.GetSubscriber();