5
\$\begingroup\$

I wanted to make a Stream that can wrap another stream and buffer all writes, to increase performance; comparable to the BufferedStream class.

However, while BufferedStream is strictly synchronous, and will block once its inner buffer is full, until the contents have been written to the underlying stream, this stream should never block on writes, and instead use a separate thread to execute the writes.

This buffer stream blocks only when .Flush() is called or the stream is disposed.

My main worries are also the .Flush() and .Dispose() methods, it seems easy to miss a corner case with multithreading there.

The whole thing is supposed to be thread safe, allowing simultaneous calls from multiple threads, although I suppose in most cases this would not make sense.

The whole thing is supposed to be called like this:

using (var stream = new AsyncBufferedStream(System.IO.File.Create(@"c:\xyz.dat")))
using (var writer = new StreamWriter(stream))
{
    // perform processor-bound work to get the next line to be written
    // then write the line
    writer.Write("abc");
}

(A short main program is at the end of the code)

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace xyz
{
    /// <summary> Wrap a stream into an asynchronous buffer. All writes will be queued to be executed asynchronously</summary>
    class AsyncBufferedStream : Stream
    {
        /// <summary> wrapped inner stream; we pass every write through to this stream</summary>
        private Stream wrappedStream;

        /// <summary> buffer, contains all items not yet written to the wrapped stream</summary>
        private BlockingCollection<BufferBlock> blockBuffer = new BlockingCollection<BufferBlock>();

        /// <summary> The Task that writes to the wrapped Stream</summary>
        private Task writeTask;

        private long countBytesInBuffer = 0;
        private long countBlocksWritten = 0;


        /// <summary> Wrap a stream into an asynchronous buffer. All writes will be queued to be executed asynchronously</summary>
        public AsyncBufferedStream(Stream stream)
        {
            this.wrappedStream = stream;
            this.writeTask = new Task(this.writeToInnerStream, TaskCreationOptions.LongRunning);
            this.writeTask.Start();
        }

        /// <summary> blocks, until the current content of the buffer has been written to the underlying stream, then flushes the underlying stream</summary>
        public override void Flush()
        {
            var buf = this.blockBuffer;
            if (buf==null || buf.IsAddingCompleted) throw new ObjectDisposedException("BlockBuffer");

            if (this.blockBuffer.Count > 0) // buffer is not empty
            {
                // insert a dummy item, and wait until it is processed.
                using (ManualResetEventSlim ev = new ManualResetEventSlim(false))
                {
                    this.blockBuffer.Add(new BufferBlock(ev));
                    ev.Wait();
                }
                // now all bytes, that where in the buffer before Flush(), have been written to the wrapped stream
            }

            var stream = this.wrappedStream;
            if (stream == null) throw new ObjectDisposedException("Stream");
            stream.Flush();

        }

        /// <summary> not supported </summary>
        public override long Seek(long offset, SeekOrigin origin) {throw new NotSupportedException();}

        /// <summary> not supported </summary>
        public override void SetLength(long value) {throw new NotSupportedException();}

        /// <summary> not supported </summary>
        public override int Read(byte[] buffer, int offset, int count) {throw new NotSupportedException();}

        /// <summary> queues the specified bytes for writing to the wrapped stream  </summary>
        public override void Write(byte[] buffer, int offset, int count)
        {
            var bytes = new byte[count];
            Array.Copy(buffer, offset, bytes, 0, count);

            Interlocked.Add(ref this.countBytesInBuffer, bytes.Length);
            this.blockBuffer.Add(new BufferBlock(bytes));
        }

        /// <summary> continuosly writes the content of the buffer to the wrapped stream</summary>
        private void writeToInnerStream()
        {
            foreach (BufferBlock block in blockBuffer.GetConsumingEnumerable())
            {
                if (block.bytes != null)
                {
                    this.wrappedStream.Write(block.bytes, 0, block.bytes.Length);
                    Interlocked.Add(ref this.countBytesInBuffer, -block.bytes.Length);
                    Interlocked.Increment(ref this.countBlocksWritten);
                }

                // notify observers such as the Flush() method
                if (block.ConsumedEvent != null)
                {
                    block.ConsumedEvent.Set();
                }
            }
        }

        /// <summary> not supported </summary>
        public override bool CanRead { get { return false; } }

        /// <summary> not supported </summary>
        public override bool CanSeek { get { return false; } }

        /// <summary> returns true, when this stream can be written to</summary>
        public override bool CanWrite
        {
            get
            {
                var buf = this.blockBuffer;
                if (buf == null) return false;
                var stream = this.wrappedStream;
                if (stream == null) return false;
                return !buf.IsAddingCompleted && stream.CanWrite; 
            }
        }

        /// <summary> returns the length of the underlying wrapped stream</summary>
        public override long Length
        {
            get
            {
                var stream = this.wrappedStream;
                if (stream == null) throw new ObjectDisposedException("stream");
                return stream.Length;
            }
        }

        /// <summary> current position</summary>
        public override long Position
        {
            get
            {
                var stream = this.wrappedStream;
                if (stream==null) throw new ObjectDisposedException("stream");
                return stream.Position + this.countBytesInBuffer;
            } 
            set{throw new NotSupportedException();} }


        /// <summary> Disposes all owned resources</summary>
        protected override void Dispose(bool disposing)
        {
            try
            {
                if (disposing)
                {
                    var buf = this.blockBuffer;
                    var task = this.writeTask;

                    if (buf != null)
                    {
                        try
                        {
                            buf.CompleteAdding(); // now the writeTask should exit
                            if (task != null) task.Wait();
                        }
                        finally
                        {
                            buf.Dispose();
                        }
                    }

                    if (task != null && task.IsCompleted) // Task can only be disposed if it has completed
                    {
                        task.Dispose();
                    }

                    var stream = this.wrappedStream;
                    if (stream != null)
                    {
                        try
                        {
                            stream.Flush();
                        }
                        finally
                        {
                            stream.Dispose();
                        }
                    }
                }
            }
            finally
            {
                this.wrappedStream = null;
                this.blockBuffer = null;
                this.writeTask = null;
                this.countBytesInBuffer = 0;

                base.Dispose(disposing);
            }
        }

        /// <summary> A block of bytes to be written</summary>
        private struct BufferBlock
        {
            /// <summary> bytes to be written</summary>
            internal readonly byte[] bytes;

            /// <summary> notification when this block has been written (optional)</summary>
            internal readonly ManualResetEventSlim ConsumedEvent;

            /// <summary> Normal constructor: just write this block to the stream</summary>
            internal BufferBlock(byte[] bytes)
            {
                this.bytes = bytes;
                this.ConsumedEvent = null;
            }
            /// <summary> Notification for .Flush(): just set an event when this block has been processed</summary>
            internal BufferBlock(ManualResetEventSlim consumedEvent)
            {
                this.bytes = null;
                this.ConsumedEvent = consumedEvent;
            }
        }

}



// -----------------------------------------------------------------------------
// test code
static void Main()
{
    Stopwatch watch = Stopwatch.StartNew();
    using (var stream = System.IO.File.Create(@"c:\temp\test.dat"))
    using (var writer = new StreamWriter(stream))
    {
        for (int i = 0; i < 100000; i++)
        {
            writer.WriteLine("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA");
            for (int j = 0; j < 1000; j++)
            {
                Math.Sin(Math.Cos(Math.Pow(j, j)));
            }
        }
    }
    Console.WriteLine("unbuffered: " + watch.ElapsedMilliseconds);

    watch = Stopwatch.StartNew();
    using (var stream = new AsyncBufferedStream(System.IO.File.Create(@"c:\temp\test.dat")))
    using (var writer = new StreamWriter(stream))
    {
        for (int i = 0; i < 100000; i++)
        {
            writer.WriteLine("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA");
            for (int j = 0; j < 1000; j++)
            {
                Math.Sin(Math.Cos(Math.Pow(j, j)));
            }
        }
    }
    Console.WriteLine("buffered:   " + watch.ElapsedMilliseconds);
    return;
}

The goal was to speed up writing files by splitting the IO-bound and the processor-bound tasks. However first performance tests show worse performance than using the native file stream. I am not sure whether this is the fault of my implementation, or whether the native file stream is already so optimized that an asynchronous wrapper will not be able to outperform it.

\$\endgroup\$
  • \$\begingroup\$ I don't understand the claim that BufferedStream is strictly synchronous; it offers asynchronous methods including WriteAsync and FlushAsync. \$\endgroup\$ – mjolka Jun 20 '15 at 2:23
  • 1
    \$\begingroup\$ It is true that the Stream class does have some default asynchronous methods like BeginWrite and in net 4.5 even WriteAsync, so saying that BufferedStream is entirely synchronous is not really correct, But what I wanted was a stream that takes care of all the asynchronicity behind the scenes, so I can use it in the same way as a normal stream, and pass it to methods that are not designed for concurrency. Perhaps once I upgrade to net 4.5 I will reconsider this issue, but on net 4.0 I found the BeginWrite methods clunky and cumbersome to use, and no good way to use a StreamWriter with them. \$\endgroup\$ – HugoRune Jun 20 '15 at 7:30
3
\$\begingroup\$

The issue is the underlying dataset being tested.

Instead of just writing the same string again and again:

writer.WriteLine("AAAAAAA...");
for (int j = 0; j < 1000; j++)
{
    Math.Sin(Math.Cos(Math.Pow(j, j)));
}

Generate a random string, then write that data:

var testString = Enumerable.Range(0, 1000).Aggregate("", (s, x) => s + (char) random.Next(33, 126));
writer.WriteLine(testString);

Full implementation of Main():

private static void Main(string[] args)
{
    var random = new Random();


    Stopwatch watch = Stopwatch.StartNew();

    using (var stream = new AsyncBufferedStream(System.IO.File.Create(@"c:\temp\test.dat")))
    using (var writer = new StreamWriter(stream))
    {
        for (int i = 0; i < 1000; i++)
        {
            writer.WriteLine(Enumerable.Range(0, 1000).Aggregate("", (s, x) => s + (char)random.Next(33, 126)));
        }
    }
    Console.WriteLine("buffered:   " + watch.ElapsedMilliseconds);

    watch = Stopwatch.StartNew();
    using (var stream = System.IO.File.Create(@"c:\temp\test.dat"))
    using (var writer = new StreamWriter(stream))
    {
        for (int i = 0; i < 10000; i++)
        {
            writer.WriteLine(Enumerable.Range(0, 1000).Aggregate("", (s, x) => s + (char)random.Next(33, 126)));
        }
    }
    Console.WriteLine("unbuffered: " + watch.ElapsedMilliseconds);
    Console.ReadLine();
    return;
}

There seems to be low level optimization for repeating data like "AAAAAAAAAAAAAAAAA", which makes sense. It could literally be an optimization at the hardware level. You need to make sure the disk controller isn't just re-using the data it already has in cache.

On average the new class is performing 5x faster for me on randomly generated data.

\$\endgroup\$
  • \$\begingroup\$ @RubberDuck Any pointers on how I could change this answer to make it more applicable (qualify)? \$\endgroup\$ – Luke Cummings Jun 19 '15 at 19:58
  • 2
    \$\begingroup\$ I changed my mind after re-reading the OP's question. I think this is a very good answer indeed. Pointing out that it was a failure in the bench marking is indeed a valid review IMO. \$\endgroup\$ – RubberDuck Jun 19 '15 at 20:01
  • \$\begingroup\$ Nice catch. It's probably worth noting that the correct solution is not to use random data, but to use representative data - if your data can be optimised (e.g. by behind the scenes compression) who are you to argue? \$\endgroup\$ – NPSF3000 Jun 20 '15 at 9:47
  • \$\begingroup\$ @HugoRune: There is no such thing as a "low level optimization" for repeating data - performance difference seems to be much smaller if you use i < 10000 in both tests. Actually, the default StreamWriter performs ~25% faster on my machine, which makes perfect sense since OP's allocates a bunch of arrays, and then still has to write all this data to disk anyway. \$\endgroup\$ – Groo May 3 '16 at 10:35

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.