From b5f8d1aa2b1e90d1e4e792d40f1b4cf8b70eba4f Mon Sep 17 00:00:00 2001 From: vsadov <8218165+VSadov@users.noreply.github.com> Date: Sun, 5 Jul 2026 00:31:33 -0700 Subject: [PATCH 1/8] rent memory outside the lock --- .../src/System.IO.Pipelines.csproj | 8 +- .../src/System/IO/Pipelines/Pipe.cs | 283 ++++++++++++------ .../src/System/IO/Pipelines/PipeOptions.cs | 20 +- .../tests/BufferSegmentPoolTest.cs | 4 +- 4 files changed, 205 insertions(+), 110 deletions(-) diff --git a/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj b/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj index e0cb743c86e261..42035d63812b5d 100644 --- a/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj +++ b/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj @@ -1,4 +1,4 @@ - + $(NetCoreAppCurrent);$(NetCoreAppPrevious);$(NetCoreAppMinimum);netstandard2.0;$(NetFrameworkMinimum) @@ -46,6 +46,12 @@ System.IO.Pipelines.PipeReader + + + diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs index e375311e545598..4bb46bb44736e7 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. using System.Buffers; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Runtime.CompilerServices; @@ -24,8 +25,10 @@ public sealed partial class Pipe private static readonly SendOrPostCallback s_syncContextExecuteWithoutExecutionContextCallback = ExecuteWithoutExecutionContext!; private static readonly Action s_scheduleWithExecutionContextCallback = ExecuteWithExecutionContext!; - // Mutable struct! Don't make this readonly - private BufferSegmentStack _bufferSegmentPool; + // Pool of reusable BufferSegment instances. + // We are using SPSC queue here to reduce interaction of reader and writer threads when + // acquiring/releasing the segments. + private readonly SingleProducerSingleConsumerQueue _bufferSegmentPool = new SingleProducerSingleConsumerQueue(); private readonly DefaultPipeReader _reader; private readonly DefaultPipeWriter _writer; @@ -97,8 +100,6 @@ public Pipe(PipeOptions options) ThrowHelper.ThrowArgumentNullException(ExceptionArgument.options); } - _bufferSegmentPool = new BufferSegmentStack(options.InitialSegmentPoolSize); - _operationState = default; _readerCompletion = default; _writerCompletion = default; @@ -171,24 +172,48 @@ private void AllocateWriteHeadIfNeeded(int sizeHint) private void AllocateWriteHeadSynchronized(int sizeHint) { + // Rent the backing memory outside the lock, but only when writing is active. + // + // While writing is active, the writer thread exclusively owns _writingHead and + // _writingHeadMemory: the reader only releases them (in AdvanceReader) when writing + // is NOT active. That makes the reads below race-free, and it makes them stable - the + // reader can only shrink _writingHeadMemory (to default) and cannot even do that while + // writing is active, and this (single) writer thread does not change it between here + // and taking the lock. So if there isn't enough room now there won't be enough under + // the lock either: a buffer rented here is guaranteed to be used and never needs to be + // returned. Keeping the (potentially contended) pool rental off the lock is the point. + // + // Note we deliberately do NOT acquire a BufferSegment here. The segment pool is a + // single-producer/single-consumer queue, so an unused segment could not be safely + // returned from this thread. Segments are taken (GetOrCreateSegment) only under the + // lock, at the exact point we are certain to use them. Acquiring via SPSC should be + // too cheap to bother that it happens under lock. + object? prerented = null; + if (_operationState.IsWritingActive && + (_writingHeadMemory.Length == 0 || _writingHeadMemory.Length < sizeHint)) + { + prerented = RentMemoryUnsynchronized(sizeHint); + } + lock (SyncObj) { _operationState.BeginWrite(); - if (_writingHead == null) + int bytesLeftInBuffer = _writingHeadMemory.Length; + if (bytesLeftInBuffer == 0 || bytesLeftInBuffer < sizeHint) { - // We need to allocate memory to write since nobody has written before - BufferSegment newSegment = AllocateSegment(sizeHint); + if (_writingHead is null) + { + // Writing wasn't active on entry, so we could not have pre-rented. + Debug.Assert(prerented is null); - // Set all the pointers - _writingHead = _readHead = _readTail = newSegment; - _lastExaminedIndex = 0; - } - else - { - int bytesLeftInBuffer = _writingHeadMemory.Length; + BufferSegment newSegment = GetOrCreateSegment(); + AttachMemory(newSegment, prerented, sizeHint); - if (bytesLeftInBuffer == 0 || bytesLeftInBuffer < sizeHint) + _writingHead = _readHead = _readTail = newSegment; + _lastExaminedIndex = 0; + } + else { if (_writingHeadBytesBuffered > 0) { @@ -197,29 +222,37 @@ private void AllocateWriteHeadSynchronized(int sizeHint) _writingHeadBytesBuffered = 0; } - if (_writingHead.Length == 0) + if (_writingHead.End == 0) { - // If we got here that means Advance was called with 0 bytes or GetMemory was called again without any writes occurring - // And, the newly requested memory size is greater than our unused segments internal memory buffer - // So we should reuse the BufferSegment and replace the memory it's holding, this way ReadAsync will not receive a buffer with one segment being empty + // Advance was called with 0 bytes, or GetMemory was called again without + // any writes occurring, and the requested size is larger than the unused + // head's buffer. Reuse the BufferSegment and swap out its memory so + // ReadAsync will not observe an empty segment. _writingHead.ResetMemory(); - RentMemory(_writingHead, sizeHint); + AttachMemory(_writingHead, prerented, sizeHint); } else { - BufferSegment newSegment = AllocateSegment(sizeHint); + BufferSegment newSegment = GetOrCreateSegment(); + AttachMemory(newSegment, prerented, sizeHint); _writingHead.SetNext(newSegment); _writingHead = newSegment; } } } + else + { + // The head still has room. Only reachable when writing wasn't active on entry; + // otherwise we would have pre-rented and this branch would not be taken. + Debug.Assert(prerented is null); + } } } private BufferSegment AllocateSegment(int sizeHint) { - BufferSegment newSegment = CreateSegmentUnsynchronized(); + BufferSegment newSegment = GetOrCreateSegment(); RentMemory(newSegment, sizeHint); @@ -256,6 +289,50 @@ private void RentMemory(BufferSegment segment, int sizeHint) _writingHeadMemory = segment.AvailableMemory; } + // Rents backing memory without touching any shared writer state, so it is safe to call + // outside the lock. Returns either an IMemoryOwner (from the configured pool) or a + // byte[] (from the shared array pool); AttachMemory understands both. + private object RentMemoryUnsynchronized(int sizeHint) + { + MemoryPool? pool = null; + int maxSize = -1; + + if (!_options.IsDefaultSharedMemoryPool) + { + pool = _options.Pool; + maxSize = pool.MaxBufferSize; + } + + return sizeHint <= maxSize + ? pool!.Rent(GetSegmentSize(sizeHint, maxSize)) + : ArrayPool.Shared.Rent(GetSegmentSize(sizeHint)); + } + + // Attaches memory to a segment and updates _writingHeadMemory. If memory was already rented + // outside the lock (prerented) it is attached; otherwise memory is rented here. Must be + // called under the lock. + private void AttachMemory(BufferSegment segment, object? prerented, int sizeHint) + { + Debug.Assert(segment.MemoryOwner is null); + + switch (prerented) + { + case IMemoryOwner owner: + segment.SetOwnedMemory(owner); + _writingHeadMemory = segment.AvailableMemory; + break; + case byte[] array: + segment.SetOwnedMemory(array); + _writingHeadMemory = segment.AvailableMemory; + break; + default: + Debug.Assert(prerented is null); + // Nothing pre-rented (writing was not active); rent under the lock. + RentMemory(segment, sizeHint); + break; + } + } + private int GetSegmentSize(int sizeHint, int maxBufferSize = int.MaxValue) { // First we need to handle case where hint is smaller than minimum segment size @@ -265,9 +342,9 @@ private int GetSegmentSize(int sizeHint, int maxBufferSize = int.MaxValue) return adjustedToMaximumSize; } - private BufferSegment CreateSegmentUnsynchronized() + private BufferSegment GetOrCreateSegment() { - if (_bufferSegmentPool.TryPop(out BufferSegment? segment)) + if (_bufferSegmentPool.TryDequeue(out BufferSegment? segment)) { return segment; } @@ -275,15 +352,17 @@ private BufferSegment CreateSegmentUnsynchronized() return new BufferSegment(); } - private void ReturnSegmentUnsynchronized(BufferSegment segment) + private void ReturnSegment(BufferSegment segment) { Debug.Assert(segment != _readHead, "Returning _readHead segment that's in use!"); Debug.Assert(segment != _readTail, "Returning _readTail segment that's in use!"); Debug.Assert(segment != _writingHead, "Returning _writingHead segment that's in use!"); + // The check for the current pooled count may race with dequeing, + // but occasional overestimating is ok here. if (_bufferSegmentPool.Count < _options.MaxSegmentPoolSize) { - _bufferSegmentPool.Push(segment); + _bufferSegmentPool.Enqueue(segment); } } @@ -474,81 +553,89 @@ private void AdvanceReader(BufferSegment? consumedSegment, int consumedIndex, Bu CompletionData completionData = default; - lock (SyncObj) + try { - var examinedEverything = false; - if (examinedSegment == _readTail) - { - examinedEverything = examinedIndex == _readTailIndex; - } - - if (examinedSegment != null && _lastExaminedIndex >= 0) + lock (SyncObj) { - // This can be negative resulting in _unconsumedBytes increasing, this should be safe because we've already checked that - // examined >= consumed above, so we can't get into a state where we un-examine too much - long examinedBytes = BufferSegment.GetLength(_lastExaminedIndex, examinedSegment, examinedIndex); - long oldLength = _unconsumedBytes; - - _unconsumedBytes -= examinedBytes; - - // Store the absolute position - _lastExaminedIndex = examinedSegment.RunningIndex + examinedIndex; - - Debug.Assert(_unconsumedBytes >= 0, "Length has gone negative"); - Debug.Assert(ResumeWriterThreshold >= 1, "ResumeWriterThreshold is less than 1"); - - if (oldLength >= ResumeWriterThreshold && - _unconsumedBytes < ResumeWriterThreshold) + var examinedEverything = false; + if (examinedSegment == _readTail) { - // Should only release backpressure if we made forward progress - Debug.Assert(examinedBytes > 0); - _writerAwaitable.Complete(out completionData); + examinedEverything = examinedIndex == _readTailIndex; } - } - if (consumedSegment != null) - { - if (_readHead == null) + if (examinedSegment != null && _lastExaminedIndex >= 0) { - ThrowHelper.ThrowInvalidOperationException_AdvanceToInvalidCursor(); - return; - } + // This can be negative resulting in _unconsumedBytes increasing, this should be safe because we've already checked that + // examined >= consumed above, so we can't get into a state where we un-examine too much + long examinedBytes = BufferSegment.GetLength(_lastExaminedIndex, examinedSegment, examinedIndex); + long oldLength = _unconsumedBytes; - returnStart = _readHead; - returnEnd = consumedSegment; + _unconsumedBytes -= examinedBytes; - void MoveReturnEndToNextBlock() - { - BufferSegment? nextBlock = returnEnd!.NextSegment; - if (_readTail == returnEnd) - { - _readTail = nextBlock; - _readTailIndex = 0; - } + // Store the absolute position + _lastExaminedIndex = examinedSegment.RunningIndex + examinedIndex; - _readHead = nextBlock; - _readHeadIndex = 0; + Debug.Assert(_unconsumedBytes >= 0, "Length has gone negative"); + Debug.Assert(ResumeWriterThreshold >= 1, "ResumeWriterThreshold is less than 1"); - returnEnd = nextBlock; + if (oldLength >= ResumeWriterThreshold && + _unconsumedBytes < ResumeWriterThreshold) + { + // Should only release backpressure if we made forward progress + Debug.Assert(examinedBytes > 0); + _writerAwaitable.Complete(out completionData); + } } - if (consumedIndex == returnEnd.Length) + if (consumedSegment != null) { - // If the writing head isn't block we're about to return, then we can move to the next one - // and return this block safely - if (_writingHead != returnEnd) + if (_readHead == null) { - MoveReturnEndToNextBlock(); + ThrowHelper.ThrowInvalidOperationException_AdvanceToInvalidCursor(); + return; } - // If the writing head is the same as the block to be returned, then we need to make sure - // there's no pending write and that there's no buffered data for the writing head - else if (_writingHeadBytesBuffered == 0 && !_operationState.IsWritingActive) + + returnStart = _readHead; + returnEnd = consumedSegment; + + void MoveReturnEndToNextBlock() { - // Reset the writing head to null if it's the return block and we've consumed everything - _writingHead = null; - _writingHeadMemory = default; + BufferSegment? nextBlock = returnEnd!.NextSegment; + if (_readTail == returnEnd) + { + _readTail = nextBlock; + _readTailIndex = 0; + } - MoveReturnEndToNextBlock(); + _readHead = nextBlock; + _readHeadIndex = 0; + + returnEnd = nextBlock; + } + + if (consumedIndex == returnEnd.Length) + { + // If the writing head isn't block we're about to return, then we can move to the next one + // and return this block safely + if (_writingHead != returnEnd) + { + MoveReturnEndToNextBlock(); + } + // If the writing head is the same as the block to be returned, then we need to make sure + // there's no pending write and that there's no buffered data for the writing head + else if (_writingHeadBytesBuffered == 0 && !_operationState.IsWritingActive) + { + // Reset the writing head to null if it's the return block and we've consumed everything + _writingHead = null; + _writingHeadMemory = default; + + MoveReturnEndToNextBlock(); + } + else + { + _readHead = consumedSegment; + _readHeadIndex = consumedIndex; + } } else { @@ -556,31 +643,29 @@ void MoveReturnEndToNextBlock() _readHeadIndex = consumedIndex; } } - else + + // We reset the awaitable to not completed if we've examined everything the producer produced so far + // but only if writer is not completed yet + if (examinedEverything && !_writerCompletion.IsCompleted) { - _readHead = consumedSegment; - _readHeadIndex = consumedIndex; - } - } + Debug.Assert(_writerAwaitable.IsCompleted, "PipeWriter.FlushAsync isn't completed and will deadlock"); - // We reset the awaitable to not completed if we've examined everything the producer produced so far - // but only if writer is not completed yet - if (examinedEverything && !_writerCompletion.IsCompleted) - { - Debug.Assert(_writerAwaitable.IsCompleted, "PipeWriter.FlushAsync isn't completed and will deadlock"); + _readerAwaitable.SetUncompleted(); + } - _readerAwaitable.SetUncompleted(); + _operationState.EndRead(); } - + } + finally + { + // outside the lock: reset and return the segments while (returnStart != null && returnStart != returnEnd) { BufferSegment? next = returnStart.NextSegment; returnStart.Reset(); - ReturnSegmentUnsynchronized(returnStart); + ReturnSegment(returnStart); returnStart = next; } - - _operationState.EndRead(); } TrySchedule(WriterScheduler, completionData); diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs index c482240d54b610..59dd17b140e49a 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs @@ -34,14 +34,18 @@ public PipeOptions( { MinimumSegmentSize = minimumSegmentSize == -1 ? DefaultMinimumSegmentSize : minimumSegmentSize; - // TODO: These *should* be computed based on how much users want to buffer and the minimum segment size. Today we don't have a way - // to let users specify the maximum buffer size, so we pick a reasonable number based on defaults. They can influence - // how much gets buffered by increasing the minimum segment size. - - // With a default segment size of 4K this maps to 16K - InitialSegmentPoolSize = 4; - - // With a default segment size of 4K this maps to 1MB. If the pipe has large segments this will be bigger than 1MB... + // Cap the per-pipe segment-object pool to bound memory in degenerate cases. + // Buffers are returned to the MemoryPool on Reset() before a segment is pooled, so a + // pooled BufferSegment holds no buffer and is ~96 bytes on 64-bit. The cap therefore + // costs at most ~24 KB per pipe (256 * ~96 bytes); the backing memory is pooled separately. + // + // Normal pipes never approach this cap. The pool only grows to the peak number of + // simultaneously-live segments, which for a throttled pipe is roughly + // PauseWriterThreshold / MinimumSegmentSize - with the defaults that is + // 64 KB / 4 KB = ~16 segments. Reaching 256 requires either a very large + // PauseWriterThreshold (>= 256 * MinimumSegmentSize, e.g. 1 MB of unconsumed data at + // 4 KB segments) or an unbounded pipe (pauseWriterThreshold: 0) whose producer + // consistently outruns the consumer. MaxSegmentPoolSize = 256; // By default, we'll throttle the writer at 64K of buffered data diff --git a/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs b/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs index c89ae0e8f20249..76b089b66a0595 100644 --- a/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs +++ b/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs @@ -94,10 +94,10 @@ public async Task BufferSegmentsPooledUpToThreshold() _pipe.Reader.AdvanceTo(result.Buffer.End); - // Assert Pipe.MaxSegmentPoolSize pooled segments + // Assert Pipe.MaxSegmentPoolSize pooled segments. (reuse is FIFO) for (int i = 0; i < PipeOptions.Default.MaxSegmentPoolSize; i++) { - Assert.Same(oldSegments[i], newSegments[PipeOptions.Default.MaxSegmentPoolSize - i - 1]); + Assert.Same(oldSegments[i], newSegments[i]); } // The last segment shouldn't exist in the new list of segments at all (it should be new) From 87b6d13fdd556bdd84a25af3d038506356a7ef1d Mon Sep 17 00:00:00 2001 From: vsadov <8218165+VSadov@users.noreply.github.com> Date: Sun, 5 Jul 2026 10:15:35 -0700 Subject: [PATCH 2/8] use Lock --- .../src/System/IO/Pipelines/Pipe.cs | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs index 4bb46bb44736e7..77488f343c834c 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs @@ -35,7 +35,17 @@ public sealed partial class Pipe // The options instance private readonly PipeOptions _options; - private readonly object _sync = new object(); + + // This lock protects the shared state between the writer and reader (most of this class). + // On .NET 9+ use System.Threading.Lock, which is faster than Monitor on object; older + // targets (netstandard2.0, .NET Framework) fall back to a plain object + Monitor. +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _sync = new(); + private System.Threading.Lock SyncObj => _sync; +#else + private readonly object _sync = new(); + private object SyncObj => _sync; +#endif // Computed state from the options instance private bool UseSynchronizationContext => _options.UseSynchronizationContext; @@ -46,9 +56,6 @@ public sealed partial class Pipe private PipeScheduler ReaderScheduler => _options.ReaderScheduler; private PipeScheduler WriterScheduler => _options.WriterScheduler; - // This sync objects protects the shared state between the writer and reader (most of this class) - private object SyncObj => _sync; - // The number of bytes flushed but not consumed by the reader private long _unconsumedBytes; From 08d9389407848f0a74f6e9f7dfc1e95fbc4a7464 Mon Sep 17 00:00:00 2001 From: vsadov <8218165+VSadov@users.noreply.github.com> Date: Sun, 5 Jul 2026 13:37:42 -0700 Subject: [PATCH 3/8] Less locking in Advance path --- .../src/System/IO/Pipelines/Pipe.cs | 33 +++++++++++++++---- 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs index 77488f343c834c..8f1e13da4a77f0 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs @@ -421,20 +421,41 @@ internal bool CommitUnsynchronized() internal void Advance(int bytes) { - lock (SyncObj) + // While writing is active, the writer thread exclusively owns _writingHead, + // _writingHeadMemory, _writingHeadBytesBuffered and _unflushedBytes: the reader + // only releases/observes them when writing is NOT active. So the bounds check and + // AdvanceCore need no lock in that state. + if (_operationState.IsWritingActive) { if ((uint)bytes > (uint)_writingHeadMemory.Length) { ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.bytes); } - // If the reader is completed we no-op Advance but leave GetMemory and FlushAsync alone - if (_readerCompletion.IsCompleted) + // Best-effort no-op if the reader completed; this check is racy even under the + // lock (the reader can complete right after), and _state is a reference so the + // read is atomic thus a lock-free read is equivalent. + if (!_readerCompletion.IsCompleted) { - return; + AdvanceCore(bytes); } + } + else + { + // Cold path (e.g. Advance(0) with no prior GetMemory): preserve exact original + // semantics under the lock. + lock (SyncObj) + { + if ((uint)bytes > (uint)_writingHeadMemory.Length) + { + ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.bytes); + } - AdvanceCore(bytes); + if (!_readerCompletion.IsCompleted) + { + AdvanceCore(bytes); + } + } } } @@ -630,7 +651,7 @@ void MoveReturnEndToNextBlock() } // If the writing head is the same as the block to be returned, then we need to make sure // there's no pending write and that there's no buffered data for the writing head - else if (_writingHeadBytesBuffered == 0 && !_operationState.IsWritingActive) + else if (!_operationState.IsWritingActive && _writingHeadBytesBuffered == 0) { // Reset the writing head to null if it's the return block and we've consumed everything _writingHead = null; From 220ace9c6fcca557b85559ed471e74735aca63ad Mon Sep 17 00:00:00 2001 From: vsadov <8218165+VSadov@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:54:28 -0700 Subject: [PATCH 4/8] more aggressive prerent --- .../src/System/IO/Pipelines/Pipe.cs | 52 +++++++++---------- 1 file changed, 25 insertions(+), 27 deletions(-) diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs index 8f1e13da4a77f0..d929ae2ff69802 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs @@ -179,16 +179,16 @@ private void AllocateWriteHeadIfNeeded(int sizeHint) private void AllocateWriteHeadSynchronized(int sizeHint) { - // Rent the backing memory outside the lock, but only when writing is active. + // Speculatively rent backing memory outside the lock. // - // While writing is active, the writer thread exclusively owns _writingHead and - // _writingHeadMemory: the reader only releases them (in AdvanceReader) when writing - // is NOT active. That makes the reads below race-free, and it makes them stable - the - // reader can only shrink _writingHeadMemory (to default) and cannot even do that while - // writing is active, and this (single) writer thread does not change it between here - // and taking the lock. So if there isn't enough room now there won't be enough under - // the lock either: a buffer rented here is guaranteed to be used and never needs to be - // returned. Keeping the (potentially contended) pool rental off the lock is the point. + // Reading _writingHeadMemory.Length can race with the reader (which sets + // _writingHeadMemory = default under the lock when writing is NOT active), but .Length + // reads a single int field, so it is atomic - stale at worst, never torn. It is only a + // hint; the authoritative decision is remade under the lock. The rented buffer is always + // consumed on non-exceptional paths: _writingHeadMemory.Length only shrinks between this + // read and the lock (the reader can only reduce it to zero, and this writer thread does + // not touch it in between), so "insufficient room" still holds under the lock. That is + // why no return path is needed - see the Debug.Assert(prerented is null) after the lock. // // Note we deliberately do NOT acquire a BufferSegment here. The segment pool is a // single-producer/single-consumer queue, so an unused segment could not be safely @@ -196,8 +196,7 @@ private void AllocateWriteHeadSynchronized(int sizeHint) // lock, at the exact point we are certain to use them. Acquiring via SPSC should be // too cheap to bother that it happens under lock. object? prerented = null; - if (_operationState.IsWritingActive && - (_writingHeadMemory.Length == 0 || _writingHeadMemory.Length < sizeHint)) + if (_writingHeadMemory.Length == 0 || _writingHeadMemory.Length < sizeHint) { prerented = RentMemoryUnsynchronized(sizeHint); } @@ -211,11 +210,8 @@ private void AllocateWriteHeadSynchronized(int sizeHint) { if (_writingHead is null) { - // Writing wasn't active on entry, so we could not have pre-rented. - Debug.Assert(prerented is null); - BufferSegment newSegment = GetOrCreateSegment(); - AttachMemory(newSegment, prerented, sizeHint); + AttachMemory(newSegment, ref prerented, sizeHint); _writingHead = _readHead = _readTail = newSegment; _lastExaminedIndex = 0; @@ -236,25 +232,24 @@ private void AllocateWriteHeadSynchronized(int sizeHint) // head's buffer. Reuse the BufferSegment and swap out its memory so // ReadAsync will not observe an empty segment. _writingHead.ResetMemory(); - AttachMemory(_writingHead, prerented, sizeHint); + AttachMemory(_writingHead, ref prerented, sizeHint); } else { BufferSegment newSegment = GetOrCreateSegment(); - AttachMemory(newSegment, prerented, sizeHint); + AttachMemory(newSegment, ref prerented, sizeHint); _writingHead.SetNext(newSegment); _writingHead = newSegment; } } } - else - { - // The head still has room. Only reachable when writing wasn't active on entry; - // otherwise we would have pre-rented and this branch would not be taken. - Debug.Assert(prerented is null); - } } + + // On every non-exceptional path the speculative rent is consumed above (AttachMemory + // nulls it), because _writingHeadMemory.Length only shrinks between the off-lock read + // and the lock. If this fires, that invariant was violated. + Debug.Assert(prerented is null); } private BufferSegment AllocateSegment(int sizeHint) @@ -316,9 +311,9 @@ private object RentMemoryUnsynchronized(int sizeHint) } // Attaches memory to a segment and updates _writingHeadMemory. If memory was already rented - // outside the lock (prerented) it is attached; otherwise memory is rented here. Must be - // called under the lock. - private void AttachMemory(BufferSegment segment, object? prerented, int sizeHint) + // outside the lock (prerented) it is attached and the reference is cleared; otherwise memory + // is rented here (race fallback). Must be called under the lock. + private void AttachMemory(BufferSegment segment, ref object? prerented, int sizeHint) { Debug.Assert(segment.MemoryOwner is null); @@ -327,14 +322,17 @@ private void AttachMemory(BufferSegment segment, object? prerented, int sizeHint case IMemoryOwner owner: segment.SetOwnedMemory(owner); _writingHeadMemory = segment.AvailableMemory; + prerented = null; break; case byte[] array: segment.SetOwnedMemory(array); _writingHeadMemory = segment.AvailableMemory; + prerented = null; break; default: Debug.Assert(prerented is null); - // Nothing pre-rented (writing was not active); rent under the lock. + // Nothing pre-rented (the racy read saw enough room but the reader reset it + // before we took the lock); rent under the lock. RentMemory(segment, sizeHint); break; } From 71eee16a597f4c3382c0d7c28346cdb360deaa44 Mon Sep 17 00:00:00 2001 From: vsadov Date: Wed, 15 Jul 2026 17:04:01 -0700 Subject: [PATCH 5/8] prefer local --- .../src/System/IO/Pipelines/ThreadPoolScheduler.netcoreapp.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/ThreadPoolScheduler.netcoreapp.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/ThreadPoolScheduler.netcoreapp.cs index 1e9cd6d7efceb5..28e509c8839227 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/ThreadPoolScheduler.netcoreapp.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/ThreadPoolScheduler.netcoreapp.cs @@ -10,12 +10,12 @@ internal sealed class ThreadPoolScheduler : PipeScheduler { public override void Schedule(Action action, object? state) { - System.Threading.ThreadPool.QueueUserWorkItem(action, state, preferLocal: false); + System.Threading.ThreadPool.QueueUserWorkItem(action, state, preferLocal: true); } internal override void UnsafeSchedule(Action action, object? state) { - System.Threading.ThreadPool.UnsafeQueueUserWorkItem(action, state, preferLocal: false); + System.Threading.ThreadPool.UnsafeQueueUserWorkItem(action, state, preferLocal: true); } } } From fa2d64a83845f47adeb5465306e347f3cd1cf4b6 Mon Sep 17 00:00:00 2001 From: vsadov <8218165+VSadov@users.noreply.github.com> Date: Thu, 16 Jul 2026 09:54:24 -0700 Subject: [PATCH 6/8] tweak comments --- .../src/System/IO/Pipelines/Pipe.cs | 14 +++++++------- .../src/System/IO/Pipelines/PipeOptions.cs | 7 ++++--- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs index d929ae2ff69802..93ef2512b87c79 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs @@ -37,8 +37,8 @@ public sealed partial class Pipe private readonly PipeOptions _options; // This lock protects the shared state between the writer and reader (most of this class). - // On .NET 9+ use System.Threading.Lock, which is faster than Monitor on object; older - // targets (netstandard2.0, .NET Framework) fall back to a plain object + Monitor. + // On .NET 9+ use System.Threading.Lock, as a monitor lock here tends to inflate into Lock anyways. + // Older targets (netstandard2.0, .NET Framework) fall back to a plain object + Monitor. #if NET9_0_OR_GREATER private readonly System.Threading.Lock _sync = new(); private System.Threading.Lock SyncObj => _sync; @@ -193,8 +193,8 @@ private void AllocateWriteHeadSynchronized(int sizeHint) // Note we deliberately do NOT acquire a BufferSegment here. The segment pool is a // single-producer/single-consumer queue, so an unused segment could not be safely // returned from this thread. Segments are taken (GetOrCreateSegment) only under the - // lock, at the exact point we are certain to use them. Acquiring via SPSC should be - // too cheap to bother that it happens under lock. + // lock, at the exact point we are certain to use them. Acquiring via SPSC is relatively + // cheap though. object? prerented = null; if (_writingHeadMemory.Length == 0 || _writingHeadMemory.Length < sizeHint) { @@ -419,7 +419,7 @@ internal bool CommitUnsynchronized() internal void Advance(int bytes) { - // While writing is active, the writer thread exclusively owns _writingHead, + // While writing is active, the writer thread (us) exclusively owns _writingHead, // _writingHeadMemory, _writingHeadBytesBuffered and _unflushedBytes: the reader // only releases/observes them when writing is NOT active. So the bounds check and // AdvanceCore need no lock in that state. @@ -440,8 +440,8 @@ internal void Advance(int bytes) } else { - // Cold path (e.g. Advance(0) with no prior GetMemory): preserve exact original - // semantics under the lock. + // Cold path (e.g. Advance(0) with no prior GetMemory): use lock, + // to get exclusive access to the writing head. lock (SyncObj) { if ((uint)bytes > (uint)_writingHeadMemory.Length) diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs index 59dd17b140e49a..f7dc8a10601920 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs @@ -34,10 +34,11 @@ public PipeOptions( { MinimumSegmentSize = minimumSegmentSize == -1 ? DefaultMinimumSegmentSize : minimumSegmentSize; - // Cap the per-pipe segment-object pool to bound memory in degenerate cases. + // Cap the per-pipe segment-object pool to bound memory in edge cases. // Buffers are returned to the MemoryPool on Reset() before a segment is pooled, so a - // pooled BufferSegment holds no buffer and is ~96 bytes on 64-bit. The cap therefore - // costs at most ~24 KB per pipe (256 * ~96 bytes); the backing memory is pooled separately. + // pooled BufferSegment holds no bytes buffers and itself is ~96 bytes on 64-bit. + // The cap therefore costs at most ~24 KB per pipe (256 * ~96 bytes); + // the backing memory is pooled separately. // // Normal pipes never approach this cap. The pool only grows to the peak number of // simultaneously-live segments, which for a throttled pipe is roughly From 37a1957dc938ef37f1ef4ec914b3b66dd730b521 Mon Sep 17 00:00:00 2001 From: Vladimir Sadov Date: Thu, 16 Jul 2026 10:03:27 -0700 Subject: [PATCH 7/8] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../System.IO.Pipelines/src/System.IO.Pipelines.csproj | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj b/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj index 42035d63812b5d..79c24f6a1b4838 100644 --- a/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj +++ b/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj @@ -1,4 +1,4 @@ - + $(NetCoreAppCurrent);$(NetCoreAppPrevious);$(NetCoreAppMinimum);netstandard2.0;$(NetFrameworkMinimum) From 640d9a6ed202c8c9d854bab61a3246db82999a81 Mon Sep 17 00:00:00 2001 From: vsadov <8218165+VSadov@users.noreply.github.com> Date: Thu, 16 Jul 2026 10:23:02 -0700 Subject: [PATCH 8/8] InitialSegmentPoolSize is now unused --- .../src/System/IO/Pipelines/PipeOptions.cs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs index f7dc8a10601920..803d0e7f8bbb20 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeOptions.cs @@ -123,11 +123,6 @@ public PipeOptions( /// internal bool IsDefaultSharedMemoryPool { get; } - /// - /// The initialize size of the segment pool - /// - internal int InitialSegmentPoolSize { get; } - /// /// The maximum number of segments to pool ///