Socket-Level Server Programming & .NET

A more sophisticated approach is to manage a pool of threads, each waiting on a semaphore to be notified when a client comes along for them. The thread then handles one request and queues it for processing. When the processing is complete, a thread is obtained from the thread pool to send the response to the client. The pool is of fixed size, limiting the number of clients competing for resources, thus preventing the server from being overloaded. The threads are created before the client request arrives, reducing the response time for a particular request since unblocking an existing thread from a semaphore is much faster than creating a new thread. However, the problem with this approach is that it is complex to implement.

The .NET Framework addresses this complexity with its ThreadPool utility class in System.Threading assembly. Some of the methods of the Socket class of the System.Net.Sockets assembly also use the same ThreadPool. An instance of the Socket class is a wrapper for a TCP socket handle to provide an object-oriented veneer. The Socket class offers several methods that wrap the Winsock 2 asynchronous functions (named WSAAsync...) into a straightforward callback mechanism similar to that used to respond to GUI events.

The .NET Framework manages the thread pool for you. If a thread blocks for I/O and work is queued for processing, another thread is unblocked to keep all CPUs in a multiprocessor system busy. If the load on the CPU from other processes is light, the timeslice for the threads gets longer to prevent unneeded context switching. Conversely, if the load on the CPU from other processes is heavy, the timeslice for each thread gets smaller. When the thread pool is initialized, it contains a small number of threads. If the threads in the pool are constantly running, more threads are added to the pool. The maximum number of threads in the pool is limited. The limit may vary with the number of CPUs in the system. ThreadPool.GetMaxThreads returns the maximum on a particular system.

In this article, I'll present code for .NET Framework sockets and asynchronous callbacks. The C# program in Listing One is equivalent to a VB.NET program that's available electronically (see "Resource Center," page 5), except that the VB.NET program does not include code for UDP threads. In both cases, the code implements an echo server.

In the Main() method, the object representing the service is constructed to listen on a specific port number. I used port 2000 as the default to avoid conflicting with the existing echo server running on port 7 on my development machine. You can specify another port as a command-line option and convert to an integer via the Int32.Parse utility function, which is essentially the same as the atoi function in C. The TCPechodAsync constructor creates a Socket object the same way the generic C code fragment using the socket API does in Listing Two. The IPEndPoint object is analogous to the struct sockaddr_in in the traditional socket API. The IPAddress.ANY in the C# code is equivalent to the INADDR_ANY macro in the C code, denoting that the socket will be associated with any and all IP addresses on any and all interfaces on the local host.

After constructing the TCPechodAsync object, the Main method invokes Run, which uses the Socket class's BeginAccept method to set up a call to the FinishAccept method. The call to FinishAccept will not be made until after a client has come along to connect to the server. Meanwhile, the main thread returns from BeginAccept, and then returns to Main. There, the example program sets up a semaphore on which to wait forever. If it did not wait, the program would terminate all the threads in the thread pool and exit. In a more involved server, Main could go on to set up a UDP-based server or an IPv6-based server before waiting forever.

A reference to FinishAccept is encapsulated inside an object of type AsyncCallback. This is the .NET way of managing type safety of pointers to functions. AsyncCallback is declared:

public delegate void AsyncCallback

(IAsyncResult ar);

This is much prettier than the analogous C/C++ declaration:

typedef void (*AsyncCallback)

(IAsyncResult).

In addition, while a function pointer in C/C++ can be typecast in an unsafe manner, a .NET delegate function cannot. Furthermore, a delegate can be wrapped around object member functions with the same syntax used to wrap one around a static class function: Using a pointer to a C++ member function is nontrivial.

Delegates are also easier to use than Java event listener classes. If the Java JDK provided asynchronous socket methods, it would have an interface called something like AcceptListener with a method named FinishAccept. You would then write an inner class that implemented this interface, create an instance of it, and hand that instance off to the class that was wrapped around a socket primitive. Delegates don't give you more expressive power than the aforementioned technique in Java, just more simplicity.

The FinishAccept method, in turn, calls the Socket class's EndAccept method to get the socket associated with a newly connected client. The result returned by EndAccept would be the same if the Accept method was invoked instead of the BeginAccept/EndAccept pair. Just as Run set up an asynchronous call to FinishAccept, FinishAccept sets up an asynchronous call to FinishReceive. The interesting feature of this asynchronous call is that state information about the client connection is passed along as a property of the IAsyncResult parameter. The ClientState inner class encapsulates both the socket for the connection and the buffer that is being used for I/O with that socket.

