我曾使用Yuric BlockingStream,但在我们的代码中运行20分钟至一小时后,性能急剧下降。我认为性能下降是由于垃圾回收器和该方法在快速流传输大量数据时创建的众多缓冲区所致(我没有时间证明这一点)。最终,我创建了一个环形缓冲区版本,与我们的代码一起使用时不会出现性能下降。
public class RingBufferedStream : Stream
{
private readonly byte[] store;
private readonly ManualResetEventAsync writeAvailable
= new ManualResetEventAsync(false);
private readonly ManualResetEventAsync readAvailable
= new ManualResetEventAsync(false);
private readonly CancellationTokenSource cancellationTokenSource
= new CancellationTokenSource();
private int readPos;
private int readAvailableByteCount;
private int writePos;
private int writeAvailableByteCount;
private bool disposed;
public RingBufferedStream(int bufferSize)
{
this.store = new byte[bufferSize];
this.writeAvailableByteCount = bufferSize;
this.readAvailableByteCount = 0;
}
public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => true;
public override long Length
{
get
{
throw new NotSupportedException(
"Cannot get length on RingBufferedStream");
}
}
public override int ReadTimeout { get; set; } = Timeout.Infinite;
public override int WriteTimeout { get; set; } = Timeout.Infinite;
public override long Position
{
get
{
throw new NotSupportedException(
"Cannot set position on RingBufferedStream");
}
set
{
throw new NotSupportedException(
"Cannot set position on RingBufferedStream");
}
}
public int BufferedByteCount => this.readAvailableByteCount;
public override void Flush()
{
}
public override void SetLength(long value)
{
throw new NotSupportedException(
"Cannot set length on RingBufferedStream");
}
public override long Seek(long offset, SeekOrigin origin)
{
throw new NotSupportedException("Cannot seek on RingBufferedStream");
}
public override void Write(byte[] buffer, int offset, int count)
{
if (this.disposed)
{
throw new ObjectDisposedException("RingBufferedStream");
}
Monitor.Enter(this.store);
bool haveLock = true;
try
{
while (count > 0)
{
if (this.writeAvailableByteCount == 0)
{
this.writeAvailable.Reset();
Monitor.Exit(this.store);
haveLock = false;
bool canceled;
if (!this.writeAvailable.Wait(
this.WriteTimeout,
this.cancellationTokenSource.Token,
out canceled) || canceled)
{
break;
}
Monitor.Enter(this.store);
haveLock = true;
}
else
{
var toWrite = this.store.Length - this.writePos;
if (toWrite > this.writeAvailableByteCount)
{
toWrite = this.writeAvailableByteCount;
}
if (toWrite > count)
{
toWrite = count;
}
Array.Copy(
buffer,
offset,
this.store,
this.writePos,
toWrite);
offset += toWrite;
count -= toWrite;
this.writeAvailableByteCount -= toWrite;
this.readAvailableByteCount += toWrite;
this.writePos += toWrite;
if (this.writePos == this.store.Length)
{
this.writePos = 0;
}
this.readAvailable.Set();
}
}
}
finally
{
if (haveLock)
{
Monitor.Exit(this.store);
}
}
}
public override void WriteByte(byte value)
{
if (this.disposed)
{
throw new ObjectDisposedException("RingBufferedStream");
}
Monitor.Enter(this.store);
bool haveLock = true;
try
{
while (true)
{
if (this.writeAvailableByteCount == 0)
{
this.writeAvailable.Reset();
Monitor.Exit(this.store);
haveLock = false;
bool canceled;
if (!this.writeAvailable.Wait(
this.WriteTimeout,
this.cancellationTokenSource.Token,
out canceled) || canceled)
{
break;
}
Monitor.Enter(this.store);
haveLock = true;
}
else
{
this.store[this.writePos] = value;
--this.writeAvailableByteCount;
++this.readAvailableByteCount;
++this.writePos;
if (this.writePos == this.store.Length)
{
this.writePos = 0;
}
this.readAvailable.Set();
break;
}
}
}
finally
{
if (haveLock)
{
Monitor.Exit(this.store);
}
}
}
public override int Read(byte[] buffer, int offset, int count)
{
if (this.disposed)
{
throw new ObjectDisposedException("RingBufferedStream");
}
Monitor.Enter(this.store);
int ret = 0;
bool haveLock = true;
try
{
while (count > 0)
{
if (this.readAvailableByteCount == 0)
{
this.readAvailable.Reset();
Monitor.Exit(this.store);
haveLock = false;
bool canceled;
if (!this.readAvailable.Wait(
this.ReadTimeout,
this.cancellationTokenSource.Token,
out canceled) || canceled)
{
break;
}
Monitor.Enter(this.store);
haveLock = true;
}
else
{
var toRead = this.store.Length - this.readPos;
if (toRead > this.readAvailableByteCount)
{
toRead = this.readAvailableByteCount;
}
if (toRead > count)
{
toRead = count;
}
Array.Copy(
this.store,
this.readPos,
buffer,
offset,
toRead);
offset += toRead;
count -= toRead;
this.readAvailableByteCount -= toRead;
this.writeAvailableByteCount += toRead;
ret += toRead;
this.readPos += toRead;
if (this.readPos == this.store.Length)
{
this.readPos = 0;
}
this.writeAvailable.Set();
}
}
}
finally
{
if (haveLock)
{
Monitor.Exit(this.store);
}
}
return ret;
}
public override int ReadByte()
{
if (this.disposed)
{
throw new ObjectDisposedException("RingBufferedStream");
}
Monitor.Enter(this.store);
int ret = -1;
bool haveLock = true;
try
{
while (true)
{
if (this.readAvailableByteCount == 0)
{
this.readAvailable.Reset();
Monitor.Exit(this.store);
haveLock = false;
bool canceled;
if (!this.readAvailable.Wait(
this.ReadTimeout,
this.cancellationTokenSource.Token,
out canceled) || canceled)
{
break;
}
Monitor.Enter(this.store);
haveLock = true;
}
else
{
ret = this.store[this.readPos];
++this.writeAvailableByteCount;
--this.readAvailableByteCount;
++this.readPos;
if (this.readPos == this.store.Length)
{
this.readPos = 0;
}
this.writeAvailable.Set();
break;
}
}
}
finally
{
if (haveLock)
{
Monitor.Exit(this.store);
}
}
return ret;
}
protected override void Dispose(bool disposing)
{
if (disposing)
{
this.disposed = true;
this.cancellationTokenSource.Cancel();
}
base.Dispose(disposing);
}
}
该类使用我们的ManualResetEventAsync
来帮助实现干净的关闭。
public sealed class ManualResetEventAsync
{
private volatile TaskCompletionSource<bool> taskCompletionSource =
new TaskCompletionSource<bool>();
public ManualResetEventAsync(bool initialState)
{
if (initialState)
{
this.Set();
}
}
public Task GetWaitTask()
{
return this.taskCompletionSource.Task;
}
public void Set()
{
var tcs = this.taskCompletionSource;
Task.Factory.StartNew(
s => ((TaskCompletionSource<bool>)s).TrySetResult(true),
tcs,
CancellationToken.None,
TaskCreationOptions.PreferFairness,
TaskScheduler.Default);
tcs.Task.Wait();
}
public void Reset()
{
while (true)
{
var tcs = this.taskCompletionSource;
if (!tcs.Task.IsCompleted
#pragma warning disable 420
|| Interlocked.CompareExchange(
ref this.taskCompletionSource,
new TaskCompletionSource<bool>(),
tcs) == tcs)
#pragma warning restore 420
{
return;
}
}
}
public void Wait()
{
this.GetWaitTask().Wait();
}
public void Wait(CancellationToken cancellationToken)
{
this.GetWaitTask().Wait(cancellationToken);
}
public void Wait(CancellationToken cancellationToken, out bool canceled)
{
try
{
this.GetWaitTask().Wait(cancellationToken);
canceled = false;
}
catch (Exception ex)
when (ex is OperationCanceledException
|| (ex is AggregateException
&& ex.InnerOf<OperationCanceledException>() != null))
{
canceled = true;
}
}
public bool Wait(TimeSpan timeout)
{
return this.GetWaitTask().Wait(timeout);
}
public bool Wait(int millisecondsTimeout)
{
return this.GetWaitTask().Wait(millisecondsTimeout);
}
public bool Wait(int millisecondsTimeout, CancellationToken cancellationToken)
{
return this.GetWaitTask().Wait(millisecondsTimeout, cancellationToken);
}
public bool Wait(
int millisecondsTimeout,
CancellationToken cancellationToken,
out bool canceled)
{
bool ret = false;
try
{
ret = this.GetWaitTask().Wait(millisecondsTimeout, cancellationToken);
canceled = false;
}
catch (Exception ex)
when (ex is OperationCanceledException
|| (ex is AggregateException
&& ex.InnerOf<OperationCanceledException>() != null))
{
canceled = true;
}
return ret;
}
}
而且,
ManualResetEventAsync
使用
InnerOf<T>
扩展......
public static class Extensions
{
public static T InnerOf<T>(this Exception ex)
where T : Exception
{
return (T)InnerOf(ex, typeof(T));
}
public static Exception InnerOf(this Exception ex, Type t)
{
if (ex == null || t.IsInstanceOfType(ex))
{
return ex;
}
var ae = ex as AggregateException;
if (ae != null)
{
foreach (var e in ae.InnerExceptions)
{
var ret = InnerOf(e, t);
if (ret != null)
{
return ret;
}
}
}
return InnerOf(ex.InnerException, t);
}
}