Result box simplify (#1064)

* simplify the whole result-box/TaskCompletionSource mess with the realization that the TCS *can be* the result-box, and simple (non-TCS) boxes can be [ThreadStatic]

* rev Pipelines.Sockets.Unofficial (removed AwaitableLockToken)

StackExchange.Redis.sln

 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio 15
-VisualStudioVersion = 15.0.26823.1
+# Visual Studio Version 16
+VisualStudioVersion = 16.0.28531.58
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{3AD17044-6BFF-4750-9AC2-2CA466375F2A}"
 	ProjectSection(SolutionItems) = preProject

src/StackExchange.Redis/Condition.cs

@@ -309,7 +309,7 @@ public static Condition StringNotEqual(RedisKey key, RedisValue value)
 
         internal abstract void CheckCommands(CommandMap commandMap);
 
-        internal abstract IEnumerable<Message> CreateMessages(int db, ResultBox resultBox);
+        internal abstract IEnumerable<Message> CreateMessages(int db, IResultBox resultBox);
 
         internal abstract int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy);
         internal abstract bool TryValidate(in RawResult result, out bool value);
@@ -439,7 +439,7 @@ public override string ToString()
 
             internal override void CheckCommands(CommandMap commandMap) => commandMap.AssertAvailable(cmd);
 
