Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 51 additions & 2 deletions src/Servers/Kestrel/Core/src/Internal/Http3/Http3Stream.cs
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,11 @@ internal abstract partial class Http3Stream : HttpProtocol, IHttp3Stream, IHttpS
private readonly ManualResetValueTaskSource<object?> _appCompletedTaskSource = new();
private readonly object _completionLock = new();

// Published under _completionLock by the abort that first transitions the stream into the Aborted state. It is completed
// once that abort has finished running its side-effects (performed outside the lock). Request finalization waits on
// it before pooling the stream so a late abort can't tear down a transport that has been reused.
private TaskCompletionSource? _abortCompletedTcs;

protected RequestHeaderParsingState _requestHeaderParsingState;

public bool EndStreamReceived => (_completionState & StreamCompletionFlags.EndStreamReceived) == StreamCompletionFlags.EndStreamReceived;
Expand Down Expand Up @@ -101,6 +106,7 @@ public void Initialize(Http3StreamContext context)
_eagerRequestHeadersParsedLimit = ServerOptions.Limits.MaxRequestHeaderCount * 2;
_isMethodConnect = false;
_completionState = default;
_abortCompletedTcs = null;
StreamTimeoutTimestamp = 0;

if (_frameWriter == null)
Expand Down Expand Up @@ -147,8 +153,17 @@ public void Abort(ConnectionAbortedException abortReason, Http3ErrorCode errorCo

private void AbortCore(Exception exception, Http3ErrorCode errorCode)
{
TaskCompletionSource abortCompleted;

lock (_completionLock)
{
// Only the completion-state transition is performed under _completionLock. The abort
// side-effects below must run *outside* the lock: _http3Output.Stop() (and the frame
// writer/transport teardown) acquire Http3OutputProducer._dataWriterLock, which is taken
// in the opposite order on the inline output path (Http3OutputProducer.FlushAsync holds
// _dataWriterLock and, via the inline data pipe pump -> QuicStreamContext.FireStreamClosed,
// calls back into AbortCore -> _completionLock). Holding _completionLock across Stop() would
// therefore create a _completionLock <-> _dataWriterLock deadlock. This mirrors Http2Stream.
if (IsCompleted || IsAborted)
{
return;
Expand All @@ -161,7 +176,14 @@ private void AbortCore(Exception exception, Http3ErrorCode errorCode)
return;
}

if (!(exception is ConnectionAbortedException abortReason))
// Publish before releasing the lock so request finalization can observe this in-flight abort
// and wait for the side-effects below to finish before the stream is pooled and reused.
abortCompleted = _abortCompletedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
}

try
{
if (exception is not ConnectionAbortedException abortReason)
{
abortReason = new ConnectionAbortedException(exception.Message, exception);
}
Expand All @@ -186,6 +208,28 @@ private void AbortCore(Exception exception, Http3ErrorCode errorCode)
// Abort framewriter and underlying transport after stopping output.
_frameWriter.Abort(abortReason);
}
finally
{
abortCompleted.TrySetResult();
}
}

// Marks the stream Completed and waits for any in-flight abort's side-effects to finish. Called by
// request finalization before the transport is drained, disposed and pooled. Setting Completed under
// _completionLock closes the door: any abort arriving afterwards sees IsCompleted and no-ops, so it
// can't mutate a transport that has been reused. An abort that won the race before the door closed is
// captured here and awaited so its Stop()/frame writer Abort() land on this stream, not a reused one.
private ValueTask CompleteAndWaitForAbortAsync()
{
TaskCompletionSource? abortCompleted;

lock (_completionLock)
{
_completionState |= StreamCompletionFlags.Completed;
abortCompleted = _abortCompletedTcs;
}

return abortCompleted is null ? default : new ValueTask(abortCompleted.Task);
}

protected override void OnErrorAfterResponseStarted()
Expand Down Expand Up @@ -703,11 +747,16 @@ public async Task ProcessRequestAsync<TContext>(IHttpApplication<TContext> appli
}
finally
{
// Wait for any in-flight abort to finish mutating this stream's transport/output before
// we drain, dispose and pool it; otherwise a late abort could tear down a stream that has
// already been reused for another request. This also marks the stream Completed so any
// subsequent abort becomes a no-op.
await CompleteAndWaitForAbortAsync();

// Drain transports and dispose.
await _context.StreamContext.DisposeAsync();

// Tells the connection to remove the stream from its active collection.
ApplyCompletionFlag(StreamCompletionFlags.Completed);
_context.StreamLifetimeHandler.OnStreamCompleted(this);

// If we have a webtransport session on this stream, end it
Expand Down
Loading