Initialize queue, pin sizes, reuse sends

Author: benaadams

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

@@ -16,6 +16,8 @@ public class SocketOutput : ISocketOutput
     {
         private const int _maxPendingWrites = 3;
         private const int _maxBytesPreCompleted = 65536;
+        private const int _maxPooledWriteContexts = 16;
+        private const int _maxPooledBufferQueues = 16;
 
         private readonly KestrelThread _thread;
         private readonly UvStreamHandle _socket;
@@ -25,6 +27,7 @@ public class SocketOutput : ISocketOutput
 
         // This locks access to to all of the below fields
         private readonly object _lockObj = new object();
+        private bool _isDisposed; 
 
         // The number of write operations that have been scheduled so far
         // but have not completed.
@@ -34,6 +37,7 @@ public class SocketOutput : ISocketOutput
         private Exception _lastWriteError;
         private WriteContext _nextWriteContext;
         private readonly Queue<TaskCompletionSource<object>> _tasksPending;
+        private readonly Queue<WriteContext> _writeContexts;
 
         public SocketOutput(
             KestrelThread thread,
@@ -47,7 +51,8 @@ public SocketOutput(
             _connection = connection;
             _connectionId = connectionId;
             _log = log;
-            _tasksPending = new Queue<TaskCompletionSource<object>>();
+            _tasksPending = new Queue<TaskCompletionSource<object>>(16);
+            _writeContexts = new Queue<WriteContext>(_maxPooledWriteContexts);
         }
 
         public Task WriteAsync(
@@ -71,7 +76,14 @@ public Task WriteAsync(
             {
                 if (_nextWriteContext == null)
                 {
-                    _nextWriteContext = new WriteContext(this);
+                    if (_writeContexts.Count > 0)
+                    {
+                        _nextWriteContext = _writeContexts.Dequeue();
+                    }
+                    else
+                    {
+                        _nextWriteContext = new WriteContext(this);
+                    }
                 }
 
                 if (buffer.Array != null)
@@ -180,13 +192,13 @@ private void WriteAllPending()
         }
 
         // This is called on the libuv event loop
-        private void OnWriteCompleted(Queue<ArraySegment<byte>> writtenBuffers, int status, Exception error)
+        private void OnWriteCompleted(WriteContext write)
         {
-            _log.ConnectionWriteCallback(_connectionId, status);
+            var status = write.WriteStatus;
 
-            if (error != null)
+            if (write.WriteError != null)
             {
-                _lastWriteError = new IOException(error.Message, error);
+                _lastWriteError = new IOException(write.WriteError.Message, write.WriteError);
 
                 // Abort the connection for any failed write.
                 _connection.Abort();
@@ -203,7 +215,7 @@ private void OnWriteCompleted(Queue<ArraySegment<byte>> writtenBuffers, int stat
                     _writesPending--;
                 }
 
-                foreach (var writeBuffer in writtenBuffers)
+                foreach (var writeBuffer in write.Buffers)
                 {
                     // _numBytesPreCompleted can temporarily go negative in the event there are
                     // completed writes that we haven't triggered callbacks for yet.
@@ -230,16 +242,31 @@ private void OnWriteCompleted(Queue<ArraySegment<byte>> writtenBuffers, int stat
                     }
                     else
                     {
+                        var error = _lastWriteError;
                         // error is closure captured 
                         ThreadPool.QueueUserWorkItem(
-                            (o) => ((TaskCompletionSource<object>)o).SetException(_lastWriteError), 
+                            (o) => ((TaskCompletionSource<object>)o).SetException(error), 
                             tcs);
                     }
                 }
 
+                if (_writeContexts.Count < _maxPooledWriteContexts 
+                    && write.Buffers.Count <= _maxPooledBufferQueues
+                    && !_isDisposed)
+                {
+                    write.Reset();
+                    _writeContexts.Enqueue(write);
+                }
+                else
+                {
+                    write.Dispose();
+                }
+
                 // Now that the while loop has completed the following invariants should hold true:
                 Debug.Assert(_numBytesPreCompleted >= 0);
             }
+
+            _log.ConnectionWriteCallback(_connectionId, status);
         }
 
         void ISocketOutput.Write(ArraySegment<byte> buffer, bool immediate)
@@ -261,8 +288,24 @@ Task ISocketOutput.WriteAsync(ArraySegment<byte> buffer, bool immediate, Cancell
             return WriteAsync(buffer, immediate);
         }
 
-        private class WriteContext
+        private void Dispose()
         {
+            lock (_lockObj)
+            {
+                _isDisposed = true;
+
+                while (_writeContexts.Count > 0)
+                {
+                    _writeContexts.Dequeue().Dispose();
+                }
+            }
+
+        }
+
+        private class WriteContext : IDisposable
+        {
+            private const int BUFFER_COUNT = 4;
+
             public SocketOutput Self;
 
             public Queue<ArraySegment<byte>> Buffers;
@@ -272,12 +315,18 @@ private class WriteContext
             public int WriteStatus;
             public Exception WriteError;
 
+            private UvWriteReq _writeReq;
+            public ArraySegment<byte>[] _segments;
+
             public int ShutdownSendStatus;
 
             public WriteContext(SocketOutput self)
             {
                 Self = self;
-                Buffers = new Queue<ArraySegment<byte>>();
+                Buffers = new Queue<ArraySegment<byte>>(_maxPooledBufferQueues);
+                _segments = new ArraySegment<byte>[BUFFER_COUNT];
+                _writeReq = new UvWriteReq(Self._log);
+                _writeReq.Init(Self._thread.Loop);
             }
 
             /// <summary>
@@ -291,19 +340,24 @@ public void DoWriteIfNeeded()
                     return;
                 }
 
-                var buffers = new ArraySegment<byte>[Buffers.Count];
+                ArraySegment<byte>[] segments;
+                if (Buffers.Count > BUFFER_COUNT)
+                {
+                    segments = new ArraySegment<byte>[Buffers.Count];
+                }
+                else
+                {
+                    segments = _segments;
+                }
 
                 var i = 0;
                 foreach (var buffer in Buffers)
                 {
-                    buffers[i++] = buffer;
+                    segments[i++] = buffer;
                 }
-
-                var writeReq = new UvWriteReq(Self._log);
-                writeReq.Init(Self._thread.Loop);
-                writeReq.Write(Self._socket, new ArraySegment<ArraySegment<byte>>(buffers), (_writeReq, status, error, state) =>
+                
+                _writeReq.Write(Self._socket, new ArraySegment<ArraySegment<byte>>(segments, 0, Buffers.Count), (_writeReq, status, error, state) =>
                 {
-                    _writeReq.Dispose();
                     var _this = (WriteContext)state;
                     _this.WriteStatus = status;
                     _this.WriteError = error;
@@ -330,7 +384,7 @@ public void DoShutdownIfNeeded()
                     var _this = (WriteContext)state;
                     _this.ShutdownSendStatus = status;
 
-                    _this.Self._log.ConnectionWroteFin(Self._connectionId, status);
+                    _this.Self._log.ConnectionWroteFin(_this.Self._connectionId, status);
 
                     _this.DoDisconnectIfNeeded();
                 }, this);
@@ -341,8 +395,14 @@ public void DoShutdownIfNeeded()
             /// </summary>
             public void DoDisconnectIfNeeded()
             {
-                if (SocketDisconnect == false || Self._socket.IsClosed)
+                if (SocketDisconnect == false)
+                {
+                    Complete();
+                    return;
+                }
+                else if (Self._socket.IsClosed)
                 {
+                    Self.Dispose();
                     Complete();
                     return;
                 }
@@ -354,7 +414,28 @@ public void DoDisconnectIfNeeded()
 
             public void Complete()
             {
-                Self.OnWriteCompleted(Buffers, WriteStatus, WriteError);
+                Self.OnWriteCompleted(this);
+            }
+
+            public void Reset()
+            {
+                Buffers.Clear();
+                SocketDisconnect = false;
+                SocketShutdownSend = false;
+                WriteStatus = 0;
+                WriteError = null;
+                ShutdownSendStatus = 0;
+
+                var segments = _segments;
+                for (var i = 0; i < segments.Length; i++)
+                {
+                    segments[i] = default(ArraySegment<byte>);
+                }
+            }
+
+            public void Dispose()
+            {
+                _writeReq.Dispose();
             }
         }
     }

src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs

@@ -24,10 +24,10 @@ public class KestrelThread
         private Thread _thread;
         private UvLoopHandle _loop;
         private UvAsyncHandle _post;
-        private Queue<Work> _workAdding = new Queue<Work>();
-        private Queue<Work> _workRunning = new Queue<Work>();
-        private Queue<CloseHandle> _closeHandleAdding = new Queue<CloseHandle>();
-        private Queue<CloseHandle> _closeHandleRunning = new Queue<CloseHandle>();
+        private Queue<Work> _workAdding = new Queue<Work>(1024);
+        private Queue<Work> _workRunning = new Queue<Work>(1024);
+        private Queue<CloseHandle> _closeHandleAdding = new Queue<CloseHandle>(256);
+        private Queue<CloseHandle> _closeHandleRunning = new Queue<CloseHandle>(256);
         private object _workSync = new Object();
         private bool _stopImmediate = false;
         private bool _initCompleted = false;

src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs

@@ -14,15 +14,15 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
     /// </summary>
     public class UvWriteReq : UvRequest
     {
-        private readonly static Libuv.uv_write_cb _uv_write_cb = UvWriteCb;
+        private readonly static Libuv.uv_write_cb _uv_write_cb = (IntPtr ptr, int status) => UvWriteCb(ptr, status); 
 
         private IntPtr _bufs;
 
         private Action<UvWriteReq, int, Exception, object> _callback;
         private object _state;
         private const int BUFFER_COUNT = 4;
 
-        private List<GCHandle> _pins = new List<GCHandle>();
+        private List<GCHandle> _pins = new List<GCHandle>(BUFFER_COUNT + 1);
 
         public UvWriteReq(IKestrelTrace logger) : base(logger)
         {