-            internal override IEnumerable<Message> CreateMessages(int db, ResultBox resultBox)
+            internal override IEnumerable<Message> CreateMessages(int db, IResultBox resultBox)
             {
                 yield return Message.Create(db, CommandFlags.None, RedisCommand.WATCH, key);
 
@@ -519,7 +519,7 @@ public override string ToString()
 
             internal override void CheckCommands(CommandMap commandMap) => commandMap.AssertAvailable(cmd);
 
-            internal sealed override IEnumerable<Message> CreateMessages(int db, ResultBox resultBox)
+            internal sealed override IEnumerable<Message> CreateMessages(int db, IResultBox resultBox)
             {
                 yield return Message.Create(db, CommandFlags.None, RedisCommand.WATCH, key);
 
@@ -601,7 +601,7 @@ internal override void CheckCommands(CommandMap commandMap)
                 commandMap.AssertAvailable(RedisCommand.LINDEX);
             }
 
-            internal sealed override IEnumerable<Message> CreateMessages(int db, ResultBox resultBox)
+            internal sealed override IEnumerable<Message> CreateMessages(int db, IResultBox resultBox)
             {
                 yield return Message.Create(db, CommandFlags.None, RedisCommand.WATCH, key);
 
@@ -700,7 +700,7 @@ internal override void CheckCommands(CommandMap commandMap)
                 commandMap.AssertAvailable(cmd);
             }
 
-            internal sealed override IEnumerable<Message> CreateMessages(int db, ResultBox resultBox)
+            internal sealed override IEnumerable<Message> CreateMessages(int db, IResultBox resultBox)
             {
                 yield return Message.Create(db, CommandFlags.None, RedisCommand.WATCH, key);
 
@@ -763,7 +763,7 @@ public override string ToString()
 
             internal override void CheckCommands(CommandMap commandMap) => commandMap.AssertAvailable(RedisCommand.ZCOUNT);
 
-            internal sealed override IEnumerable<Message> CreateMessages(int db, ResultBox resultBox)
+            internal sealed override IEnumerable<Message> CreateMessages(int db, IResultBox resultBox)
             {
                 yield return Message.Create(db, CommandFlags.None, RedisCommand.WATCH, key);
 
@@ -799,14 +799,14 @@ public sealed class ConditionResult
     {
         internal readonly Condition Condition;
 
-        private ResultBox<bool> resultBox;
+        private IResultBox<bool> resultBox;
 
         private volatile bool wasSatisfied;
 
         internal ConditionResult(Condition condition)
         {
             Condition = condition;
-            resultBox = ResultBox<bool>.Get(condition);
+            resultBox = SimpleResultBox<bool>.Create();
         }
 
         /// <summary>
@@ -816,12 +816,12 @@ internal ConditionResult(Condition condition)
 
         internal IEnumerable<Message> CreateMessages(int db) => Condition.CreateMessages(db, resultBox);
 
-        internal ResultBox<bool> GetBox() { return resultBox; }
+        internal IResultBox<bool> GetBox() { return resultBox; }
         internal bool UnwrapBox()
         {
             if (resultBox != null)
             {
-                ResultBox<bool>.UnwrapAndRecycle(resultBox, false, out bool val, out Exception ex);
+                bool val = resultBox.GetResult(out var ex);
                 resultBox = null;
                 wasSatisfied = ex == null && val;
             }

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -1910,7 +1910,7 @@ internal ServerEndPoint SelectServer(RedisCommand command, CommandFlags flags, i
             return ServerSelectionStrategy.Select(command, key, flags);
         }
 
-        private bool PrepareToPushMessageToBridge<T>(Message message, ResultProcessor<T> processor, ResultBox<T> resultBox, ref ServerEndPoint server)
+        private bool PrepareToPushMessageToBridge<T>(Message message, ResultProcessor<T> processor, IResultBox<T> resultBox, ref ServerEndPoint server)
         {
             message.SetSource(processor, resultBox);
 
@@ -1964,12 +1964,12 @@ private bool PrepareToPushMessageToBridge<T>(Message message, ResultProcessor<T>
             Trace("No server or server unavailable - aborting: " + message);
             return false;
         }
-        private ValueTask<WriteResult> TryPushMessageToBridgeAsync<T>(Message message, ResultProcessor<T> processor, ResultBox<T> resultBox, ref ServerEndPoint server)
+        private ValueTask<WriteResult> TryPushMessageToBridgeAsync<T>(Message message, ResultProcessor<T> processor, IResultBox<T> resultBox, ref ServerEndPoint server)
             => PrepareToPushMessageToBridge(message, processor, resultBox, ref server) ? server.TryWriteAsync(message) : new ValueTask<WriteResult>(WriteResult.NoConnectionAvailable);
 
         [Obsolete("prefer async")]
 #pragma warning disable CS0618
-        private WriteResult TryPushMessageToBridgeSync<T>(Message message, ResultProcessor<T> processor, ResultBox<T> resultBox, ref ServerEndPoint server)
+        private WriteResult TryPushMessageToBridgeSync<T>(Message message, ResultProcessor<T> processor, IResultBox<T> resultBox, ref ServerEndPoint server)
             => PrepareToPushMessageToBridge(message, processor, resultBox, ref server) ? server.TryWriteSync(message) : WriteResult.NoConnectionAvailable;
 #pragma warning restore CS0618
 
@@ -2127,11 +2127,10 @@ internal Task<T> ExecuteAsyncImpl<T>(Message message, ResultProcessor<T> process
             }
 
             TaskCompletionSource<T> tcs = null;
-            ResultBox<T> source = null;
+            IResultBox<T> source = null;
             if (!message.IsFireAndForget)
             {
-                tcs = TaskSource.Create<T>(state);
-                source = ResultBox<T>.Get(tcs);
+                source = TaskResultBox<T>.Create(out tcs, state);
             }
             var write = TryPushMessageToBridgeAsync(message, processor, source, ref server);
             if (!write.IsCompletedSuccessfully) return ExecuteAsyncImpl_Awaited<T>(this, write, tcs, message, server);
@@ -2231,7 +2230,7 @@ internal T ExecuteSyncImpl<T>(Message message, ResultProcessor<T> processor, Ser
             }
             else
             {
-                var source = ResultBox<T>.Get(null);
+                var source = SimpleResultBox<T>.Get();
 
                 lock (source)
                 {
@@ -2256,7 +2255,7 @@ internal T ExecuteSyncImpl<T>(Message message, ResultProcessor<T> processor, Ser
                     }
                 }
                 // snapshot these so that we can recycle the box
-                ResultBox<T>.UnwrapAndRecycle(source, true, out T val, out Exception ex); // now that we aren't locking it...
+                var  val = source.GetResult(out var ex, canRecycle: true); // now that we aren't locking it...
                 if (ex != null) throw ex;
                 Trace(message + " received " + val);
                 return val;

src/StackExchange.Redis/Message.cs

@@ -75,7 +75,7 @@ internal abstract class Message : ICompletable
                                                        | CommandFlags.FireAndForget
                                                        | CommandFlags.NoRedirect
                                                        | CommandFlags.NoScriptCache;
-        private ResultBox resultBox;
+        private IResultBox resultBox;
 
         private ResultProcessor resultProcessor;
 
@@ -206,7 +206,7 @@ internal void SetScriptUnavailable()
         public bool IsFireAndForget => (Flags & CommandFlags.FireAndForget) != 0;
         public bool IsInternalCall => (Flags & InternalCallFlag) != 0;
 
-        public ResultBox ResultBox => resultBox;
+        public IResultBox ResultBox => resultBox;
 
         public abstract int ArgCount { get; } // note: over-estimate if necessary
 
@@ -617,7 +617,15 @@ internal virtual void SetExceptionAndComplete(Exception exception, PhysicalBridg
             bridge.CompleteSyncOrAsync(this);
         }
 
-        internal bool TrySetResult<T>(T value) => resultBox is ResultBox<T> typed && typed.TrySetResult(value);
+        internal bool TrySetResult<T>(T value)
+        {
+            if (resultBox is IResultBox<T> typed && !typed.IsFaulted)
+            {
+                typed.SetResult(value);
+                return true;
+            }
+            return false;
+        }
 
         internal void SetEnqueued(PhysicalConnection connection)
         {
@@ -711,14 +719,14 @@ internal void SetPreferSlave()
             Flags = (Flags & ~MaskMasterServerPreference) | CommandFlags.PreferSlave;
         }
 
-        internal void SetSource(ResultProcessor resultProcessor, ResultBox resultBox)
+        internal void SetSource(ResultProcessor resultProcessor, IResultBox resultBox)
         { // note order here reversed to prevent overload resolution errors
             if (resultBox != null && resultBox.IsAsync) SetNeedsTimeoutCheck();
             this.resultBox = resultBox;
             this.resultProcessor = resultProcessor;
         }
 
-        internal void SetSource<T>(ResultBox<T> resultBox, ResultProcessor<T> resultProcessor)
+        internal void SetSource<T>(IResultBox<T> resultBox, ResultProcessor<T> resultProcessor)
         {
             if (resultBox != null && resultBox.IsAsync) SetNeedsTimeoutCheck();
             this.resultBox = resultBox;

src/StackExchange.Redis/PhysicalBridge.cs

@@ -8,6 +8,7 @@
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using Pipelines.Sockets.Unofficial.Threading;
+using static Pipelines.Sockets.Unofficial.Threading.MutexSlim;
 using PendingSubscriptionState = global::StackExchange.Redis.ConnectionMultiplexer.Subscription.PendingSubscriptionState;
 
 namespace StackExchange.Redis
@@ -684,7 +685,7 @@ private WriteResult WriteMessageInsideLock(PhysicalConnection physical, Message
             }
         }
 
-        private async ValueTask<WriteResult> WriteMessageTakingDelayedWriteLockAsync(MutexSlim.AwaitableLockToken pendingLock, PhysicalConnection physical, Message message)
+        private async ValueTask<WriteResult> WriteMessageTakingDelayedWriteLockAsync(ValueTask<LockToken> pendingLock, PhysicalConnection physical, Message message)
         {
             try
             {
@@ -763,7 +764,7 @@ internal ValueTask<WriteResult> WriteMessageTakingWriteLockAsync(PhysicalConnect
             message.SetEnqueued(physical); // this also records the read/write stats at this point
 
             bool releaseLock = false;
-            MutexSlim.LockToken token = default;
+            LockToken token = default;
             try
             {
                 // try to acquire it synchronously
@@ -772,7 +773,7 @@ internal ValueTask<WriteResult> WriteMessageTakingWriteLockAsync(PhysicalConnect
                 if (!pending.IsCompletedSuccessfully) return WriteMessageTakingDelayedWriteLockAsync(pending, physical, message);
 
                 releaseLock = true;
-                token = pending.GetResult(); // we can't use "using" for this, because we might not want to kill it yet
+                token = pending.Result; // we can't use "using" for this, because we might not want to kill it yet
                 if (!token.Success) // (in particular, me might hand the lifetime to CompleteWriteAndReleaseLockAsync)
                 {
                     message.Cancel();
@@ -807,7 +808,7 @@ internal ValueTask<WriteResult> WriteMessageTakingWriteLockAsync(PhysicalConnect
             }
         }
 
-        private async ValueTask<WriteResult> CompleteWriteAndReleaseLockAsync(MutexSlim.LockToken lockToken, ValueTask<WriteResult> flush, Message message)
+        private async ValueTask<WriteResult> CompleteWriteAndReleaseLockAsync(LockToken lockToken, ValueTask<WriteResult> flush, Message message)
         {
             using (lockToken)
             {

src/StackExchange.Redis/RedisBatch.cs

@@ -76,10 +76,9 @@ internal override Task<T> ExecuteAsync<T>(Message message, ResultProcessor<T> pr
             }
             else
             {
-                var tcs = TaskSource.Create<T>(asyncState);
-                var source = ResultBox<T>.Get(tcs);
-                message.SetSource(source, processor);
+                var source = TaskResultBox<T>.Create(out var tcs, asyncState);
                 task = tcs.Task;
+                message.SetSource(source, processor);
             }
 
             // store it

src/StackExchange.Redis/RedisDatabase.cs

@@ -3714,7 +3714,7 @@ protected override SortedSetEntry[] Parse(in RawResult result)
         private class StringGetWithExpiryMessage : Message.CommandKeyBase, IMultiMessage
         {
             private readonly RedisCommand ttlCommand;
-            private ResultBox<TimeSpan?> box;
+            private IResultBox<TimeSpan?> box;
 
             public StringGetWithExpiryMessage(int db, CommandFlags flags, RedisCommand ttlCommand, in RedisKey key)
                 : base(db, flags, RedisCommand.GET, key)
@@ -3726,7 +3726,7 @@ public StringGetWithExpiryMessage(int db, CommandFlags flags, RedisCommand ttlCo
 
             public IEnumerable<Message> GetMessages(PhysicalConnection connection)
             {
-                box = ResultBox<TimeSpan?>.Get(null);
+                box = SimpleResultBox<TimeSpan?>.Create();
                 var ttl = Message.Create(Db, Flags, ttlCommand, Key);
                 var proc = ttlCommand == RedisCommand.PTTL ? ResultProcessor.TimeSpanFromMilliseconds : ResultProcessor.TimeSpanFromSeconds;
                 ttl.SetSource(proc, box);
@@ -3738,7 +3738,7 @@ public bool UnwrapValue(out TimeSpan? value, out Exception ex)
             {
                 if (box != null)
                 {
-                    ResultBox<TimeSpan?>.UnwrapAndRecycle(box, false, out value, out ex);
+                    value = box.GetResult(out ex);
                     box = null;
                     return ex == null;
                 }

src/StackExchange.Redis/RedisSubscriber.cs

@@ -259,12 +259,11 @@ private PendingSubscriptionState(object asyncState, RedisChannel channel, Subscr
                         : (channel.IsPatternBased ? RedisCommand.PUNSUBSCRIBE : RedisCommand.UNSUBSCRIBE);
                     var msg = Message.Create(-1, flags, cmd, channel);
                     if (internalCall) msg.SetInternalCall();
-                    var taskSource = TaskSource.Create<bool>(asyncState);
-                    var source = ResultBox<bool>.Get(taskSource);
+
+                    var source = TaskResultBox<bool>.Create(out _taskSource, asyncState);
                     msg.SetSource(ResultProcessor.TrackSubscriptions, source);
 
                     Subscription = subscription;
-                    _taskSource = taskSource;
                     Message = msg;
                     IsSlave = isSlave;
                 }

src/StackExchange.Redis/RedisTransaction.cs

@@ -73,15 +73,14 @@ internal override Task<T> ExecuteAsync<T>(Message message, ResultProcessor<T> pr
             }
             else
             {
-                var tcs = TaskSource.Create<T>(asyncState, TaskCreationOptions.RunContinuationsAsynchronously);
-                var source = ResultBox<T>.Get(tcs);
+                var source = TaskResultBox<T>.Create(out var tcs, asyncState, TaskCreationOptions.RunContinuationsAsynchronously);
                 message.SetSource(source, processor);
                 task = tcs.Task;
             }
 
             // prepare an outer-command that decorates that, but expects QUEUED
             var queued = new QueuedMessage(message);
-            var wasQueued = ResultBox<bool>.Get(null);
+            var wasQueued = SimpleResultBox<bool>.Create();
             queued.SetSource(wasQueued, QueuedProcessor.Default);
 
             // store it, and return the task of the *outer* command
@@ -100,7 +99,7 @@ internal override Task<T> ExecuteAsync<T>(Message message, ResultProcessor<T> pr
                         // think it should be!
                         var sel = PhysicalConnection.GetSelectDatabaseCommand(message.Db);
                         queued = new QueuedMessage(sel);
-                        wasQueued = ResultBox<bool>.Get(null);
+                        wasQueued = SimpleResultBox<bool>.Create();
                         queued.SetSource(wasQueued, QueuedProcessor.Default);
                         _pending.Add(queued);
                         break;
@@ -240,7 +239,7 @@ public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy)
 
             public IEnumerable<Message> GetMessages(PhysicalConnection connection)
             {
-                ResultBox lastBox = null;
+                IResultBox lastBox = null;
                 var bridge = connection.BridgeCouldBeNull;
                 if (bridge == null) throw new ObjectDisposedException(connection.ToString());
 
@@ -268,7 +267,7 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                             {
                                 // need to have locked them before sending them
                                 // to guarantee that we see the pulse
-                                ResultBox latestBox = conditions[i].GetBox();
+                                IResultBox latestBox = conditions[i].GetBox();
                                 Monitor.Enter(latestBox);
                                 if (lastBox != null) Monitor.Exit(lastBox);
                                 lastBox = latestBox;
@@ -328,7 +327,7 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                                 if (explicitCheckForQueued)
                                 {   // need to have locked them before sending them
                                     // to guarantee that we see the pulse
-                                    ResultBox thisBox = op.ResultBox;
+                                    IResultBox thisBox = op.ResultBox;
                                     if (thisBox != null)
                                     {
                                         Monitor.Enter(thisBox);

src/StackExchange.Redis/ResultBox.cs

@@ -4,146 +4,153 @@
 
 namespace StackExchange.Redis
 {
-    internal abstract class ResultBox
+    internal interface IResultBox
     {
-        protected Exception _exception;
-        public abstract bool IsAsync { get; }
-        public bool IsFaulted => _exception != null;
+        bool IsAsync { get; }
+        bool IsFaulted { get; }
+        void SetException(Exception ex);
+        bool TryComplete(bool isAsync);
+        void Cancel();
+    }
+    internal interface IResultBox<T> : IResultBox
+    {
+        T GetResult(out Exception ex, bool canRecycle = false);
+        void SetResult(T value);
+    }
 
-        public void SetException(Exception exception) => _exception = exception ?? s_cancelled;
+    internal abstract class SimpleResultBox : IResultBox
+    {
+        private volatile Exception _exception;
 
-        public abstract bool TryComplete(bool isAsync);
+        bool IResultBox.IsAsync => false;
+        bool IResultBox.IsFaulted => _exception != null;
+        void IResultBox.SetException(Exception exception) => _exception = exception ?? CancelledException;
+        void IResultBox.Cancel() => _exception = CancelledException;
 
-        public void Cancel() => _exception = s_cancelled;
+        bool IResultBox.TryComplete(bool isAsync)
+        {
+            lock (this)
+            { // tell the waiting thread that we're done
+                Monitor.PulseAll(this);
+            }
+            ConnectionMultiplexer.TraceWithoutContext("Pulsed", "Result");
+            return true;
+        }
 
         // in theory nobody should directly observe this; the only things
         // that call Cancel are transactions etc - TCS-based, and we detect
         // that and use TrySetCanceled instead
         // about any confusion in stack-trace
-        private static readonly Exception s_cancelled = new TaskCanceledException();
+        internal static readonly Exception CancelledException = new TaskCanceledException();
+
+        protected Exception Exception
+        {
+            get => _exception;
+            set => _exception = value;
+        }
     }
 
-    internal sealed class ResultBox<T> : ResultBox
+    internal sealed class SimpleResultBox<T> : SimpleResultBox, IResultBox<T>
     {
-        private static readonly ResultBox<T>[] store = new ResultBox<T>[64];
-        private object stateOrCompletionSource;
-        private int _usageCount;
-        private T value;
+        private SimpleResultBox() { }
+        private T _value;
 
-        public ResultBox(object stateOrCompletionSource)
-        {
-            this.stateOrCompletionSource = stateOrCompletionSource;
-            _usageCount = 1;
-        }
+        [ThreadStatic]
+        private static SimpleResultBox<T> _perThreadInstance;
 
-        public static ResultBox<T> Get(object stateOrCompletionSource)
+        public static IResultBox<T> Create() => new SimpleResultBox<T>();
+        public static IResultBox<T> Get() // includes recycled boxes; used from sync, so makes re-use easy
         {
-            ResultBox<T> found;
-            for (int i = 0; i < store.Length; i++)
-            {
-                if ((found = Interlocked.Exchange(ref store[i], null)) != null)
-                {
-                    found.Reset(stateOrCompletionSource);
-                    return found;
-                }
-            }
-
-            return new ResultBox<T>(stateOrCompletionSource);
+            var obj = _perThreadInstance ?? new SimpleResultBox<T>();
+            _perThreadInstance = null; // in case of oddness; only set back when recycled
+            return obj;
         }
+        void IResultBox<T>.SetResult(T value) => _value = value;
 
-        public static void UnwrapAndRecycle(ResultBox<T> box, bool recycle, out T value, out Exception exception)
+        T IResultBox<T>.GetResult(out Exception ex, bool canRecycle)
         {
-            if (box == null)
-            {
-                value = default(T);
-                exception = null;
-            }
-            else
+            var value = _value;
+            ex = Exception;
+            if (canRecycle)
             {
-                value = box.value;
-                exception = box._exception;
-                box.value = default(T);
-                box._exception = null;
-                if (recycle)
-                {
-                    var newCount = Interlocked.Decrement(ref box._usageCount);
-                    if (newCount != 0)
-                        throw new InvalidOperationException($"Result box count error: is {newCount} in UnwrapAndRecycle (should be 0)");
-
-                    // Clear state prior to recycling, so as not to root it
-                    box.stateOrCompletionSource = null;
-                    for (int i = 0; i < store.Length; i++)
-                    {
-                        if (Interlocked.CompareExchange(ref store[i], box, null) == null) return;
-                    }
-                }
+                Exception = null;
+                _value = default;
+                _perThreadInstance = this;
             }
+            return value;
         }
+    }
 
-        public void SetResult(T value)
-        {
-            this.value = value;
-        }
+    internal sealed class TaskResultBox<T> : TaskCompletionSource<T>, IResultBox<T>
+    {
+        // you might be asking "wait, doesn't the Task own these?", to which
+        // I say: no; we can't set *immediately* due to thread-theft etc, hence
+        // the fun TryComplete indirection - so we need somewhere to buffer them
+        private volatile Exception _exception;
+        private T _value;
+
+        private TaskResultBox(object asyncState, TaskCreationOptions creationOptions) : base(asyncState, creationOptions)
+        { }
+
+        bool IResultBox.IsAsync => true;
+
+        bool IResultBox.IsFaulted => _exception != null;
 
-        internal bool TrySetResult(T value)
+        void IResultBox.Cancel() => _exception = SimpleResultBox.CancelledException;
+
+        void IResultBox.SetException(Exception ex) => _exception = ex ?? SimpleResultBox.CancelledException;
+
+        void IResultBox<T>.SetResult(T value) => _value = value;
+
+        T IResultBox<T>.GetResult(out Exception ex, bool _)
         {
-            if (_exception != null) return false;
-            this.value = value;
-            return true;
+            ex = _exception;
+            return _value;
+            // nothing to do re recycle: TaskCompletionSource<T> cannot be recycled
         }
 
-        public override bool IsAsync => stateOrCompletionSource is TaskCompletionSource<T>;
-
-        public override bool TryComplete(bool isAsync)
+        bool IResultBox.TryComplete(bool isAsync)
         {
-            if (stateOrCompletionSource is TaskCompletionSource<T> tcs)
+            if (isAsync || (Task.CreationOptions & TaskCreationOptions.RunContinuationsAsynchronously) != 0)
             {
-                if (isAsync || (tcs.Task.CreationOptions & TaskCreationOptions.RunContinuationsAsynchronously) != 0)
+                // either on the async completion step, or the task is guarded
+                // againsts thread-stealing; complete it directly
+                // (note: RunContinuationsAsynchronously is only usable from NET46)
+                var val = _value;
+                var ex = _exception;
+
+                if (ex == null)
                 {
-                    // either on the async completion step, or the task is guarded
-                    // againsts thread-stealing; complete it directly
-                    // (note: RunContinuationsAsynchronously is only usable from NET46)
-                    UnwrapAndRecycle(this, true, out T val, out Exception ex);
-
-                    if (ex == null)
-                    {
-                        tcs.TrySetResult(val);
-                    }
-                    else
-                    {
-                        if (ex is TaskCanceledException) tcs.TrySetCanceled();
-                        else tcs.TrySetException(ex);
-                        // mark it as observed
-                        GC.KeepAlive(tcs.Task.Exception);
-                        GC.SuppressFinalize(tcs.Task);
-                    }
-                    return true;
+                    TrySetResult(val);
                 }
                 else
                 {
-                    // could be thread-stealing continuations; push to async to preserve the reader thread
-                    return false;
+                    if (ex is TaskCanceledException) TrySetCanceled();
+                    else TrySetException(ex);
+                    // mark any exception as observed
+                    var task = Task;
+                    GC.KeepAlive(task.Exception);
+                    GC.SuppressFinalize(task);
                 }
+                return true;
             }
             else
             {
-                lock (this)
-                { // tell the waiting thread that we're done
-                    Monitor.PulseAll(this);
-                }
-                ConnectionMultiplexer.TraceWithoutContext("Pulsed", "Result");
-                return true;
+                // could be thread-stealing continuations; push to async to preserve the reader thread
+                return false;
             }
         }
 
-        private void Reset(object stateOrCompletionSource)
+        public static IResultBox<T> Create(out TaskCompletionSource<T> source, object asyncState, TaskCreationOptions creationOptions = TaskCreationOptions.None)
         {
-            var newCount = Interlocked.Increment(ref _usageCount);
-            if (newCount != 1) throw new InvalidOperationException($"Result box count error: is {newCount} in Reset (should be 1)");
-            value = default(T);
-            _exception = null;
-
-            this.stateOrCompletionSource = stateOrCompletionSource;
+            // it might look a little odd to return the same object as two different things,
+            // but that's because it is serving two purposes, and I want to make it clear
+            // how it is being used in those 2 different ways; also, the *fact* that they
+            // are the same underlying object is an implementation detail that the rest of
+            // the code doesn't need to know about
+            var obj = new TaskResultBox<T>(asyncState, creationOptions);
+            source = obj;
+            return obj;
         }
     }
 }

src/StackExchange.Redis/ResultProcessor.cs

@@ -2032,7 +2032,7 @@ internal abstract class ResultProcessor<T> : ResultProcessor
         protected void SetResult(Message message, T value)
         {
             if (message == null) return;
-            var box = message.ResultBox as ResultBox<T>;
+            var box = message.ResultBox as IResultBox<T>;
             message.SetResponseReceived();
 
             box?.SetResult(value);

src/StackExchange.Redis/ServerEndPoint.cs

@@ -574,8 +574,7 @@ private static async Task<T> WriteDirectAsync_Awaited<T>(ServerEndPoint @this, M
 
         internal Task<T> WriteDirectAsync<T>(Message message, ResultProcessor<T> processor, object asyncState = null, PhysicalBridge bridge = null)
         {
-            var tcs = TaskSource.Create<T>(asyncState);
-            var source = ResultBox<T>.Get(tcs);
+            var source = TaskResultBox<T>.Create(out var tcs, asyncState);
             message.SetSource(processor, source);
             if (bridge == null) bridge = GetBridge(message.Command);
 

src/StackExchange.Redis/StackExchange.Redis.csproj

@@ -15,7 +15,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Pipelines.Sockets.Unofficial" Version="1.0.18" />
+    <PackageReference Include="Pipelines.Sockets.Unofficial" Version="1.0.20" />
     <PackageReference Include="System.Diagnostics.PerformanceCounter" Version="4.5.0" />
     <PackageReference Include="System.IO.Pipelines" Version="4.5.1" />
     <PackageReference Include="System.Threading.Channels" Version="4.5.0" />