Implement a redis-server, to help testing (#894)

* v0.0 of test server

* server very much doesn't work, but: pieces are in place

* it's alive!

* implement MemoryCache server

* be a little pickier about which faults we actually report

* implement **very badly** pre-RESP protocol for server; this is just to test PING_INLINE etc, and it *finds a catastrophic case*

* make shutdown much more graceful; Databases should be immutable; enable ConnectionExecute vs ServerExecute concept; add awaitable server shutdown task

* make MemoryCache usage standalone, and enable flush; implement cusstom box/unbox operations on RedisValue

* implement basic "info"

* limit scope of GetCurrentProcess

* set slaveof in config to make clients happier

* rename server types; implement MEMORY PURGE and TIME; make types explicit throughout in RedisServer

* implement KEYS; prefer NotSupportedException to NotImplementedException, but recognize both

* implement UNLINK; better handling of nil requests

* implement basic set ops

* implement STRLEN; handle WRONGTYPE

* convention / reflection based command registration

* overhaul how arity works so we can implement COMMAND; support null arrays

* make sure we can parse null arras is RedisResult

* set server socket options

* fix error handling incomplete lines in the "inline" protocol

* move ParseInlineProtocol out, but: still not implemented

* implement CLIENT REPLY and add the "inline" protocol

* need to support either quote tokenizer

* add readme to the server code

* accessibility on sample code

* implement the last of the commands needed for redis-benchmark

* fix naming

StackExchange.Redis.Server/MemoryCacheRedisServer.cs

+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Runtime.Caching;
+using System.Runtime.CompilerServices;
+
+namespace StackExchange.Redis.Server
+{
+    public class MemoryCacheRedisServer : RedisServer
+    {
+        public MemoryCacheRedisServer(TextWriter output = null) : base(1, output)
+            => CreateNewCache();
+
+        private MemoryCache _cache;
+
+        private void CreateNewCache()
+        {
+            var old = _cache;
+            _cache = new MemoryCache(GetType().Name);
+            if (old != null) old.Dispose();
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing) _cache.Dispose();
+            base.Dispose(disposing);
+        }
+
+        protected override long Dbsize(int database) => _cache.GetCount();
+        protected override RedisValue Get(int database, RedisKey key)
+            => RedisValue.Unbox(_cache[key]);
+        protected override void Set(int database, RedisKey key, RedisValue value)
+            => _cache[key] = value.Box();
+        protected override bool Del(int database, RedisKey key)
+            => _cache.Remove(key) != null;
+        protected override void Flushdb(int database)
+            => CreateNewCache();
+
+        protected override bool Exists(int database, RedisKey key)
+            => _cache.Contains(key);
+
+        protected override IEnumerable<RedisKey> Keys(int database, RedisKey pattern)
+        {
+            string s = pattern;
+            foreach (var pair in _cache)
+            {
+                if (IsMatch(pattern, pair.Key)) yield return pair.Key;
+            }
+        }
+        protected override bool Sadd(int database, RedisKey key, RedisValue value)
+            => GetSet(key, true).Add(value);
+
+        protected override bool Sismember(int database, RedisKey key, RedisValue value)
+            => GetSet(key, false)?.Contains(value) ?? false;
+
+        protected override bool Srem(int database, RedisKey key, RedisValue value)
+        {
+            var set = GetSet(key, false);
+            if (set != null && set.Remove(value))
+            {
+                if (set.Count == 0) _cache.Remove(key);
+                return true;
+            }
+            return false;
+        }
+        protected override long Scard(int database, RedisKey key)
+            => GetSet(key, false)?.Count ?? 0;
+
+        HashSet<RedisValue> GetSet(RedisKey key, bool create)
+        {
+            var set = (HashSet<RedisValue>)_cache[key];
+            if (set == null && create)
+            {
+                set = new HashSet<RedisValue>();
+                _cache[key] = set;
+            }
+            return set;
+        }
+
+        protected override RedisValue Spop(int database, RedisKey key)
+        {
+            var set = GetSet(key, false);
+            if (set == null) return RedisValue.Null;
+
+            var result = set.First();
+            set.Remove(result);
+            if (set.Count == 0) _cache.Remove(key);
+            return result;
+        }
+
+        protected override long Lpush(int database, RedisKey key, RedisValue value)
+        {
+            var stack = GetStack(key, true);
+            stack.Push(value);
+            return stack.Count;
+        }
+        protected override RedisValue Lpop(int database, RedisKey key)
+        {
+            var stack = GetStack(key, false);
+            if (stack == null) return RedisValue.Null;
+
+            var val = stack.Pop();
+            if(stack.Count == 0) _cache.Remove(key);
+            return val;
+        }
+
+        protected override long Llen(int database, RedisKey key)
+            => GetStack(key, false)?.Count ?? 0;
+
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        static void ThrowArgumentOutOfRangeException() => throw new ArgumentOutOfRangeException();
+        protected override void LRange(int database, RedisKey key, long start, RedisValue[] arr)
+        {
+            var stack = GetStack(key, false);
+
+            using (var iter = stack.GetEnumerator())
+            {
+                // skip
+                while (start-- > 0) if (!iter.MoveNext()) ThrowArgumentOutOfRangeException();
+
+                // take
+                for (int i = 0; i < arr.Length; i++)
+                {
+                    if (!iter.MoveNext()) ThrowArgumentOutOfRangeException();
+                    arr[i] = iter.Current;
+                }
+            }
+        }
+
+        Stack<RedisValue> GetStack(RedisKey key, bool create)
+        {
+            var stack = (Stack<RedisValue>)_cache[key];
+            if (stack == null && create)
+            {
+                stack = new Stack<RedisValue>();
+                _cache[key] = stack;
+            }
+            return stack;
+        }
+
+
+    }
+}

StackExchange.Redis.Server/RedisClient.cs

+using System;
+using System.Collections.Generic;
+using System.IO.Pipelines;
+
+namespace StackExchange.Redis.Server
+{
+    public sealed class RedisClient : IDisposable
+    {
+        internal int SkipReplies { get; set; }
+        internal bool ShouldSkipResponse()
+        {
+            if (SkipReplies > 0)
+            {
+                SkipReplies--;
+                return true;
+            }
+            return false;
+        }
+        private HashSet<RedisChannel> _subscripions;
+        public int SubscriptionCount => _subscripions?.Count ?? 0;
+        internal int Subscribe(RedisChannel channel)
+        {
+            if (_subscripions == null) _subscripions = new HashSet<RedisChannel>();
+            _subscripions.Add(channel);
+            return _subscripions.Count;
+        }
+        internal int Unsubscribe(RedisChannel channel)
+        {
+            if (_subscripions == null) return 0;
+            _subscripions.Remove(channel);
+            return _subscripions.Count;
+        }
+        public int Database { get; set; }
+        public string Name { get; set; }
+        internal IDuplexPipe LinkedPipe { get; set; }
+        public bool Closed { get; internal set; }
+
+        public void Dispose()
+        {
+            Closed = true;
+            var pipe = LinkedPipe;
+            LinkedPipe = null;
+            if (pipe != null)
+            {
+                try { pipe.Input.CancelPendingRead(); } catch { }
+                try { pipe.Input.Complete(); } catch { }
+                try { pipe.Output.CancelPendingFlush(); } catch { }
+                try { pipe.Output.Complete(); } catch { }
+                if (pipe is IDisposable d) try { d.Dispose(); } catch { }
+            }
+        }
+    }
+}

StackExchange.Redis.Server/RedisRequest.cs

+using System;
+
+namespace StackExchange.Redis.Server
+{
+    public readonly ref struct RedisRequest
+    {   // why ref? don't *really* need it, but: these things are "in flight"
+        // based on an open RawResult (which is just the detokenized ReadOnlySequence<byte>)
+        // so: using "ref" makes it clear that you can't expect to store these and have
+        // them keep working
+        private readonly RawResult _inner;
+
+        public int Count { get; }
+        public string Command { get; }
+        public override string ToString() => Command;
+        public override bool Equals(object obj) => throw new NotSupportedException();
+
+        public RedisResult WrongArgCount() => RedisResult.Create($"ERR wrong number of arguments for '{Command}' command", ResultType.Error);
+
+        public RedisResult UnknownSubcommandOrArgumentCount() => RedisResult.Create($"ERR Unknown subcommand or wrong number of arguments for '{Command}'.", ResultType.Error);
+
+        public string GetString(int index)
+            => _inner[index].GetString();
+        
+        internal RedisResult GetResult(int index)
+            => RedisResult.Create(_inner[index].AsRedisValue());
+
+        public bool IsString(int index, string value) // TODO: optimize
+            => string.Equals(value, _inner[index].GetString(), StringComparison.OrdinalIgnoreCase);
+
+        public override int GetHashCode() => throw new NotSupportedException();
+        internal RedisRequest(RawResult result)
+            : this(result, result.ItemsCount, result[0].GetString()) { }
+        private RedisRequest(RawResult inner, int count, string command)
+        {
+            _inner = inner;
+            Count = count;
+            Command = command;
+        }
+        internal RedisRequest AsCommand(string command)
+            => new RedisRequest(_inner, Count, command);
+
+        
+        public void Recycle() => _inner.Recycle();
+
+        public RedisValue GetValue(int index)
+            => _inner[index].AsRedisValue();
+
+        public int GetInt32(int index)
+            => (int)_inner[index].AsRedisValue();
+
+        public long GetInt64(int index) => (long)_inner[index].AsRedisValue();
+    
+        public RedisKey GetKey(int index) => _inner[index].AsRedisKey();
+        
+        public RedisChannel GetChannel(int index, RedisChannel.PatternMode mode)
+            => _inner[index].AsRedisChannel(null, mode);
+    }
+}

StackExchange.Redis.Server/StackExchange.Redis.Server.csproj

+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>$(LibraryTargetFrameworks)</TargetFrameworks>
+    <Description>Basic redis server based on StackExchange.Redis</Description>
+    <AssemblyTitle>StackExchange.Redis</AssemblyTitle>
+    <GenerateDocumentationFile>true</GenerateDocumentationFile>
+    <AssemblyName>StackExchange.Redis.Server</AssemblyName>
+    <PackageId>StackExchange.Redis.Server</PackageId>
+    <PackageTags>Server;Async;Redis;Cache;PubSub;Messaging</PackageTags>
+    <OutputTypeEx>Library</OutputTypeEx>
+    <SignAssembly>true</SignAssembly>
+    <PublicSign Condition=" '$(OS)' != 'Windows_NT' ">true</PublicSign>
+    <LangVersion>latest</LangVersion>
+    <NoWarn>$(NoWarn);CS1591</NoWarn>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(Configuration)|$(TargetFramework)|$(Platform)'=='Debug|netstandard2.0|AnyCPU'">
+    
+    <AllowUnsafeBlocks>false</AllowUnsafeBlocks>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\StackExchange.Redis\StackExchange.Redis.csproj" />
+    <PackageReference Include="System.Runtime.Caching" Version="4.5.0" />
+  </ItemGroup>
+</Project>

StackExchange.Redis.Server/readme.md

+# Wait, what is this?
+
+This is **not** a replacement for redis!
+
+This is some example code that illustrates using "pipelines" to implement a server, in this case a server that works like 'redis',
+implementing the same protocol, and offering similar services.
+
+What it isn't:
+
+- supported
+- as good as redis
+- feature complete
+- bug free
+
+What it is:
+
+- useful for me to test my protocol handling
+- useful for debugging
+- useful for anyone looking for reference code for implementing a custom server based on pipelines
+- fun
+
+Example usage:
+
+```c#
+using System;
+using System.Net;
+using System.Threading.Tasks;
+using StackExchange.Redis.Server;
+
+static class Program
+{
+    static async Task Main()
+    {
+        using (var server = new MemoryCacheRedisServer(Console.Out))
+        {
+            server.Listen(new IPEndPoint(IPAddress.Loopback, 6379));
+            await server.Shutdown;
+        }
+    }
+}
+```
\ No newline at end of file

StackExchange.Redis.Tests/Cluster.cs

@@ -677,11 +677,11 @@ public void MovedProfiling()
             var Key = Me();
             const string Value = "redirected-value";
 
-            var profiler = new ProfilingSession();
+            var profiler = new Profiling.PerThreadProfiler();
 
             using (var conn = Create())
             {
-                conn.RegisterProfiler(() => profiler);
+                conn.RegisterProfiler(profiler.GetSession);
 
                 var endpoints = conn.GetEndPoints();
                 var servers = endpoints.Select(e => conn.GetServer(e));
@@ -705,7 +705,7 @@ public void MovedProfiling()
                 string b = (string)conn.GetServer(wrongMasterNode.EndPoint).Execute("GET", Key);
                 Assert.Equal(Value, b); // wrong master, allow redirect
 
-                var msgs = profiler.FinishProfiling().ToList();
+                var msgs = profiler.GetSession().FinishProfiling().ToList();
 
                 // verify that things actually got recorded properly, and the retransmission profilings are connected as expected
                 {

StackExchange.Redis.Tests/Profiling.cs

@@ -202,13 +202,13 @@ public void ManyContexts()
             }
         }
 
-        private class PerThreadProfiler
+        internal class PerThreadProfiler
         {
             ThreadLocal<ProfilingSession> perThreadSession = new ThreadLocal<ProfilingSession>(() => new ProfilingSession());
 
             public ProfilingSession GetSession() => perThreadSession.Value;
         }
-        private class AsyncLocalProfiler
+        internal class AsyncLocalProfiler
         {
             AsyncLocal<ProfilingSession> perThreadSession = new AsyncLocal<ProfilingSession>();
 

StackExchange.Redis.Tests/StackExchange.Redis.Tests.csproj

@@ -7,6 +7,7 @@
     <GenerateDocumentationFile>false</GenerateDocumentationFile>
     <SignAssembly>true</SignAssembly>
     <DebugType>full</DebugType>
+    <LangVersion>latest</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>

StackExchange.Redis.sln

@@ -79,6 +79,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Failover", "Failover", "{D0
 		RedisConfigs\Failover\slave-6383.conf = RedisConfigs\Failover\slave-6383.conf
 	EndProjectSection
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "StackExchange.Redis.Server", "StackExchange.Redis.Server\StackExchange.Redis.Server.csproj", "{8375813E-FBAF-4DA3-A2C7-E4645B39B931}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -117,6 +119,10 @@ Global
 		{769640F3-889C-4E8A-A7DF-916AE9B432A6}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{769640F3-889C-4E8A-A7DF-916AE9B432A6}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{769640F3-889C-4E8A-A7DF-916AE9B432A6}.Release|Any CPU.Build.0 = Release|Any CPU
+		{8375813E-FBAF-4DA3-A2C7-E4645B39B931}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{8375813E-FBAF-4DA3-A2C7-E4645B39B931}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{8375813E-FBAF-4DA3-A2C7-E4645B39B931}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{8375813E-FBAF-4DA3-A2C7-E4645B39B931}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

StackExchange.Redis/StackExchange/AssemblyInfoHack.cs

@@ -2,4 +2,5 @@
 // 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.Runtime.CompilerServices;
+[assembly: InternalsVisibleTo("StackExchange.Redis.Server, PublicKey=00240000048000009400000006020000002400005253413100040000010001007791a689e9d8950b44a9a8886baad2ea180e7a8a854f158c9b98345ca5009cdd2362c84f368f1c3658c132b3c0f74e44ff16aeb2e5b353b6e0fe02f923a050470caeac2bde47a2238a9c7125ed7dab14f486a5a64558df96640933b9f2b6db188fc4a820f96dce963b662fa8864adbff38e5b4542343f162ecdc6dad16912fff")]
 [assembly: InternalsVisibleTo("StackExchange.Redis.Tests, PublicKey=00240000048000009400000006020000002400005253413100040000010001007791a689e9d8950b44a9a8886baad2ea180e7a8a854f158c9b98345ca5009cdd2362c84f368f1c3658c132b3c0f74e44ff16aeb2e5b353b6e0fe02f923a050470caeac2bde47a2238a9c7125ed7dab14f486a5a64558df96640933b9f2b6db188fc4a820f96dce963b662fa8864adbff38e5b4542343f162ecdc6dad16912fff")]

StackExchange.Redis/StackExchange/Redis/BufferReader.cs

+using System;
+using System.Buffers;
+using System.IO;
+
+namespace StackExchange.Redis
+{
+    internal enum ConsumeResult
+    {
+        Failure,
+        Success,
+        NeedMoreData,
+    }
+    internal ref struct BufferReader
+    {
+        private ReadOnlySequence<byte>.Enumerator _iterator;
+        private ReadOnlySpan<byte> _current;
+
+        public ReadOnlySpan<byte> OversizedSpan => _current;
+
+        public ReadOnlySpan<byte> SlicedSpan => _current.Slice(OffsetThisSpan, RemainingThisSpan);
+        public int OffsetThisSpan { get; private set; }
+        private int TotalConsumed { get; set; } // hide this; callers should use the snapshot-aware methods instead
+        public int RemainingThisSpan { get; private set; }
+
+        public bool IsEmpty => RemainingThisSpan == 0;
+
+        private bool FetchNextSegment()
+        {
+            do
+            {
+                if (!_iterator.MoveNext())
+                {
+                    OffsetThisSpan = RemainingThisSpan = 0;
+                    return false;
+                }
+
+                _current = _iterator.Current.Span;
+                OffsetThisSpan = 0;
+                RemainingThisSpan = _current.Length;
+            } while (IsEmpty); // skip empty segments, they don't help us!
+
+            return true;
+        }
+
+        public BufferReader(ReadOnlySequence<byte> buffer)
+        {
+            _buffer = buffer;
+            _lastSnapshotPosition = buffer.Start;
+            _lastSnapshotBytes = 0;
+            _iterator = buffer.GetEnumerator();
+            _current = default;
+            OffsetThisSpan = RemainingThisSpan = TotalConsumed = 0;
+
+            FetchNextSegment();
+        }
+
+        private static readonly byte[] CRLF = { (byte)'\r', (byte)'\n' };
+
+        /// <summary>
+        /// Note that in results other than success, no guarantees are made about final state; if you care: snapshot
+        /// </summary>
+        public ConsumeResult TryConsumeCRLF()
+        {
+            switch (RemainingThisSpan)
+            {
+                case 0:
+                    return ConsumeResult.NeedMoreData;
+                case 1:
+                    if (_current[OffsetThisSpan] != (byte)'\r') return ConsumeResult.Failure;
+                    Consume(1);
+                    if (IsEmpty) return ConsumeResult.NeedMoreData;
+                    var next = _current[OffsetThisSpan];
+                    Consume(1);
+                    return next == '\n' ? ConsumeResult.Success : ConsumeResult.Failure;
+                default:
+                    var offset = OffsetThisSpan;
+                    var result = _current[offset++] == (byte)'\r' && _current[offset] == (byte)'\n'
+                        ? ConsumeResult.Success : ConsumeResult.Failure;
+                    Consume(2);
+                    return result;
+            }
+        }
+        public bool TryConsume(int count)
+        {
+            if (count < 0) throw new ArgumentOutOfRangeException(nameof(count));
+            do
+            {
+                var available = RemainingThisSpan;
+                if (count <= available)
+                {
+                    // consume part of this span
+                    TotalConsumed += count;
+                    RemainingThisSpan -= count;
+                    OffsetThisSpan += count;
+
+                    if (count == available) FetchNextSegment(); // burned all of it; fetch next
+                    return true;
+                }
+
+                // consume all of this span
+                TotalConsumed += available;
+                count -= available;
+            } while (FetchNextSegment());
+            return false;
+        }
+
+        private readonly ReadOnlySequence<byte> _buffer;
+        private SequencePosition _lastSnapshotPosition;
+        private long _lastSnapshotBytes;
+
+        // makes an internal note of where we are, as a SequencePosition; useful
+        // to avoid having to use buffer.Slice on huge ranges
+        private SequencePosition SnapshotPosition()
+        {
+            var consumed = TotalConsumed;
+            var delta = consumed - _lastSnapshotBytes;
+            if (delta == 0) return _lastSnapshotPosition;
+
+            var pos = _buffer.GetPosition(delta, _lastSnapshotPosition);
+            _lastSnapshotBytes = consumed;
+            return _lastSnapshotPosition = pos;
+        }
+        public ReadOnlySequence<byte> ConsumeAsBuffer(int count)
+        {
+            if (!TryConsumeAsBuffer(count, out var buffer)) throw new EndOfStreamException();
+            return buffer;
+        }
+        public ReadOnlySequence<byte> ConsumeToEnd()
+        {
+            var from = SnapshotPosition();
+            var result = _buffer.Slice(from);
+            while (FetchNextSegment()) { } // consume all
+            return result;
+        }
+        public bool TryConsumeAsBuffer(int count, out ReadOnlySequence<byte> buffer)
+        {
+            var from = SnapshotPosition();
+            if (!TryConsume(count))
+            {
+                buffer = default;
+                return false;
+            }
+            var to = SnapshotPosition();
+            buffer = _buffer.Slice(from, to);
+            return true;
+        }
+        public void Consume(int count)
+        {
+            if (!TryConsume(count)) throw new EndOfStreamException();
+        }
+
+        internal static int FindNext(BufferReader reader, byte value) // very deliberately not ref; want snapshot
+        {
+            int totalSkipped = 0;
+            do
+            {
+                if (reader.RemainingThisSpan == 0) continue;
+
+                var span = reader.SlicedSpan;
+                int found = span.IndexOf(value);
+                if (found >= 0) return totalSkipped + found;
+
+                totalSkipped += span.Length;
+            } while (reader.FetchNextSegment());
+            return -1;
+        }
+        internal static int FindNextCrLf(BufferReader reader) // very deliberately not ref; want snapshot
+        {
+            // is it in the current span? (we need to handle the offsets differently if so)
+
+            int totalSkipped = 0;
+            bool haveTrailingCR = false;
+            do
+            {
+                if (reader.RemainingThisSpan == 0) continue;
+
+                var span = reader.SlicedSpan;
+                if (haveTrailingCR)
+                {
+                    if (span[0] == '\n') return totalSkipped - 1;
+                    haveTrailingCR = false;
+                }
+
+                int found = span.IndexOf(CRLF);
+                if (found >= 0) return totalSkipped + found;
+
+                haveTrailingCR = span[span.Length - 1] == '\r';
+                totalSkipped += span.Length;
+            }
+            while (reader.FetchNextSegment());
+            return -1;
+        }
+
+        //internal static bool HasBytes(BufferReader reader, int count) // very deliberately not ref; want snapshot
+        //{
+        //    if (count < 0) throw new ArgumentOutOfRangeException(nameof(count));
+        //    do
+        //    {
+        //        var available = reader.RemainingThisSpan;
+        //        if (count <= available) return true;
+        //        count -= available;
+        //    } while (reader.FetchNextSegment());
+        //    return false;
+        //}
+
+        public int ConsumeByte()
+        {
+            if (IsEmpty) return -1;
+            var value = _current[OffsetThisSpan];
+            Consume(1);
+            return value;
+        }
+        public int PeekByte() => IsEmpty ? -1 : _current[OffsetThisSpan];
+
+        public ReadOnlySequence<byte> SliceFromCurrent()
+        {
+            var from = SnapshotPosition();
+            return _buffer.Slice(from);
+        }
+    }
+}

StackExchange.Redis/StackExchange/Redis/Condition.cs

@@ -308,7 +308,7 @@ protected override void WriteImpl(PhysicalConnection physical)
                     {
                         physical.WriteHeader(command, 2);
                         physical.Write(Key);
-                        physical.Write(value);
+                        physical.WriteBulkString(value);
                     }
                 }
             }

