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);
}
}
}
}
}
NotImplementedExceptions in this code at all. As forNotSupportedException- standard streams do this too. It's ok for Seek to throw as long asCanSeekis false - it's explicitly part of the contract. \$\endgroup\$readingAction/writingActionthat fulfills the contract (readingActionshould only read andwritingActionshould 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\$