Do not call into MsQuic inside a lock #67037
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -119,6 +119,7 @@ internal MsQuicStream(MsQuicConnection.State connectionState, SafeMsQuicStreamHa | |
| _state.StateGCHandle = GCHandle.Alloc(_state); | ||
| try | ||
| { | ||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| MsQuicApi.Api.SetCallbackHandlerDelegate( | ||
| _state.Handle, | ||
| s_streamDelegate, | ||
|
|
@@ -164,6 +165,7 @@ internal MsQuicStream(MsQuicConnection.State connectionState, QUIC_STREAM_OPEN_F | |
|
|
||
| try | ||
| { | ||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| uint status = MsQuicApi.Api.StreamOpenDelegate( | ||
| connectionState.Handle, | ||
| flags, | ||
|
|
@@ -173,6 +175,7 @@ internal MsQuicStream(MsQuicConnection.State connectionState, QUIC_STREAM_OPEN_F | |
|
|
||
| QuicExceptionHelpers.ThrowIfFailed(status, "Failed to open stream to peer."); | ||
|
|
||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| status = MsQuicApi.Api.StreamStartDelegate(_state.Handle, QUIC_STREAM_START_FLAGS.FAIL_BLOCKED); | ||
| QuicExceptionHelpers.ThrowIfFailed(status, "Could not start stream."); | ||
| } | ||
|
|
@@ -227,7 +230,7 @@ internal override int WriteTimeout | |
| get | ||
| { | ||
| ThrowIfDisposed(); | ||
| return _writeTimeout; | ||
| } | ||
| set | ||
| { | ||
|
|
@@ -420,6 +423,8 @@ internal override ValueTask<int> ReadAsync(Memory<byte> destination, Cancellatio | |
| long abortError; | ||
| bool preCanceled = false; | ||
|
|
||
| int bytesRead = -1; | ||
| bool reenableReceive = false; | ||
| lock (_state) | ||
| { | ||
| initialReadState = _state.ReadState; | ||
|
|
@@ -482,22 +487,32 @@ internal override ValueTask<int> ReadAsync(Memory<byte> destination, Cancellatio | |
| { | ||
| _state.ReadState = ReadState.None; | ||
|
|
||
| bytesRead = CopyMsQuicBuffersToUserBuffer(_state.ReceiveQuicBuffers.AsSpan(0, _state.ReceiveQuicBuffersCount), destination.Span); | ||
|
|
||
| if (bytesRead != _state.ReceiveQuicBuffersTotalBytes) | ||
| { | ||
| // Need to re-enable receives because MsQuic will pause them when we don't consume the entire buffer. | ||
| reenableReceive = true; | ||
| } | ||
| else if (_state.ReceiveIsFinal) | ||
| { | ||
| // This was a final message and we've consumed everything. We can complete the state without waiting for PEER_SEND_SHUTDOWN | ||
| _state.ReadState = ReadState.ReadsCompleted; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // methods below need to be called outside of the lock | ||
| if (bytesRead > -1) | ||
|
There was a problem hiding this comment. Can this ever be <= -1? We didn't have check like this before if I understand it correctly. There was a problem hiding this comment. it will be -1 if There was a problem hiding this comment. I see now, it's its initial value set before the lock. |
||
| { | ||
| ReceiveComplete(bytesRead); | ||
|
|
||
| if (reenableReceive) | ||
| { | ||
| EnableReceive(); | ||
|
There was a problem hiding this comment. If we're calling this outside of the lock, is there something else that ensures we're not racing with another thread to disable receives? There was a problem hiding this comment. Stephen's right. Just a hint here (I haven't thought this fully through), we're using the cc @CarnaViire There was a problem hiding this comment. TL;DR: RECV event is the only thing disabling receives. It seems to me that it should be ok due to additional guard from msquic side (no new RECV event will come until ReceiveComplete+EnableReceive is called). But I don't like how fragile it looks 😢 Note that the change is only in a branch for "IndividualReadComplete" state. It means it happens AFTER some data already arrived in RECV event. We do have a guard for parallel reads which looks exactly at ReadState. So it will throw if ReadAsync is called while state is PendingRead (no data available and there's already a waiting read). By moving ReceiveComplete+EnableReceive out of the lock, we allow a time where state is already changed from "IndividualReadComplete" to "None", but ongoing "first" ReceiveAsync function is not exited yet and ReceiveComplete+EnableReceive are not called yet (but all data is already copied). So a second ReceiveAsync might enter. It would see state "None" which would mean data is not available. If it grabs the lock before ReceiveComplete+EnableReceive are called (i.e. RECV event is still not possible), it would change the state to PendingRead (waiting for data), store the destination buffer reference, register cancellation and return a task to wait. All the things "None" branch touches are not touched in the remainder of the first ReceiveAsync (between exiting from the lock and returning new value task). The bottom line is I don't think there's any problem in ReadAsync vs HandleEventRecv race. But we might want to rethink guard against parallel reads. There was a problem hiding this comment. So if I understand correctly, the only think that can go wrong with this change is parallel ReadAsync operations, which user code is not supposed to do anyway... Should I add the guards against parallel operations in this PR, or should we do that separately? @ManickaP @stephentoub , thoughts? There was a problem hiding this comment. I would say leave it as is for now. We have a separate issue for proper guards against parallel reads and writes #52627 |
||
| } | ||
|
|
||
| return new ValueTask<int>(bytesRead); | ||
| } | ||
|
|
||
| // All success scenarios returned at this point. Failure scenarios below: | ||
|
|
@@ -510,7 +525,7 @@ internal override ValueTask<int> ReadAsync(Memory<byte> destination, Cancellatio | |
| ex = new InvalidOperationException("Only one read is supported at a time."); | ||
| break; | ||
| case ReadState.Aborted: | ||
| ex = preCanceled ? new OperationCanceledException(cancellationToken) : | ||
| ThrowHelper.GetStreamAbortedException(abortError); | ||
| break; | ||
| case ReadState.ConnectionClosed: | ||
|
|
@@ -609,6 +624,7 @@ internal override void AbortWrite(long errorCode) | |
|
|
||
| private void StartShutdown(QUIC_STREAM_SHUTDOWN_FLAGS flags, long errorCode) | ||
| { | ||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| uint status = MsQuicApi.Api.StreamShutdownDelegate(_state.Handle, flags, errorCode); | ||
| QuicExceptionHelpers.ThrowIfFailed(status, "StreamShutdown failed."); | ||
| } | ||
|
|
@@ -818,15 +834,17 @@ private void Dispose(bool disposing) | |
| { | ||
| // Handle race condition when stream can be closed handling SHUTDOWN_COMPLETE. | ||
| StartShutdown(QUIC_STREAM_SHUTDOWN_FLAGS.GRACEFUL, errorCode: 0); | ||
| } | ||
| catch (ObjectDisposedException) { }; | ||
| } | ||
|
|
||
| if (abortRead) | ||
| { | ||
| try | ||
| { | ||
| StartShutdown(QUIC_STREAM_SHUTDOWN_FLAGS.ABORT_RECEIVE, 0xffffffff); | ||
| } | ||
| catch (ObjectDisposedException) { }; | ||
| } | ||
|
|
||
| if (completeRead) | ||
|
|
@@ -845,6 +863,7 @@ private void Dispose(bool disposing) | |
|
|
||
| private void EnableReceive() | ||
| { | ||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
|
There was a problem hiding this comment. Instead of repeating all this, it might be a little more readable to just call a method? [Conditional("DEBUG")]
internal void AssertMonitorNotEntered(object obj)
{
Debug.Assert(!Monitor.IsEntered(obj), "Monitor was unexpectedly held");
}There was a problem hiding this comment. We're using asserts like this all over the place, including S.N.Http. So I'm fine with this as-is, it's still single line. What I do think is that the message is unnecessary here and we rarely include something like this. There was a problem hiding this comment. (If we do #65965, it'll also be redundant in this case, as with or without the message you'd get the same assert.) |
||
| uint status = MsQuicApi.Api.StreamReceiveSetEnabledDelegate(_state.Handle, enabled: true); | ||
| QuicExceptionHelpers.ThrowIfFailed(status, "StreamReceiveSetEnabled failed."); | ||
| } | ||
|
|
@@ -1289,6 +1308,7 @@ private unsafe ValueTask SendReadOnlyMemoryAsync( | |
| _state.BufferArrays[0] = handle; | ||
| _state.SendBufferCount = 1; | ||
|
|
||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| uint status = MsQuicApi.Api.StreamSendDelegate( | ||
| _state.Handle, | ||
| quicBuffers, | ||
|
|
@@ -1352,6 +1372,7 @@ private unsafe ValueTask SendReadOnlySequenceAsync( | |
| ++count; | ||
| } | ||
|
|
||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| uint status = MsQuicApi.Api.StreamSendDelegate( | ||
| _state.Handle, | ||
| quicBuffers, | ||
|
|
@@ -1412,6 +1433,7 @@ private unsafe ValueTask SendReadOnlyMemoryListAsync( | |
| _state.BufferArrays[i] = handle; | ||
| } | ||
|
|
||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| uint status = MsQuicApi.Api.StreamSendDelegate( | ||
| _state.Handle, | ||
| quicBuffers, | ||
|
|
@@ -1434,6 +1456,7 @@ private unsafe ValueTask SendReadOnlyMemoryListAsync( | |
|
|
||
| private void ReceiveComplete(int bufferLength) | ||
| { | ||
| Debug.Assert(!Monitor.IsEntered(_state), "!Monitor.IsEntered(_state)"); | ||
| uint status = MsQuicApi.Api.StreamReceiveCompleteDelegate(_state.Handle, (ulong)bufferLength); | ||
| QuicExceptionHelpers.ThrowIfFailed(status, "Could not complete receive call."); | ||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What led you to even expand more the messages in asserts after Stephen's comment?
It'll become redundant.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I thought it would be more useful to have a message handy when something crashes, I thought we generally tend to include the message with the assert in this repo
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's exactly the other way around, just grep for
Debug.Assertin the repo. And if it contains the message, I mostly see some additional info and not just the copy of the condition.