less work in locks
This commit is contained in:
parent
feb40402de
commit
81dba3961d
|
|
@ -16,6 +16,9 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
{
|
{
|
||||||
private const int _maxPendingWrites = 3;
|
private const int _maxPendingWrites = 3;
|
||||||
private const int _maxBytesPreCompleted = 65536;
|
private const int _maxBytesPreCompleted = 65536;
|
||||||
|
private const int _initialTaskQueues = 64;
|
||||||
|
|
||||||
|
private static WaitCallback _returnBlocks = (state) => ReturnBlocks((MemoryPoolBlock2)state);
|
||||||
|
|
||||||
private readonly KestrelThread _thread;
|
private readonly KestrelThread _thread;
|
||||||
private readonly UvStreamHandle _socket;
|
private readonly UvStreamHandle _socket;
|
||||||
|
|
@ -44,6 +47,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
private Exception _lastWriteError;
|
private Exception _lastWriteError;
|
||||||
private WriteContext _nextWriteContext;
|
private WriteContext _nextWriteContext;
|
||||||
private readonly Queue<TaskCompletionSource<object>> _tasksPending;
|
private readonly Queue<TaskCompletionSource<object>> _tasksPending;
|
||||||
|
private readonly Queue<TaskCompletionSource<object>> _tasksCompleted;
|
||||||
|
|
||||||
public SocketOutput(
|
public SocketOutput(
|
||||||
KestrelThread thread,
|
KestrelThread thread,
|
||||||
|
|
@ -58,7 +62,8 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
_connection = connection;
|
_connection = connection;
|
||||||
_connectionId = connectionId;
|
_connectionId = connectionId;
|
||||||
_log = log;
|
_log = log;
|
||||||
_tasksPending = new Queue<TaskCompletionSource<object>>();
|
_tasksPending = new Queue<TaskCompletionSource<object>>(_initialTaskQueues);
|
||||||
|
_tasksCompleted = new Queue<TaskCompletionSource<object>>(_initialTaskQueues);
|
||||||
|
|
||||||
_head = memory.Lease();
|
_head = memory.Lease();
|
||||||
_tail = _head;
|
_tail = _head;
|
||||||
|
|
@ -79,6 +84,8 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
}
|
}
|
||||||
TaskCompletionSource<object> tcs = null;
|
TaskCompletionSource<object> tcs = null;
|
||||||
|
|
||||||
|
var scheduleWrite = false;
|
||||||
|
|
||||||
lock (_contextLock)
|
lock (_contextLock)
|
||||||
{
|
{
|
||||||
if (_nextWriteContext == null)
|
if (_nextWriteContext == null)
|
||||||
|
|
@ -118,11 +125,16 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
|
|
||||||
if (_writesPending < _maxPendingWrites && immediate)
|
if (_writesPending < _maxPendingWrites && immediate)
|
||||||
{
|
{
|
||||||
ScheduleWrite();
|
scheduleWrite = true;
|
||||||
_writesPending++;
|
_writesPending++;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (scheduleWrite)
|
||||||
|
{
|
||||||
|
ScheduleWrite();
|
||||||
|
}
|
||||||
|
|
||||||
// Return TaskCompletionSource's Task if set, otherwise completed Task
|
// Return TaskCompletionSource's Task if set, otherwise completed Task
|
||||||
return tcs?.Task ?? TaskUtilities.CompletedTask;
|
return tcs?.Task ?? TaskUtilities.CompletedTask;
|
||||||
}
|
}
|
||||||
|
|
@ -164,6 +176,9 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
|
|
||||||
public void ProducingComplete(MemoryPoolIterator2 end, int count)
|
public void ProducingComplete(MemoryPoolIterator2 end, int count)
|
||||||
{
|
{
|
||||||
|
var decreasePreCompleted = false;
|
||||||
|
MemoryPoolBlock2 blockToReturn = null;
|
||||||
|
|
||||||
lock (_returnLock)
|
lock (_returnLock)
|
||||||
{
|
{
|
||||||
Debug.Assert(_isProducing);
|
Debug.Assert(_isProducing);
|
||||||
|
|
@ -176,26 +191,40 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
|
|
||||||
if (count != 0)
|
if (count != 0)
|
||||||
{
|
{
|
||||||
lock (_contextLock)
|
decreasePreCompleted = true;
|
||||||
{
|
|
||||||
_numBytesPreCompleted += count;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
var block = _returnFromOnProducingComplete;
|
blockToReturn = _returnFromOnProducingComplete;
|
||||||
while (block != null)
|
|
||||||
{
|
|
||||||
var returnBlock = block;
|
|
||||||
block = block.Next;
|
|
||||||
|
|
||||||
returnBlock.Pool?.Return(returnBlock);
|
|
||||||
}
|
|
||||||
|
|
||||||
_returnFromOnProducingComplete = null;
|
_returnFromOnProducingComplete = null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (decreasePreCompleted)
|
||||||
|
{
|
||||||
|
lock (_contextLock)
|
||||||
|
{
|
||||||
|
_numBytesPreCompleted += count;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
if (blockToReturn != null)
|
||||||
|
{
|
||||||
|
ThreadPool.QueueUserWorkItem(_returnBlocks, blockToReturn);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void ReturnBlocks(MemoryPoolBlock2 block)
|
||||||
|
{
|
||||||
|
while(block != null)
|
||||||
|
{
|
||||||
|
var returningBlock = block;
|
||||||
|
block = returningBlock.Next;
|
||||||
|
|
||||||
|
returningBlock.Pool?.Return(returningBlock);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void ScheduleWrite()
|
private void ScheduleWrite()
|
||||||
|
|
@ -252,11 +281,13 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
_connection.Abort();
|
_connection.Abort();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
bool scheduleWrite = false;
|
||||||
|
|
||||||
lock (_contextLock)
|
lock (_contextLock)
|
||||||
{
|
{
|
||||||
if (_nextWriteContext != null)
|
if (_nextWriteContext != null)
|
||||||
{
|
{
|
||||||
ScheduleWrite();
|
scheduleWrite = true;
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
|
|
@ -279,21 +310,36 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
_numBytesPreCompleted += bytesToWrite;
|
_numBytesPreCompleted += bytesToWrite;
|
||||||
bytesLeftToBuffer -= bytesToWrite;
|
bytesLeftToBuffer -= bytesToWrite;
|
||||||
|
|
||||||
if (_lastWriteError == null)
|
_tasksCompleted.Enqueue(tcs);
|
||||||
{
|
|
||||||
ThreadPool.QueueUserWorkItem(
|
|
||||||
(o) => ((TaskCompletionSource<object>)o).SetResult(null),
|
|
||||||
tcs);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
// error is closure captured
|
|
||||||
ThreadPool.QueueUserWorkItem(
|
|
||||||
(o) => ((TaskCompletionSource<object>)o).SetException(_lastWriteError),
|
|
||||||
tcs);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
while (_tasksCompleted.Count > 0)
|
||||||
|
{
|
||||||
|
var tcs = _tasksCompleted.Dequeue();
|
||||||
|
if (_lastWriteError == null)
|
||||||
|
{
|
||||||
|
ThreadPool.QueueUserWorkItem(
|
||||||
|
(o) => ((TaskCompletionSource<object>)o).SetResult(null),
|
||||||
|
tcs);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
// error is closure captured
|
||||||
|
ThreadPool.QueueUserWorkItem(
|
||||||
|
(o) => ((TaskCompletionSource<object>)o).SetException(_lastWriteError),
|
||||||
|
tcs);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (scheduleWrite)
|
||||||
|
{
|
||||||
|
// ScheduleWrite();
|
||||||
|
// on right thread, fairness issues?
|
||||||
|
WriteAllPending();
|
||||||
|
}
|
||||||
|
|
||||||
|
_tasksCompleted.Clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
// This is called on the libuv event loop
|
// This is called on the libuv event loop
|
||||||
|
|
@ -345,6 +391,8 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
|
|
||||||
private class WriteContext
|
private class WriteContext
|
||||||
{
|
{
|
||||||
|
private static WaitCallback _returnWrittenBlocks = (state) => ReturnWrittenBlocks((MemoryPoolBlock2)state);
|
||||||
|
|
||||||
private MemoryPoolIterator2 _lockedStart;
|
private MemoryPoolIterator2 _lockedStart;
|
||||||
private MemoryPoolIterator2 _lockedEnd;
|
private MemoryPoolIterator2 _lockedEnd;
|
||||||
private int _bufferCount;
|
private int _bufferCount;
|
||||||
|
|
@ -385,7 +433,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
{
|
{
|
||||||
_writeReq.Dispose();
|
_writeReq.Dispose();
|
||||||
var _this = (WriteContext)state;
|
var _this = (WriteContext)state;
|
||||||
_this.ReturnFullyWrittenBlocks();
|
_this.ScheduleReturnFullyWrittenBlocks();
|
||||||
_this.WriteStatus = status;
|
_this.WriteStatus = status;
|
||||||
_this.WriteError = error;
|
_this.WriteError = error;
|
||||||
_this.DoShutdownIfNeeded();
|
_this.DoShutdownIfNeeded();
|
||||||
|
|
@ -442,10 +490,29 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
Self.OnWriteCompleted(_byteCount, WriteStatus, WriteError);
|
Self.OnWriteCompleted(_byteCount, WriteStatus, WriteError);
|
||||||
}
|
}
|
||||||
|
|
||||||
private void ReturnFullyWrittenBlocks()
|
private void ScheduleReturnFullyWrittenBlocks()
|
||||||
{
|
{
|
||||||
var block = _lockedStart.Block;
|
var block = _lockedStart.Block;
|
||||||
while (block != _lockedEnd.Block)
|
var end = _lockedEnd.Block;
|
||||||
|
if (block == end)
|
||||||
|
{
|
||||||
|
end.Unpin();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
while (block.Next != end)
|
||||||
|
{
|
||||||
|
block = block.Next;
|
||||||
|
block.Unpin();
|
||||||
|
}
|
||||||
|
block.Next = null;
|
||||||
|
|
||||||
|
ThreadPool.QueueUserWorkItem(_returnWrittenBlocks, _lockedStart.Block);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void ReturnWrittenBlocks(MemoryPoolBlock2 block)
|
||||||
|
{
|
||||||
|
while (block != null)
|
||||||
{
|
{
|
||||||
var returnBlock = block;
|
var returnBlock = block;
|
||||||
block = block.Next;
|
block = block.Next;
|
||||||
|
|
@ -453,8 +520,6 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
|
||||||
returnBlock.Unpin();
|
returnBlock.Unpin();
|
||||||
returnBlock.Pool?.Return(returnBlock);
|
returnBlock.Pool?.Return(returnBlock);
|
||||||
}
|
}
|
||||||
|
|
||||||
_lockedEnd.Block.Unpin();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private void LockWrite()
|
private void LockWrite()
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue