// ==++== // // Copyright (c) Microsoft Corporation. All rights reserved. // // ==--== /*============================================================ ** ** Class: BufferedStream ** ** gpaperin ** ** Purpose: A composable Stream that buffers reads & writes to the underlying stream. ** ** ===========================================================*/ using System; using System.Runtime.InteropServices; using System.Globalization; using System.Diagnostics.Contracts; using System.Runtime.CompilerServices; using System.Threading; #if FEATURE_ASYNC_IO using System.Collections.ObjectModel; using System.Security; using System.Threading.Tasks; #endif // !FEATURE_PAL && FEATURE_ASYNC_IO namespace System.IO { /// /// One of the design goals here is to prevent the buffer from getting in the way and slowing /// down underlying stream accesses when it is not needed. If you always read & write for sizes /// greater than the internal buffer size, then this class may not even allocate the internal buffer. /// See a large comment in Write for the details of the write buffer heuristic. /// /// This class buffers reads & writes in a shared buffer. /// (If you maintained two buffers separately, one operation would always trash the other buffer /// anyways, so we might as well use one buffer.) /// The assumption here is you will almost always be doing a series of reads or writes, but rarely /// alternate between the two of them on the same stream. /// /// Class Invariants: /// The class has one buffer, shared for reading & writing. /// It can only be used for one or the other at any point in time - not both. /// The following should be true: /// 0 implies the read buffer is valid, but we're at the end of the buffer. /// * _readPos == _readLen == 0 means the read buffer contains garbage. /// * Either _writePos can be greater than 0, or _readLen & _readPos can be greater than zero, /// but neither can be greater than zero at the same time. /// ]]> /// This class will never cache more bytes than the max specified buffer size. /// However, it may use a temporary buffer of up to twice the size in order to combine several IO operations on /// the underlying stream into a single operation. This is because we assume that memory copies are significantly /// faster than IO operations on the underlying stream (if this was not true, using buffering is never appropriate). /// The max size of this "shadow" buffer is limited as to not allocate it on the LOH. /// Shadowing is always transient. Even when using this technique, this class still guarantees that the number of /// bytes cached (not yet written to the target stream or not yet consumed by the user) is never larger than the /// actual specified buffer size. /// [ComVisible(true)] public sealed class BufferedStream : Stream { private const Int32 _DefaultBufferSize = 4096; private Stream _stream; // Underlying stream. Close sets _stream to null. private Byte[] _buffer; // Shared read/write buffer. Alloc on first use. private readonly Int32 _bufferSize; // Length of internal buffer (not counting the shadow buffer). private Int32 _readPos; // Read pointer within shared buffer. private Int32 _readLen; // Number of bytes read in buffer from _stream. private Int32 _writePos; // Write pointer within shared buffer. #if FEATURE_ASYNC_IO private BeginEndAwaitableAdapter _beginEndAwaitable; // Used to be able to await a BeginXxx call and thus to share code // between the APM and Async pattern implementations private Task _lastSyncCompletedReadTask; // The last successful Task returned from ReadAsync // (perf optimization for successive reads of the same size) #endif // !FEATURE_PAL && FEATURE_ASYNC_IO // Removing a private default constructor is a breaking change for the DataContractSerializer. // Because this ctor was here previously we need to keep it around. private BufferedStream() { } public BufferedStream(Stream stream) : this(stream, _DefaultBufferSize) { } public BufferedStream(Stream stream, Int32 bufferSize) { if (stream == null) throw new ArgumentNullException("stream"); if (bufferSize <= 0) throw new ArgumentOutOfRangeException("bufferSize", Environment.GetResourceString("ArgumentOutOfRange_MustBePositive", "bufferSize")); Contract.EndContractBlock(); BCLDebug.Perf(!(stream is FileStream), "FileStream is buffered - don't wrap it in a BufferedStream"); BCLDebug.Perf(!(stream is MemoryStream), "MemoryStream shouldn't be wrapped in a BufferedStream!"); BCLDebug.Perf(!(stream is BufferedStream), "BufferedStream shouldn't be wrapped in another BufferedStream!"); _stream = stream; _bufferSize = bufferSize; // Allocate _buffer on its first use - it will not be used if all reads // & writes are greater than or equal to buffer size. if (!_stream.CanRead && !_stream.CanWrite) __Error.StreamIsClosed(); } private void EnsureNotClosed() { if (_stream == null) __Error.StreamIsClosed(); } private void EnsureCanSeek() { Contract.Requires(_stream != null); if (!_stream.CanSeek) __Error.SeekNotSupported(); } private void EnsureCanRead() { Contract.Requires(_stream != null); if (!_stream.CanRead) __Error.ReadNotSupported(); } private void EnsureCanWrite() { Contract.Requires(_stream != null); if (!_stream.CanWrite) __Error.WriteNotSupported(); } #if FEATURE_ASYNC_IO private void EnsureBeginEndAwaitableAllocated() { // We support only a single ongoing async operation and enforce this with a semaphore, // so singleton is fine and no need to worry about a ---- here. if (_beginEndAwaitable == null) _beginEndAwaitable = new BeginEndAwaitableAdapter(); } #endif // !FEATURE_PAL && FEATURE_ASYNC_IO /// MaxShadowBufferSize is chosed such that shadow buffers are not allocated on the Large Object Heap. /// Currently, an object is allocated on the LOH if it is larger than 85000 bytes. See LARGE_OBJECT_SIZE in ndp\clr\src\vm\gc.h /// We will go with exactly 80 KBytes, although this is somewhat arbitrary. private const Int32 MaxShadowBufferSize = 81920; // Make sure not to get to the Large Object Heap. private void EnsureShadowBufferAllocated() { Contract.Assert(_buffer != null); Contract.Assert(_bufferSize > 0); // Already have shadow buffer? if (_buffer.Length != _bufferSize || _bufferSize >= MaxShadowBufferSize) return; Byte[] shadowBuffer = new Byte[Math.Min(_bufferSize + _bufferSize, MaxShadowBufferSize)]; Buffer.InternalBlockCopy(_buffer, 0, shadowBuffer, 0, _writePos); _buffer = shadowBuffer; } private void EnsureBufferAllocated() { Contract.Assert(_bufferSize > 0); // BufferedStream is not intended for multi-threaded use, so no worries about the get/set ---- on _buffer. if (_buffer == null) _buffer = new Byte[_bufferSize]; } internal Stream UnderlyingStream { [FriendAccessAllowed] [Pure] get { return _stream; } } internal Int32 BufferSize { [FriendAccessAllowed] [Pure] get { return _bufferSize; } } public override bool CanRead { [Pure] get { return _stream != null && _stream.CanRead; } } public override bool CanWrite { [Pure] get { return _stream != null && _stream.CanWrite; } } public override bool CanSeek { [Pure] get { return _stream != null && _stream.CanSeek; } } public override Int64 Length { get { EnsureNotClosed(); if (_writePos > 0) FlushWrite(); return _stream.Length; } } public override Int64 Position { get { EnsureNotClosed(); EnsureCanSeek(); Contract.Assert(! (_writePos > 0 && _readPos != _readLen), "Read and Write buffers cannot both have data in them at the same time."); return _stream.Position + (_readPos - _readLen + _writePos); } set { if (value < 0) throw new ArgumentOutOfRangeException("value", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); Contract.EndContractBlock(); EnsureNotClosed(); EnsureCanSeek(); if (_writePos > 0) FlushWrite(); _readPos = 0; _readLen = 0; _stream.Seek(value, SeekOrigin.Begin); } } protected override void Dispose(bool disposing) { try { if (disposing && _stream != null) { try { Flush(); } finally { _stream.Close(); } } } finally { _stream = null; _buffer = null; #if FEATURE_ASYNC_IO _lastSyncCompletedReadTask = null; #endif // !FEATURE_PAL && FEATURE_ASYNC_IO // Call base.Dispose(bool) to cleanup async IO resources base.Dispose(disposing); } } public override void Flush() { EnsureNotClosed(); // Has WRITE data in the buffer: if (_writePos > 0) { FlushWrite(); Contract.Assert(_writePos == 0 && _readPos == 0 && _readLen == 0); return; } // Has READ data in the buffer: if (_readPos < _readLen) { // If the underlying stream is not seekable AND we have something in the read buffer, then FlushRead would throw. // We can either throw away the buffer resulting in data loss (!) or ignore the Flush. // (We cannot throw becasue it would be a breaking change.) We opt into ignoring the Flush in that situation. if (!_stream.CanSeek) return; FlushRead(); // User streams may have opted to throw from Flush if CanWrite is false (although the abstract Stream does not do so). // However, if we do not forward the Flush to the underlying stream, we may have problems when chaining several streams. // Let us make a best effort attempt: if (_stream.CanWrite || _stream is BufferedStream) _stream.Flush(); Contract.Assert(_writePos == 0 && _readPos == 0 && _readLen == 0); return; } // We had no data in the buffer, but we still need to tell the underlying stream to flush. if (_stream.CanWrite || _stream is BufferedStream) _stream.Flush(); _writePos = _readPos = _readLen = 0; } #if FEATURE_ASYNC_IO public override Task FlushAsync(CancellationToken cancellationToken) { if (cancellationToken.IsCancellationRequested) return Task.FromCancellation(cancellationToken); EnsureNotClosed(); return FlushAsyncInternal(cancellationToken, this, _stream, _writePos, _readPos, _readLen); } private static async Task FlushAsyncInternal(CancellationToken cancellationToken, BufferedStream _this, Stream stream, Int32 writePos, Int32 readPos, Int32 readLen) { // We bring instance fields down as local parameters to this async method becasue BufferedStream is derived from MarshalByRefObject. // Field access would be from the async state machine i.e., not via the this pointer and would require runtime checking to see // if we are talking to a remote object, whcih is currently very slow (Dev11 bug #365921). // Field access from whithin Asserts is, of course, irrelevant. Contract.Assert(stream != null); SemaphoreSlim sem = _this.EnsureAsyncActiveSemaphoreInitialized(); await sem.WaitAsync().ConfigureAwait(false); try { if (writePos > 0) { await _this.FlushWriteAsync(cancellationToken).ConfigureAwait(false); Contract.Assert(_this._writePos == 0 && _this._readPos == 0 && _this._readLen == 0); return; } if (readPos < readLen) { // If the underlying stream is not seekable AND we have something in the read buffer, then FlushRead would throw. // We can either throw away the buffer resulting in date loss (!) or ignore the Flush. (We cannot throw becasue it // would be a breaking change.) We opt into ignoring the Flush in that situation. if (!stream.CanSeek) return; _this.FlushRead(); // not async; it uses Seek, but there's no SeekAsync // User streams may have opted to throw from Flush if CanWrite is false (although the abstract Stream does not do so). // However, if we do not forward the Flush to the underlying stream, we may have problems when chaining several streams. // Let us make a best effort attempt: if (stream.CanRead || stream is BufferedStream) await stream.FlushAsync(cancellationToken).ConfigureAwait(false); Contract.Assert(_this._writePos == 0 && _this._readPos == 0 && _this._readLen == 0); return; } // We had no data in the buffer, but we still need to tell the underlying stream to flush. if (stream.CanWrite || stream is BufferedStream) await stream.FlushAsync(cancellationToken).ConfigureAwait(false); // There was nothing in the buffer: Contract.Assert(_this._writePos == 0 && _this._readPos == _this._readLen); } finally { sem.Release(); } } #endif // !FEATURE_PAL && FEATURE_ASYNC_IO // Reading is done in blocks, but someone could read 1 byte from the buffer then write. // At that point, the underlying stream's pointer is out of [....] with this stream's position. // All write functions should call this function to ensure that the buffered data is not lost. private void FlushRead() { Contract.Assert(_writePos == 0, "BufferedStream: Write buffer must be empty in FlushRead!"); if (_readPos - _readLen != 0) _stream.Seek(_readPos - _readLen, SeekOrigin.Current); _readPos = 0; _readLen = 0; } private void ClearReadBufferBeforeWrite() { // This is called by write methods to clear the read buffer. Contract.Assert(_readPos <= _readLen, "_readPos <= _readLen [" + _readPos +" <= " + _readLen + "]"); // No READ data in the buffer: if (_readPos == _readLen) { _readPos = _readLen = 0; return; } // Must have READ data. Contract.Assert(_readPos < _readLen); // If the underlying stream cannot seek, FlushRead would end up throwing NotSupported. // However, since the user did not call a method that is intuitively expected to seek, a better message is in order. // Ideally, we would throw an InvalidOperation here, but for backward compat we have to stick with NotSupported. if (!_stream.CanSeek) throw new NotSupportedException(Environment.GetResourceString("NotSupported_CannotWriteToBufferedStreamIfReadBufferCannotBeFlushed")); FlushRead(); } private void FlushWrite() { Contract.Assert(_readPos == 0 && _readLen == 0, "BufferedStream: Read buffer must be empty in FlushWrite!"); Contract.Assert(_buffer != null && _bufferSize >= _writePos, "BufferedStream: Write buffer must be allocated and write position must be in the bounds of the buffer in FlushWrite!"); _stream.Write(_buffer, 0, _writePos); _writePos = 0; _stream.Flush(); } #if FEATURE_ASYNC_IO private async Task FlushWriteAsync(CancellationToken cancellationToken) { Contract.Assert(_readPos == 0 && _readLen == 0, "BufferedStream: Read buffer must be empty in FlushWrite!"); Contract.Assert(_buffer != null && _bufferSize >= _writePos, "BufferedStream: Write buffer must be allocated and write position must be in the bounds of the buffer in FlushWrite!"); await _stream.WriteAsync(_buffer, 0, _writePos, cancellationToken).ConfigureAwait(false); _writePos = 0; await _stream.FlushAsync(cancellationToken).ConfigureAwait(false); } #endif // !FEATURE_PAL && FEATURE_ASYNC_IO private Int32 ReadFromBuffer(Byte[] array, Int32 offset, Int32 count) { Int32 readBytes = _readLen - _readPos; Contract.Assert(readBytes >= 0); if (readBytes == 0) return 0; Contract.Assert(readBytes > 0); if (readBytes > count) readBytes = count; Buffer.InternalBlockCopy(_buffer, _readPos, array, offset, readBytes); _readPos += readBytes; return readBytes; } private Int32 ReadFromBuffer(Byte[] array, Int32 offset, Int32 count, out Exception error) { try { error = null; return ReadFromBuffer(array, offset, count); } catch (Exception ex) { error = ex; return 0; } } public override int Read([In, Out] Byte[] array, Int32 offset, Int32 count) { if (array == null) throw new ArgumentNullException("array", Environment.GetResourceString("ArgumentNull_Buffer")); if (offset < 0) throw new ArgumentOutOfRangeException("offset", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (count < 0) throw new ArgumentOutOfRangeException("count", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (array.Length - offset < count) throw new ArgumentException(Environment.GetResourceString("Argument_InvalidOffLen")); Contract.EndContractBlock(); EnsureNotClosed(); EnsureCanRead(); Int32 bytesFromBuffer = ReadFromBuffer(array, offset, count); // We may have read less than the number of bytes the user asked for, but that is part of the Stream contract. // Reading again for more data may cause us to block if we're using a device with no clear end of file, // such as a serial port or pipe. If we blocked here and this code was used with redirected pipes for a // process's standard output, this can lead to deadlocks involving two processes. // BUT - this is a breaking change. // So: If we could not read all bytes the user asked for from the buffer, we will try once from the underlying // stream thus ensuring the same blocking behaviour as if the underlying stream was not wrapped in this BufferedStream. if (bytesFromBuffer == count) return bytesFromBuffer; Int32 alreadySatisfied = bytesFromBuffer; if (bytesFromBuffer > 0) { count -= bytesFromBuffer; offset += bytesFromBuffer; } // So the READ buffer is empty. Contract.Assert(_readLen == _readPos); _readPos = _readLen = 0; // If there was anything in the WRITE buffer, clear it. if (_writePos > 0) FlushWrite(); // If the requested read is larger than buffer size, avoid the buffer and still use a single read: if (count >= _bufferSize) { return _stream.Read(array, offset, count) + alreadySatisfied; } // Ok. We can fill the buffer: EnsureBufferAllocated(); _readLen = _stream.Read(_buffer, 0, _bufferSize); bytesFromBuffer = ReadFromBuffer(array, offset, count); // We may have read less than the number of bytes the user asked for, but that is part of the Stream contract. // Reading again for more data may cause us to block if we're using a device with no clear end of stream, // such as a serial port or pipe. If we blocked here & this code was used with redirected pipes for a process's // standard output, this can lead to deadlocks involving two processes. Additionally, translating one read on the // BufferedStream to more than one read on the underlying Stream may defeat the whole purpose of buffering of the // underlying reads are significantly more expensive. return bytesFromBuffer + alreadySatisfied; } #if FEATURE_ASYNC_IO public override IAsyncResult BeginRead(Byte[] buffer, Int32 offset, Int32 count, AsyncCallback callback, Object state) { if (buffer == null) throw new ArgumentNullException("buffer", Environment.GetResourceString("ArgumentNull_Buffer")); if (offset < 0) throw new ArgumentOutOfRangeException("offset", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (count < 0) throw new ArgumentOutOfRangeException("count", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (buffer.Length - offset < count) throw new ArgumentException(Environment.GetResourceString("Argument_InvalidOffLen")); Contract.EndContractBlock(); // Previous version incorrectly threw NotSupported instead of ObjectDisposed. We keep that behaviour for back-compat. // EnsureNotClosed(); if (_stream == null) __Error.ReadNotSupported(); EnsureCanRead(); Int32 bytesFromBuffer = 0; // Try to satisfy the request from the buffer synchronously. But still need a sem-lock in case that another // Async IO Task accesses the buffer concurrently. If we fail to acquire the lock without waiting, make this // an Async operation. SemaphoreSlim sem = base.EnsureAsyncActiveSemaphoreInitialized(); Task semaphoreLockTask = sem.WaitAsync(); if (semaphoreLockTask.Status == TaskStatus.RanToCompletion) { bool completeSynchronously = true; try { Exception error; bytesFromBuffer = ReadFromBuffer(buffer, offset, count, out error); // If we satistied enough data from the buffer, we can complete synchronously. // Reading again for more data may cause us to block if we're using a device with no clear end of file, // such as a serial port or pipe. If we blocked here and this code was used with redirected pipes for a // process's standard output, this can lead to deadlocks involving two processes. // BUT - this is a breaking change. // So: If we could not read all bytes the user asked for from the buffer, we will try once from the underlying // stream thus ensuring the same blocking behaviour as if the underlying stream was not wrapped in this BufferedStream. completeSynchronously = (bytesFromBuffer == count || error != null); if (completeSynchronously) { SynchronousAsyncResult asyncResult = (error == null) ? new SynchronousAsyncResult(bytesFromBuffer, state) : new SynchronousAsyncResult(error, state, isWrite: false); if (callback != null) callback(asyncResult); return asyncResult; } } finally { if (completeSynchronously) // if this is FALSE, we will be entering ReadFromUnderlyingStreamAsync and releasing there. sem.Release(); } } // Delegate to the async implementation. return BeginReadFromUnderlyingStream(buffer, offset + bytesFromBuffer, count - bytesFromBuffer, callback, state, bytesFromBuffer, semaphoreLockTask); } private IAsyncResult BeginReadFromUnderlyingStream(Byte[] buffer, Int32 offset, Int32 count, AsyncCallback callback, Object state, Int32 bytesAlreadySatisfied, Task semaphoreLockTask) { Task readOp = ReadFromUnderlyingStreamAsync(buffer, offset, count, CancellationToken.None, bytesAlreadySatisfied, semaphoreLockTask, useApmPattern: true); return TaskToApm.Begin(readOp, callback, state); } public override Int32 EndRead(IAsyncResult asyncResult) { if (asyncResult == null) throw new ArgumentNullException("asyncResult"); Contract.Ensures(Contract.Result() >= 0); Contract.EndContractBlock(); var sAR = asyncResult as SynchronousAsyncResult; if (sAR != null) return SynchronousAsyncResult.EndRead(asyncResult); return TaskToApm.End(asyncResult); } private Task LastSyncCompletedReadTask(Int32 val) { Task t = _lastSyncCompletedReadTask; Contract.Assert(t == null || t.Status == TaskStatus.RanToCompletion); if (t != null && t.Result == val) return t; t = Task.FromResult(val); _lastSyncCompletedReadTask = t; return t; } public override Task ReadAsync(Byte[] buffer, Int32 offset, Int32 count, CancellationToken cancellationToken) { if (buffer == null) throw new ArgumentNullException("buffer", Environment.GetResourceString("ArgumentNull_Buffer")); if (offset < 0) throw new ArgumentOutOfRangeException("offset", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (count < 0) throw new ArgumentOutOfRangeException("count", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (buffer.Length - offset < count) throw new ArgumentException(Environment.GetResourceString("Argument_InvalidOffLen")); Contract.EndContractBlock(); // Fast path check for cancellation already requested if (cancellationToken.IsCancellationRequested) return Task.FromCancellation(cancellationToken); EnsureNotClosed(); EnsureCanRead(); Int32 bytesFromBuffer = 0; // Try to satisfy the request from the buffer synchronously. But still need a sem-lock in case that another // Async IO Task accesses the buffer concurrently. If we fail to acquire the lock without waiting, make this // an Async operation. SemaphoreSlim sem = base.EnsureAsyncActiveSemaphoreInitialized(); Task semaphoreLockTask = sem.WaitAsync(); if (semaphoreLockTask.Status == TaskStatus.RanToCompletion) { bool completeSynchronously = true; try { Exception error; bytesFromBuffer = ReadFromBuffer(buffer, offset, count, out error); // If we satistied enough data from the buffer, we can complete synchronously. // Reading again for more data may cause us to block if we're using a device with no clear end of file, // such as a serial port or pipe. If we blocked here and this code was used with redirected pipes for a // process's standard output, this can lead to deadlocks involving two processes. // BUT - this is a breaking change. // So: If we could not read all bytes the user asked for from the buffer, we will try once from the underlying // stream thus ensuring the same blocking behaviour as if the underlying stream was not wrapped in this BufferedStream. completeSynchronously = (bytesFromBuffer == count || error != null); if (completeSynchronously) { return (error == null) ? LastSyncCompletedReadTask(bytesFromBuffer) : Task.FromException(error); } } finally { if (completeSynchronously) // if this is FALSE, we will be entering ReadFromUnderlyingStreamAsync and releasing there. sem.Release(); } } // Delegate to the async implementation. return ReadFromUnderlyingStreamAsync(buffer, offset + bytesFromBuffer, count - bytesFromBuffer, cancellationToken, bytesFromBuffer, semaphoreLockTask, useApmPattern: false); } /// BufferedStream should be as thin a wrapper as possible. We want that ReadAsync delegates to /// ReadAsync of the underlying _stream and that BeginRead delegates to BeginRead of the underlying stream, /// rather than calling the base Stream which implements the one in terms of the other. This allows BufferedStream /// to affect the semantics of the stream it wraps as little as possible. At the same time, we want to share as /// much code between the APM and the Async pattern implementations as possible. This method is called by both with /// a corresponding useApmPattern value. Recall that Task implements IAsyncResult. /// -2 if _bufferSize was set to 0 while waiting on the semaphore; otherwise num of bytes read. private async Task ReadFromUnderlyingStreamAsync(Byte[] array, Int32 offset, Int32 count, CancellationToken cancellationToken, Int32 bytesAlreadySatisfied, Task semaphoreLockTask, bool useApmPattern) { // Same conditions validated with exceptions in ReadAsync: // (These should be Contract.Requires(..) but that method had some issues in async methods; using Assert(..) for now.) Contract.Assert(array != null); Contract.Assert(offset >= 0); Contract.Assert(count >= 0); Contract.Assert(array.Length - offset >= count); Contract.Assert(_stream != null); Contract.Assert(_stream.CanRead); Contract.Assert(_bufferSize > 0); Contract.Assert(semaphoreLockTask != null); // Employ async waiting based on the same synchronization used in BeginRead of the abstract Stream. await semaphoreLockTask.ConfigureAwait(false); try { // The buffer might have been changed by another async task while we were waiting on the semaphore. // Check it now again. Int32 bytesFromBuffer = ReadFromBuffer(array, offset, count); if (bytesFromBuffer == count) return bytesAlreadySatisfied + bytesFromBuffer; if (bytesFromBuffer > 0) { count -= bytesFromBuffer; offset += bytesFromBuffer; bytesAlreadySatisfied += bytesFromBuffer; } Contract.Assert(_readLen == _readPos); _readPos = _readLen = 0; // If there was anything in the WRITE buffer, clear it. if (_writePos > 0) await FlushWriteAsync(cancellationToken).ConfigureAwait(false); // no Begin-End read version for Flush. Use Async. // If the requested read is larger than buffer size, avoid the buffer and still use a single read: if (count >= _bufferSize) { if (useApmPattern) { EnsureBeginEndAwaitableAllocated(); _stream.BeginRead(array, offset, count, BeginEndAwaitableAdapter.Callback, _beginEndAwaitable); return bytesAlreadySatisfied + _stream.EndRead(await _beginEndAwaitable); } else { return bytesAlreadySatisfied + await _stream.ReadAsync(array, offset, count, cancellationToken).ConfigureAwait(false); } } // Ok. We can fill the buffer: EnsureBufferAllocated(); if (useApmPattern) { EnsureBeginEndAwaitableAllocated(); _stream.BeginRead(_buffer, 0, _bufferSize, BeginEndAwaitableAdapter.Callback, _beginEndAwaitable); _readLen = _stream.EndRead(await _beginEndAwaitable); } else { _readLen = await _stream.ReadAsync(_buffer, 0, _bufferSize, cancellationToken).ConfigureAwait(false); } bytesFromBuffer = ReadFromBuffer(array, offset, count); return bytesAlreadySatisfied + bytesFromBuffer; } finally { SemaphoreSlim sem = base.EnsureAsyncActiveSemaphoreInitialized(); sem.Release(); } } #endif // !FEATURE_PAL && FEATURE_ASYNC_IO public override Int32 ReadByte() { EnsureNotClosed(); EnsureCanRead(); if (_readPos == _readLen) { if (_writePos > 0) FlushWrite(); EnsureBufferAllocated(); _readLen = _stream.Read(_buffer, 0, _bufferSize); _readPos = 0; } if (_readPos == _readLen) return -1; Int32 b = _buffer[_readPos++]; return b; } private void WriteToBuffer(Byte[] array, ref Int32 offset, ref Int32 count) { Int32 bytesToWrite = Math.Min(_bufferSize - _writePos, count); if (bytesToWrite <= 0) return; EnsureBufferAllocated(); Buffer.InternalBlockCopy(array, offset, _buffer, _writePos, bytesToWrite); _writePos += bytesToWrite; count -= bytesToWrite; offset += bytesToWrite; } private void WriteToBuffer(Byte[] array, ref Int32 offset, ref Int32 count, out Exception error) { try { error = null; WriteToBuffer(array, ref offset, ref count); } catch (Exception ex) { error = ex; } } public override void Write(Byte[] array, Int32 offset, Int32 count) { if (array == null) throw new ArgumentNullException("array", Environment.GetResourceString("ArgumentNull_Buffer")); if (offset < 0) throw new ArgumentOutOfRangeException("offset", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (count < 0) throw new ArgumentOutOfRangeException("count", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (array.Length - offset < count) throw new ArgumentException(Environment.GetResourceString("Argument_InvalidOffLen")); Contract.EndContractBlock(); EnsureNotClosed(); EnsureCanWrite(); if (_writePos == 0) ClearReadBufferBeforeWrite(); #region Write algorithm comment // We need to use the buffer, while avoiding unnecessary buffer usage / memory copies. // We ASSUME that memory copies are much cheaper than writes to the underlying stream, so if an extra copy is // guaranteed to reduce the number of writes, we prefer it. // We pick a simple strategy that makes degenerate cases rare if our assumptions are right. // // For ever write, we use a simple heuristic (below) to decide whether to use the buffer. // The heuristic has the desirable property (*) that if the specified user data can fit into the currently available // buffer space without filling it up completely, the heuristic will always tell us to use the buffer. It will also // tell us to use the buffer in cases where the current write would fill the buffer, but the remaining data is small // enough such that subsequent operations can use the buffer again. // // Algorithm: // Determine whether or not to buffer according to the heuristic (below). // If we decided to use the buffer: // Copy as much user data as we can into the buffer. // If we consumed all data: We are finished. // Otherwise, write the buffer out. // Copy the rest of user data into the now cleared buffer (no need to write out the buffer again as the heuristic // will prevent it from being filled twice). // If we decided not to use the buffer: // Can the data already in the buffer and current user data be combines to a single write // by allocating a "shadow" buffer of up to twice the size of _bufferSize (up to a limit to avoid LOH)? // Yes, it can: // Allocate a larger "shadow" buffer and ensure the buffered data is moved there. // Copy user data to the shadow buffer. // Write shadow buffer to the underlying stream in a single operation. // No, it cannot (amount of data is still too large): // Write out any data possibly in the buffer. // Write out user data directly. // // Heuristic: // If the subsequent write operation that follows the current write operation will result in a write to the // underlying stream in case that we use the buffer in the current write, while it would not have if we avoided // using the buffer in the current write (by writing current user data to the underlying stream directly), then we // prefer to avoid using the buffer since the corresponding memory copy is wasted (it will not reduce the number // of writes to the underlying stream, which is what we are optimising for). // ASSUME that the next write will be for the same amount of bytes as the current write (most common case) and // determine if it will cause a write to the underlying stream. If the next write is actually larger, our heuristic // still yields the right behaviour, if the next write is actually smaller, we may making an unnecessary write to // the underlying stream. However, this can only occur if the current write is larger than half the buffer size and // we will recover after one iteration. // We have: // useBuffer = (_writePos + count + count < _bufferSize + _bufferSize) // // Example with _bufferSize = 20, _writePos = 6, count = 10: // // +---------------------------------------+---------------------------------------+ // | current buffer | next iteration's "future" buffer | // +---------------------------------------+---------------------------------------+ // |0| | | | | | | | | |1| | | | | | | | | |2| | | | | | | | | |3| | | | | | | | | | // |0|1|2|3|4|5|6|7|8|9|0|1|2|3|4|5|6|7|8|9|0|1|2|3|4|5|6|7|8|9|0|1|2|3|4|5|6|7|8|9| // +-----------+-------------------+-------------------+---------------------------+ // | _writePos | current count | assumed next count|avail buff after next write| // +-----------+-------------------+-------------------+---------------------------+ // // A nice property (*) of this heuristic is that it will always succeed if the user data completely fits into the // available buffer, i.e. if count < (_bufferSize - _writePos). #endregion Write algorithm comment Contract.Assert(_writePos < _bufferSize); Int32 totalUserBytes; bool useBuffer; checked { // We do not expect buffer sizes big enough for an overflow, but if it happens, lets fail early: totalUserBytes = _writePos + count; useBuffer = (totalUserBytes + count < (_bufferSize + _bufferSize)); } if (useBuffer) { WriteToBuffer(array, ref offset, ref count); if (_writePos < _bufferSize) { Contract.Assert(count == 0); return; } Contract.Assert(count >= 0); Contract.Assert(_writePos == _bufferSize); Contract.Assert(_buffer != null); _stream.Write(_buffer, 0, _writePos); _writePos = 0; WriteToBuffer(array, ref offset, ref count); Contract.Assert(count == 0); Contract.Assert(_writePos < _bufferSize); } else { // if (!useBuffer) // Write out the buffer if necessary. if (_writePos > 0) { Contract.Assert(_buffer != null); Contract.Assert(totalUserBytes >= _bufferSize); // Try avoiding extra write to underlying stream by combining previously buffered data with current user data: if (totalUserBytes <= (_bufferSize + _bufferSize) && totalUserBytes <= MaxShadowBufferSize) { EnsureShadowBufferAllocated(); Buffer.InternalBlockCopy(array, offset, _buffer, _writePos, count); _stream.Write(_buffer, 0, totalUserBytes); _writePos = 0; return; } _stream.Write(_buffer, 0, _writePos); _writePos = 0; } // Write out user data. _stream.Write(array, offset, count); } } #if FEATURE_ASYNC_IO public override IAsyncResult BeginWrite(Byte[] buffer, Int32 offset, Int32 count, AsyncCallback callback, Object state) { if (buffer == null) throw new ArgumentNullException("buffer", Environment.GetResourceString("ArgumentNull_Buffer")); if (offset < 0) throw new ArgumentOutOfRangeException("offset", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (count < 0) throw new ArgumentOutOfRangeException("count", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (buffer.Length - offset < count) throw new ArgumentException(Environment.GetResourceString("Argument_InvalidOffLen")); Contract.EndContractBlock(); // Previous version incorrectly threw NotSupported instead of ObjectDisposed. We keep that behaviour for back-compat. // EnsureNotClosed(); if (_stream == null) __Error.ReadNotSupported(); EnsureCanWrite(); // Try to satisfy the request from the buffer synchronously. But still need a sem-lock in case that another // Async IO Task accesses the buffer concurrently. If we fail to acquire the lock without waiting, make this // an Async operation. SemaphoreSlim sem = base.EnsureAsyncActiveSemaphoreInitialized(); Task semaphoreLockTask = sem.WaitAsync(); if (semaphoreLockTask.Status == TaskStatus.RanToCompletion) { bool completeSynchronously = true; try { if (_writePos == 0) ClearReadBufferBeforeWrite(); // If the write completely fits into the buffer, we can complete synchronously. Contract.Assert(_writePos < _bufferSize); completeSynchronously = (count < _bufferSize - _writePos); if (completeSynchronously) { Exception error; WriteToBuffer(buffer, ref offset, ref count, out error); Contract.Assert(count == 0); SynchronousAsyncResult asyncResult = (error == null) ? new SynchronousAsyncResult(state) : new SynchronousAsyncResult(error, state, isWrite: true); if (callback != null) callback(asyncResult); return asyncResult; } } finally { if (completeSynchronously) // if this is FALSE, we will be entering WriteToUnderlyingStreamAsync and releasing there. sem.Release(); } } // Delegate to the async implementation. return BeginWriteToUnderlyingStream(buffer, offset, count, callback, state, semaphoreLockTask); } private IAsyncResult BeginWriteToUnderlyingStream(Byte[] buffer, Int32 offset, Int32 count, AsyncCallback callback, Object state, Task semaphoreLockTask) { Task writeOp = WriteToUnderlyingStreamAsync(buffer, offset, count, CancellationToken.None, semaphoreLockTask, useApmPattern: true); return TaskToApm.Begin(writeOp, callback, state); } public override void EndWrite(IAsyncResult asyncResult) { if (asyncResult == null) throw new ArgumentNullException("asyncResult"); Contract.EndContractBlock(); var sAR = asyncResult as SynchronousAsyncResult; if (sAR != null) { SynchronousAsyncResult.EndWrite(asyncResult); return; } TaskToApm.End(asyncResult); } public override Task WriteAsync(Byte[] buffer, Int32 offset, Int32 count, CancellationToken cancellationToken) { if (buffer == null) throw new ArgumentNullException("buffer", Environment.GetResourceString("ArgumentNull_Buffer")); if (offset < 0) throw new ArgumentOutOfRangeException("offset", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (count < 0) throw new ArgumentOutOfRangeException("count", Environment.GetResourceString("ArgumentOutOfRange_NeedNonNegNum")); if (buffer.Length - offset < count) throw new ArgumentException(Environment.GetResourceString("Argument_InvalidOffLen")); Contract.EndContractBlock(); // Fast path check for cancellation already requested if (cancellationToken.IsCancellationRequested) return Task.FromCancellation(cancellationToken); EnsureNotClosed(); EnsureCanWrite(); // Try to satisfy the request from the buffer synchronously. But still need a sem-lock in case that another // Async IO Task accesses the buffer concurrently. If we fail to acquire the lock without waiting, make this // an Async operation. SemaphoreSlim sem = base.EnsureAsyncActiveSemaphoreInitialized(); Task semaphoreLockTask = sem.WaitAsync(); if (semaphoreLockTask.Status == TaskStatus.RanToCompletion) { bool completeSynchronously = true; try { if (_writePos == 0) ClearReadBufferBeforeWrite(); Contract.Assert(_writePos < _bufferSize); // If the write completely fits into the buffer, we can complete synchronously: completeSynchronously = (count < _bufferSize - _writePos); if (completeSynchronously) { Exception error; WriteToBuffer(buffer, ref offset, ref count, out error); Contract.Assert(count == 0); return (error == null) ? Task.CompletedTask : Task.FromException(error); } } finally { if (completeSynchronously) // if this is FALSE, we will be entering WriteToUnderlyingStreamAsync and releasing there. sem.Release(); } } // Delegate to the async implementation. return WriteToUnderlyingStreamAsync(buffer, offset, count, cancellationToken, semaphoreLockTask, useApmPattern: false); } /// BufferedStream should be as thin a wrapper as possible. We want that WriteAsync delegates to /// WriteAsync of the underlying _stream and that BeginWrite delegates to BeginWrite of the underlying stream, /// rather than calling the base Stream which implements the one in terms of the other. This allows BufferedStream /// to affect the semantics of the stream it wraps as little as possible. At the same time, we want to share as /// much code between the APM and the Async pattern implementations as possible. This method is called by both with /// a corresponding useApmPattern value. Recall that Task implements IAsyncResult. private async Task WriteToUnderlyingStreamAsync(Byte[] array, Int32 offset, Int32 count, CancellationToken cancellationToken, Task semaphoreLockTask, bool useApmPattern) { // (These should be Contract.Requires(..) but that method had some issues in async methods; using Assert(..) for now.) Contract.Assert(array != null); Contract.Assert(offset >= 0); Contract.Assert(count >= 0); Contract.Assert(array.Length - offset >= count); Contract.Assert(_stream != null); Contract.Assert(_stream.CanWrite); Contract.Assert(_bufferSize > 0); Contract.Assert(semaphoreLockTask != null); // See the LARGE COMMENT in Write(..) for the explanation of the write buffer algorithm. await semaphoreLockTask.ConfigureAwait(false); try { // The buffer might have been changed by another async task while we were waiting on the semaphore. // However, note that if we recalculate the [....] completion condition to TRUE, then useBuffer will also be TRUE. if (_writePos == 0) ClearReadBufferBeforeWrite(); Int32 totalUserBytes; bool useBuffer; checked { // We do not expect buffer sizes big enough for an overflow, but if it happens, lets fail early: totalUserBytes = _writePos + count; useBuffer = (totalUserBytes + count < (_bufferSize + _bufferSize)); } if (useBuffer) { WriteToBuffer(array, ref offset, ref count); if (_writePos < _bufferSize) { Contract.Assert(count == 0); return; } Contract.Assert(count >= 0); Contract.Assert(_writePos == _bufferSize); Contract.Assert(_buffer != null); if (useApmPattern) { EnsureBeginEndAwaitableAllocated(); _stream.BeginWrite(_buffer, 0, _writePos, BeginEndAwaitableAdapter.Callback, _beginEndAwaitable); _stream.EndWrite(await _beginEndAwaitable); } else { await _stream.WriteAsync(_buffer, 0, _writePos, cancellationToken).ConfigureAwait(false); } _writePos = 0; WriteToBuffer(array, ref offset, ref count); Contract.Assert(count == 0); Contract.Assert(_writePos < _bufferSize); } else { // if (!useBuffer) // Write out the buffer if necessary. if (_writePos > 0) { Contract.Assert(_buffer != null); Contract.Assert(totalUserBytes >= _bufferSize); // Try avoiding extra write to underlying stream by combining previously buffered data with current user data: if (totalUserBytes <= (_bufferSize + _bufferSize) && totalUserBytes <= MaxShadowBufferSize) { EnsureShadowBufferAllocated(); Buffer.InternalBlockCopy(array, offset, _buffer, _writePos, count); if (useApmPattern) { EnsureBeginEndAwaitableAllocated(); _stream.BeginWrite(_buffer, 0, totalUserBytes, BeginEndAwaitableAdapter.Callback, _beginEndAwaitable); _stream.EndWrite(await _beginEndAwaitable); } else { await _stream.WriteAsync(_buffer, 0, totalUserBytes, cancellationToken).ConfigureAwait(false); } _writePos = 0; return; } if (useApmPattern) { EnsureBeginEndAwaitableAllocated(); _stream.BeginWrite(_buffer, 0, _writePos, BeginEndAwaitableAdapter.Callback, _beginEndAwaitable); _stream.EndWrite(await _beginEndAwaitable); } else { await _stream.WriteAsync(_buffer, 0, _writePos, cancellationToken).ConfigureAwait(false); } _writePos = 0; } // Write out user data. if (useApmPattern) { EnsureBeginEndAwaitableAllocated(); _stream.BeginWrite(array, offset, count, BeginEndAwaitableAdapter.Callback, _beginEndAwaitable); _stream.EndWrite(await _beginEndAwaitable); } else { await _stream.WriteAsync(array, offset, count, cancellationToken).ConfigureAwait(false); } } } finally { SemaphoreSlim sem = base.EnsureAsyncActiveSemaphoreInitialized(); sem.Release(); } } #endif // !FEATURE_PAL && FEATURE_ASYNC_IO public override void WriteByte(Byte value) { EnsureNotClosed(); if (_writePos == 0) { EnsureCanWrite(); ClearReadBufferBeforeWrite(); EnsureBufferAllocated(); } // We should not be flushing here, but only writing to the underlying stream, but previous version flushed, so we keep this. if (_writePos >= _bufferSize - 1) FlushWrite(); _buffer[_writePos++] = value; Contract.Assert(_writePos < _bufferSize); } public override Int64 Seek(Int64 offset, SeekOrigin origin) { EnsureNotClosed(); EnsureCanSeek(); // If we have bytes in the WRITE buffer, flush them out, seek and be done. if (_writePos > 0) { // We should be only writing the buffer and not flushing, // but the previous version did flush and we stick to it for back-compat reasons. FlushWrite(); return _stream.Seek(offset, origin); } // The buffer is either empty or we have a buffered READ. if (_readLen - _readPos > 0 && origin == SeekOrigin.Current) { // If we have bytes in the READ buffer, adjust the seek offset to account for the resulting difference // between this stream's position and the underlying stream's position. offset -= (_readLen - _readPos); } Int64 oldPos = Position; Contract.Assert(oldPos == _stream.Position + (_readPos - _readLen)); Int64 newPos = _stream.Seek(offset, origin); // If the seek destination is still within the data currently in the buffer, we want to keep the buffer data and continue using it. // Otherwise we will throw away the buffer. This can only happen on READ, as we flushed WRITE data above. // The offset of the new/updated seek pointer within _buffer: _readPos = (Int32) (newPos - (oldPos - _readPos)); // If the offset of the updated seek pointer in the buffer is still legal, then we can keep using the buffer: if (0 <= _readPos && _readPos < _readLen) { // Adjust the seek pointer of the underlying stream to reflect the amount of useful bytes in the read buffer: _stream.Seek(_readLen - _readPos, SeekOrigin.Current); } else { // The offset of the updated seek pointer is not a legal offset. Loose the buffer. _readPos = _readLen = 0; } Contract.Assert(newPos == Position, "newPos (=" + newPos + ") == Position (=" + Position + ")"); return newPos; } public override void SetLength(Int64 value) { if (value < 0) throw new ArgumentOutOfRangeException("value", Environment.GetResourceString("ArgumentOutOfRange_NegFileSize")); Contract.EndContractBlock(); EnsureNotClosed(); EnsureCanSeek(); EnsureCanWrite(); Flush(); _stream.SetLength(value); } } // class BufferedStream } // namespace