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 e2c35f4efca031..db4b9fb79535bd 100644 --- a/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj +++ b/src/libraries/System.IO.Pipelines/src/System.IO.Pipelines.csproj @@ -7,7 +7,7 @@ + Link="Common\System\Threading\Tasks\TaskToApm.cs" /> @@ -48,7 +48,7 @@ - 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 08837ce80b6b2d..caf0767c2e14b2 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.Generic; using System.Diagnostics; using System.Runtime.CompilerServices; using System.Threading; @@ -13,9 +14,6 @@ namespace System.IO.Pipelines /// The default and implementation. public sealed partial class Pipe { - internal const int InitialSegmentPoolSize = 16; // 65K - internal const int MaxSegmentPoolSize = 256; // 1MB - private static readonly Action s_signalReaderAwaitable = state => ((Pipe)state!).ReaderCancellationRequested(); private static readonly Action s_signalWriterAwaitable = state => ((Pipe)state!).WriterCancellationRequested(); private static readonly Action s_invokeCompletionCallbacks = state => ((PipeCompletionCallbacks)state!).Execute(); @@ -26,24 +24,27 @@ public sealed partial class Pipe private static readonly SendOrPostCallback s_syncContextExecuteWithoutExecutionContextCallback = ExecuteWithoutExecutionContext!; private static readonly Action s_scheduleWithExecutionContextCallback = ExecuteWithExecutionContext!; - // This sync objects protects the shared state between the writer and reader (most of this class) - private readonly object _sync = new object(); - - private readonly MemoryPool? _pool; - private readonly int _minimumSegmentSize; - private readonly long _pauseWriterThreshold; - private readonly long _resumeWriterThreshold; - - private readonly PipeScheduler _readerScheduler; - private readonly PipeScheduler _writerScheduler; - // Mutable struct! Don't make this readonly private BufferSegmentStack _bufferSegmentPool; private readonly DefaultPipeReader _reader; private readonly DefaultPipeWriter _writer; - private readonly bool _useSynchronizationContext; + // The options instance + private readonly PipeOptions _options; + private readonly object _sync = new object(); + + // Computed state from the options instance + private bool UseSynchronizationContext => _options.UseSynchronizationContext; + private int MinimumSegmentSize => _options.MinimumSegmentSize; + private long PauseWriterThreshold => _options.PauseWriterThreshold; + private long ResumeWriterThreshold => _options.ResumeWriterThreshold; + + 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; @@ -65,7 +66,6 @@ public sealed partial class Pipe private BufferSegment? _readHead; private int _readHeadIndex; - private readonly int _maxPooledBufferSize; private bool _disposed; // The extent of the bytes available to the PipeReader to consume @@ -80,7 +80,6 @@ public sealed partial class Pipe // Determines what current operation is in flight (reading/writing) private PipeOperationState _operationState; - internal long Length => _unconsumedBytes; /// Initializes a new instance of the class using as options. @@ -97,24 +96,15 @@ public Pipe(PipeOptions options) ThrowHelper.ThrowArgumentNullException(ExceptionArgument.options); } - _bufferSegmentPool = new BufferSegmentStack(InitialSegmentPoolSize); + _bufferSegmentPool = new BufferSegmentStack(options.InitialSegmentPoolSize); _operationState = default; _readerCompletion = default; _writerCompletion = default; - // If we're using the default pool then mark it as null since we're just going to use the - // array pool under the covers - _pool = options.Pool == MemoryPool.Shared ? null : options.Pool; - _maxPooledBufferSize = _pool?.MaxBufferSize ?? -1; - _minimumSegmentSize = options.MinimumSegmentSize; - _pauseWriterThreshold = options.PauseWriterThreshold; - _resumeWriterThreshold = options.ResumeWriterThreshold; - _readerScheduler = options.ReaderScheduler; - _writerScheduler = options.WriterScheduler; - _useSynchronizationContext = options.UseSynchronizationContext; - _readerAwaitable = new PipeAwaitable(completed: false, _useSynchronizationContext); - _writerAwaitable = new PipeAwaitable(completed: true, _useSynchronizationContext); + _options = options; + _readerAwaitable = new PipeAwaitable(completed: false, UseSynchronizationContext); + _writerAwaitable = new PipeAwaitable(completed: true, UseSynchronizationContext); _reader = new DefaultPipeReader(this); _writer = new DefaultPipeWriter(this); } @@ -123,8 +113,8 @@ private void ResetState() { _readerCompletion.Reset(); _writerCompletion.Reset(); - _readerAwaitable = new PipeAwaitable(completed: false, _useSynchronizationContext); - _writerAwaitable = new PipeAwaitable(completed: true, _useSynchronizationContext); + _readerAwaitable = new PipeAwaitable(completed: false, UseSynchronizationContext); + _writerAwaitable = new PipeAwaitable(completed: true, UseSynchronizationContext); _readTailIndex = 0; _readHeadIndex = 0; _lastExaminedIndex = -1; @@ -180,7 +170,7 @@ private void AllocateWriteHeadIfNeeded(int sizeHint) private void AllocateWriteHeadSynchronized(int sizeHint) { - lock (_sync) + lock (SyncObj) { _operationState.BeginWrite(); @@ -220,11 +210,19 @@ private BufferSegment AllocateSegment(int sizeHint) Debug.Assert(sizeHint >= 0); BufferSegment newSegment = CreateSegmentUnsynchronized(); - int maxSize = _maxPooledBufferSize; + MemoryPool? pool = null; + int maxSize = -1; + + if (!_options.IsDefaultSharedMemoryPool) + { + pool = _options.Pool; + maxSize = pool.MaxBufferSize; + } + if (sizeHint <= maxSize) { // Use the specified pool as it fits. Specified pool is not null as maxSize == -1 if _pool is null. - newSegment.SetOwnedMemory(_pool!.Rent(GetSegmentSize(sizeHint, maxSize))); + newSegment.SetOwnedMemory(pool!.Rent(GetSegmentSize(sizeHint, maxSize))); } else { @@ -241,7 +239,7 @@ private BufferSegment AllocateSegment(int sizeHint) private int GetSegmentSize(int sizeHint, int maxBufferSize = int.MaxValue) { // First we need to handle case where hint is smaller than minimum segment size - sizeHint = Math.Max(_minimumSegmentSize, sizeHint); + sizeHint = Math.Max(MinimumSegmentSize, sizeHint); // After that adjust it to fit into pools max buffer size int adjustedToMaximumSize = Math.Min(maxBufferSize, sizeHint); return adjustedToMaximumSize; @@ -263,7 +261,7 @@ private void ReturnSegmentUnsynchronized(BufferSegment segment) Debug.Assert(segment != _readTail, "Returning _readTail segment that's in use!"); Debug.Assert(segment != _writingHead, "Returning _writingHead segment that's in use!"); - if (_bufferSegmentPool.Count < MaxSegmentPoolSize) + if (_bufferSegmentPool.Count < _options.MaxSegmentPoolSize) { _bufferSegmentPool.Push(segment); } @@ -291,9 +289,9 @@ internal bool CommitUnsynchronized() _unconsumedBytes += _unflushedBytes; // Do not reset if reader is complete - if (_pauseWriterThreshold > 0 && - oldLength < _pauseWriterThreshold && - _unconsumedBytes >= _pauseWriterThreshold && + if (PauseWriterThreshold > 0 && + oldLength < PauseWriterThreshold && + _unconsumedBytes >= PauseWriterThreshold && !_readerCompletion.IsCompleted) { _writerAwaitable.SetUncompleted(); @@ -307,7 +305,7 @@ internal bool CommitUnsynchronized() internal void Advance(int bytes) { - lock (_sync) + lock (SyncObj) { if ((uint)bytes > (uint)_writingHeadMemory.Length) { @@ -336,12 +334,12 @@ internal ValueTask FlushAsync(CancellationToken cancellationToken) { CompletionData completionData; ValueTask result; - lock (_sync) + lock (SyncObj) { PrepareFlush(out completionData, out result, cancellationToken); } - TrySchedule(_readerScheduler, completionData); + TrySchedule(ReaderScheduler, completionData); return result; } @@ -389,7 +387,7 @@ internal void CompleteWriter(Exception? exception) PipeCompletionCallbacks? completionCallbacks; bool readerCompleted; - lock (_sync) + lock (SyncObj) { // Commit any pending buffers CommitUnsynchronized(); @@ -406,10 +404,10 @@ internal void CompleteWriter(Exception? exception) if (completionCallbacks != null) { - ScheduleCallbacks(_readerScheduler, completionCallbacks); + ScheduleCallbacks(ReaderScheduler, completionCallbacks); } - TrySchedule(_readerScheduler, completionData); + TrySchedule(ReaderScheduler, completionData); } internal void AdvanceReader(in SequencePosition consumed) @@ -443,7 +441,7 @@ private void AdvanceReader(BufferSegment? consumedSegment, int consumedIndex, Bu CompletionData completionData = default; - lock (_sync) + lock (SyncObj) { var examinedEverything = false; if (examinedSegment == _readTail) @@ -468,8 +466,8 @@ private void AdvanceReader(BufferSegment? consumedSegment, int consumedIndex, Bu Debug.Assert(_unconsumedBytes >= 0, "Length has gone negative"); - if (oldLength >= _resumeWriterThreshold && - _unconsumedBytes < _resumeWriterThreshold) + if (oldLength >= ResumeWriterThreshold && + _unconsumedBytes < ResumeWriterThreshold) { _writerAwaitable.Complete(out completionData); } @@ -552,7 +550,7 @@ void MoveReturnEndToNextBlock() _operationState.EndRead(); } - TrySchedule(_writerScheduler, completionData); + TrySchedule(WriterScheduler, completionData); } internal void CompleteReader(Exception? exception) @@ -561,7 +559,7 @@ internal void CompleteReader(Exception? exception) CompletionData completionData; bool writerCompleted; - lock (_sync) + lock (SyncObj) { // If we're reading, treat clean up that state before continuting if (_operationState.IsReadingActive) @@ -584,10 +582,10 @@ internal void CompleteReader(Exception? exception) if (completionCallbacks != null) { - ScheduleCallbacks(_writerScheduler, completionCallbacks); + ScheduleCallbacks(WriterScheduler, completionCallbacks); } - TrySchedule(_writerScheduler, completionData); + TrySchedule(WriterScheduler, completionData); } internal void OnWriterCompleted(Action callback, object? state) @@ -598,35 +596,35 @@ internal void OnWriterCompleted(Action callback, object? st } PipeCompletionCallbacks? completionCallbacks; - lock (_sync) + lock (SyncObj) { completionCallbacks = _writerCompletion.AddCallback(callback, state); } if (completionCallbacks != null) { - ScheduleCallbacks(_readerScheduler, completionCallbacks); + ScheduleCallbacks(ReaderScheduler, completionCallbacks); } } internal void CancelPendingRead() { CompletionData completionData; - lock (_sync) + lock (SyncObj) { _readerAwaitable.Cancel(out completionData); } - TrySchedule(_readerScheduler, completionData); + TrySchedule(ReaderScheduler, completionData); } internal void CancelPendingFlush() { CompletionData completionData; - lock (_sync) + lock (SyncObj) { _writerAwaitable.Cancel(out completionData); } - TrySchedule(_writerScheduler, completionData); + TrySchedule(WriterScheduler, completionData); } internal void OnReaderCompleted(Action callback, object? state) @@ -637,14 +635,14 @@ internal void OnReaderCompleted(Action callback, object? st } PipeCompletionCallbacks? completionCallbacks; - lock (_sync) + lock (SyncObj) { completionCallbacks = _readerCompletion.AddCallback(callback, state); } if (completionCallbacks != null) { - ScheduleCallbacks(_writerScheduler, completionCallbacks); + ScheduleCallbacks(WriterScheduler, completionCallbacks); } } @@ -656,7 +654,7 @@ internal ValueTask ReadAsync(CancellationToken token) } ValueTask result; - lock (_sync) + lock (SyncObj) { _readerAwaitable.BeginOperation(token, s_signalReaderAwaitable, this); @@ -678,7 +676,7 @@ internal ValueTask ReadAsync(CancellationToken token) internal bool TryRead(out ReadResult result) { - lock (_sync) + lock (SyncObj) { if (_readerCompletion.IsCompleted) { @@ -776,7 +774,7 @@ private static void ExecuteWithExecutionContext(object state) private void CompletePipe() { - lock (_sync) + lock (SyncObj) { if (_disposed) { @@ -821,7 +819,7 @@ internal void OnReadAsyncCompleted(Action continuation, object? state, { CompletionData completionData; bool doubleCompletion; - lock (_sync) + lock (SyncObj) { _readerAwaitable.OnCompleted(continuation, state, flags, out completionData, out doubleCompletion); } @@ -829,7 +827,7 @@ internal void OnReadAsyncCompleted(Action continuation, object? state, { Writer.Complete(ThrowHelper.CreateInvalidOperationException_NoConcurrentOperation()); } - TrySchedule(_readerScheduler, completionData); + TrySchedule(ReaderScheduler, completionData); } internal ReadResult GetReadAsyncResult() @@ -839,7 +837,7 @@ internal ReadResult GetReadAsyncResult() CancellationToken cancellationToken = default; try { - lock (_sync) + lock (SyncObj) { if (!_readerAwaitable.IsCompleted) { @@ -914,7 +912,7 @@ internal FlushResult GetFlushAsyncResult() try { - lock (_sync) + lock (SyncObj) { if (!_writerAwaitable.IsCompleted) { @@ -963,7 +961,7 @@ internal ValueTask WriteAsync(ReadOnlyMemory source, Cancella CompletionData completionData; ValueTask result; - lock (_sync) + lock (SyncObj) { // Allocate whatever the pool gives us so we can write, this also marks the // state as writing @@ -984,7 +982,7 @@ internal ValueTask WriteAsync(ReadOnlyMemory source, Cancella PrepareFlush(out completionData, out result, cancellationToken); } - TrySchedule(_readerScheduler, completionData); + TrySchedule(ReaderScheduler, completionData); return result; } @@ -1024,7 +1022,7 @@ internal void OnFlushAsyncCompleted(Action continuation, object? state, { CompletionData completionData; bool doubleCompletion; - lock (_sync) + lock (SyncObj) { _writerAwaitable.OnCompleted(continuation, state, flags, out completionData, out doubleCompletion); } @@ -1032,27 +1030,27 @@ internal void OnFlushAsyncCompleted(Action continuation, object? state, { Reader.Complete(ThrowHelper.CreateInvalidOperationException_NoConcurrentOperation()); } - TrySchedule(_writerScheduler, completionData); + TrySchedule(WriterScheduler, completionData); } private void ReaderCancellationRequested() { CompletionData completionData; - lock (_sync) + lock (SyncObj) { _readerAwaitable.CancellationTokenFired(out completionData); } - TrySchedule(_readerScheduler, completionData); + TrySchedule(ReaderScheduler, completionData); } private void WriterCancellationRequested() { CompletionData completionData; - lock (_sync) + lock (SyncObj) { _writerAwaitable.CancellationTokenFired(out completionData); } - TrySchedule(_writerScheduler, completionData); + TrySchedule(WriterScheduler, completionData); } /// Gets the for this pipe. @@ -1066,7 +1064,7 @@ private void WriterCancellationRequested() /// Resets the pipe. public void Reset() { - lock (_sync) + lock (SyncObj) { if (!_disposed) { diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeAwaitable.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeAwaitable.cs index dab70646c1df49..17427bb0a1203f 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeAwaitable.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeAwaitable.cs @@ -14,9 +14,9 @@ internal struct PipeAwaitable private AwaitableState _awaitableState; private Action? _completion; private object? _completionState; + // It's rare to have to capture custom context here + private SchedulingContext? _schedulingContext; private CancellationTokenRegistration _cancellationTokenRegistration; - private SynchronizationContext? _synchronizationContext; - private ExecutionContext? _executionContext; #if (!NETSTANDARD2_0 && !NETFRAMEWORK) private CancellationToken CancellationToken => _cancellationTokenRegistration.Token; @@ -32,8 +32,7 @@ public PipeAwaitable(bool completed, bool useSynchronizationContext) _completion = null; _completionState = null; _cancellationTokenRegistration = default; - _synchronizationContext = null; - _executionContext = null; + _schedulingContext = null; #if (NETSTANDARD2_0 || NETFRAMEWORK) _cancellationToken = CancellationToken.None; #endif @@ -73,13 +72,13 @@ private void ExtractCompletion(out CompletionData completionData) { Action? currentCompletion = _completion; object? currentState = _completionState; - ExecutionContext? executionContext = _executionContext; - SynchronizationContext? synchronizationContext = _synchronizationContext; + SchedulingContext? schedulingContext = _schedulingContext; + ExecutionContext? executionContext = schedulingContext?.ExecutionContext; + SynchronizationContext? synchronizationContext = schedulingContext?.SynchronizationContext; _completion = null; _completionState = null; - _synchronizationContext = null; - _executionContext = null; + _schedulingContext = null; completionData = currentCompletion != null ? new CompletionData(currentCompletion, currentState, executionContext, synchronizationContext) : @@ -91,8 +90,7 @@ public void SetUncompleted() { Debug.Assert(_completion == null); Debug.Assert(_completionState == null); - Debug.Assert(_synchronizationContext == null); - Debug.Assert(_executionContext == null); + Debug.Assert(_schedulingContext == null); _awaitableState &= ~AwaitableState.Completed; } @@ -104,7 +102,7 @@ public void OnCompleted(Action continuation, object? state, ValueTaskSo if (IsCompleted || doubleCompletion) { - completionData = new CompletionData(continuation, state, _executionContext, _synchronizationContext); + completionData = new CompletionData(continuation, state, _schedulingContext?.ExecutionContext, _schedulingContext?.SynchronizationContext); return; } @@ -118,14 +116,16 @@ public void OnCompleted(Action continuation, object? state, ValueTaskSo SynchronizationContext? sc = SynchronizationContext.Current; if (sc != null && sc.GetType() != typeof(SynchronizationContext)) { - _synchronizationContext = sc; + _schedulingContext ??= new SchedulingContext(); + _schedulingContext.SynchronizationContext = sc; } } // Capture the execution context if ((flags & ValueTaskSourceOnCompletedFlags.FlowExecutionContext) != 0) { - _executionContext = ExecutionContext.Capture(); + _schedulingContext ??= new SchedulingContext(); + _schedulingContext.ExecutionContext = ExecutionContext.Capture(); } } @@ -185,5 +185,11 @@ private enum AwaitableState Canceled = 4, UseSynchronizationContext = 8 } + + private class SchedulingContext + { + public SynchronizationContext? SynchronizationContext { get; set; } + public ExecutionContext? ExecutionContext { get; set; } + } } } diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletion.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletion.cs index a37292d52a44e6..5c0ff7b5915428 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletion.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletion.cs @@ -1,7 +1,7 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. -using System.Buffers; +using System.Collections.Generic; using System.Diagnostics; using System.Runtime.CompilerServices; using System.Runtime.ExceptionServices; @@ -11,51 +11,36 @@ namespace System.IO.Pipelines [DebuggerDisplay("IsCompleted: {" + nameof(IsCompleted) + "}")] internal struct PipeCompletion { - private static readonly ArrayPool s_completionCallbackPool = ArrayPool.Shared; + private static readonly object s_completedSuccessfully = new object(); - private const int InitialCallbacksSize = 1; + private object? _state; + private List? _callbacks; - private bool _isCompleted; - private ExceptionDispatchInfo? _exceptionInfo; + public bool IsCompleted => _state != null; - private PipeCompletionCallback _firstCallback; - private PipeCompletionCallback[]? _callbacks; - private int _callbackCount; - - public bool IsCompleted => _isCompleted; - - public bool IsFaulted => _exceptionInfo != null; + public bool IsFaulted => _state is ExceptionDispatchInfo; public PipeCompletionCallbacks? TryComplete(Exception? exception = null) { - if (!_isCompleted) + if (_state == null) { - _isCompleted = true; if (exception != null) { - _exceptionInfo = ExceptionDispatchInfo.Capture(exception); + _state = ExceptionDispatchInfo.Capture(exception); + } + else + { + _state = s_completedSuccessfully; } } + return GetCallbacks(); } public PipeCompletionCallbacks? AddCallback(Action callback, object? state) { - if (_callbackCount == 0) - { - _firstCallback = new PipeCompletionCallback(callback, state); - _callbackCount++; - } - else - { - EnsureSpace(); - - // -1 to adjust for _firstCallback - var callbackIndex = _callbackCount - 1; - _callbackCount++; - Debug.Assert(_callbacks != null); - _callbacks[callbackIndex] = new PipeCompletionCallback(callback, state); - } + _callbacks ??= new List(); + _callbacks.Add(new PipeCompletionCallback(callback, state)); if (IsCompleted) { @@ -65,35 +50,17 @@ internal struct PipeCompletion return null; } - private void EnsureSpace() - { - if (_callbacks == null) - { - _callbacks = s_completionCallbackPool.Rent(InitialCallbacksSize); - } - - int newLength = _callbackCount - 1; - - if (newLength == _callbacks.Length) - { - PipeCompletionCallback[] newArray = s_completionCallbackPool.Rent(_callbacks.Length * 2); - Array.Copy(_callbacks, newArray, _callbacks.Length); - s_completionCallbackPool.Return(_callbacks, clearArray: true); - _callbacks = newArray; - } - } - [MethodImpl(MethodImplOptions.AggressiveInlining)] public bool IsCompletedOrThrow() { - if (!_isCompleted) + if (!IsCompleted) { return false; } - if (_exceptionInfo != null) + if (_state is ExceptionDispatchInfo edi) { - ThrowLatchedException(); + edi.Throw(); } return true; @@ -102,36 +69,23 @@ public bool IsCompletedOrThrow() private PipeCompletionCallbacks? GetCallbacks() { Debug.Assert(IsCompleted); - if (_callbackCount == 0) + + var callbacks = _callbacks; + if (callbacks == null) { return null; } - var callbacks = new PipeCompletionCallbacks(s_completionCallbackPool, - _callbackCount, - _exceptionInfo?.SourceException, - _firstCallback, - _callbacks); - - _firstCallback = default; _callbacks = null; - _callbackCount = 0; - return callbacks; + + return new PipeCompletionCallbacks(callbacks, _state as ExceptionDispatchInfo); } public void Reset() { Debug.Assert(IsCompleted); Debug.Assert(_callbacks == null); - _isCompleted = false; - _exceptionInfo = null; - } - - [MethodImpl(MethodImplOptions.NoInlining)] - private void ThrowLatchedException() - { - Debug.Assert(_exceptionInfo != null); - _exceptionInfo.Throw(); + _state = null; } public override string ToString() diff --git a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletionCallbacks.cs b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletionCallbacks.cs index 9bba043440c021..47c9d064ac5738 100644 --- a/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletionCallbacks.cs +++ b/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletionCallbacks.cs @@ -1,52 +1,38 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. -using System.Buffers; using System.Collections.Generic; +using System.Runtime.ExceptionServices; namespace System.IO.Pipelines { internal sealed class PipeCompletionCallbacks { - private readonly ArrayPool _pool; - private readonly int _count; + private readonly List _callbacks; private readonly Exception? _exception; - private readonly PipeCompletionCallback _firstCallback; - private readonly PipeCompletionCallback[]? _callbacks; - public PipeCompletionCallbacks(ArrayPool pool, int count, Exception? exception, PipeCompletionCallback firstCallback, PipeCompletionCallback[]? callbacks) + public PipeCompletionCallbacks(List callbacks, ExceptionDispatchInfo? edi) { - _pool = pool; - _count = count; - _exception = exception; - _firstCallback = firstCallback; _callbacks = callbacks; + _exception = edi?.SourceException; } public void Execute() { - if (_count == 0) + var count = _callbacks.Count; + if (count == 0) { return; } List? exceptions = null; - Execute(_firstCallback, ref exceptions); - if (_callbacks != null) { - try - { - for (var i = 0; i < _count - 1; i++) - { - var callback = _callbacks[i]; - Execute(callback, ref exceptions); - } - } - finally + for (int i = 0; i < count; i++) { - _pool.Return(_callbacks, clearArray: true); + var callback = _callbacks[i]; + Execute(callback, ref exceptions); } } 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 6790f7ecf25f44..998845f804dd33 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 @@ -11,10 +11,6 @@ public class PipeOptions { private const int DefaultMinimumSegmentSize = 4096; - private const int DefaultResumeWriterThreshold = DefaultMinimumSegmentSize * Pipe.InitialSegmentPoolSize / 2; - - private const int DefaultPauseWriterThreshold = DefaultMinimumSegmentSize * Pipe.InitialSegmentPoolSize; - /// Gets the default instance of . /// A object initialized with default parameters. public static PipeOptions Default { get; } = new PipeOptions(); @@ -36,6 +32,24 @@ public PipeOptions( int minimumSegmentSize = -1, bool useSynchronizationContext = true) { + 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 defaukt segment size of 4K this maps to 16K + InitialSegmentPoolSize = 4; + + // With a defaukt segment size of 4K this maps to 1MB. If the pipe has large segments this will be bigger than 1MB... + MaxSegmentPoolSize = 256; + + // By default, we'll throttle the writer at 64K of buffered data + const int DefaultPauseWriterThreshold = 65536; + + // Resume threshold is 1/2 of the pause threshold to prevent thrashing at the limit + const int DefaultResumeWriterThreshold = DefaultPauseWriterThreshold / 2; + if (pauseWriterThreshold == -1) { pauseWriterThreshold = DefaultPauseWriterThreshold; @@ -55,11 +69,11 @@ public PipeOptions( } Pool = pool ?? MemoryPool.Shared; + IsDefaultSharedMemoryPool = Pool == MemoryPool.Shared; ReaderScheduler = readerScheduler ?? PipeScheduler.ThreadPool; WriterScheduler = writerScheduler ?? PipeScheduler.ThreadPool; PauseWriterThreshold = pauseWriterThreshold; ResumeWriterThreshold = resumeWriterThreshold; - MinimumSegmentSize = minimumSegmentSize == -1 ? DefaultMinimumSegmentSize : minimumSegmentSize; UseSynchronizationContext = useSynchronizationContext; } @@ -90,5 +104,20 @@ public PipeOptions( /// Gets the object used for buffer management. /// A pool of memory blocks used for buffer management. public MemoryPool Pool { get; } + + /// + /// Returns true if Pool is .Shared + /// + internal bool IsDefaultSharedMemoryPool { get; } + + /// + /// The initialize size of the segment pool + /// + internal int InitialSegmentPoolSize { get; } + + /// + /// The maximum number of segments to pool + /// + internal int MaxSegmentPoolSize { get; } } } diff --git a/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs b/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs index 421c31c36ee77d..c89ae0e8f20249 100644 --- a/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs +++ b/src/libraries/System.IO.Pipelines/tests/BufferSegmentPoolTest.cs @@ -60,7 +60,7 @@ public async Task BufferSegmentsAreReused() [Fact] public async Task BufferSegmentsPooledUpToThreshold() { - int blockCount = Pipe.MaxSegmentPoolSize + 1; + int blockCount = PipeOptions.Default.MaxSegmentPoolSize + 1; // Write 256 blocks to ensure they get reused for (int i = 0; i < blockCount; i++) @@ -95,9 +95,9 @@ public async Task BufferSegmentsPooledUpToThreshold() _pipe.Reader.AdvanceTo(result.Buffer.End); // Assert Pipe.MaxSegmentPoolSize pooled segments - for (int i = 0; i < Pipe.MaxSegmentPoolSize; i++) + for (int i = 0; i < PipeOptions.Default.MaxSegmentPoolSize; i++) { - Assert.Same(oldSegments[i], newSegments[Pipe.MaxSegmentPoolSize - i - 1]); + Assert.Same(oldSegments[i], newSegments[PipeOptions.Default.MaxSegmentPoolSize - i - 1]); } // The last segment shouldn't exist in the new list of segments at all (it should be new)