The first thing FinishAccept does to the socket returned by EndAccept is to set some socket options. Again, the code is very much analogous to the standard socket API. I wanted to increase the size of the operating system's send/receive buffers to increase the potential throughput of the server. To optimize throughput, these buffers need to have the same number of bytes as can be sent before TCP receives an acknowledgment. This would be equal to the product of the round-trip delay in seconds and the effective bandwidth of the network in bytes per second. In a 10-megabit Ethernet, you'd expect round trips to take about one millisecond and an effective bandwidth of a megabyte per second. Hence, a kilobyte buffer is big enough to optimize the connection. However, on higher bandwidth connections with longer round-trip delays, such as a transcontinental ATM, you need proportionally bigger buffers in the OS for optimum performance.

After asking the operating system to begin the Receive operation and call FinishReceive when the operation completes, FinishAccept sets up an asynchronous call to itself to process the next client that comes along. The operating system maintains a queue of connections established with clients that the application program has not yet accepted. The maximum size of this queue is specified as a parameter to the socket Listen function. If too many clients attempt to connect at virtually the same time, the operating system will not have room for them in the queue and will refuse the connection. To avoid this, I was tempted to have Run begin multiple Accept operations on the listening socket. However, the documentation is quite clear that nonstatic members of the same instance of the Socket class are not threadsafe.

After asking the operating system to begin the Accept operation for the next client, FinishAccept sets up an asynchronous call to doInverseDNS by interacting directly with the thread pool using the ThreadPool.QueueUserWorkItem utility function. Here, a WaitCallback delegate wraps the function pointer to doInverseDNS rather than an AsyncCallback delegate. The difference is that a function wrapped in a WaitCallback has a parameter that represents state information associated with the callback when QueueUserWorkItem was invoked rather than the more involved IAsyncResult, which has much more than just application state information.

Figure 1 illustrates how each method of the class is called asynchronously from other methods via a dataflow diagram. The listening socket follows the dashed arrow, client sockets follow the solid arrow, and the inverse DNS lookup follows the dotted arrow. FinishReceive sets up an asynchronous call to FinishSend, which returns the favor by forming an endless loop that is broken only when the client breaks the connection or some exception is thrown. An exception may be thrown in odd circumstances such as when the client's host crashes and subsequently reboots. The client's host then has no idea what to do with the data the server is trying to send to it and responds with a TCP reset.

Just as BeginAccept/EndAccept generates the same result as Accept, BeginReceive/EndReceive generates the same result as Receive, and BeginSend/EndSend generates the same result as Send. BeginReceive and Receive take essentially the same arguments as the standard socket API recv function. However, the .NET API adds more flags that can be used as final parameters. These are enumerated in as constants in the SocketFlags class. The SocketFlags parameter to Receive is a bitmask (multiple flags can be combined with a bitwise-OR). I use SocketFlags.Partial because I want to respond to echo requests containing a line of text smaller than the full buffer size. This behavior is the same as the socket API recv function with zero as the flag for the final parameter. Use of SocketFlags.None simplifies the common task of reading an amount of data known in advance into an appropriately sized buffer. The socket API recv function is called in a loop until the number of bytes returned by each invocation of recv totals to the buffer size.

For the echo protocol, the server needs to take care when sending data back to the client to deal gracefully with a form of deadlock known as "netlock." Suppose a naively written client attempts to send a megabyte of data with one system call and receive the reply after the system call returns. The client will be blocked on the call to send until the OS can transfer most of the data to the server and get it acknowledged. Only when the remaining unacknowledged data can fit in the operating system's send buffer is the client unblocked from a standard call to send the data. The problem is that the server is sending the reply stream at the same time the client is still sending the request stream. Now, as the client's OS receive buffer fills up, it won't be emptied because the client is still trying to send. Eventually, the server's OS send buffer fills up and any more calls to send are blocked.

One thing the server could do is maintain a large application-level buffer inside its program. But how large is large enough? The server has no way of knowing, since the echo protocol doesn't specify how long the request stream will be. A rogue client may even send data continuously without ever stopping to read the reply; requiring a buffer of infinite size. The enlarged OS buffers you created by setting the socket options in FinishAccept will help some, but the server can't prevent netlock from happening. The best you can do is to make sure that such clients don't end up consuming large amounts of server resources. Using a thread pool instead of one thread or process per client does this fairly well. A large number of such clients would still force the server to maintain a lot of state information.

Conclusion