StackExchange.Redis/StackExchange/Redis/Message.cs

@@ -756,7 +756,7 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 2);
                 physical.Write(Channel);
-                physical.Write(value);
+                physical.WriteBulkString(value);
             }
         }
 
@@ -857,7 +857,7 @@ protected override void WriteImpl(PhysicalConnection physical)
                 physical.WriteHeader(Command, 3);
                 physical.Write(Key);
                 physical.Write(key1);
-                physical.Write(value);
+                physical.WriteBulkString(value);
             }
         }
 
@@ -889,7 +889,7 @@ protected override void WriteImpl(PhysicalConnection physical)
                 physical.WriteHeader(command, values.Length);
                 for (int i = 0; i < values.Length; i++)
                 {
-                    physical.Write(values[i]);
+                    physical.WriteBulkString(values[i]);
                 }
             }
         }
@@ -939,7 +939,7 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 2);
                 physical.Write(Key);
-                physical.Write(value);
+                physical.WriteBulkString(value);
             }
         }
 
@@ -968,7 +968,7 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, values.Length + 2);
                 physical.Write(Key);
-                for (int i = 0; i < values.Length; i++) physical.Write(values[i]);
+                for (int i = 0; i < values.Length; i++) physical.WriteBulkString(values[i]);
                 physical.Write(key1);
             }
         }
@@ -989,7 +989,7 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, values.Length + 1);
                 physical.Write(Key);
-                for (int i = 0; i < values.Length; i++) physical.Write(values[i]);
+                for (int i = 0; i < values.Length; i++) physical.WriteBulkString(values[i]);
             }
         }
 
@@ -1008,8 +1008,8 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 3);
                 physical.Write(Key);
-                physical.Write(value0);
-                physical.Write(value1);
+                physical.WriteBulkString(value0);
+                physical.WriteBulkString(value1);
             }
         }
 
@@ -1030,9 +1030,9 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 4);
                 physical.Write(Key);
-                physical.Write(value0);
-                physical.Write(value1);
-                physical.Write(value2);
+                physical.WriteBulkString(value0);
+                physical.WriteBulkString(value1);
+                physical.WriteBulkString(value2);
             }
         }
 
@@ -1055,10 +1055,10 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 5);
                 physical.Write(Key);
-                physical.Write(value0);
-                physical.Write(value1);
-                physical.Write(value2);
-                physical.Write(value3);
+                physical.WriteBulkString(value0);
+                physical.WriteBulkString(value1);
+                physical.WriteBulkString(value2);
+                physical.WriteBulkString(value3);
             }
         }
 
@@ -1097,7 +1097,7 @@ protected override void WriteImpl(PhysicalConnection physical)
                 physical.WriteHeader(command, values.Length);
                 for (int i = 0; i < values.Length; i++)
                 {
-                    physical.Write(values[i]);
+                    physical.WriteBulkString(values[i]);
                 }
             }
         }
@@ -1114,7 +1114,7 @@ public CommandValueChannelMessage(int db, CommandFlags flags, RedisCommand comma
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 2);
-                physical.Write(value);
+                physical.WriteBulkString(value);
                 physical.Write(Channel);
             }
         }
@@ -1138,7 +1138,7 @@ public override void AppendStormLog(StringBuilder sb)
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 2);
-                physical.Write(value);
+                physical.WriteBulkString(value);
                 physical.Write(Key);
             }
         }
@@ -1155,7 +1155,7 @@ public CommandValueMessage(int db, CommandFlags flags, RedisCommand command, Red
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 1);
-                physical.Write(value);
+                physical.WriteBulkString(value);
             }
         }
 
@@ -1173,8 +1173,8 @@ public CommandValueValueMessage(int db, CommandFlags flags, RedisCommand command
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 2);
-                physical.Write(value0);
-                physical.Write(value1);
+                physical.WriteBulkString(value0);
+                physical.WriteBulkString(value1);
             }
         }
 
@@ -1194,9 +1194,9 @@ public CommandValueValueValueMessage(int db, CommandFlags flags, RedisCommand co
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 3);
-                physical.Write(value0);
-                physical.Write(value1);
-                physical.Write(value2);
+                physical.WriteBulkString(value0);
+                physical.WriteBulkString(value1);
+                physical.WriteBulkString(value2);
             }
         }
 
@@ -1220,11 +1220,11 @@ public CommandValueValueValueValueValueMessage(int db, CommandFlags flags, Redis
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 5);
-                physical.Write(value0);
-                physical.Write(value1);
-                physical.Write(value2);
-                physical.Write(value3);
-                physical.Write(value4);
+                physical.WriteBulkString(value0);
+                physical.WriteBulkString(value1);
+                physical.WriteBulkString(value2);
+                physical.WriteBulkString(value3);
+                physical.WriteBulkString(value4);
             }
         }
 
@@ -1237,7 +1237,7 @@ public SelectMessage(int db, CommandFlags flags) : base(db, flags, RedisCommand.
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 1);
-                physical.Write(Db);
+                physical.WriteBulkString(Db);
             }
         }
     }

StackExchange.Redis/StackExchange/Redis/RawResult.cs

@@ -7,6 +7,15 @@ namespace StackExchange.Redis
 {
     internal readonly struct RawResult
     {
+        internal RawResult this[int index]
+        {
+            get
+            {
+                if (index >= _itemsCount) throw new IndexOutOfRangeException();
+                return _itemsOversized[index];
+            }
+        }
+        internal int ItemsCount => _itemsCount;
         internal static readonly RawResult NullMultiBulk = new RawResult(null, 0);
         internal static readonly RawResult EmptyMultiBulk = new RawResult(Array.Empty<RawResult>(), 0);
         internal static readonly RawResult Nil = default;
@@ -72,6 +81,58 @@ public override string ToString()
             }
         }
 
+        public Tokenizer GetInlineTokenizer() => new Tokenizer(_payload);
+
+        internal ref struct Tokenizer
+        {
+            // tokenizes things according to the inline protocol
+            // specifically; the line: abc    "def ghi" jkl
+            // is 3 tokens: "abc", "def ghi" and "jkl"
+            public Tokenizer GetEnumerator() => this;
+            BufferReader _value;
+            
+            public Tokenizer(ReadOnlySequence<byte> value)
+            {
+                _value = new BufferReader(value);
+                Current = default;
+            }
+
+            public bool MoveNext()
+            {
+                Current = default;
+                // take any white-space
+                while (_value.PeekByte() == (byte)' ') { _value.Consume(1); }
+
+                byte terminator = (byte)' ';
+                var first = _value.PeekByte();
+                if (first < 0) return false; // EOF
+
+                switch (_value.PeekByte())
+                {
+                    case (byte)'"':
+                    case (byte)'\'':
+                        // start of string
+                        terminator = (byte)first;
+                        _value.Consume(1);
+                        break;
+                }
+                   
+                int end = BufferReader.FindNext(_value, terminator);
+                if (end < 0)
+                {
+                    Current = _value.ConsumeToEnd();
+                }
+                else
+                {
+                    Current = _value.ConsumeAsBuffer(end);
+                    _value.Consume(1); // drop the terminator itself;
+                }
+                return true;
+
+            }
+            public ReadOnlySequence<byte> Current { get; private set; }
+
+        }
         internal RedisChannel AsRedisChannel(byte[] channelPrefix, RedisChannel.PatternMode mode)
         {
             switch (Type)
@@ -202,6 +263,12 @@ internal ReadOnlySpan<RawResult> GetItems()
                 return new ReadOnlySpan<RawResult>(_itemsOversized, 0, _itemsCount);
             throw new InvalidOperationException();
         }
+        internal ReadOnlyMemory<RawResult> GetItemsMemory()
+        {
+            if (Type == ResultType.MultiBulk)
+                return new ReadOnlyMemory<RawResult>(_itemsOversized, 0, _itemsCount);
+            throw new InvalidOperationException();
+        }
 
         internal RedisKey[] GetItemsAsKeys()
         {

StackExchange.Redis/StackExchange/Redis/RedisDatabase.cs

@@ -698,13 +698,13 @@ protected override void WriteImpl(PhysicalConnection physical)
                 bool isCopy = (migrateOptions & MigrateOptions.Copy) != 0;
                 bool isReplace = (migrateOptions & MigrateOptions.Replace) != 0;
                 physical.WriteHeader(Command, 5 + (isCopy ? 1 : 0) + (isReplace ? 1 : 0));
-                physical.Write(toHost);
-                physical.Write(toPort);
+                physical.WriteBulkString(toHost);
+                physical.WriteBulkString(toPort);
                 physical.Write(Key);
-                physical.Write(toDatabase);
-                physical.Write(timeoutMilliseconds);
-                if (isCopy) physical.Write(RedisLiterals.COPY);
-                if (isReplace) physical.Write(RedisLiterals.REPLACE);
+                physical.WriteBulkString(toDatabase);
+                physical.WriteBulkString(timeoutMilliseconds);
+                if (isCopy) physical.WriteBulkString(RedisLiterals.COPY);
+                if (isReplace) physical.WriteBulkString(RedisLiterals.REPLACE);
             }
         }
 
@@ -3169,8 +3169,8 @@ public ScriptLoadMessage(CommandFlags flags, string script)
             protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 2);
-                physical.Write(RedisLiterals.LOAD);
-                physical.Write((RedisValue)Script);
+                physical.WriteBulkString(RedisLiterals.LOAD);
+                physical.WriteBulkString((RedisValue)Script);
             }
         }
 
@@ -3237,7 +3237,7 @@ protected override void WriteImpl(PhysicalConnection physical)
                     {   // recognises well-known types
                         var val = RedisValue.TryParse(arg);
                         if (val.IsNull && arg != null) throw new InvalidCastException($"Unable to parse value: '{arg}'");
-                        physical.Write(val);
+                        physical.WriteBulkString(val);
                     }
                 }
             }
@@ -3334,18 +3334,18 @@ protected override void WriteImpl(PhysicalConnection physical)
                 else if (asciiHash != null)
                 {
                     physical.WriteHeader(RedisCommand.EVALSHA, 2 + keys.Length + values.Length);
-                    physical.Write((RedisValue)asciiHash);
+                    physical.WriteBulkString((RedisValue)asciiHash);
                 }
                 else
                 {
                     physical.WriteHeader(RedisCommand.EVAL, 2 + keys.Length + values.Length);
-                    physical.Write((RedisValue)script);
+                    physical.WriteBulkString((RedisValue)script);
                 }
-                physical.Write(keys.Length);
+                physical.WriteBulkString(keys.Length);
                 for (int i = 0; i < keys.Length; i++)
                     physical.Write(keys[i]);
                 for (int i = 0; i < values.Length; i++)
-                    physical.Write(values[i]);
+                    physical.WriteBulkString(values[i]);
             }
         }
 
@@ -3386,11 +3386,11 @@ protected override void WriteImpl(PhysicalConnection physical)
             {
                 physical.WriteHeader(Command, 2 + keys.Length + values.Length);
                 physical.Write(Key);
-                physical.Write(keys.Length);
+                physical.WriteBulkString(keys.Length);
                 for (int i = 0; i < keys.Length; i++)
                     physical.Write(keys[i]);
                 for (int i = 0; i < values.Length; i++)
-                    physical.Write(values[i]);
+                    physical.WriteBulkString(values[i]);
             }
         }
 

StackExchange.Redis/StackExchange/Redis/RedisValue.cs

@@ -26,8 +26,32 @@ private RedisValue(long overlappedValue64, ReadOnlyMemory<byte> memory, object o
         }
 
         private readonly static object Sentinel_Integer = new object();
+
+        
+
         private readonly static object Sentinel_Raw = new object();
         private readonly static object Sentinel_Double = new object();
+
+        /// <summary>
+        /// Obtain this value as an object - to be used alongside Unbox
+        /// </summary>
+        public object Box()
+        {
+            var obj = _objectOrSentinel;
+            if (obj is null || obj is string || obj is byte[]) return obj;
+            return this;
+        }
+        /// <summary>
+        /// Parse this object as a value - to be used alongside Box
+        /// </summary>
+        public static RedisValue Unbox(object value)
+        {
+            if (value == null) return RedisValue.Null;
+            if (value is string s) return s;
+            if (value is byte[] b) return b;
+            return (RedisValue)value;
+        }
+
         /// <summary>
         /// Represents the string <c>""</c>
         /// </summary>
@@ -305,6 +329,20 @@ internal StorageType Type
             }
         }
 
+        /// <summary>
+        /// Get the size of this value in bytes
+        /// </summary>
+        public long Length()
+        {
+            switch(Type)
+            {
+                case StorageType.Null: return 0;
+                case StorageType.Raw: return _memory.Length;
+                case StorageType.String: return Encoding.UTF8.GetByteCount((string)_objectOrSentinel);
+                default: throw new InvalidOperationException("Unable to compute length of type: " + Type);
+            }
+        }
+
         /// <summary>
         /// Compare against a RedisValue for relative order
         /// </summary>

StackExchange.Redis/StackExchange/Redis/ResultProcessor.cs

@@ -364,7 +364,7 @@ protected override void WriteImpl(PhysicalConnection physical)
                     else
                     {
                         physical.WriteHeader(command, 1);
-                        physical.Write(value);
+                        physical.WriteBulkString(value);
                     }
                 }
             }

TestConsole/Program.cs

 using System;
-using System.Diagnostics;
+using System.Net;
+using System.Threading.Tasks;
+using StackExchange.Redis.Server;
 
-namespace TestConsole
+static class Program
 {
-    internal static class Program
+    static async Task Main()
     {
-        private static int Main()
+        using (var server = new MemoryCacheRedisServer(Console.Out))
         {
-            try
-            {
-                using (var obj = new BasicTest.RedisBenchmarks())
-                {
-                    var watch = Stopwatch.StartNew();
-                    obj.ExecuteIncrBy();
-                    watch.Stop();
-                    Console.WriteLine($"{watch.ElapsedMilliseconds}ms");
-                }
-                return 0;
-            }
-            catch (Exception ex)
-            {
-                Console.Error.WriteLine(ex.Message);
-                return -1;
-            }
+            server.Listen(new IPEndPoint(IPAddress.Loopback, 6378));
+            await server.Shutdown;
         }
     }
 }

TestConsole/TestConsole.csproj

@@ -12,6 +12,8 @@
 
   <ItemGroup>
     <ProjectReference Include="..\BasicTest\BasicTest.csproj" />
+    <ProjectReference Include="..\StackExchange.Redis.Server\StackExchange.Redis.Server.csproj" />
+    <ProjectReference Include="..\StackExchange.Redis.Tests\StackExchange.Redis.Tests.csproj" />
     <ProjectReference Include="..\StackExchange.Redis\StackExchange.Redis.csproj" />
   </ItemGroup>
 </Project>