From 07bded5736d9a5fe734ac8ddf78575935ad9de83 Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Wed, 1 Jul 2020 17:31:22 +0200 Subject: [PATCH 01/12] SocketPal does not need to be partial --- .../System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs index 8cec72ec9bf1c2..ffc0abd01695f9 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs @@ -13,7 +13,7 @@ namespace System.Net.Sockets { - internal static partial class SocketPal + internal static class SocketPal { public const bool SupportsMultipleConnectAttempts = false; public static readonly int MaximumAddressSize = Interop.Sys.GetMaximumAddressSize(); From 2d9a71314d69ec5fa935416bfa5cd3500a58b3a3 Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Wed, 1 Jul 2020 17:50:22 +0200 Subject: [PATCH 02/12] after a succesfull partial read, set queue state to waiting so next read does not peform a syscall but rather waits for epoll notification --- .../src/System/Net/Sockets/SocketAsyncContext.Unix.cs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index 2a5c3c3587d8d1..63c6796941de28 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -311,6 +311,8 @@ public void DoAbort() ErrorCode = SocketError.OperationAborted; } + internal virtual bool IsPartialSuccess => false; + protected abstract void Abort(); protected abstract bool DoTryComplete(SocketAsyncContext context); @@ -457,6 +459,8 @@ private sealed class BufferMemoryReceiveOperation : ReceiveOperation public BufferMemoryReceiveOperation(SocketAsyncContext context) : base(context) { } + internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && Buffer.Length > BytesTransferred; + protected override bool DoTryComplete(SocketAsyncContext context) { // Zero byte read is performed to know when data is available. @@ -536,6 +540,8 @@ private sealed unsafe class BufferPtrReceiveOperation : ReceiveOperation public BufferPtrReceiveOperation(SocketAsyncContext context) : base(context) { } + internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && Length > BytesTransferred; + protected override bool DoTryComplete(SocketAsyncContext context) => SocketPal.TryCompleteReceiveFrom(context._socket, new Span(BufferPtr, Length), null, Flags, SocketAddress, ref SocketAddressLen, out BytesTransferred, out ReceivedFlags, out ErrorCode); } @@ -1035,7 +1041,7 @@ public OperationResult ProcessQueuedOperation(TOperation op) // No more operations to process _tail = null; _isNextOperationSynchronous = false; - _state = QueueState.Ready; + _state = op.IsPartialSuccess ? QueueState.Waiting : QueueState.Ready; _sequenceNumber++; Trace(context, $"Exit (finished queue)"); } From f882ea2cdc7c360feaf5e9102c0d12f5d8bc1a29 Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Wed, 1 Jul 2020 21:21:06 +0200 Subject: [PATCH 03/12] when epoll notification is received and the state is waiting and the queue is empty, change the state to ready --- .../src/System/Net/Sockets/SocketAsyncContext.Unix.cs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index 63c6796941de28..25ea4d64891a29 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -866,7 +866,12 @@ public bool StartAsyncOperation(SocketAsyncContext context, TOperation operation return null; case QueueState.Waiting: - Debug.Assert(_tail != null, "State == Waiting but queue is empty!"); + if (_tail == null) + { + _state = QueueState.Ready; + return null; + } + op = _tail.Next; Debug.Assert(_isNextOperationSynchronous == (op.Event != null)); if (skipAsyncEvents && !_isNextOperationSynchronous) From 99ddeaf0513de9b0b57edd14908168f5b85533bb Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Thu, 2 Jul 2020 14:21:56 +0200 Subject: [PATCH 04/12] add dummy hacked logging --- .../src/System/Net/Sockets/SocketPal.Unix.cs | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs index ffc0abd01695f9..84101adee43fea 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs @@ -6,6 +6,7 @@ using System.Collections; using System.Collections.Generic; using System.Diagnostics; +using System.Diagnostics.Tracing; using System.IO; using System.Runtime.InteropServices; using System.Threading; @@ -13,6 +14,38 @@ namespace System.Net.Sockets { + internal sealed class LinuxSocketsEventSource : EventSource + { + internal static readonly LinuxSocketsEventSource Log = new LinuxSocketsEventSource(); + + private LinuxSocketsEventSource() : base("Npgsql") { } // using an existing name that is recognized by aspnet/benchmarks backend + + private PollingCounter? _totalReadsNonBlockingCounter; + private PollingCounter? _totalReadsWouldBlockCounter; + //private PollingCounter? _totalWritesNonBlockingCounter; + //private PollingCounter? _totalWritesWouldBlockCounter; + + private long _readsWouldBlockCount, _readsNonBlockingCount, _writesWouldBlockCount, _writesNonBlockingCount; + + protected override void OnEventCommand(EventCommandEventArgs command) + { + if (command.Command != EventCommand.Enable) + { + return; + } + + _totalReadsNonBlockingCounter = new PollingCounter("total-commands", this, () => _readsNonBlockingCount) { DisplayName = "Total Commands" }; + _totalReadsWouldBlockCounter = new PollingCounter("failed-commands", this, () => _readsWouldBlockCount) { DisplayName = "Failed Commands" }; + //_totalWritesNonBlockingCounter = new PollingCounter("linux-sockets/writes-non-blocking", this, () => _writesNonBlockingCount) { DisplayName = "Writes Non Blocking" }; + //_totalWritesWouldBlockCounter = new PollingCounter("linux-sockets/writes-would-block", this, () => _writesWouldBlockCount) { DisplayName = "Writes Would Block" }; + } + + internal void OnReadNonBlocking() => Interlocked.Increment(ref _readsNonBlockingCount); + internal void OnReadWoudlBlock() => Interlocked.Increment(ref _readsWouldBlockCount); + internal void OnWriteNonBlocking() => Interlocked.Increment(ref _writesNonBlockingCount); + internal void OnWriteWoudlBlock() => Interlocked.Increment(ref _writesWouldBlockCount); + } + internal static class SocketPal { public const bool SupportsMultipleConnectAttempts = false; @@ -711,6 +744,7 @@ public static unsafe bool TryCompleteReceive(SafeSocketHandle socket, Span { bytesReceived = received; errorCode = SocketError.Success; + LinuxSocketsEventSource.Log.OnReadNonBlocking(); return true; } @@ -722,6 +756,7 @@ public static unsafe bool TryCompleteReceive(SafeSocketHandle socket, Span return true; } + LinuxSocketsEventSource.Log.OnReadWoudlBlock(); errorCode = SocketError.Success; return false; } @@ -768,6 +803,7 @@ public static unsafe bool TryCompleteReceiveFrom(SafeSocketHandle socket, Span return true; } + LinuxSocketsEventSource.Log.OnWriteWoudlBlock(); errorCode = successfulSend ? SocketError.Success : SocketError.WouldBlock; return false; } @@ -895,6 +933,7 @@ public static bool TryCompleteSendTo(SafeSocketHandle socket, ReadOnlySpan if (isComplete) { errorCode = SocketError.Success; + LinuxSocketsEventSource.Log.OnWriteNonBlocking(); return true; } From e524076af17a97cf0f9f4cae9d2c13a307a9ef9b Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Thu, 2 Jul 2020 17:55:45 +0200 Subject: [PATCH 05/12] some improvements --- .../System/Net/Sockets/SocketAsyncContext.Unix.cs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index 25ea4d64891a29..4001c05f157369 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -459,7 +459,7 @@ private sealed class BufferMemoryReceiveOperation : ReceiveOperation public BufferMemoryReceiveOperation(SocketAsyncContext context) : base(context) { } - internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && Buffer.Length > BytesTransferred; + internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Buffer.Length; protected override bool DoTryComplete(SocketAsyncContext context) { @@ -540,7 +540,7 @@ private sealed unsafe class BufferPtrReceiveOperation : ReceiveOperation public BufferPtrReceiveOperation(SocketAsyncContext context) : base(context) { } - internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && Length > BytesTransferred; + internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Length; protected override bool DoTryComplete(SocketAsyncContext context) => SocketPal.TryCompleteReceiveFrom(context._socket, new Span(BufferPtr, Length), null, Flags, SocketAddress, ref SocketAddressLen, out BytesTransferred, out ReceivedFlags, out ErrorCode); @@ -727,12 +727,12 @@ private enum QueueState : int public bool IsNextOperationSynchronous_Speculative => _isNextOperationSynchronous; - public void Init() + public void Init(bool isReady) { Debug.Assert(_queueLock == null); _queueLock = new object(); - _state = QueueState.Ready; + _state = isReady ? QueueState.Ready : QueueState.Waiting; _sequenceNumber = 0; } @@ -869,6 +869,7 @@ public bool StartAsyncOperation(SocketAsyncContext context, TOperation operation if (_tail == null) { _state = QueueState.Ready; + _sequenceNumber++; return null; } @@ -1204,8 +1205,8 @@ public SocketAsyncContext(SafeSocketHandle socket) { _socket = socket; - _receiveQueue.Init(); - _sendQueue.Init(); + _receiveQueue.Init(isReady: false); + _sendQueue.Init(isReady: true); } public bool PreferInlineCompletions From 570de52d297384c61b24f73a114d1b8b2b4ddc7f Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Fri, 3 Jul 2020 12:31:26 +0200 Subject: [PATCH 06/12] Revert "add dummy hacked logging" This reverts commit 99ddeaf0513de9b0b57edd14908168f5b85533bb. --- .../src/System/Net/Sockets/SocketPal.Unix.cs | 39 ------------------- 1 file changed, 39 deletions(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs index 84101adee43fea..ffc0abd01695f9 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs @@ -6,7 +6,6 @@ using System.Collections; using System.Collections.Generic; using System.Diagnostics; -using System.Diagnostics.Tracing; using System.IO; using System.Runtime.InteropServices; using System.Threading; @@ -14,38 +13,6 @@ namespace System.Net.Sockets { - internal sealed class LinuxSocketsEventSource : EventSource - { - internal static readonly LinuxSocketsEventSource Log = new LinuxSocketsEventSource(); - - private LinuxSocketsEventSource() : base("Npgsql") { } // using an existing name that is recognized by aspnet/benchmarks backend - - private PollingCounter? _totalReadsNonBlockingCounter; - private PollingCounter? _totalReadsWouldBlockCounter; - //private PollingCounter? _totalWritesNonBlockingCounter; - //private PollingCounter? _totalWritesWouldBlockCounter; - - private long _readsWouldBlockCount, _readsNonBlockingCount, _writesWouldBlockCount, _writesNonBlockingCount; - - protected override void OnEventCommand(EventCommandEventArgs command) - { - if (command.Command != EventCommand.Enable) - { - return; - } - - _totalReadsNonBlockingCounter = new PollingCounter("total-commands", this, () => _readsNonBlockingCount) { DisplayName = "Total Commands" }; - _totalReadsWouldBlockCounter = new PollingCounter("failed-commands", this, () => _readsWouldBlockCount) { DisplayName = "Failed Commands" }; - //_totalWritesNonBlockingCounter = new PollingCounter("linux-sockets/writes-non-blocking", this, () => _writesNonBlockingCount) { DisplayName = "Writes Non Blocking" }; - //_totalWritesWouldBlockCounter = new PollingCounter("linux-sockets/writes-would-block", this, () => _writesWouldBlockCount) { DisplayName = "Writes Would Block" }; - } - - internal void OnReadNonBlocking() => Interlocked.Increment(ref _readsNonBlockingCount); - internal void OnReadWoudlBlock() => Interlocked.Increment(ref _readsWouldBlockCount); - internal void OnWriteNonBlocking() => Interlocked.Increment(ref _writesNonBlockingCount); - internal void OnWriteWoudlBlock() => Interlocked.Increment(ref _writesWouldBlockCount); - } - internal static class SocketPal { public const bool SupportsMultipleConnectAttempts = false; @@ -744,7 +711,6 @@ public static unsafe bool TryCompleteReceive(SafeSocketHandle socket, Span { bytesReceived = received; errorCode = SocketError.Success; - LinuxSocketsEventSource.Log.OnReadNonBlocking(); return true; } @@ -756,7 +722,6 @@ public static unsafe bool TryCompleteReceive(SafeSocketHandle socket, Span return true; } - LinuxSocketsEventSource.Log.OnReadWoudlBlock(); errorCode = SocketError.Success; return false; } @@ -803,7 +768,6 @@ public static unsafe bool TryCompleteReceiveFrom(SafeSocketHandle socket, Span return true; } - LinuxSocketsEventSource.Log.OnWriteWoudlBlock(); errorCode = successfulSend ? SocketError.Success : SocketError.WouldBlock; return false; } @@ -933,7 +895,6 @@ public static bool TryCompleteSendTo(SafeSocketHandle socket, ReadOnlySpan if (isComplete) { errorCode = SocketError.Success; - LinuxSocketsEventSource.Log.OnWriteNonBlocking(); return true; } From b0d01e5596ae1eb6a55b11dba1c9c63895e0f865 Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Fri, 3 Jul 2020 14:34:51 +0200 Subject: [PATCH 07/12] the receive queue can not be initialized with Waiting state as it affects first non-blocking sync call --- .../src/System/Net/Sockets/SocketAsyncContext.Unix.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index 4001c05f157369..d19ee159e3d309 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -727,12 +727,12 @@ private enum QueueState : int public bool IsNextOperationSynchronous_Speculative => _isNextOperationSynchronous; - public void Init(bool isReady) + public void Init() { Debug.Assert(_queueLock == null); _queueLock = new object(); - _state = isReady ? QueueState.Ready : QueueState.Waiting; + _state = QueueState.Ready; _sequenceNumber = 0; } @@ -1205,8 +1205,8 @@ public SocketAsyncContext(SafeSocketHandle socket) { _socket = socket; - _receiveQueue.Init(isReady: false); - _sendQueue.Init(isReady: true); + _receiveQueue.Init(); + _sendQueue.Init(); } public bool PreferInlineCompletions From a48f383ad50f5ffdce6d2d5a0d5be28a9ab826d9 Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Fri, 3 Jul 2020 19:09:07 +0200 Subject: [PATCH 08/12] polishing --- .../Net/Sockets/SocketAsyncContext.Unix.cs | 44 +++++++++++++++---- 1 file changed, 36 insertions(+), 8 deletions(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index d19ee159e3d309..04ee251ef92adb 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -311,7 +311,8 @@ public void DoAbort() ErrorCode = SocketError.OperationAborted; } - internal virtual bool IsPartialSuccess => false; + // returns true when read()|write() returned lower number of bytes than the size of buffer + internal virtual bool HasExhaustedIoSpace => false; protected abstract void Abort(); @@ -371,6 +372,8 @@ private sealed class BufferMemorySendOperation : SendOperation public BufferMemorySendOperation(SocketAsyncContext context) : base(context) { } + internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Buffer.Length; + protected override bool DoTryComplete(SocketAsyncContext context) { int bufferIndex = 0; @@ -429,6 +432,8 @@ private sealed unsafe class BufferPtrSendOperation : SendOperation public BufferPtrSendOperation(SocketAsyncContext context) : base(context) { } + internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && Count > 0; // TryCompleteSendTo modifies Count + protected override bool DoTryComplete(SocketAsyncContext context) { int bufferIndex = 0; @@ -459,7 +464,7 @@ private sealed class BufferMemoryReceiveOperation : ReceiveOperation public BufferMemoryReceiveOperation(SocketAsyncContext context) : base(context) { } - internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Buffer.Length; + internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Buffer.Length; protected override bool DoTryComplete(SocketAsyncContext context) { @@ -540,7 +545,7 @@ private sealed unsafe class BufferPtrReceiveOperation : ReceiveOperation public BufferPtrReceiveOperation(SocketAsyncContext context) : base(context) { } - internal override bool IsPartialSuccess => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Length; + internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Length; protected override bool DoTryComplete(SocketAsyncContext context) => SocketPal.TryCompleteReceiveFrom(context._socket, new Span(BufferPtr, Length), null, Flags, SocketAddress, ref SocketAddressLen, out BytesTransferred, out ReceivedFlags, out ErrorCode); @@ -1047,15 +1052,38 @@ public OperationResult ProcessQueuedOperation(TOperation op) // No more operations to process _tail = null; _isNextOperationSynchronous = false; - _state = op.IsPartialSuccess ? QueueState.Waiting : QueueState.Ready; - _sequenceNumber++; + + if (op.HasExhaustedIoSpace) + { + // the buffer that given operation was using was bigger than + // the number of bytes returned from read()|write() + // so instead of setting the state to Ready and having the next read()|write() + // return EWOULDBLOCK, we set the state to Waiting and just wait + // for new epoll notification which is going to set it to Ready + // see the Q&A section of https://www.man7.org/linux/man-pages/man7/epoll.7.html for more + _state = QueueState.Waiting; + } + else + { + _state = QueueState.Ready; + _sequenceNumber++; + } Trace(context, $"Exit (finished queue)"); } else { - // Pop current operation and advance to next - nextOp = _tail.Next = op.Next; - _isNextOperationSynchronous = nextOp.Event != null; + // Pop current operation + _tail.Next = op.Next; + _isNextOperationSynchronous = op.Next != null; + + if (op.HasExhaustedIoSpace) + { + _state = QueueState.Waiting; + } + else + { + nextOp = _tail.Next; // will be dispatched outside of lock + } } } } From 44cf3c2faf6ff2ef2111110883307cb14b2fd776 Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Fri, 3 Jul 2020 19:20:45 +0200 Subject: [PATCH 09/12] bug fix? --- .../src/System/Net/Sockets/SocketAsyncContext.Unix.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index 04ee251ef92adb..a944d93d29ac4a 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -437,7 +437,7 @@ public BufferPtrSendOperation(SocketAsyncContext context) : base(context) { } protected override bool DoTryComplete(SocketAsyncContext context) { int bufferIndex = 0; - return SocketPal.TryCompleteSendTo(context._socket, new ReadOnlySpan(BufferPtr, Offset + Count), null, ref bufferIndex, ref Offset, ref Count, Flags, SocketAddress, SocketAddressLen, ref BytesTransferred, out ErrorCode); + return SocketPal.TryCompleteSendTo(context._socket, new ReadOnlySpan(BufferPtr + Offset, Count), null, ref bufferIndex, ref Offset, ref Count, Flags, SocketAddress, SocketAddressLen, ref BytesTransferred, out ErrorCode); } } From 7793da07a5e890447aa71c82e5edefa78c588145 Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Mon, 6 Jul 2020 12:01:33 +0200 Subject: [PATCH 10/12] fixing an error that Ive introduced --- .../src/System/Net/Sockets/SocketAsyncContext.Unix.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index a944d93d29ac4a..58502afe4e0680 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -1074,7 +1074,7 @@ public OperationResult ProcessQueuedOperation(TOperation op) { // Pop current operation _tail.Next = op.Next; - _isNextOperationSynchronous = op.Next != null; + _isNextOperationSynchronous = op.Next.Event != null; if (op.HasExhaustedIoSpace) { From 79085f766391ae95ba16783d4d5e2513d93760eb Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Mon, 6 Jul 2020 15:28:53 +0200 Subject: [PATCH 11/12] it was not a typo --- .../src/System/Net/Sockets/SocketAsyncContext.Unix.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index 58502afe4e0680..e8aa62bd3ced5a 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -437,7 +437,7 @@ public BufferPtrSendOperation(SocketAsyncContext context) : base(context) { } protected override bool DoTryComplete(SocketAsyncContext context) { int bufferIndex = 0; - return SocketPal.TryCompleteSendTo(context._socket, new ReadOnlySpan(BufferPtr + Offset, Count), null, ref bufferIndex, ref Offset, ref Count, Flags, SocketAddress, SocketAddressLen, ref BytesTransferred, out ErrorCode); + return SocketPal.TryCompleteSendTo(context._socket, new ReadOnlySpan(BufferPtr, Offset + Count), null, ref bufferIndex, ref Offset, ref Count, Flags, SocketAddress, SocketAddressLen, ref BytesTransferred, out ErrorCode); } } From afe8086fc94ff7b5ace07b8421a731106cc2f0dc Mon Sep 17 00:00:00 2001 From: Adam Sitnik Date: Tue, 7 Jul 2020 14:51:04 +0200 Subject: [PATCH 12/12] state can be set to Waiting only if IO space has been exhausted AND we did not receive any new epoll notifications in the meantime (state==processing, _sequenceNumber++) --- .../Net/Sockets/SocketAsyncContext.Unix.cs | 45 ++++++++----------- 1 file changed, 18 insertions(+), 27 deletions(-) diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs index e8aa62bd3ced5a..72f865c238986e 100644 --- a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs +++ b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs @@ -120,6 +120,7 @@ private enum State } private int _state; // Actually AsyncOperation.State. + protected int _observedSequenceNumber; #if DEBUG private int _callbackQueued; // When non-zero, the callback has been queued. @@ -312,7 +313,7 @@ public void DoAbort() } // returns true when read()|write() returned lower number of bytes than the size of buffer - internal virtual bool HasExhaustedIoSpace => false; + internal virtual bool HasExhaustedIoSpace(int currentSequenceNumber) => false; protected abstract void Abort(); @@ -320,6 +321,8 @@ public void DoAbort() public abstract void InvokeCallback(bool allowPooling); + public void SetSequenceNumber(int sequenceNumber) => _observedSequenceNumber = sequenceNumber; + [Conditional("SOCKETASYNCCONTEXT_TRACE")] public void Trace(string message, [CallerMemberName] string? memberName = null) { @@ -372,7 +375,8 @@ private sealed class BufferMemorySendOperation : SendOperation public BufferMemorySendOperation(SocketAsyncContext context) : base(context) { } - internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Buffer.Length; + internal override bool HasExhaustedIoSpace(int currentSequenceNumber) => _observedSequenceNumber == currentSequenceNumber + && ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Buffer.Length; protected override bool DoTryComplete(SocketAsyncContext context) { @@ -432,7 +436,8 @@ private sealed unsafe class BufferPtrSendOperation : SendOperation public BufferPtrSendOperation(SocketAsyncContext context) : base(context) { } - internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && Count > 0; // TryCompleteSendTo modifies Count + internal override bool HasExhaustedIoSpace(int currentSequenceNumber) => _observedSequenceNumber == currentSequenceNumber + && ErrorCode == SocketError.Success && BytesTransferred > 0 && Count > 0; // TryCompleteSendTo modifies Count protected override bool DoTryComplete(SocketAsyncContext context) { @@ -464,7 +469,8 @@ private sealed class BufferMemoryReceiveOperation : ReceiveOperation public BufferMemoryReceiveOperation(SocketAsyncContext context) : base(context) { } - internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Buffer.Length; + internal override bool HasExhaustedIoSpace(int currentSequenceNumber) => _observedSequenceNumber == currentSequenceNumber + && ErrorCode == SocketError.Success && BytesTransferred >= 0 && BytesTransferred < Buffer.Length; protected override bool DoTryComplete(SocketAsyncContext context) { @@ -545,7 +551,8 @@ private sealed unsafe class BufferPtrReceiveOperation : ReceiveOperation public BufferPtrReceiveOperation(SocketAsyncContext context) : base(context) { } - internal override bool HasExhaustedIoSpace => ErrorCode == SocketError.Success && BytesTransferred > 0 && BytesTransferred < Length; + internal override bool HasExhaustedIoSpace(int currentSequenceNumber) => _observedSequenceNumber == currentSequenceNumber + && ErrorCode == SocketError.Success && BytesTransferred >= 0 && BytesTransferred < Length; protected override bool DoTryComplete(SocketAsyncContext context) => SocketPal.TryCompleteReceiveFrom(context._socket, new Span(BufferPtr, Length), null, Flags, SocketAddress, ref SocketAddressLen, out BytesTransferred, out ReceivedFlags, out ErrorCode); @@ -745,27 +752,11 @@ public void Init() // observedSequenceNumber must be passed to StartAsyncOperation. public bool IsReady(SocketAsyncContext context, out int observedSequenceNumber) { - // It is safe to read _state and _sequence without using Lock. - // - The return value is soley based on Volatile.Read of _state. - // - The Volatile.Read of _sequenceNumber ensures we read a value before executing the operation. - // This is needed to retry the operation in StartAsyncOperation in case the _sequenceNumber incremented. - // - Because no Lock is taken, it is possible we observe a sequence number increment before the state - // becomes Ready. When that happens, observedSequenceNumber is decremented, and StartAsyncOperation will - // execute the operation because the sequence number won't match. - - Debug.Assert(sizeof(QueueState) == sizeof(int)); - QueueState state = (QueueState)Volatile.Read(ref Unsafe.As(ref _state)); - observedSequenceNumber = Volatile.Read(ref _sequenceNumber); - - bool isReady = state == QueueState.Ready || state == QueueState.Stopped; - if (!isReady) + using (Lock()) { - observedSequenceNumber--; + observedSequenceNumber = _sequenceNumber; + return _state == QueueState.Ready || _state == QueueState.Stopped; } - - Trace(context, $"{isReady}"); - - return isReady; } // Return true for pending, false for completed synchronously (including failure and abort) @@ -879,6 +870,7 @@ public bool StartAsyncOperation(SocketAsyncContext context, TOperation operation } op = _tail.Next; + op.SetSequenceNumber(_sequenceNumber); Debug.Assert(_isNextOperationSynchronous == (op.Event != null)); if (skipAsyncEvents && !_isNextOperationSynchronous) { @@ -1053,7 +1045,7 @@ public OperationResult ProcessQueuedOperation(TOperation op) _tail = null; _isNextOperationSynchronous = false; - if (op.HasExhaustedIoSpace) + if (op.HasExhaustedIoSpace(currentSequenceNumber: _sequenceNumber)) { // the buffer that given operation was using was bigger than // the number of bytes returned from read()|write() @@ -1066,7 +1058,6 @@ public OperationResult ProcessQueuedOperation(TOperation op) else { _state = QueueState.Ready; - _sequenceNumber++; } Trace(context, $"Exit (finished queue)"); } @@ -1076,7 +1067,7 @@ public OperationResult ProcessQueuedOperation(TOperation op) _tail.Next = op.Next; _isNextOperationSynchronous = op.Next.Event != null; - if (op.HasExhaustedIoSpace) + if (op.HasExhaustedIoSpace(currentSequenceNumber: _sequenceNumber)) { _state = QueueState.Waiting; }