C# Tutorial - Asynchronous Stream Operations [Intermediate]

So earlier this week, we got a tutorial request to go over how to do asynchronous reading from a stream in C#. Looking around on the site, while we have a number of tutorials that deal with and use streams, we don't generally use the asynchronous methods. It is a good topic, though, so I decided to write something up.

The example we are going to take a look at today is pretty simple - it reads and writes text files. The only special part is that the reading and writing is done asynchronously, using the BeginRead and BeginWrite methods on the Stream class. Below you can see a screen shot of the app - although really there is not much to look at.

We aren't going to go over the XAML for the interface, I'm pretty sure you can already guess what it looks like. All that we care about is that the "Read" button is hooked up to a method called ReadClick and the "Write" button is hooked up to a method called WriteClick. And so first, let's dive into that read method:

private void ReadClick(object sender, RoutedEventArgs e)
{
  OpenFileDialog ofd = new OpenFileDialog();
  ofd.Multiselect = false;
  if (!(bool)ofd.ShowDialog(this))
  { return; }

  if (!File.Exists(ofd.FileName))
  { 
    MessageBox.Show("Invalid File.");
    return; 
  }

  _TextBlock.Text = "";

  FileStream stream = null;
  try
  { stream = File.OpenRead(ofd.FileName); }
  catch (IOException)
  {
    MessageBox.Show("Unable to open file.");
    return;
  }

  if (!stream.CanRead)
  {
    MessageBox.Show("Cannot read stream.");
    stream.Close();
    return;
  }

  Byte[] myByteArray = new Byte[ChunkSize];

  try
  {
    stream.BeginRead(myByteArray, 0, myByteArray.Length, 
      ReadAsyncCallback, new MyAsyncInfo(myByteArray, stream));
  }
  catch (IOException)
  {
    MessageBox.Show("Unable to start read.");
    stream.Close();
  }
}

We start off by showing an open file dialog and wait for the result. If the result is a valid file, we try to open it (making sure to catch possible errors). Once we have the file open, we want to make sure we can read it, and if not we close it and bail. And then finally, we get down to the actual business of reading. And even though the call to BeginRead only starts the reading process, it can still throw a number of different exceptions, which is why we have it surrounded here by a try-catch.

BeginRead first takes some information about what to read into - namely, the byte array and the position in the byte array to start writing data at. Speaking of that byte array, the variable ChunkSize (which we used to set the size of the byte array) is just a constant defined at the top of the class. It has the value of 1024 in this case, but it can be whatever value makes sense for the data you are reading in.

The third argument is the amount of data to try and read, in the case, the length of the byte array. The fourth argument is the callback - this is the function that will get called when the read completes. Finally, the fifth argument is any piece of state that you want to be available to the callback. In this case, we are passing a custom object that looks like the following:

public class MyAsyncInfo
{
  public Byte[] ByteArray { get; set; }
  public Stream MyStream { get; set; }

  public MyAsyncInfo(Byte[] array, Stream stream)
  {
    ByteArray = array;
    MyStream = stream;
  }
}

We are using this object so that we can get to the byte array that has been filled and the stream once we are inside of the read callback function. The read callback function has to match the signature of the AsyncCallback delegate - it is essentially a method that takes one argument (a IAsyncResult) and has no return value. Below you can see what the callback we are using in this case looks like:

private void ReadAsyncCallback(IAsyncResult ar)
{
  MyAsyncInfo info = ar.AsyncState as MyAsyncInfo;

  int amountRead = 0;
  try
  { amountRead = info.MyStream.EndRead(ar); }
  catch (IOException)
  {
    Dispatcher.Invoke(new Action(() => { 
      MessageBox.Show("Unable to complete read."); }));
    info.MyStream.Close();
    return;
  }

  string text = Encoding.UTF8.GetString(info.ByteArray, 0, amountRead);

  Dispatcher.Invoke(new Action(() => { _TextBlock.Text += text; }), null);

  if (info.MyStream.Position < info.MyStream.Length)
  {
    try
    {
      info.MyStream.BeginRead(info.ByteArray, 0, 
        info.ByteArray.Length, ReadAsyncCallback, info);
    }
    catch (IOException)
    {
      Dispatcher.Invoke(new Action(() => { 
        MessageBox.Show("Unable to start next read."); }));
      info.MyStream.Close();
    }
  }
  else
  {
    info.MyStream.Close();
    Dispatcher.Invoke(new Action(() => { MessageBox.Show("Done reading!"); }));
  }
}

