Converting Microsoft's Asynchronous Server Socket Sample to Async CTP

Question

I have successfully implemented and tested  the Microsoft approach described at http://msdn.microsoft.com/en-us/library/fx6588te.aspx

I have downloaded the async CTP.

I am attempting to convert the sample listed to the above to the async/wait model but keep going in circles. I need some help in getting off the dime here. Could someone please show me how to reconstruct this successful sample into the new paradigm  Just the first part would be a great help.

 

using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;

// State object for reading client data asynchronously
public class StateObject {
  // Client socket.
  public Socket workSocket = null;
  // Size of receive buffer.
  public const int BufferSize = 1024;
  // Receive buffer.
  public byte[] buffer = new byte[BufferSize];
// Received data string.
  public StringBuilder sb = new StringBuilder(); 
}

public class AsynchronousSocketListener {
  // Thread signal.
  public static ManualResetEvent allDone = new ManualResetEvent(false);

  public AsynchronousSocketListener() {
  }

  public static void StartListening() {
    // Data buffer for incoming data.
    byte[] bytes = new Byte[1024];

    // Establish the local endpoint for the socket.
    // The DNS name of the computer
    // running the listener is "host.contoso.com".
    IPHostEntry ipHostInfo = Dns.Resolve(Dns.GetHostName());
    IPAddress ipAddress = ipHostInfo.AddressList[0];
    IPEndPoint localEndPoint = new IPEndPoint(ipAddress, 11000);

    // Create a TCP/IP socket.
    Socket listener = new Socket(AddressFamily.InterNetwork,
      SocketType.Stream, ProtocolType.Tcp );

    // Bind the socket to the local endpoint and listen for incoming connections.
    try {
      listener.Bind(localEndPoint);
      listener.Listen(100);

      while (true) {
        // Set the event to nonsignaled state.
        allDone.Reset();

        // Start an asynchronous socket to listen for connections.
        Console.WriteLine("Waiting for a connection...");
        listener.BeginAccept( 
          new AsyncCallback(AcceptCallback),
          listener );

        // Wait until a connection is made before continuing.
        allDone.WaitOne();
      }

    } catch (Exception e) {
      Console.WriteLine(e.ToString());
    }

    Console.WriteLine("\nPress ENTER to continue...");
    Console.Read();
    
  }

  public static void AcceptCallback(IAsyncResult ar) {
    // Signal the main thread to continue.
    allDone.Set();

    // Get the socket that handles the client request.
    Socket listener = (Socket) ar.AsyncState;
    Socket handler = listener.EndAccept(ar);

    // Create the state object.
    StateObject state = new StateObject();
    state.workSocket = handler;
    handler.BeginReceive( state.buffer, 0, StateObject.BufferSize, 0,
      new AsyncCallback(ReadCallback), state);
  }

  public static void ReadCallback(IAsyncResult ar) {
    String content = String.Empty;
    
    // Retrieve the state object and the handler socket
    // from the asynchronous state object.
    StateObject state = (StateObject) ar.AsyncState;
    Socket handler = state.workSocket;

    // Read data from the client socket. 
    int bytesRead = handler.EndReceive(ar);

    if (bytesRead > 0) {
      // There might be more data, so store the data received so far.
      state.sb.Append(Encoding.ASCII.GetString(
        state.buffer,0,bytesRead));

      // Check for end-of-file tag. If it is not there, read 
      // more data.
      content = state.sb.ToString();
      if (content.IndexOf("<EOF>") > -1) {
        // All the data has been read from the 
        // client. Display it on the console.
        Console.WriteLine("Read {0} bytes from socket. \n Data : {1}",
          content.Length, content );
        // Echo the data back to the client.
        Send(handler, content);
      } else {
        // Not all data received. Get more.
        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0,
        new AsyncCallback(ReadCallback), state);
      }
    }
  }
  

 

Thanks.

 

using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;

// State object for reading client data asynchronously
public class StateObject {
  // Client socket.
  public Socket workSocket = null;
  // Size of receive buffer.
  public const int BufferSize = 1024;
  // Receive buffer.
  public byte[] buffer = new byte[BufferSize];
// Received data string.
  public StringBuilder sb = new StringBuilder(); 
}

public class AsynchronousSocketListener {
  // Thread signal.
  public static ManualResetEvent allDone = new ManualResetEvent(false);

  public AsynchronousSocketListener() {
  }

  public static void StartListening() {
    // Data buffer for incoming data.
    byte[] bytes = new Byte[1024];

    // Establish the local endpoint for the socket.
    // The DNS name of the computer
    // running the listener is "host.contoso.com".
    IPHostEntry ipHostInfo = Dns.Resolve(Dns.GetHostName());
    IPAddress ipAddress = ipHostInfo.AddressList[0];
    IPEndPoint localEndPoint = new IPEndPoint(ipAddress, 11000);

    // Create a TCP/IP socket.
    Socket listener = new Socket(AddressFamily.InterNetwork,
      SocketType.Stream, ProtocolType.Tcp );

    // Bind the socket to the local endpoint and listen for incoming connections.
    try {
      listener.Bind(localEndPoint);
      listener.Listen(100);

      while (true) {
        // Set the event to nonsignaled state.
        allDone.Reset();

        // Start an asynchronous socket to listen for connections.
        Console.WriteLine("Waiting for a connection...");
        listener.BeginAccept( 
          new AsyncCallback(AcceptCallback),
          listener );

        // Wait until a connection is made before continuing.
        allDone.WaitOne();
      }

    } catch (Exception e) {
      Console.WriteLine(e.ToString());
    }

    Console.WriteLine("\nPress ENTER to continue...");
    Console.Read();
    
  }

  public static void AcceptCallback(IAsyncResult ar) {
    // Signal the main thread to continue.
    allDone.Set();

    // Get the socket that handles the client request.
    Socket listener = (Socket) ar.AsyncState;
    Socket handler = listener.EndAccept(ar);

    // Create the state object.
    StateObject state = new StateObject();
    state.workSocket = handler;
    handler.BeginReceive( state.buffer, 0, StateObject.BufferSize, 0,
      new AsyncCallback(ReadCallback), state);
  }

  public static void ReadCallback(IAsyncResult ar) {
    String content = String.Empty;
    
    // Retrieve the state object and the handler socket
    // from the asynchronous state object.
    StateObject state = (StateObject) ar.AsyncState;
    Socket handler = state.workSocket;

    // Read data from the client socket. 
    int bytesRead = handler.EndReceive(ar);

    if (bytesRead > 0) {
      // There might be more data, so store the data received so far.
      state.sb.Append(Encoding.ASCII.GetString(
        state.buffer,0,bytesRead));

      // Check for end-of-file tag. If it is not there, read 
      // more data.
      content = state.sb.ToString();
      if (content.IndexOf("<EOF>") > -1) {
        // All the data has been read from the 
        // client. Display it on the console.
        Console.WriteLine("Read {0} bytes from socket. \n Data : {1}",
          content.Length, content );
        // Echo the data back to the client.
        Send(handler, content);
      } else {
        // Not all data received. Get more.
        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0,
        new AsyncCallback(ReadCallback), state);
      }
    }
  }

 

 

Answer 1

Hi Skylistener-

The main processing loop would end up looking something like:

listener.Bind(localEndPoint);
listener.Listen(100);
while (true)
{
    using (Socket handler = await listener.AcceptAsync())
    {
        var sb = new StringBuilder();
        var content = string.Empty;
        int bytesRead;
        while ((bytesRead = await Task<int>.Factory.FromAsync(
            handler.BeginReceive(buffer, 0, buffer.Length, 0, null, null), handler.EndReceive)) > 0)
        {
            sb.Append(Encoding.ASCII.GetString(buffer, 0, bytesRead));
            content = sb.ToString();
            if (content.IndexOf("<EOF>") > -1)
            {
                Console.WriteLine("Read {0} bytes from socket. \n Data : {1}", content.Length, content);
                break;
            }
        }
 
        byte[] byteData = Encoding.ASCII.GetBytes(content);
        int bytesSent = await Task<int>.Factory.FromAsync(
            handler.BeginSend(byteData, 0, byteData.Length, 0, null, null), handler.EndSend);
        Console.WriteLine("Sent {0} bytes to client.", bytesSent);
 
        handler.Shutdown(SocketShutdown.Both);
    }
}

The Async CTP is missing some extension methods for Socket, so I had to use FromAsync directly, and in this case because the missing APIs take more than three parameters (the maximum the delegate-based overloads of FromAsync work with), I used the IAsyncResult-based overloads of FromAsync, which are a bit more expensive.  That said, you could always create your own wrappers for these, e.g.

public static Task<int> ReceiveAsync(this Socket socket, byte[] buffer, int offset, int count, SocketFlags flags)
{
    var tcs = new TaskCompletionSource<int>(socket);
    socket.BeginReceive(buffer, offset, count, flags, iar =>
    {
        try { tcs.SetResult(socket.EndReceive(iar)); }
        catch (Exception exc) { tcs.SetException(exc); }
    }, null);
    return tcs.Task;
}

or if you wanted to avoid a few more allocations:

public static Task<int> ReceiveAsync(this Socket socket, byte[] buffer, int offset, int count, SocketFlags flags)
{
    var tcs = new TaskCompletionSource<int>(socket);
    socket.BeginReceive(buffer, offset, count, flags, iar =>
    {
        var t = (TaskCompletionSource<int>)iar.AsyncState;
        var s = (Socket)t.Task.AsyncState;
        try { t.SetResult(s.EndReceive(iar)); }
        catch (Exception exc) { t.SetException(exc); }
    }, tcs);
    return tcs.Task;
}

I hope that helps.

Answer 2

Stephen Toub can you please explain how second snipped helps with less allocations?

Answer 3

The second snippet uses a lambda that doesn't close over any captured state.  As such, the compiler will likely be able to cache the delegate and reuse it for every call, rather than needing to allocate a closure and a delegate to a method on that closure object for each invocation.  See http://blogs.msdn.com/b/pfxteam/archive/2012/02/03/10263921.aspx for some more info.