diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs index 906faa986..4593dbc70 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs @@ -35,8 +35,8 @@ public Connection(ListenerContext context, UvStreamHandle socket) : base(context _connectionId = Interlocked.Increment(ref _lastConnectionId); - _rawSocketInput = new SocketInput(Memory2); - _rawSocketOutput = new SocketOutput(Thread, _socket, _connectionId, Log); + _rawSocketInput = new SocketInput(InputMemory); + _rawSocketOutput = new SocketOutput(OutputMemory, Thread, _socket, _connectionId, Log); } public void Start() @@ -89,7 +89,7 @@ public void Start() private void ApplyConnectionFilter() { - var filteredStreamAdapter = new FilteredStreamAdapter(_filterContext.Connection, Memory2, Log); + var filteredStreamAdapter = new FilteredStreamAdapter(_filterContext.Connection, InputMemory, Log); SocketInput = filteredStreamAdapter.SocketInput; SocketOutput = filteredStreamAdapter.SocketOutput; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs index 13546271e..ddd8578cc 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs @@ -19,6 +19,9 @@ namespace Microsoft.AspNet.Server.Kestrel.Http { public partial class Frame : FrameContext, IFrameControl { + private const string _endLine = "\r\n"; + private const string _endDelimiter = ": "; + private static readonly Encoding _ascii = Encoding.ASCII; private static readonly ArraySegment _endChunkBytes = CreateAsciiByteArraySegment("\r\n"); private static readonly ArraySegment _endChunkedResponseBytes = CreateAsciiByteArraySegment("0\r\n\r\n"); @@ -26,9 +29,10 @@ public partial class Frame : FrameContext, IFrameControl private static readonly ArraySegment _emptyData = new ArraySegment(new byte[0]); private static readonly byte[] _hex = Encoding.ASCII.GetBytes("0123456789abcdef"); - private readonly object _onStartingSync = new Object(); - private readonly object _onCompletedSync = new Object(); + private readonly object _onStartingSync = new object(); + private readonly object _onCompletedSync = new object(); private readonly FrameRequestHeaders _requestHeaders = new FrameRequestHeaders(); + private readonly byte[] _nullBuffer = new byte[4096]; private readonly FrameResponseHeaders _responseHeaders = new FrameResponseHeaders(); private List, object>> _onStarting; @@ -197,8 +201,10 @@ public async Task RequestProcessingAsync() await ProduceEnd(); - // Finish reading the request body in case the app did not. - await RequestBody.CopyToAsync(Stream.Null); + while (await RequestBody.ReadAsync(_nullBuffer, 0, _nullBuffer.Length) != 0) + { + // Finish reading the request body in case the app did not. + } } terminated = !_keepAlive; @@ -492,8 +498,7 @@ private Tuple, IDisposable> CreateResponseHeader( writer.Write(HttpVersion); writer.Write(' '); writer.Write(status); - writer.Write('\r'); - writer.Write('\n'); + writer.Write(_endLine); var hasConnection = false; var hasTransferEncoding = false; @@ -521,11 +526,9 @@ private Tuple, IDisposable> CreateResponseHeader( foreach (var value in header.Value) { writer.Write(header.Key); - writer.Write(':'); - writer.Write(' '); + writer.Write(_endDelimiter); writer.Write(value); - writer.Write('\r'); - writer.Write('\n'); + writer.Write(_endLine); if (isConnection && value.IndexOf("close", StringComparison.OrdinalIgnoreCase) != -1) { @@ -572,8 +575,7 @@ private Tuple, IDisposable> CreateResponseHeader( } else { - writer.Write('\r'); - writer.Write('\n'); + writer.Write(_endLine); } writer.Flush(); return new Tuple, IDisposable>(writer.Buffer, writer); @@ -751,7 +753,7 @@ public static bool TakeMessageHeaders(SocketInput input, FrameRequestHeaders req var value = beginValue.GetString(endValue); if (wrapping) { - value = value.Replace("\r\n", " "); + value = value.Replace(_endLine, " "); } consumed = scan; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerContext.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerContext.cs index 79a096d4d..88e1dc6af 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerContext.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerContext.cs @@ -11,13 +11,15 @@ public class ListenerContext : ServiceContext { public ListenerContext() { - Memory2 = new MemoryPool2(); + InputMemory = new MemoryPool2(); + OutputMemory = new MemoryPool2(); } public ListenerContext(ServiceContext serviceContext) : base(serviceContext) { - Memory2 = new MemoryPool2(); + InputMemory = new MemoryPool2(); + OutputMemory = new MemoryPool2(); } public ListenerContext(ListenerContext listenerContext) @@ -26,7 +28,8 @@ public ListenerContext(ListenerContext listenerContext) ServerAddress = listenerContext.ServerAddress; Thread = listenerContext.Thread; Application = listenerContext.Application; - Memory2 = listenerContext.Memory2; + InputMemory = listenerContext.InputMemory; + OutputMemory = listenerContext.OutputMemory; Log = listenerContext.Log; } @@ -36,6 +39,8 @@ public ListenerContext(ListenerContext listenerContext) public Func Application { get; set; } - public MemoryPool2 Memory2 { get; set; } + public MemoryPool2 InputMemory { get; set; } + + public MemoryPool2 OutputMemory { get; set; } } } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs index 41aeeb3bf..0ed0f366b 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs @@ -21,7 +21,7 @@ abstract public class ListenerPrimary : Listener // this message is passed to write2 because it must be non-zero-length, // but it has no other functional significance - private readonly ArraySegment> _dummyMessage = new ArraySegment>(new[] { new ArraySegment(new byte[] { 1, 2, 3, 4 }) }); + private readonly byte[] _dummyBuffer = new byte[] { 1, 2, 3, 4 }; protected ListenerPrimary(ServiceContext serviceContext) : base(serviceContext) { @@ -79,14 +79,17 @@ protected override void DispatchConnection(UvStreamHandle socket) } else { + var msg = MemoryPoolBlock2.Create(new ArraySegment(_dummyBuffer), IntPtr.Zero, null, null); + msg.End = msg.Start + _dummyBuffer.Length; + var dispatchPipe = _dispatchPipes[index]; var write = new UvWriteReq(Log); write.Init(Thread.Loop); write.Write2( dispatchPipe, - _dummyMessage, + new ArraySegment(new[] { msg }), socket, - (write2, status, error, state) => + (write2, status, error, bytesWritten, state) => { write2.Dispose(); ((UvStreamHandle)state).Dispose(); diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPool.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPool.cs index 0fc5c390e..eb3beed5b 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPool.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPool.cs @@ -2,7 +2,7 @@ // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information. using System; -using System.Collections.Generic; +using System.Collections.Concurrent; namespace Microsoft.AspNet.Server.Kestrel.Http { @@ -95,29 +95,23 @@ public void FreeSegment(ArraySegment segment) class Pool { - private readonly Stack _stack = new Stack(); - private readonly object _sync = new object(); + private readonly ConcurrentQueue _queue = new ConcurrentQueue(); public T[] Alloc(int size) { - lock (_sync) + T[] item; + if (_queue.TryDequeue(out item)) { - if (_stack.Count != 0) - { - return _stack.Pop(); - } + return item; } return new T[size]; } public void Free(T[] value, int limit) { - lock (_sync) + if (_queue.Count < limit) { - if (_stack.Count < limit) - { - _stack.Push(value); - } + _queue.Enqueue(value); } } } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPoolTextWriter.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPoolTextWriter.cs index 2f549dc15..647e27d63 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPoolTextWriter.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/MemoryPoolTextWriter.cs @@ -7,7 +7,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http { - public class MemoryPoolTextWriter : TextWriter + public class MemoryPoolTextWriter : IDisposable { private readonly IMemoryPool _memory; @@ -39,7 +39,7 @@ public ArraySegment Buffer } } - public override Encoding Encoding + public Encoding Encoding { get { @@ -47,27 +47,17 @@ public override Encoding Encoding } } - protected override void Dispose(bool disposing) + public void Dispose() { - try + if (_textArray != null) { - if (disposing) - { - if (_textArray != null) - { - _memory.FreeChar(_textArray); - _textArray = null; - } - if (_dataArray != null) - { - _memory.FreeByte(_dataArray); - _dataArray = null; - } - } + _memory.FreeChar(_textArray); + _textArray = null; } - finally + if (_dataArray != null) { - base.Dispose(disposing); + _memory.FreeByte(_dataArray); + _dataArray = null; } } @@ -107,7 +97,7 @@ private void Grow(int minimumAvailable) _dataArray = newArray; } - public override void Write(char value) + public void Write(char value) { if (_textLength == _textEnd) { @@ -121,7 +111,7 @@ public override void Write(char value) _textArray[_textEnd++] = value; } - public override void Write(string value) + public void Write(string value) { var sourceIndex = 0; var sourceLength = value.Length; @@ -144,7 +134,7 @@ public override void Write(string value) } } - public override void Flush() + public void Flush() { while (_textBegin != _textEnd) { diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs index d0d218ac5..0807ed410 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs @@ -2,8 +2,8 @@ // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information. using System; +using System.Collections.Concurrent; using System.Collections.Generic; -using System.Diagnostics; using System.Threading; using System.Threading.Tasks; using Microsoft.AspNet.Server.Kestrel.Infrastructure; @@ -13,104 +13,215 @@ namespace Microsoft.AspNet.Server.Kestrel.Http { public class SocketOutput : ISocketOutput { - private const int _maxPendingWrites = 3; - private const int _maxBytesPreCompleted = 65536; + internal const int MaxBytesPreCompleted = 3 * (MemoryPool2.NativeBlockSize * UvWriteReq.BUFFER_COUNT) - 1; + + private static MemoryPoolBlock2[] _emptyBlocks = new MemoryPoolBlock2[0]; private readonly KestrelThread _thread; private readonly UvStreamHandle _socket; private readonly long _connectionId; private readonly IKestrelTrace _log; - // This locks access to to all of the below fields - private readonly object _lockObj = new object(); + private readonly MemoryPool2 _memory; + + private WriteBlock _currentWriteBlock; + private ConcurrentQueue _memoryBlocks; - // The number of write operations that have been scheduled so far - // but have not completed. - private int _writesPending = 0; + private long _bytesQueued; + private long _bytesWritten; private int _numBytesPreCompleted = 0; private Exception _lastWriteError; - private WriteContext _nextWriteContext; - private readonly Queue _callbacksPending; + private readonly ConcurrentQueue _callbacksPending; + + public int ShutdownSendStatus; - public SocketOutput(KestrelThread thread, UvStreamHandle socket, long connectionId, IKestrelTrace log) + private int pendingWriteBitFlag = 0; + private bool SocketDisconnected; + private bool SocketShutdownSent; + + public SocketOutput(MemoryPool2 memory, KestrelThread thread, UvStreamHandle socket, long connectionId, IKestrelTrace log) { _thread = thread; _socket = socket; _connectionId = connectionId; _log = log; - _callbacksPending = new Queue(); + _callbacksPending = new ConcurrentQueue(); + _memory = memory; + _memoryBlocks = new ConcurrentQueue(); } - public void Write( + internal void Write( ArraySegment buffer, - Action callback, + Action callback, object state, bool immediate = true, bool socketShutdownSend = false, bool socketDisconnect = false) { - //TODO: need buffering that works - if (buffer.Array != null) - { - var copy = new byte[buffer.Count]; - Array.Copy(buffer.Array, buffer.Offset, copy, 0, buffer.Count); - buffer = new ArraySegment(copy); - _log.ConnectionWrite(_connectionId, buffer.Count); - } + bool triggerCallbackNow; + var queuedBytes = _bytesQueued; - bool triggerCallbackNow = false; + var inputLength = buffer.Array != null ? buffer.Count : 0; + MemoryPoolBlock2 memoryBlock; - lock (_lockObj) + if (inputLength > 0) { - if (_nextWriteContext == null) + memoryBlock = Interlocked.Exchange(ref _currentWriteBlock.Block, null); + + _log.ConnectionWrite(_connectionId, inputLength); + + int blockRemaining = memoryBlock != null ? memoryBlock.Data.Count - (memoryBlock.End - memoryBlock.Start) : 0; + + var remaining = inputLength; + var offset = buffer.Offset; + + while (remaining > 0) { - _nextWriteContext = new WriteContext(this); + if (memoryBlock == null) + { + memoryBlock = _memory.Lease(MemoryPool2.NativeBlockSize); + blockRemaining = memoryBlock.Data.Count; + } + + var copyAmount = blockRemaining >= remaining ? remaining : blockRemaining; + Buffer.BlockCopy(buffer.Array, offset, memoryBlock.Array, memoryBlock.End, copyAmount); + + remaining -= copyAmount; + blockRemaining -= copyAmount; + memoryBlock.End += copyAmount; + offset += copyAmount; + + if (blockRemaining == 0) + { + _memoryBlocks.Enqueue(new WriteBlock() { Block = memoryBlock }); + memoryBlock = null; + } } + + Interlocked.Exchange(ref _currentWriteBlock.Block, memoryBlock); - if (buffer.Array != null) + if (immediate) { - _nextWriteContext.Buffers.Enqueue(buffer); + _currentWriteBlock.SocketDisconnect |= socketDisconnect; + _currentWriteBlock.SocketShutdownSend |= socketShutdownSend; } - if (socketShutdownSend) + } + + if (immediate) + { + SendBufferedData(); + } + + lock (_callbacksPending) + { + if (_lastWriteError == null && + _callbacksPending.Count == 0 && + inputLength + _numBytesPreCompleted <= MaxBytesPreCompleted) { - _nextWriteContext.SocketShutdownSend = true; + triggerCallbackNow = true; } - if (socketDisconnect) + else { - _nextWriteContext.SocketDisconnect = true; + triggerCallbackNow = false; } - // 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. - triggerCallbackNow = _lastWriteError == null && - _callbacksPending.Count == 0 && - _numBytesPreCompleted + buffer.Count <= _maxBytesPreCompleted; - if (triggerCallbackNow) + } + + if (triggerCallbackNow) + { + callback(null, state, 0, true); + } + else + { + _callbacksPending.Enqueue(new CallbackContext { - _numBytesPreCompleted += buffer.Count; - } - else + Callback = callback, + State = state, + BytesWrittenThreshold = queuedBytes + }); + } + + Interlocked.Add(ref _numBytesPreCompleted, inputLength); + } + + private WriteContext GetContext() + { + MemoryPoolBlock2[] data = null; + + var count = 0; + var dataLength = 0; + + bool socketDisconnect = false; + bool socketShutdownSend = false; + + WriteBlock writeBlock; + while (_memoryBlocks.TryDequeue(out writeBlock)) + { + var block = writeBlock.Block; + if (block != null) { - _callbacksPending.Enqueue(new CallbackContext + if (data == null) { - Callback = callback, - State = state, - BytesToWrite = buffer.Count - }); + data = new MemoryPoolBlock2[UvWriteReq.BUFFER_COUNT]; + } + var length = block.End - block.Start; + data[count] = block; + dataLength += length; + count++; } - if (_writesPending < _maxPendingWrites && immediate) + socketDisconnect |= writeBlock.SocketDisconnect; + socketShutdownSend |= writeBlock.SocketShutdownSend; + + if (count == UvWriteReq.BUFFER_COUNT) { - ScheduleWrite(); - _writesPending++; + break; } } - // Make sure we call user code outside of the lock. - if (triggerCallbackNow) + if (count < UvWriteReq.BUFFER_COUNT) { - // callback(error, state, calledInline) - callback(null, state, true); + var block = Interlocked.Exchange(ref _currentWriteBlock.Block, null); ; + + if (block != null) + { + if (data == null) + { + data = new MemoryPoolBlock2[UvWriteReq.BUFFER_COUNT]; + } + var length = block.End - block.Start; + data[count] = block; + dataLength += length; + count++; + } + else + { + socketDisconnect = _currentWriteBlock.SocketDisconnect; + socketShutdownSend = _currentWriteBlock.SocketShutdownSend; + } + } + + if (count == 0 && !socketDisconnect && !socketShutdownSend) + { + return null; + } + + Interlocked.Add(ref _bytesQueued, dataLength); + + return new WriteContext() + { + Data = new ArraySegment(data ?? _emptyBlocks, 0, count), + Output = this, + SocketDisconnect = socketDisconnect, + SocketShutdownSend = socketShutdownSend + }; + } + + private void SendBufferedData() + { + if (Interlocked.CompareExchange(ref pendingWriteBitFlag, 1, 0) == 0) + { + _thread.Post(so => WriteContext.DoWriteIfNeeded(so), this); } } @@ -119,13 +230,13 @@ public void End(ProduceEndType endType) switch (endType) { case ProduceEndType.SocketShutdownSend: - Write(default(ArraySegment), (error, state, calledInline) => { }, null, + Write(default(ArraySegment), (error, state, status, calledInline) => { }, null, immediate: true, socketShutdownSend: true, socketDisconnect: false); break; case ProduceEndType.SocketDisconnect: - Write(default(ArraySegment), (error, state, calledInline) => { }, null, + Write(default(ArraySegment), (error, state, status, calledInline) => { }, null, immediate: true, socketShutdownSend: false, socketDisconnect: true); @@ -133,90 +244,21 @@ public void End(ProduceEndType endType) } } - private void ScheduleWrite() - { - _thread.Post(_this => _this.WriteAllPending(), this); - } - - // This is called on the libuv event loop - private void WriteAllPending() - { - WriteContext writingContext; - - lock (_lockObj) - { - if (_nextWriteContext != null) - { - writingContext = _nextWriteContext; - _nextWriteContext = null; - } - else - { - _writesPending--; - return; - } - } - - try - { - writingContext.DoWriteIfNeeded(); - } - catch - { - lock (_lockObj) - { - // Lock instead of using Interlocked.Decrement so _writesSending - // doesn't change in the middle of executing other synchronized code. - _writesPending--; - } - - throw; - } - } - // This is called on the libuv event loop - private void OnWriteCompleted(Queue> writtenBuffers, int status, Exception error) + private void OnWriteCompleted(int status, Exception error) { - _log.ConnectionWriteCallback(_connectionId, status); - lock (_lockObj) + CallbackContext callbackContext; + while (_callbacksPending.TryPeek(out callbackContext) && callbackContext.BytesWrittenThreshold <= _bytesWritten) { _lastWriteError = error; + _log.ConnectionWriteCallback(_connectionId, status); - if (_nextWriteContext != null) - { - ScheduleWrite(); - } - else - { - _writesPending--; - } - - foreach (var writeBuffer in writtenBuffers) + if (_callbacksPending.TryDequeue(out callbackContext)) { - // _numBytesPreCompleted can temporarily go negative in the event there are - // completed writes that we haven't triggered callbacks for yet. - _numBytesPreCompleted -= writeBuffer.Count; - } - - - // bytesLeftToBuffer can be greater than _maxBytesPreCompleted - // This allows large writes to complete once they've actually finished. - var bytesLeftToBuffer = _maxBytesPreCompleted - _numBytesPreCompleted; - while (_callbacksPending.Count > 0 && - _callbacksPending.Peek().BytesToWrite <= bytesLeftToBuffer) - { - var callbackContext = _callbacksPending.Dequeue(); - - _numBytesPreCompleted += callbackContext.BytesToWrite; - // callback(error, state, calledInline) - callbackContext.Callback(_lastWriteError, callbackContext.State, false); + callbackContext.Callback(error, callbackContext.State, status, false); } - - // Now that the while loop has completed the following invariants should hold true: - Debug.Assert(_numBytesPreCompleted >= 0); - Debug.Assert(_numBytesPreCompleted <= _maxBytesPreCompleted); } } @@ -228,9 +270,10 @@ void ISocketOutput.Write(ArraySegment buffer, bool immediate) // to be a subsequent immediate==true call which will go down the following code-path Write( buffer, - (error, state, calledInline) => { }, + (error, state, status, calledInline) => { }, null, immediate: false); + return; } @@ -239,15 +282,16 @@ void ISocketOutput.Write(ArraySegment buffer, bool immediate) Write( buffer, - (error, state, calledInline) => + (error, state, status, calledInline) => { + var cs = (TaskCompletionSource)state; if (error != null) { - tcs.SetException(error); + cs.SetException(error); } else { - tcs.SetResult(0); + cs.SetResult(0); } }, tcs, @@ -267,9 +311,10 @@ Task ISocketOutput.WriteAsync(ArraySegment buffer, bool immediate, Cancell // to be a subsequent immediate==true call which will go down the following code-path Write( buffer, - (error, state, calledInline) => { }, + (error, state, status, calledInline) => { }, null, immediate: false); + return TaskUtilities.CompletedTask; } @@ -278,33 +323,50 @@ Task ISocketOutput.WriteAsync(ArraySegment buffer, bool immediate, Cancell Write( buffer, - (error, state, calledInline) => + (error, state, status, calledInline) => { - if (!calledInline) + if (status < 0) { - ThreadPool.QueueUserWorkItem(state2 => + if (!calledInline) + { + ThreadPool.QueueUserWorkItem((state2) => + { + var tcs2 = (TaskCompletionSource)state2; + if (error != null) + { + tcs2.SetException(error); + } + else + { + tcs2.SetResult(0); + } + }, tcs); + } + else { - var tcs2 = (TaskCompletionSource)state2; if (error != null) { - tcs2.SetException(error); + tcs.SetException(error); } else { - tcs2.SetResult(0); + tcs.SetResult(0); } - }, state); + } } else { - var tcs2 = (TaskCompletionSource)state; - if (error != null) + if (!calledInline) { - tcs2.SetException(error); + ThreadPool.QueueUserWorkItem((state2) => + { + var tcs2 = (TaskCompletionSource)state2; + tcs.SetResult(0); + }, tcs); } else { - tcs2.SetResult(0); + tcs.SetResult(0); } } }, @@ -314,118 +376,121 @@ Task ISocketOutput.WriteAsync(ArraySegment buffer, bool immediate, Cancell return tcs.Task; } - private class CallbackContext - { - // callback(error, state, calledInline) - public Action Callback; - public object State; - public int BytesToWrite; - } - private class WriteContext { - public SocketOutput Self; - - public Queue> Buffers; + public ArraySegment Data; + public SocketOutput Output; public bool SocketShutdownSend; public bool SocketDisconnect; - public int WriteStatus; - public Exception WriteError; - - public int ShutdownSendStatus; - - public WriteContext(SocketOutput self) - { - Self = self; - Buffers = new Queue>(); - } - - /// - /// Perform any actions needed by this work item. The individual tasks are non-blocking and - /// will continue through to each other in order. - /// - public void Execute() - { - DoWriteIfNeeded(); - } + private int Status; + private Exception Error; /// /// First step: initiate async write if needed, otherwise go to next step /// - public void DoWriteIfNeeded() + public static void DoWriteIfNeeded(SocketOutput socketOutput) { - if (Buffers.Count == 0 || Self._socket.IsClosed) + socketOutput.pendingWriteBitFlag = 0; + Interlocked.MemoryBarrier(); + + WriteContext context; + while ((context = socketOutput.GetContext()) != null) { - DoShutdownIfNeeded(); - return; + var data = context.Data; + + if (data.Count == 0 || socketOutput._socket.IsClosed) + { + DoShutdownIfNeeded(context); + return; + } + var writeReq = new UvWriteReq(socketOutput._log); + writeReq.Init(socketOutput._thread.Loop); + writeReq.Write(socketOutput._socket, + data, + (_writeReq, status, error, bytesWritten, state) => WriteCallback(_writeReq, status, error, bytesWritten, state), + context); } + } - var buffers = new ArraySegment[Buffers.Count]; + private static void WriteCallback(UvWriteReq writeReq, int status, Exception error, int bytesWritten, object state) + { + writeReq.Dispose(); + var _this = (WriteContext)state; + var socketOutput = _this.Output; - var i = 0; - foreach (var buffer in Buffers) - { - buffers[i++] = buffer; - } + Interlocked.Add(ref socketOutput._bytesWritten, bytesWritten); + Interlocked.Add(ref socketOutput._numBytesPreCompleted, -bytesWritten); - var writeReq = new UvWriteReq(Self._log); - writeReq.Init(Self._thread.Loop); - writeReq.Write(Self._socket, new ArraySegment>(buffers), (_writeReq, status, error, state) => - { - _writeReq.Dispose(); - var _this = (WriteContext)state; - _this.WriteStatus = status; - _this.WriteError = error; - DoShutdownIfNeeded(); - }, this); + _this.Error = error; + _this.Status = status; + + DoShutdownIfNeeded(_this); } /// /// Second step: initiate async shutdown if needed, otherwise go to next step /// - public void DoShutdownIfNeeded() + private static void DoShutdownIfNeeded(WriteContext req) { - if (SocketShutdownSend == false || Self._socket.IsClosed) + var socketOutput = req.Output; + if (socketOutput.SocketShutdownSent || req.SocketShutdownSend == false || socketOutput._socket.IsClosed) { - DoDisconnectIfNeeded(); + DoDisconnectIfNeeded(req); return; } - var shutdownReq = new UvShutdownReq(Self._log); - shutdownReq.Init(Self._thread.Loop); - shutdownReq.Shutdown(Self._socket, (_shutdownReq, status, state) => - { - _shutdownReq.Dispose(); - var _this = (WriteContext)state; - _this.ShutdownSendStatus = status; + socketOutput.SocketShutdownSent = true; + var shutdownReq = new UvShutdownReq(req.Output._log); + shutdownReq.Init(socketOutput._thread.Loop); + shutdownReq.Shutdown(socketOutput._socket, + (shutdownReq2, status2, state) => ShutdownCallback(shutdownReq2, status2, state), req); + } - Self._log.ConnectionWroteFin(Self._connectionId, status); + private static void ShutdownCallback(UvShutdownReq shutdownReq, int status, object state) + { + shutdownReq.Dispose(); + var _this = (WriteContext)state; + var socketOutput = _this.Output; + socketOutput.ShutdownSendStatus = status; + + socketOutput._log.ConnectionWroteFin(socketOutput._connectionId, status); - DoDisconnectIfNeeded(); - }, this); + DoDisconnectIfNeeded(_this); } /// /// Third step: disconnect socket if needed, otherwise this work item is complete /// - public void DoDisconnectIfNeeded() + private static void DoDisconnectIfNeeded(WriteContext req) { - if (SocketDisconnect == false || Self._socket.IsClosed) + var socketOutput = req.Output; + if (socketOutput.SocketDisconnected || req.SocketDisconnect == false || socketOutput._socket.IsClosed) { - Complete(); + socketOutput.OnWriteCompleted(req.Status, req.Error); return; } - Self._socket.Dispose(); - Self._log.ConnectionStop(Self._connectionId); - Complete(); + socketOutput.SocketDisconnected = true; + socketOutput._socket.Dispose(); + socketOutput._log.ConnectionStop(socketOutput._connectionId); + socketOutput.OnWriteCompleted(req.Status, req.Error); } + } - public void Complete() - { - Self.OnWriteCompleted(Buffers, WriteStatus, WriteError); - } + private struct CallbackContext + { + // callback(error, state, calledInline) + public Action Callback; + public object State; + public long BytesWrittenThreshold; + } + + private struct WriteBlock + { + public MemoryPoolBlock2 Block; + public bool SocketShutdownSend; + public bool SocketDisconnect; } } } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs index 6c26ec71b..a9dca4b0f 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs @@ -10,6 +10,7 @@ using Microsoft.AspNet.Server.Kestrel.Infrastructure; using Microsoft.AspNet.Server.Kestrel.Networking; using Microsoft.Extensions.Logging; +using System.Collections.Concurrent; namespace Microsoft.AspNet.Server.Kestrel { @@ -24,11 +25,8 @@ public class KestrelThread private Thread _thread; private UvLoopHandle _loop; private UvAsyncHandle _post; - private Queue _workAdding = new Queue(); - private Queue _workRunning = new Queue(); - private Queue _closeHandleAdding = new Queue(); - private Queue _closeHandleRunning = new Queue(); - private object _workSync = new Object(); + private ConcurrentQueue _workAdding = new ConcurrentQueue(); + private ConcurrentQueue _closeHandleAdding = new ConcurrentQueue(); private bool _stopImmediate = false; private ExceptionDispatchInfo _closeError; private IKestrelTrace _log; @@ -107,40 +105,36 @@ private void OnStopImmediate(object obj) public void Post(Action callback, object state) { - lock (_workSync) - { - _workAdding.Enqueue(new Work { CallbackAdapter = _objectCallbackAdapter, Callback = callback, State = state }); - } + _workAdding.Enqueue(new Work { CallbackAdapter = _objectCallbackAdapter, Callback = callback, State = state }); + _post.Send(); } public void Post(Action callback, T state) { - lock (_workSync) + + _workAdding.Enqueue(new Work { - _workAdding.Enqueue(new Work - { - CallbackAdapter = (callback2, state2) => ((Action)callback2).Invoke((T)state2), - Callback = callback, - State = state - }); - } + CallbackAdapter = (callback2, state2) => ((Action)callback2).Invoke((T)state2), + Callback = callback, + State = state + }); + _post.Send(); } public Task PostAsync(Action callback, object state) { var tcs = new TaskCompletionSource(); - lock (_workSync) + + _workAdding.Enqueue(new Work { - _workAdding.Enqueue(new Work - { - CallbackAdapter = _objectCallbackAdapter, - Callback = callback, - State = state, - Completion = tcs - }); - } + CallbackAdapter = _objectCallbackAdapter, + Callback = callback, + State = state, + Completion = tcs + }); + _post.Send(); return tcs.Task; } @@ -148,16 +142,15 @@ public Task PostAsync(Action callback, object state) public Task PostAsync(Action callback, T state) { var tcs = new TaskCompletionSource(); - lock (_workSync) + + _workAdding.Enqueue(new Work { - _workAdding.Enqueue(new Work - { - CallbackAdapter = (state1, state2) => ((Action)state1).Invoke((T)state2), - Callback = callback, - State = state, - Completion = tcs - }); - } + CallbackAdapter = (state1, state2) => ((Action)state1).Invoke((T)state2), + Callback = callback, + State = state, + Completion = tcs + }); + _post.Send(); return tcs.Task; } @@ -176,10 +169,8 @@ public void Send(Action callback, object state) private void PostCloseHandle(Action callback, IntPtr handle) { - lock (_workSync) - { - _closeHandleAdding.Enqueue(new CloseHandle { Callback = callback, Handle = handle }); - } + _closeHandleAdding.Enqueue(new CloseHandle { Callback = callback, Handle = handle }); + _post.Send(); } @@ -244,16 +235,9 @@ private void OnPost() private void DoPostWork() { - Queue queue; - lock (_workSync) - { - queue = _workAdding; - _workAdding = _workRunning; - _workRunning = queue; - } - while (queue.Count != 0) + Work work; + while (_workAdding.TryDequeue(out work)) { - var work = queue.Dequeue(); try { work.CallbackAdapter(work.Callback, work.State); @@ -283,16 +267,9 @@ private void DoPostWork() } private void DoPostCloseHandle() { - Queue queue; - lock (_workSync) - { - queue = _closeHandleAdding; - _closeHandleAdding = _closeHandleRunning; - _closeHandleRunning = queue; - } - while (queue.Count != 0) + CloseHandle closeHandle; + while (_closeHandleAdding.TryDequeue(out closeHandle)) { - var closeHandle = queue.Dequeue(); try { closeHandle.Callback(closeHandle.Handle); diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/MemoryPool2.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/MemoryPool2.cs index 2ee01c0d1..3df6b061b 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/MemoryPool2.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/MemoryPool2.cs @@ -31,6 +31,8 @@ public class MemoryPool2 : IDisposable /// private const int _blockLength = _blockStride - _blockUnused; + public const int NativeBlockSize = _blockLength; + /// /// 4096 * 32 gives you a slabLength of 128k contiguous bytes allocated per slab /// @@ -40,13 +42,13 @@ public class MemoryPool2 : IDisposable /// Thread-safe collection of blocks which are currently in the pool. A slab will pre-allocate all of the block tracking objects /// and add them to this collection. When memory is requested it is taken from here first, and when it is returned it is re-added. /// - private readonly ConcurrentStack _blocks = new ConcurrentStack(); + private readonly ConcurrentQueue _blocks = new ConcurrentQueue(); /// /// Thread-safe collection of slabs which have been allocated by this pool. As long as a slab is in this collection and slab.IsActive, /// the blocks will be added to _blocks when returned. /// - private readonly ConcurrentStack _slabs = new ConcurrentStack(); + private readonly ConcurrentQueue _slabs = new ConcurrentQueue(); /// /// This is part of implementing the IDisposable pattern. @@ -74,33 +76,34 @@ public MemoryPoolBlock2 Lease(int minimumSize) slab: null); } - while (true) + MemoryPoolBlock2 block; + if (_blocks.TryDequeue(out block)) { - MemoryPoolBlock2 block; - if (_blocks.TryPop(out block)) - { - // block successfully taken from the stack - return it - return block; - } - // no blocks available - grow the pool and try again - AllocateSlab(); + // block successfully taken from the stack - return it + return block; } + // no blocks available - grow the pool + return AllocateSlab(); + } /// /// Internal method called when a block is requested and the pool is empty. It allocates one additional slab, creates all of the /// block tracking objects, and adds them all to the pool. /// - private void AllocateSlab() + private MemoryPoolBlock2 AllocateSlab() { var slab = MemoryPoolSlab2.Create(_slabLength); - _slabs.Push(slab); + _slabs.Enqueue(slab); var basePtr = slab.ArrayPtr; var firstOffset = (int)((_blockStride - 1) - ((ulong)(basePtr + _blockStride - 1) % _blockStride)); - for (var offset = firstOffset; - offset + _blockLength <= _slabLength; + var poolAllocationLength = _slabLength - (_blockLength + _blockStride); + + var offset = firstOffset; + for (; + offset < poolAllocationLength; offset += _blockStride) { var block = MemoryPoolBlock2.Create( @@ -110,6 +113,15 @@ private void AllocateSlab() slab); Return(block); } + + // return last block rather than adding to pool + var newBlock = MemoryPoolBlock2.Create( + new ArraySegment(slab.Array, offset, _blockLength), + basePtr, + this, + slab); + + return newBlock; } /// @@ -123,7 +135,7 @@ private void AllocateSlab() public void Return(MemoryPoolBlock2 block) { block.Reset(); - _blocks.Push(block); + _blocks.Enqueue(block); } protected virtual void Dispose(bool disposing) @@ -133,7 +145,7 @@ protected virtual void Dispose(bool disposing) if (disposing) { MemoryPoolSlab2 slab; - while (_slabs.TryPop(out slab)) + while (_slabs.TryDequeue(out slab)) { // dispose managed state (managed objects). slab.Dispose(); diff --git a/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs b/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs index 9a68cd9df..bd0410652 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs @@ -7,10 +7,8 @@ using System.Threading.Tasks; using Microsoft.AspNet.Http.Features; using Microsoft.AspNet.Server.Kestrel.Http; -using Microsoft.AspNet.Server.Kestrel.Infrastructure; using Microsoft.AspNet.Server.Kestrel.Networking; using Microsoft.Extensions.PlatformAbstractions; -using Microsoft.Extensions.Logging; namespace Microsoft.AspNet.Server.Kestrel { diff --git a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs index 103ddfc6e..7280c6b46 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs @@ -14,15 +14,17 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking /// public class UvWriteReq : UvRequest { - private readonly static Libuv.uv_write_cb _uv_write_cb = UvWriteCb; + public const int BUFFER_COUNT = 8; - private IntPtr _bufs; + private readonly static Libuv.uv_write_cb _uv_write_cb = (ptr, status) => UvWriteCb(ptr, status); - private Action _callback; - private object _state; - private const int BUFFER_COUNT = 4; + private IntPtr _nativeBufPointers; - private List _pins = new List(); + private Action _callback; + private object _state; + + private ArraySegment _buffers; + private GCHandle _pin; public UvWriteReq(IKestrelTrace logger) : base(logger) { @@ -32,45 +34,37 @@ public void Init(UvLoopHandle loop) { var requestSize = loop.Libuv.req_size(Libuv.RequestType.WRITE); var bufferSize = Marshal.SizeOf() * BUFFER_COUNT; + CreateMemory( loop.Libuv, loop.ThreadId, requestSize + bufferSize); - _bufs = handle + requestSize; + _nativeBufPointers = handle + requestSize; } public unsafe void Write( UvStreamHandle handle, - ArraySegment> bufs, - Action callback, + ArraySegment bufs, + Action callback, object state) { try { + _buffers = bufs; // add GCHandle to keeps this SafeHandle alive while request processing - _pins.Add(GCHandle.Alloc(this, GCHandleType.Normal)); + _pin = GCHandle.Alloc(this, GCHandleType.Normal); - var pBuffers = (Libuv.uv_buf_t*)_bufs; + var pBuffers = (Libuv.uv_buf_t*)_nativeBufPointers; var nBuffers = bufs.Count; - if (nBuffers > BUFFER_COUNT) - { - // create and pin buffer array when it's larger than the pre-allocated one - var bufArray = new Libuv.uv_buf_t[nBuffers]; - var gcHandle = GCHandle.Alloc(bufArray, GCHandleType.Pinned); - _pins.Add(gcHandle); - pBuffers = (Libuv.uv_buf_t*)gcHandle.AddrOfPinnedObject(); - } for (var index = 0; index < nBuffers; index++) { - // create and pin each segment being written var buf = bufs.Array[bufs.Offset + index]; - - var gcHandle = GCHandle.Alloc(buf.Array, GCHandleType.Pinned); - _pins.Add(gcHandle); + var len = buf.End - buf.Start; + // create and pin each segment being written pBuffers[index] = Libuv.buf_init( - gcHandle.AddrOfPinnedObject() + buf.Offset, - buf.Count); + buf.Pin() - len, + buf.End - buf.Start); } _callback = callback; @@ -82,43 +76,36 @@ public unsafe void Write( _callback = null; _state = null; Unpin(this); + ProcessBlocks(); throw; } } public unsafe void Write2( UvStreamHandle handle, - ArraySegment> bufs, + ArraySegment bufs, UvStreamHandle sendHandle, - Action callback, + Action callback, object state) { try { + _buffers = bufs; // add GCHandle to keeps this SafeHandle alive while request processing - _pins.Add(GCHandle.Alloc(this, GCHandleType.Normal)); + _pin = GCHandle.Alloc(this, GCHandleType.Normal); - var pBuffers = (Libuv.uv_buf_t*)_bufs; + var pBuffers = (Libuv.uv_buf_t*)_nativeBufPointers; var nBuffers = bufs.Count; - if (nBuffers > BUFFER_COUNT) - { - // create and pin buffer array when it's larger than the pre-allocated one - var bufArray = new Libuv.uv_buf_t[nBuffers]; - var gcHandle = GCHandle.Alloc(bufArray, GCHandleType.Pinned); - _pins.Add(gcHandle); - pBuffers = (Libuv.uv_buf_t*)gcHandle.AddrOfPinnedObject(); - } for (var index = 0; index < nBuffers; index++) { - // create and pin each segment being written var buf = bufs.Array[bufs.Offset + index]; + var len = buf.End - buf.Start; - var gcHandle = GCHandle.Alloc(buf.Array, GCHandleType.Pinned); - _pins.Add(gcHandle); + // create and pin each segment being written pBuffers[index] = Libuv.buf_init( - gcHandle.AddrOfPinnedObject() + buf.Offset, - buf.Count); + buf.Pin() - len, + buf.End - buf.Start); } _callback = callback; @@ -130,23 +117,41 @@ public unsafe void Write2( _callback = null; _state = null; Unpin(this); + ProcessBlocks(); throw; } } - - private static void Unpin(UvWriteReq req) + + private int ProcessBlocks() { - foreach (var pin in req._pins) + var bytesWritten = 0; + var end = _buffers.Offset + _buffers.Count; + for (var i = _buffers.Offset; i < end; i++) { - pin.Free(); + var block = _buffers.Array[i]; + bytesWritten += block.End - block.Start; + + block.Unpin(); + + if (block.Pool != null) + { + block.Pool.Return(block); + } } - req._pins.Clear(); + + return bytesWritten; } + private static void Unpin(UvWriteReq req) + { + req._pin.Free(); + } + private static void UvWriteCb(IntPtr ptr, int status) { var req = FromIntPtr(ptr); Unpin(req); + var bytesWritten = req.ProcessBlocks(); var callback = req._callback; req._callback = null; @@ -162,7 +167,7 @@ private static void UvWriteCb(IntPtr ptr, int status) try { - callback(req, status, error, state); + callback(req, status, error, bytesWritten, state); } catch (Exception ex) { diff --git a/test/Microsoft.AspNet.Server.KestrelTests/MultipleLoopTests.cs b/test/Microsoft.AspNet.Server.KestrelTests/MultipleLoopTests.cs index cb870ea14..b8821082d 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/MultipleLoopTests.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/MultipleLoopTests.cs @@ -81,12 +81,16 @@ public void ServerPipeListenForConnections() return; } + var buffer = new byte[] { 1, 2, 3, 4 }; + var msg = MemoryPoolBlock2.Create(new ArraySegment(buffer), IntPtr.Zero, null, null); + msg.End = msg.Start + buffer.Length; + var writeRequest = new UvWriteReq(new KestrelTrace(new TestKestrelTrace())); writeRequest.Init(loop); writeRequest.Write( serverConnectionPipe, - new ArraySegment>(new ArraySegment[] { new ArraySegment(new byte[] { 1, 2, 3, 4 }) }), - (_3, status2, error2, _4) => + new ArraySegment(new[] { msg }), + (socket2, status2, error2, bytesWritten, state) => { writeRequest.Dispose(); serverConnectionPipe.Dispose(); @@ -175,13 +179,17 @@ public void ServerPipeDispatchConnections() serverConnectionPipeAcceptedEvent.WaitOne(); + var buffer = new byte[] { 1, 2, 3, 4 }; + var msg = MemoryPoolBlock2.Create(new ArraySegment(buffer), IntPtr.Zero, null, null); + msg.End = msg.Start + buffer.Length; + var writeRequest = new UvWriteReq(new KestrelTrace(new TestKestrelTrace())); writeRequest.Init(loop); writeRequest.Write2( serverConnectionPipe, - new ArraySegment>(new ArraySegment[] { new ArraySegment(new byte[] { 1, 2, 3, 4 }) }), + new ArraySegment(new[] { msg }), serverConnectionTcp, - (_3, status2, error2, _4) => + (socket2, status2, error2, bytesWritten, state) => { writeRequest.Dispose(); serverConnectionTcp.Dispose(); diff --git a/test/Microsoft.AspNet.Server.KestrelTests/NetworkingTests.cs b/test/Microsoft.AspNet.Server.KestrelTests/NetworkingTests.cs index aea327447..5247f7e15 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/NetworkingTests.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/NetworkingTests.cs @@ -208,14 +208,16 @@ public async Task SocketCanReadAndWrite() { for (var x = 0; x < 2; x++) { + var buffer = new byte[] { 65, 66, 67, 68, 69 }; + var msg = MemoryPoolBlock2.Create(new ArraySegment(buffer), IntPtr.Zero, null, null); + msg.End = msg.Start + buffer.Length; + var req = new UvWriteReq(new KestrelTrace(new TestKestrelTrace())); req.Init(loop); req.Write( tcp2, - new ArraySegment>( - new[] { new ArraySegment(new byte[] { 65, 66, 67, 68, 69 }) } - ), - (_1, _2, _3, _4) => { }, + new ArraySegment(new[] { msg }), + (_1, _2, _3, _4, _5) => { }, null); } } diff --git a/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs b/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs index 8e0425ce5..8cad2ae96 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs @@ -16,7 +16,7 @@ namespace Microsoft.AspNet.Server.KestrelTests public class SocketOutputTests { [Fact] - public void CanWrite1MB() + public void CanWriteOver1MB() { // This test was added because when initially implementing write-behind buffering in // SocketOutput, the write callback would never be invoked for writes larger than @@ -39,24 +39,27 @@ public void CanWrite1MB() var kestrelThread = kestrelEngine.Threads[0]; var socket = new MockSocket(kestrelThread.Loop.ThreadId, new TestKestrelTrace()); var trace = new KestrelTrace(new TestKestrelTrace()); - var socketOutput = new SocketOutput(kestrelThread, socket, 0, trace); - - // I doubt _maxBytesPreCompleted will ever be over a MB. If it is, we should change this test. - var bufferSize = 1048576; - var buffer = new ArraySegment(new byte[bufferSize], 0, bufferSize); - var completedWh = new ManualResetEventSlim(); - Action onCompleted = (ex, state, calledInline) => + using (var memory2 = new MemoryPool2()) { - Assert.Null(ex); - Assert.Null(state); - completedWh.Set(); - }; - - // Act - socketOutput.Write(buffer, onCompleted, null); - - // Assert - Assert.True(completedWh.Wait(1000)); + var socketOutput = new SocketOutput(memory2, kestrelThread, socket, 0, trace); + + // I doubt _maxBytesPreCompleted will ever be over a MB. If it is, we should change this test. + var bufferSize = 1048576 + SocketOutput.MaxBytesPreCompleted; + var buffer = new ArraySegment(new byte[bufferSize], 0, bufferSize); + var completedWh = new ManualResetEventSlim(); + Action onCompleted = (ex, state, status, calledInline) => + { + Assert.Null(ex); + Assert.Null(state); + completedWh.Set(); + }; + + // Act + socketOutput.Write(buffer, onCompleted, null); + + // Assert + Assert.True(completedWh.Wait(1000)); + } } } @@ -64,7 +67,7 @@ public void CanWrite1MB() public void WritesDontCompleteImmediatelyWhenTooManyBytesAreAlreadyPreCompleted() { // This should match _maxBytesPreCompleted in SocketOutput - var maxBytesPreCompleted = 65536; + var maxBytesPreCompleted = SocketOutput.MaxBytesPreCompleted; var completeQueue = new Queue>(); // Arrange @@ -84,35 +87,38 @@ public void WritesDontCompleteImmediatelyWhenTooManyBytesAreAlreadyPreCompleted( var kestrelThread = kestrelEngine.Threads[0]; var socket = new MockSocket(kestrelThread.Loop.ThreadId, new TestKestrelTrace()); var trace = new KestrelTrace(new TestKestrelTrace()); - var socketOutput = new SocketOutput(kestrelThread, socket, 0, trace); - - var bufferSize = maxBytesPreCompleted; - var buffer = new ArraySegment(new byte[bufferSize], 0, bufferSize); - var completedWh = new ManualResetEventSlim(); - Action onCompleted = (ex, state, calledInline) => + using (var memory2 = new MemoryPool2()) { - Assert.Null(ex); - Assert.Null(state); - completedWh.Set(); - }; - - // Act - socketOutput.Write(buffer, onCompleted, null); - // Assert - // The first write should pre-complete since it is <= _maxBytesPreCompleted. - Assert.True(completedWh.Wait(1000)); - // Arrange - completedWh.Reset(); - // Act - socketOutput.Write(buffer, onCompleted, null); - // Assert - // Too many bytes are already pre-completed for the second write to pre-complete. - Assert.False(completedWh.Wait(1000)); - // Act - completeQueue.Dequeue()(0); - // Assert - // Finishing the first write should allow the second write to pre-complete. - Assert.True(completedWh.Wait(1000)); + var socketOutput = new SocketOutput(memory2, kestrelThread, socket, 0, trace); + + var bufferSize = maxBytesPreCompleted; + var buffer = new ArraySegment(new byte[bufferSize], 0, bufferSize); + var completedWh = new ManualResetEventSlim(); + Action onCompleted = (ex, state, status, calledInline) => + { + Assert.Null(ex); + Assert.Null(state); + completedWh.Set(); + }; + + // Act + socketOutput.Write(buffer, onCompleted, null); + // Assert + // The first write should pre-complete since it is <= _maxBytesPreCompleted. + Assert.True(completedWh.Wait(1000)); + // Arrange + completedWh.Reset(); + // Act + socketOutput.Write(buffer, onCompleted, null); + // Assert + // Too many bytes are already pre-completed for the second write to pre-complete. + Assert.False(completedWh.Wait(1000)); + // Act + completeQueue.Dequeue()(0); + // Assert + // Finishing the first write should allow the second write to pre-complete. + Assert.True(completedWh.Wait(1000)); + } } }