7
\$\begingroup\$

I am not very good with thread-safety and often fall prey to subtle issues in concurrency. Therefore, I hope that someone here might be able to tell me whether there is a subtle concurrency issue (race condition etc.) in the following code, or whether it’s fine. In particular, have I used Monitor.Wait and Monitor.PulseAll correctly?

Of course, if you can reason about the code and come to the conclusion that it is already correct, that would be a welcome answer too.

This code is intended to implement the costreams pattern (I invented that name, so you won’t find it in Google). It runs two methods (passed in as delegates) in parallel. It provides one of those methods with a write-only stream and the other with a read-only stream. The idea is for one of them to generate data and write it to the stream, and the other one to read from the stream and consume the data. (The reading/writing methods intended to be passed in could be anything that reads to/writes from a stream; they are not likely to be specifically written to be used with costreams. If they were, they could probably be rewritten so that they wouldn’t need to use streams at all.)

While reviewing the Read method, remember that the contract for Stream.Read is slightly counterintuitive: it is allowed to read and return fewer bytes than requested (as long as it returns the number of bytes actually read). Thus the fact that it sometimes returns fewer bytes than the count parameter requests is not a bug. Of course it must not return 0 except when the end of the stream is reached.

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;

namespace MyLibrary
{
    public static class Costreams
    {
        /// <summary>Runs the two specified processes in parallel, allowing one to generate data by writing it to a stream, and the other to consume the data by reading it from a stream.</summary>
        /// <param name="writingAction">An action that generates data and writes it to a stream.</param>
        /// <param name="readingAction">An action that will want to read information from a stream.</param>
        public static void RunCostreams(Action<Stream> writingAction, Action<Stream> readingAction)
        {
            // Everything the writingAction writes will be enqueued in here and dequeued by the readingAction
            var queue = new Queue<byteChunk>();

            writingCostream writer = new writingCostream(queue);
            readingCostream reader = new readingCostream(queue);

            // Start reading in a new thread. The first call to reader.Read() will block until there is something in the queue to read.
            var thread = new Thread(() => readingAction(reader));
            thread.Start();

            // Start writing. Calls to writer.Write() will place the data in the queue and signal the reading thread.
            writingAction(writer);

            // Insert a null at the end of the queue to signal to the reader that this is where the data ends.
            queue.Enqueue(null);

            // Wait for the reader to consume all the remaining data.
            thread.Join();
        }

        private sealed class byteChunk
        {
            public byte[] Buffer;
            public int Offset;
            public int Count;
        }

        private sealed class readingCostream : Stream
        {
            private Queue<byteChunk> _queue;
            public readingCostream(Queue<byteChunk> queue) { _queue = queue; }

            public override bool CanRead { get { return true; } }
            public override bool CanSeek { get { return false; } }
            public override bool CanWrite { get { return false; } }
            public override void Flush() { }
            public override long Length { get { throw new NotSupportedException(); } }
            public override long Position { get { throw new NotSupportedException(); } set { 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(); }

            public override int Read(byte[] buffer, int offset, int count)
            {
                lock (_queue)
                {
                    // If there is no data waiting to be read, wait for it.
                    while (_queue.Count == 0)
                        Monitor.Wait(_queue);

                    var peeked = _queue.Peek();

                    // A null element in the queue signals the end of the stream. Don't dequeue this item.
                    if (peeked == null)
                        return 0;

                    if (peeked.Count <= count)
                    {
                        // If we can return the complete item, dequeue it
                        Buffer.BlockCopy(peeked.Buffer, peeked.Offset, buffer, offset, peeked.Count);
                        _queue.Dequeue();
                        return peeked.Count;
                    }
                    else
                    {
                        // If we can only return part of the item, modify it accordingly
                        Buffer.BlockCopy(peeked.Buffer, peeked.Offset, buffer, offset, count);
                        peeked.Offset += count;
                        peeked.Count -= count;
                        return count;
                    }
                }
            }
        }

        private sealed class writingCostream : Stream
        {
            private Queue<byteChunk> _queue;
            public writingCostream(Queue<byteChunk> queue) { _queue = queue; }

            public override bool CanRead { get { return false; } }
            public override bool CanSeek { get { return false; } }
            public override bool CanWrite { get { return true; } }
            public override void Flush() { }
            public override long Length { get { throw new NotSupportedException(); } }
            public override long Position { get { throw new NotSupportedException(); } set { 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(); }

            public override void Write(byte[] buffer, int offset, int count)
            {
                // Ignore zero-length writes
                if (count == 0)
                    return;

                lock (_queue)
                {
                    // We have to take a copy of the data because the calling thread might re-use the same buffer multiple times.
                    var bufferCopy = new byte[count];
                    Buffer.BlockCopy(buffer, offset, bufferCopy, 0, count);

                    // Put the data in the queue
                    _queue.Enqueue(new byteChunk { Buffer = bufferCopy, Offset = 0, Count = count });

                    // Signal the reading thread(s) that the queue has changed (in case it's waiting)
                    Monitor.PulseAll(_queue);
                }
            }
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ This seems instinctively to be a serious misuse of Streams. When you start calling NotImplementedException in so many places, you need to question your approach. Given that you can't call Seek, Length, Position or Flush, why does your readingAction need a Stream at all? If you add the calling code to the question, someone might have a much better design. \$\endgroup\$ – pdr Feb 23 '11 at 18:56
  • \$\begingroup\$ @pdr I see no NotImplementedExceptions in this code at all. As for NotSupportedException - standard streams do this too. It's ok for Seek to throw as long as CanSeek is false - it's explicitly part of the contract. \$\endgroup\$ – Roman Starkov Feb 23 '11 at 18:58
  • \$\begingroup\$ @romkyns You're right, of course, that it's NotSupportedException. And also that it's ok to do so in the right circumstances. I'm just suggesting that if everything about a stream is unsupported except Read/Write then there has to be a question over what use the Stream class is to the calling code \$\endgroup\$ – pdr Feb 23 '11 at 19:12
  • \$\begingroup\$ @pdr: This is reusable library code which is supposed to work for any readingAction/writingAction that fulfills the contract (readingAction should only read and writingAction should only write). I assure you that I am perfectly able to avoid using costreams in cases where my particular reading/writing actions don’t actually need streams. \$\endgroup\$ – Timwi Feb 23 '11 at 19:15
  • \$\begingroup\$ @Timwi Fair enough \$\endgroup\$ – pdr Feb 23 '11 at 19:30
5
\$\begingroup\$

Depends what you're trying to do. Is it really possible to have multiple Read threads, as the comment suggests? If so then you're going to run a risk of having several Reads released at once and return in the wrong order. If not then why use PulseAll as opposed to Pulse?

It feels wrong to me that if Read is called and there are items on the Queue then further calls to Write are locked out, but if Read Waits until Write releases it, then Write is called again, Write will be able to lock the Queue and use it. That said, it shouldn't be a problem given that Read will be dealing with another item at the time.

But that raises the question, why bother with the Wait? Why not replace the whole thing with a ManualResetEvent? That way, you only have to lock the queue while you're updating it, so that you can Set the event when you add data, Reset it when you remove the last item.

I haven't tested it, but it will look a lot like this:

public static class Costreams
{
    /// <summary>Runs the two specified processes in parallel, allowing one to generate data by writing it to a stream, and the other to consume the data by reading it from a stream.</summary>
    /// <param name="writingAction">An action that generates data and writes it to a stream.</param>
    /// <param name="readingAction">An action that will want to read information from a stream.</param>
    public static void RunCostreams(Action<Stream> writingAction, Action<Stream> readingAction)
    {
        // Everything the writingAction writes will be enqueued in here and dequeued by the readingAction
        var queue = new Queue<byteChunk>();
        using (var hasData = new ManualResetEvent(false))
        {
            writingCostream writer = new writingCostream(queue, hasData);
            readingCostream reader = new readingCostream(queue, hasData);

            // Start reading in a new thread. The first call to reader.Read() will block until there is something in the queue to read.
            var thread = new Thread(() => readingAction(reader));
            thread.Start();

            // Start writing. Calls to writer.Write() will place the data in the queue and signal the reading thread.
            writingAction(writer);

            // Insert a null at the end of the queue to signal to the reader that this is where the data ends.
            lock(queue)
            {
                queue.Enqueue(null);
                hasData.Set();
            }

            // Wait for the reader to consume all the remaining data.
            thread.Join();
        }
    }

    private sealed class byteChunk
    {
        public byte[] Buffer;
        public int Offset;
        public int Count;
    }

    private sealed class readingCostream : Stream
    {
        private Queue<byteChunk> _queue;
        private ManualResetEvent _hasData;

        public readingCostream(Queue<byteChunk> queue, ManualResetEvent hasData)
        {
            _queue = queue;
            _hasData = hasData;
        }

        public override bool CanRead { get { return true; } }
        public override bool CanSeek { get { return false; } }
        public override bool CanWrite { get { return false; } }
        public override void Flush() { }
        public override long Length { get { throw new NotSupportedException(); } }
        public override long Position { get { throw new NotSupportedException(); } set { 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(); }

        public override int Read(byte[] buffer, int offset, int count)
        {
            // If there is no data waiting to be read, wait for it.
            _hasData.WaitOne();

            byteChunk peeked;
            lock(_queue)
                peeked = _queue.Peek();

            // A null element in the queue signals the end of the stream. Don't dequeue this item.
            if (peeked == null)
                return 0;

            if (peeked.Count <= count)
            {
                // If we can return the complete item, dequeue it
                Buffer.BlockCopy(peeked.Buffer, peeked.Offset, buffer, offset, peeked.Count);
                lock(_queue)
                {
                    _queue.Dequeue();
                    // If this has emptied the queue, tell the next call to read
                    if (_queue.Count == 0)
                        _hasData.Reset();
                }

                return peeked.Count;
            }

            // If we can only return part of the item, modify it accordingly
            Buffer.BlockCopy(peeked.Buffer, peeked.Offset, buffer, offset, count);
            peeked.Offset += count;
            peeked.Count -= count;
            return count;
        }
    }

    private sealed class writingCostream : Stream
    {
        private Queue<byteChunk> _queue;
        private ManualResetEvent _hasData;
        public writingCostream(Queue<byteChunk> queue, ManualResetEvent _hasData)
        {
            _queue = queue;
            _hasData = hasData;
        }

        public override bool CanRead { get { return false; } }
        public override bool CanSeek { get { return false; } }
        public override bool CanWrite { get { return true; } }
        public override void Flush() { }
        public override long Length { get { throw new NotSupportedException(); } }
        public override long Position { get { throw new NotSupportedException(); } set { 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(); }

        public override void Write(byte[] buffer, int offset, int count)
        {
            // Ignore zero-length writes
            if (count == 0)
                return;

            // We have to take a copy of the data because the calling thread might re-use the same buffer multiple times.
            var bufferCopy = new byte[count];
            Buffer.BlockCopy(buffer, offset, bufferCopy, 0, count);

            // Put the data in the queue
            lock (_queue)
            {
                _queue.Enqueue(new byteChunk { Buffer = bufferCopy, Offset = 0, Count = count });

                // Inform the reading thread that the queue now has data
                _hasData.Set();
            }
        }
    }
}

A couple of other comments from a Review standpoint:

  • Microsoft's Naming Standards for C# suggest that class names should be PascalCase
  • You might also want to look at another ManualResetEvent for memory overloads - if Write gets called many times and you run out of memory, you may want to wait until Read has removed some data from the Queue, rather than failing with an OutOfMemoryException
\$\endgroup\$
  • \$\begingroup\$ I guess I can reduce the time spent inside a lock in my original to exactly the same as yours. Apart from that, what is the advantage of using ManualResetEvent over Monitor.Pulse? \$\endgroup\$ – Timwi Feb 24 '11 at 18:44
  • \$\begingroup\$ (BTW your code is missing a lock around the call to Peek, which causes a subtle bug.) \$\endgroup\$ – Timwi Feb 24 '11 at 18:46
  • \$\begingroup\$ @Timwi Using a ResetEvent is much easier to understand and manage. It's like a door in your code and you can only come through it if a certain condition is Set, which is exactly what you're trying to achieve. Using Wait and Pulse is different, it's using a complex dual queue system, making it buggy and fragile and leading to subtle situations like Write being locked out while Read completes. Plus, I don't think you can shorten your locks, at least not all of them \$\endgroup\$ – pdr Feb 24 '11 at 19:29
  • \$\begingroup\$ I'm not sure why you'd need to lock the Queue to Peek at it, assuming there is only one Read thread (as I said, if you have more than one, you will have bigger problems to solve). As there must be one item in the Queue on the line before that, the first item will not change if the Write thread adds another item. \$\endgroup\$ – pdr Feb 24 '11 at 19:34
  • \$\begingroup\$ @pdr: You need a lock around the Peek because Queue<T> is not thread-safe and cannot handle it interleaving with a simultaneous Enqueue. In particular, if the Enqueue needs to grow the queue, which it does in SetCapacity, and the Peek happens after the _array = destinationArray; but before the _head = 0;, then Peek will return the wrong item. \$\endgroup\$ – Timwi Feb 24 '11 at 22:22

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.