Since I have experience with the Java JDK socket class and the standard UNIX socket API in C, I encountered few surprises with the .NET framework Socket classes. .NET is able to take advantage of the asynchronous Winsock2 functions that are peculiar to Windows and not available on most other operating systems. The Java JDK is forced to either use only the more traditional socket API internally or emulate behaviors that are more sophisticated. Still, the .NET framework faces the same challenge of emulating Winsock2 functionality on other operating systems as it makes claims for cross-platform compatibility.

DDJ

Listing One

// TCPechodAsync.cs
// Author: Paul Buis, Computer Science Dept., Ball State University

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

/// <summary>
/// Example of asynchronous callbacks for network socket I/O in .NET framework
/// </summary>
public class TCPechodAsync 
{
    /// <summary>
    /// max number of unaccepted clients
    /// </summary>
   private const int osListenQueueSize = 64;

    /// <summary>
    /// size of OS's buffer for unsent & unacknowledged data
    /// </summary>
    private const int osSendBufferSize = 0x7ffe;

    /// <summary>
    /// size of OS's buffer for data that has arrived
    /// from client but not yet received by program
    /// </summary>
    private const int osReceiveBufferSize = 0x7ffe;

    /// <summary>
    ///  socket bound to a specific TCP port waiting for clients to connect
    /// </summary>
    protected Socket listenSocket;

    /// <summary>
    /// socket bound to a specific UDP socket
    /// </summary>
    protected Socket udpSocket;

    /// <summary>
    /// initializes listenSocket to listen for clients on specified port #
    /// </summary>
    /// <param name="port">port number to use</param>
    public TCPechodAsync(int port)
    {
        const AddressFamily af = AddressFamily.InterNetwork;
        try
        {
           IPEndPoint localEndPoint = new IPEndPoint(IPAddress.Any, port);
           listenSocket = new Socket(af, SocketType.Stream, ProtocolType.Tcp);
           listenSocket.Bind(localEndPoint);
           listenSocket.Listen(osListenQueueSize);

           udpSocket = new Socket(af, SocketType.Dgram, ProtocolType.Udp);
           udpSocket.Bind(localEndPoint);
        }
        catch (Exception e)
        {
            Console.Error.WriteLine(e.Message);
            Console.Error.WriteLine(e.StackTrace);
        }
    }
    /// <summary>
    ///  registers an event handler to respond to client connections
    ///  and starts a thread for the UDP echo service
    /// </summary>
    public void Run()
    {
        listenSocket.BeginAccept(new AsyncCallback(finishAccept), null);
        ThreadStart threadStart = new ThreadStart(doUDP);
        Thread udpThread = new Thread(threadStart);
        udpThread.Start();
    }
    /// <summary>
    /// handles UDP echo service
    /// </summary>
    public void doUDP()
    {
        const int udpBuffSize = 0xffff;
        byte[] buffer = new byte[udpBuffSize];
        int nBytes; // number of bytes actually read
        while (true)
        {
            IPEndPoint remoteIPEndPoint = new IPEndPoint(IPAddress.Any, 0);
            EndPoint remoteEndPoint = (EndPoint)remoteIPEndPoint;
            try
            {
                nBytes = udpSocket.ReceiveFrom(buffer, 
                    ref remoteEndPoint);
                udpSocket.SendTo(buffer, nBytes, SocketFlags.None,
                    remoteEndPoint);
            }
            catch (Exception e)
            {
                Console.Error.WriteLine(e.Message);
                Console.Error.WriteLine(e.StackTrace);
            }
        }
    }
    /// <summary>
    /// event handler to respond to client connections. 
    /// It accepts the connection, re-registers itself and 
    /// registers a handler to respond received data.
    /// </summary>
    /// <param name="ar">result of accept operation</param>
    protected void finishAccept(IAsyncResult ar)
    {
        Socket clientSocket = null;
        try
        {
            clientSocket = listenSocket.EndAccept(ar);
            listenSocket.BeginAccept(new AsyncCallback(finishAccept), null);
            clientSocket.SetSocketOption(SocketOptionLevel.Socket,
                SocketOptionName.SendBuffer, osSendBufferSize);
            clientSocket.SetSocketOption(SocketOptionLevel.Socket,
                SocketOptionName.ReceiveBuffer, osReceiveBufferSize);
            ClientState state = new ClientState(clientSocket);
            AsyncCallback callback = new AsyncCallback(finishReceive);
            clientSocket.BeginReceive(state.buffer, 0, state.buffer.Length,
               SocketFlags.Partial, callback, state);
            WaitCallback dnsCallback = new WaitCallback(doInverseDNS);
            ThreadPool.QueueUserWorkItem(dnsCallback, 
                                  clientSocket.RemoteEndPoint);
        }
        catch (Exception e)
        {
            Console.Error.WriteLine(e.Message);
            Console.Error.WriteLine(e.StackTrace);
            if (clientSocket != null)
                clientSocket.Close();
        }
    }
    /// <summary>
    /// A work item to do an inverse DNS lookup on an endpoint for logging.
    /// This is done concurrently with finishRead() and finishSend()
    /// </summary>
    /// <param name="state">The remote endpoint containing the address that
    ///  requires the reverse lookup</param>
    protected void doInverseDNS(Object state)
    {
        IPEndPoint ep = (IPEndPoint)state;
        IPHostEntry he = Dns.GetHostByAddress(ep.Address);
        Console.Out.WriteLine("Connection from {0} ({1}) port {2}",
            he.HostName, ep.Address, ep.Port);
    }
    /// <summary>
    /// event handler to process data received from client.
    /// It starts sending data back to client and registers
    /// handler to be notified when send is complete.
    /// </summary>
    /// <param name="ar">result of receive operation</param>
    protected void finishReceive(IAsyncResult ar)
    {
        ClientState state = (ClientState)ar.AsyncState;
        try
        {
            int numBytes = state.socket.EndReceive(ar);
            if (numBytes <= 0)
            {
                state.socket.Close();
                return;
            }
            AsyncCallback callback = new AsyncCallback(finishSend);
            state.socket.BeginSend(state.buffer, 0, numBytes,
                SocketFlags.None, callback, state);
        }
        catch (Exception e)
        {
            Console.Error.WriteLine(e.Message);
            Console.Error.WriteLine(e.StackTrace);
            state.socket.Close();
        }
   }
    /// <summary>
    /// Event handler for completion of sending data to client.
    /// It registers handler for next data recieve.
    /// </summary>
    /// <param name="ar">result of send operation</param>
    protected void finishSend(IAsyncResult ar)
    {
        ClientState state = (ClientState)ar.AsyncState;
        try
        {
            state.socket.EndSend(ar);
            AsyncCallback callback = new AsyncCallback(finishReceive);
            state.socket.BeginReceive(state.buffer, 0, state.buffer.Length,
                SocketFlags.Partial, callback, state);
        }
        catch (Exception e)
        {
            Console.Error.WriteLine(e.Message);
            Console.Error.WriteLine(e.StackTrace);
            state.socket.Close();
        }
    }
    /// <summary>
    /// Parses command line and sets up the service to run
    /// on the port specified in the command line.
    /// </summary>
    /// <param name="args">command-line argument for port number to use</param>
    public static void Main(string[] args)
    {
        int port;
        switch (args.Length)
        {
            case 0:
                port = 2000;
                break;
            case 1:
                port = Int32.Parse(args[0]);
                break;
            default:
                Console.WriteLine("Usage: TCPechodAsync [port]");
                return;
        }
        TCPechodAsync echod = new TCPechodAsync(port);
        echod.Run();

        //wait forever to prevent other threads from getting killed
        WaitHandle waiter = new ManualResetEvent(false);
        waiter.WaitOne();
    }
    /// <summary>
   /// Class to represent the state of interaction with one client
    /// </summary>
    protected class ClientState
    {
        /// <summary>
        /// size of buffer to use for all clients
        /// </summary>
        private const int clientBufferSize = 4096;

        /// <summary>
        /// buffer to use for this client
        /// </summary>
        public byte[] buffer = new byte[clientBufferSize];

        /// <summary>
        /// client socket
        /// </summary>
        public Socket socket;

        /// <summary>
        ///  constructs a state object for specified client socket
        /// </summary>
        /// <param name="socket">socket to use for this client</param>
        public ClientState(Socket socket)
        {
            this.socket = socket;
        }
    }
}

Back to Article

Listing Two

/* generic C code to bind a socket to a port and mark it as listening */
SOCKET listenSocket;
struct sockaddr_in localEndPoint;
listenSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
memset(&localEndPoint, 0, sizeof(localEndPoint));
localEndPoint.sa_family=AF_INET;
localEndPoint.sin_addr.s_addr = INADDR_ANY;
localEndPoint.sin_port = htons((u_short) port)
bind(listenSocket, &localEndPoint, sizeof(localEndPoint));
listen(s, 5);

Back to Article