The AsyncState property of IAsyncResult is where the state object passed into BeginRead is stored. So the first thing we do here is pull it out. Then, we call EndRead.

For every call to BeginRead, there must be a corresponding call to EndRead. If there were any exceptions during the read process, they will be thrown when EndRead gets called, which is why it is surrounded by a try-catch here.

EndRead always takes an IAsyncResult - it uses that argument to know what asynchronous read to end. You don't always have to call it in the AsyncCallback function either - BeginRead returns a reference to its IAsyncResult (although we don't save it in the code above, since we aren't using it for anything). If you do hold onto that reference, however, you can turn an asynchronous read into a synchronous read at any time just by calling EndRead - the EndRead call will block until the read is complete (it just never blocks when called inside the AsyncCallback, since at that point, the read is already complete).

Another thing to note is that EndRead returns the number of bytes actually read. It could be any value between 0 and the length that was passed into BeginRead. If the value is 0, this means that we are at the end of the stream and so nothing more was read.

The callback function is actually run on a different thread (generally, a thread from the thread pool), which means that we can't directly access elements owned by the main thread. This is why all the error messages, as well as the actual addition of text to the textbox, are all dispatched. If we tried to do any of those actions directly, we would hit InvalidOperationExceptions.

Ok, so back to actually walking through the code. If we successfully EndRead, we decode the text using a UTF8 encoder. Then we dispatch an action to add that text to the textbox in the user interface. If at that point, we are not done reading the file, we dispatch another asynchronous read, checking for errors appropriately. If we are done, close the stream, and pop up a happy done message.

That is it for reading - now we should have a populated text box in the interface. The work flow for writing is extremely similar - for the most part we are just swapping out BeginRead and EndRead for BeginWrite and EndWrite.

private void WriteClick(object sender, RoutedEventArgs e)
{
  SaveFileDialog sfd = new SaveFileDialog();
  sfd.OverwritePrompt = true;
  if (!(bool)sfd.ShowDialog(this))
  { return; }

  FileStream stream = null;
  try
  {
    stream = File.Open(sfd.FileName, FileMode.Create, FileAccess.Write);
  }
  catch (IOException)
  {
    MessageBox.Show("Unable to open file.");
    return;
  }

  if (!stream.CanWrite)
  {
    MessageBox.Show("Cannot write stream.");
    stream.Close();
    return;
  }

  Byte[] myByteArray = Encoding.UTF8.GetBytes(_TextBlock.Text);

  try
  {
    stream.BeginWrite(myByteArray, 0, myByteArray.Length, 
      WriteAsyncCallback, new MyAsyncInfo(myByteArray, stream));
  }
  catch (IOException)
  {
    MessageBox.Show("Unable to start write.");
    stream.Close();
  }
}

BeginWrite and EndWrite have the same semantics as their companion read methods - they take the same type of arguments , including the callback and the state object. You always need to call EndWrite for every BeginWrite, and calling EndWrite before an operation has completed will cause it to block, just like with the read calls.

private void WriteAsyncCallback(IAsyncResult ar)
{
  MyAsyncInfo info = ar.AsyncState as MyAsyncInfo;
  string msg = "Done writing!!";

  try
  {
    info.MyStream.EndWrite(ar);
  }
  catch (IOException)
  {
    msg = "Unable to complete write.";
  }
  finally
  {
    info.MyStream.Close();
  }

  Dispatcher.Invoke(new Action(() => { MessageBox.Show(msg); }));
}

For writing, we are writing the file out all at once, so a lot of the complicated logic from that read callback goes away. You could potentially do reading all at once as well, but generally reading in chunks is advantageous, because you can continually update the interface with the new information.

One difference between EndWrite and EndRead is that EndWrite does not return anything. There are no 'successful partial writes' (unlike with EndRead, where there can be 'sucessful partial reads') - if the write was unable to complete (say, because of disk space), you should expect an exception.

There is one other big thing to note about the asynchronous methods on the stream class. While these methods exist on the base stream class, not every type of stream fully supports them. The base implementation is actually blocking - it fact, all it does is call the regular old Read and Write methods. Certain types of streams, like FileStream (which we were using above) and NetworkStream fully support asynchronous operations, but you should check your particular type of stream before you expect asynchronous behavior out of these methods.

And that is it for working with streams asynchronously! You can grab the Visual Studio project for the example application in the source file section below. 

Source Files:

• StreamAsync.zip