Skip to content

Instantly share code, notes, and snippets.

@ayende
Last active Jul 12, 2022
Embed
What would you like to do?
using System.Diagnostics;
const int TotalSize = 1024 * 1024 * 64;
var duration = Stopwatch.StartNew();
long maxMemory = 0;
new Thread(() =>
{
var p = Process.GetCurrentProcess();
var max = p.WorkingSet64;
while (true)
{
var cur = p.WorkingSet64;
if (cur > max)
{
max = cur;
Interlocked.Exchange(ref maxMemory, max);
}
Thread.Sleep(300);
p.Refresh();
}
})
{
IsBackground = true
}.Start();
var readingDone = new TaskCompletionSource();
var server = new FastServerStream(TotalSize, readingDone);
var client = new SlowClientStream(readingDone.Task);
var buffer = new byte[64 * 1024];
while (true)
{
var read = server.Read(buffer, 0, buffer.Length);
if (read == 0)
break;
client.Write(buffer, 0, read);
}
Console.Write("\r \r");
if (maxMemory > 1024 * 1024 * 32)
{
Console.WriteLine($"Failed, too much memory: {maxMemory / 1024:#,#;;0} kb");
}
Console.WriteLine($"Done in {duration.Elapsed} with {maxMemory / 1024:#,#;;0} kb memory");
public class SlowClientStream : Stream
{
private Stopwatch _stopwatch = Stopwatch.StartNew();
private bool _firstRead = true;
private readonly Task _readingDone;
public SlowClientStream(Task readingDone)
{
_readingDone = readingDone;
}
public override void Write(byte[] buffer, int offset, int count)
{
if (_firstRead)
{
_firstRead = false;
if (_readingDone.IsCompleted)
throw new InvalidOperationException("The reads should be interleaved with the writes");
}
for (int i = 0; i <= count / (64 * 1024); i++)
{
Thread.Sleep(1);
}
if (_stopwatch.ElapsedMilliseconds > 5000)
{
_stopwatch.Restart();
Thread.Sleep(750);
}
}
#region Not Interesting
public override bool CanRead => false;
public override bool CanSeek => false;
public override bool CanWrite => true;
public override long Length => throw new NotSupportedException();
public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
public override void Flush()
{
throw new NotSupportedException();
}
public override int Read(byte[] buffer, int offset, int count)
{
throw new NotSupportedException();
}
public override long Seek(long offset, SeekOrigin origin)
{
throw new NotSupportedException();
}
public override void SetLength(long value)
{
throw new NotSupportedException();
}
#endregion
}
public class FastServerStream : Stream
{
private Stopwatch _stopwatch = Stopwatch.StartNew();
private long _readBytes;
private Random _src = new Random(37101);
private Random _sizes = new Random(3423);
private readonly long _size;
private readonly TaskCompletionSource readingDone;
public FastServerStream(long size, TaskCompletionSource readingDone)
{
_size = size;
this.readingDone = readingDone;
}
public override int Read(byte[] buffer, int offset, int count)
{
if (_readBytes >= _size)
{
readingDone.TrySetResult();
return 0;
}
var toRead = (int)Math.Min(count, _size - _readBytes);
if (_sizes.Next(0, 10) == 9)
{
toRead = _sizes.Next(1, toRead);
}
_src.NextBytes(new Span<byte>(buffer, offset, toRead));
_readBytes += toRead;
var minRate = _stopwatch.ElapsedMilliseconds * 1024 * 2;
if (minRate > _readBytes + (1024 * 64) /* grace */)
{
throw new InvalidOperationException($"Too slow: min of {minRate:#,#;;0} vs actual {_readBytes:#,#;;0} in {_stopwatch.Elapsed}");
}
Console.Write($"\rWrote {_readBytes / 1024} kb out of {_size / 1024:#,#;;0} kb - minimum {minRate / 1024:#,#;;0} kb");
return toRead;
}
#region Not Interesting
public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => false;
public override long Length => throw new NotSupportedException();
public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
public override void Flush()
{
throw new NotSupportedException();
}
public override long Seek(long offset, SeekOrigin origin)
{
throw new NotSupportedException();
}
public override void SetLength(long value)
{
throw new NotSupportedException();
}
public override void Write(byte[] buffer, int offset, int count)
{
throw new NotSupportedException();
}
#endregion
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment