Sequential asynchronous reads using Stream.BeginRead

Question

I'm writing a class that exposes a subsection of a stream for reading. Since data may be read from multiple different subsections at the same time, only one operation may be active at any one time.

I had the idea of locking the underlying stream before every operation. Is locking the stream around the BeginRead call sufficient to ensure that concurrent asynchronous reads from different positions in the underlying stream happen correctly?

public sealed class SubStream : Stream
{
    // ...

    public override IAsyncResult BeginRead(byte[] buffer, int offset, int count,
        AsyncCallback callback, object state)
    {
        lock (this.baseStream)
        {
            this.baseStream.Seek(this.offset + this.position, SeekOrigin.Begin);
            return this.baseStream.BeginRead(buffer, offset, count,
                callback, state);
        }
    }

    public override int EndRead(IAsyncResult asyncResult)
    {
        int read;
        lock (this.baseStream)
        {
            read = baseStream.EndRead(asyncResult);
            this.position += read;
        }
        return read;
    }

    // Read() and ReadByte() also lock on this.baseStream (not shown).

    // ...
}

For example, if thread A calls BeginRead, a lock on the base stream is acquired. Now thread B calls BeginRead and has to wait for the lock to be released. Thread A sets the position of the base stream and starts an asynchronous read operation. Then releases the lock. Thread B then acquires the lock and changes the position of the base stream and starts another asynchronous read operation. Then, sometime later, the asynchronous read from thread A completes. Can I be sure that this reads from the original position in the base stream? If not, how do I fix it?

Have you considered using a ReaderWriterLockSlim? Especially the EnterWriteLock part of it fits your description of forcing other threads to wait perfectly. — aevitas
@aevitas I don't think that would work. Since reading may happen at different positions in the base stream, only one read operation can be active at any one time. — Daniel A.A. Pelsmaeker
I see, that sounds like you may need an asynchronous semaphore then. Your current code should work if you used async and await, in which case the lock would hold until your operation returns. In this case, I don't think it does. The SemaphoreSlim class has a WaitAsync method which is probably interesting for you. — aevitas
No, this is quite problematic. For one, locking on the stream object buys you deadlock potential, you don't own that object. You should wrap it so you can associate it with a private lock object. For another, if the lock is ever taken then the odds that BeginRead() will throw is very, very high. Which is a bug, the client code isn't negotiating access to the stream properly. You should always make lots and lots of noise when you detect a threading bug. Wrap a semaphore as well. Which gives you the option for Wait(0) + throw to report the bug. Or Wait() to always make it safe. — Hans Passant
@HansPassant I looked in the implementation of Stream+SyncStream in mscorlib, and there they also lock on the wrapped stream and also wrap the call to BeginRead in a lock. I may not own the object, but by locking on it I can be sure that other objects that lock on the stream (other instances of my SubStream, or an instance of .NET's SyncStream) work correctly. Also, why would BeginRead throw? When the lock is taken, the thread will block until it is released. Otherwise, the execution is no different from the same code without the lock. — Daniel A.A. Pelsmaeker

Sreejith Nair Sreejith Nair · Accepted Answer · 2014-04-01T08:38:08

Here you might end up with multiple threads calling BeginRead on the same instance of resource (baseStream). As per MSDN, the "EndRead must be called exactly once for every call to BeginRead. Failing to end a read process before beginning another read can cause undesirable behavior such as deadlock." In you case, I recon a trouble 'if Thread B is on Seek (on baseStream) while Thread A in the middle of executing their EndRead(callback)'.

Due to the nature of requirement, you are better off with wrapping multi-threaded access with synchronous I/O. This means, the current implementation can be amended with synchronous I/O instead of asynchronous I/O. Also, you may want to consider informing queuing threads about the completion of former threads using Monitor.WaitOne (baseStream) and Monitor.Pulse(baseStream) or Monitor.PulseAll(baseStream).

Alternately, I would like to throw another idea of Memory-Mapped file for segmented style.

Sequential asynchronous reads using Stream.BeginRead

2 Answers