1339 lines
57 KiB
C#
1339 lines
57 KiB
C#
// ==++==
|
|
//
|
|
// Copyright (c) Microsoft Corporation. All rights reserved.
|
|
//
|
|
// ==--==
|
|
/*============================================================
|
|
**
|
|
** Class: BufferedStream
|
|
**
|
|
** <OWNER>gpaperin</OWNER>
|
|
**
|
|
** 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 {
|
|
|
|
/// <summary>
|
|
/// 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:
|
|
/// <![CDATA[
|
|
/// * 0 <= _readPos <= _readLen < _bufferSize
|
|
/// * 0 <= _writePos < _bufferSize
|
|
/// * _readPos == _readLen && _readPos > 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.
|
|
/// </summary>
|
|
[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<Int32> _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
|
|
|
|
|
|
/// <summary><code>MaxShadowBufferSize</code> 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.</summary>
|
|
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<Int32>(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<Int32> 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<Int32>() >= 0);
|
|
Contract.EndContractBlock();
|
|
|
|
var sAR = asyncResult as SynchronousAsyncResult;
|
|
if (sAR != null)
|
|
return SynchronousAsyncResult.EndRead(asyncResult);
|
|
return TaskToApm.End<Int32>(asyncResult);
|
|
}
|
|
|
|
|
|
private Task<Int32> LastSyncCompletedReadTask(Int32 val) {
|
|
|
|
Task<Int32> t = _lastSyncCompletedReadTask;
|
|
Contract.Assert(t == null || t.Status == TaskStatus.RanToCompletion);
|
|
|
|
if (t != null && t.Result == val)
|
|
return t;
|
|
|
|
t = Task.FromResult<Int32>(val);
|
|
_lastSyncCompletedReadTask = t;
|
|
return t;
|
|
}
|
|
|
|
|
|
public override Task<Int32> 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<Int32>(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<Int32>(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);
|
|
}
|
|
|
|
|
|
/// <summary>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.</summary>
|
|
/// <returns>-2 if _bufferSize was set to 0 while waiting on the semaphore; otherwise num of bytes read.</returns>
|
|
private async Task<Int32> 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<Int32>(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);
|
|
}
|
|
|
|
|
|
/// <summary>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.</summary>
|
|
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
|