10
\$\begingroup\$

Here's my code:

public static void CopyStream(Stream input, Stream output)
{
    var size = 8192;

    var inBuffer = new byte[size];
    var outBuffer = new byte[size];

    IAsyncResult writeAsync = null;

    int count;
    while (true)
    {
        var readAsync = input.BeginRead(inBuffer, 0, size, null, null);

        if (writeAsync != null)
        {
            output.EndWrite(writeAsync);
            writeAsync = null;
        }

        count = input.EndRead(readAsync);
        inBuffer.CopyTo(outBuffer, 0);

        if (count > 0)
        {
            writeAsync = output.BeginWrite(outBuffer, 0, count, null, null);
        }
        else
        {
            break;
        }
    }
}

This really is my first piece of code using the Begin/End asynchronous model. I'm just wondering if I have any glaring issues.

Or, are there any improvements that you can see to make this even better?

\$\endgroup\$
2
  • \$\begingroup\$ This is a cross-post of stackoverflow.com/questions/5061345, but I'm not sure how to handle the closing of the other site's question. \$\endgroup\$ Feb 21 '11 at 1:34
  • \$\begingroup\$ Also, could somebody with more rep add the 'asynchronous' and 'stream' tags to this? \$\endgroup\$ Feb 21 '11 at 1:34
7
\$\begingroup\$

It's not entirely clear what your goals are with this bit of code. Three possibilities that come to mind:

  • CopyStream should be asynchronous. As written, it does not meet this goal since it blocks on the results of EndRead and EndWrite. If this is your goal, you need to use callback methods and not call the End methods until your callbacks have been called. You also need to modify CopyStream's signature to return IAsyncResult and take a callback and state as parameters.
  • CopyStream should hide the details of the underlying asynchronous calls from its users, but is itself synchronous. Your code accomplishes this goal, but there are a few things that could use cleaning up.
  • CopyStream should hide the details of the underlying asynchronous calls, and also maximize performance by making sure the writer is kept busy at all times and never sitting around idle waiting for reads, to the extent possible. It seems like this may have been your goal due to the attempt to read/write in parallel, but by only allowing a single outstanding read or write, you're not really getting much out of it. In order to implement this correctly, you need to use a reader thread and a writer thread, with a queue of items to be written. The reader reads as fast as it can and inserts chunks of data into the queue, which the writer then writes as fast as it can or as fast as they show up.

EDIT:

As an example of how to write this code using a queue, see the following. Note that I didn't need to spin up a thread because I can just make use of the EndRead/EndWrite callbacks and have the original thread wait for a signal to be raised. Also, this is a lightly-tested first draft, so beware that there could be bugs lurking in there. This type of code is complex and difficult to get right, and should only be used when really necessary for valid perf reasons. If the complexity is not justified, I would just use the synchronous Read and Write methods to simplify the code as much as possible.

public static void CopyStream(Stream input, Stream output)
{
    AutoResetEvent completed = new AutoResetEvent(false);
    BeginRead(new CopyStreamState(completed, input, output));

    completed.WaitOne();
}

private static void BeginRead(CopyStreamState state)
{
    const int bufferSize = 8192;

    state.InputBuffer = new byte[bufferSize];
    state.Input.BeginRead(state.InputBuffer, 0, bufferSize, ReadCallback, state);
}

private static void ReadCallback(IAsyncResult ar)
{
    CopyStreamState state = (CopyStreamState)ar.AsyncState;

    int bytesRead = state.Input.EndRead(ar);
    if (bytesRead > 0)
    {
        byte[] dataToWrite = state.InputBuffer;

        if (bytesRead < state.InputBuffer.Length)
        {
            dataToWrite = new byte[bytesRead];
            Array.Copy(state.InputBuffer, dataToWrite, bytesRead);
        }

        EnqueueWriteData(state, dataToWrite);
        BeginRead(state);
    }
    else
    {
        state.FinishedReading = true;
    }

    BeginWriteOrComplete(state);
}

private static void EnqueueWriteData(CopyStreamState state, byte[] data)
{
    lock (state)
    {
        state.WriteQueue.Enqueue(data);
    }
}

private static void BeginWriteOrComplete(CopyStreamState state)
{
    lock (state)
    {
        if (!state.WriteInProgress)
        {
            if (state.WriteQueue.Count > 0)
            {
                byte[] outputBuffer = state.WriteQueue.Dequeue();
                state.WriteInProgress = true;
                state.Output.BeginWrite(outputBuffer, 0, outputBuffer.Length, WriteCallback, state);
            }
            else if (state.FinishedReading)
            {
                state.Completed.Set();
            }
        }
    }
}

private static void WriteCallback(IAsyncResult ar)
{
    CopyStreamState state = (CopyStreamState)ar.AsyncState;
    state.Output.EndWrite(ar);

    lock (state)
    {
        state.WriteInProgress = false;
        BeginWriteOrComplete(state);
    }
}

private class CopyStreamState
{
    public CopyStreamState(
        AutoResetEvent completed, 
        Stream input, 
        Stream output)
    {
        this.Completed = completed;
        this.Input = input;
        this.Output = output;
        this.WriteQueue = new Queue<byte[]>();
    }

    public AutoResetEvent Completed { get; private set; }
    public Stream Input { get; private set; }
    public Stream Output { get; private set; }
    public Queue<byte[]> WriteQueue { get; private set; }

    public byte[] InputBuffer { get; set; }
    public bool FinishedReading { get; set; }
    public bool WriteInProgress { get; set; }
}
\$\endgroup\$
8
  • \$\begingroup\$ Ah, cool, thanks for the info on swapping buffers. Anyways, the second or third bullet point are what I was looking for. \$\endgroup\$ Feb 21 '11 at 3:31
  • \$\begingroup\$ However, I'm not seeing how having a Queue in between the reader and writer will actually help the overall time of the method go down. I mean, it will ensure that the faster operation is done as soon as possible, but even with the Queue, you are still limited by the slowest of the two... \$\endgroup\$ Feb 21 '11 at 3:33
  • \$\begingroup\$ You are still limited by the slowest of the two (writing), but the queue ensures that you can write continuously instead of having to move in lock-step with reading. As it is now, your reader will sit idle while waiting for the writer, and then writer has to sit idle while waiting for the reader. By using a queue, you can allow the reader to move as fast as it can, and ensure that the writer always has something to write, so it never sits idle. Certainly, the reader will finish first and the writer will take longer, but the overall time can go down considerably. \$\endgroup\$
    – Saeed
    Feb 21 '11 at 3:38
  • \$\begingroup\$ I would say that using a queue is the single most important thing here. Using a queue would eliminate any and all swapping/copying of buffers. At least to me, that was the first thing that caught my mind when reading the code. \$\endgroup\$ Feb 21 '11 at 10:21
  • \$\begingroup\$ Ok, fair enough. However, this method needs to be synchronous. What is the simplest way to do this with a queue? Certainly we are going to need to spin up at least 1 Thread (possibly from the pool), and will need locking around the queue, and cross thread signaling, and.... this is getting complex. Not difficult, just complex. \$\endgroup\$ Feb 21 '11 at 18:15
1
\$\begingroup\$

Just a quick note: If I'm following your code correctly, I just can't see how the last if (writeAsync != null) could ever be true.

\$\endgroup\$
1
  • \$\begingroup\$ You're right. My bad. I originally had the EndRead call at the top of the loop. \$\endgroup\$ Feb 21 '11 at 18:11

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.