Don't use QueueUserWorkItem to trigger write callbacks immediately

- In this case we are off the event loop, so we can invoke the callback directly.

Author: halter73

src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs

@@ -45,17 +45,7 @@ public void Write(ArraySegment<byte> buffer, Action<Exception, object> callback,
 
             KestrelTrace.Log.ConnectionWrite(0, buffer.Count);
 
-            var writeOp = new WriteOperation
-            {
-                Buffer = buffer
-            };
-
-            var callbackContext = new CallbackContext
-            {
-                Callback = callback,
-                State = state,
-                BytesToWrite = buffer.Count
-            };
+            bool triggerCallbackNow = false;
 
             lock (_lockObj)
             {
@@ -64,20 +54,22 @@ public void Write(ArraySegment<byte> buffer, Action<Exception, object> callback,
                     _nextWriteContext = new WriteContext(this);
                 }
 
-                _nextWriteContext.Operations.Enqueue(writeOp);
+                _nextWriteContext.Buffers.Enqueue(buffer);
                 _numBytesBuffered += buffer.Count;
 
                 // Complete the write task immediately if all previous write tasks have been completed,
                 // the buffers haven't grown too large, and the last write to the socket succeeded.
-                if (_lastWriteError == null &&
-                    _callbacksPending.Count == 0 &&
-                    _numBytesBuffered < _maxBytesBufferedBeforeThrottling)
-                {
-                    TriggerCallback(callbackContext);
-                }
-                else
+                triggerCallbackNow = _lastWriteError == null &&
+                                     _callbacksPending.Count == 0 &&
+                                     _numBytesBuffered < _maxBytesBufferedBeforeThrottling;
+                if (!triggerCallbackNow)
                 {
-                    _callbacksPending.Enqueue(callbackContext);
+                    _callbacksPending.Enqueue(new CallbackContext
+                    {
+                        Callback = callback,
+                        State = state,
+                        BytesToWrite = buffer.Count
+                    });
                 }
 
                 if (_writesPending < _maxPendingWrites)
@@ -86,6 +78,11 @@ public void Write(ArraySegment<byte> buffer, Action<Exception, object> callback,
                     _writesPending++;
                 }
             }
+
+            if (triggerCallbackNow)
+            {
+                callback(null, state);
+            }
         }
 
         private void ScheduleWrite()
@@ -118,19 +115,21 @@ private void WriteAllPending()
 
             try
             {
-                var buffers = new ArraySegment<byte>[writingContext.Operations.Count];
+                var buffers = new ArraySegment<byte>[writingContext.Buffers.Count];
 
                 var i = 0;
-                foreach (var writeOp in writingContext.Operations)
+                foreach (var buffer in writingContext.Buffers)
                 {
-                    buffers[i] = writeOp.Buffer;
-                    i++;
+                    buffers[i++] = buffer;
                 }
 
-                writingContext.WriteReq.Write(_socket, new ArraySegment<ArraySegment<byte>>(buffers), (r, status, error, state) =>
+                var writeReq = new UvWriteReq();
+                writeReq.Init(_thread.Loop);
+
+                writeReq.Write(_socket, new ArraySegment<ArraySegment<byte>>(buffers), (r, status, error, state) =>
                 {
                     var writtenContext = (WriteContext)state;
-                    writtenContext.Self.OnWriteCompleted(writtenContext.Operations, r, status, error);
+                    writtenContext.Self.OnWriteCompleted(writtenContext.Buffers, r, status, error);
                 }, writingContext);
             }
             catch
@@ -147,7 +146,7 @@ private void WriteAllPending()
         }
 
         // This is called on the libuv event loop
-        private void OnWriteCompleted(Queue<WriteOperation> completedWrites, UvWriteReq req, int status, Exception error)
+        private void OnWriteCompleted(Queue<ArraySegment<byte>> writtenBuffers, UvWriteReq req, int status, Exception error)
         {
             lock (_lockObj)
             {
@@ -162,16 +161,16 @@ private void OnWriteCompleted(Queue<WriteOperation> completedWrites, UvWriteReq
                     _writesPending--;
                 }
 
-                foreach (var writeOp in completedWrites)
+                foreach (var writeBuffer in writtenBuffers)
                 {
-                    _numBytesBuffered -= writeOp.Buffer.Count;
+                    _numBytesBuffered -= writeBuffer.Count;
                 }
 
                 var bytesLeftToBuffer = _maxBytesBufferedBeforeThrottling - _numBytesBuffered;
-                while (_callbacksPending.Count > 0 && _callbacksPending.Peek().BytesToWrite < bytesLeftToBuffer)
+                while (_callbacksPending.Count > 0 &&
+                       _callbacksPending.Peek().BytesToWrite < bytesLeftToBuffer)
                 {
-                    var context = _callbacksPending.Dequeue();
-                    TriggerCallback(context);
+                    TriggerCallback(_callbacksPending.Dequeue());
                 }
             }
 
@@ -196,27 +195,16 @@ private class CallbackContext
             public int BytesToWrite;
         }
 
-        private class WriteOperation
-        {
-            public ArraySegment<byte> Buffer;
-        }
-
         private class WriteContext
         {
             public WriteContext(SocketOutput self)
             {
                 Self = self;
-
-                WriteReq = new UvWriteReq();
-                WriteReq.Init(self._thread.Loop);
-
-                Operations = new Queue<WriteOperation>();
+                Buffers = new Queue<ArraySegment<byte>>();
             }
 
             public SocketOutput Self;
-
-            public UvWriteReq WriteReq;
-            public Queue<WriteOperation> Operations;
+            public Queue<ArraySegment<byte>> Buffers;
         }
     }
 }