Non-blocking TCP and UDP in Unity3D Game Engine

Question

I would like constructive criticism of a networking plugin I've written for Unity 3D. The use-case of the code is for people in "creative coding," where simple TCP and UDP messages are used for communication with lighting controllers, micro-controllers, and other devices, in addition to other computers. The requirement was to make it totally non-blocking, even in the event of a delayed or failed TCP message, so that it would be suitable for use in live performances or guest-experiences in theme parks. It does not support persistent TCP.

A design choice was to separate sending and receiving into different classes, in a similar pattern to the graphical language Max. Each of these classes is wrapped by a Unity class (MonoBehaviour derived), messages and errors being passed back and forth from the main Unity thread with ConcurrentQueue.

TCP Sending

public class TcpThreadedClient
{

    // For returning error messages to main 
    private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();

    public void Send(string address, int port, string message)
    {
        var sendTask = new Task(() => SendTask(address, port, message));
        sendTask.Start();
    }

    public void SendTask(string address, int port, string message)
    {
        using (TcpClient client = new TcpClient())
        {
            try
            {
                // Connect to endpoint.
                Task connect = client.ConnectAsync(address, port);
                connect.Wait();
                // Convert the string data to byte data using ASCII encoding.  
                byte[] byteData = Encoding.ASCII.GetBytes(message);
                // Get the NetworkStream and send,
                // with NoDelay (disable caching).
                client.NoDelay = true;
                NetworkStream stream = client.GetStream();
                stream.Write(byteData, 0, message.Length);
            }
            catch (SocketException se)
            {
                PrintError("SocketException connecting to " + address + ":" + port + ": " + se.ToString());
            }
            catch (Exception e)
            {
                PrintError("Unexpected exception connecting to " + address + ":" + port + ": " + e.ToString());
            }
        }       
    }

    private void PrintError(string e)
    {
        errorQueue.Enqueue(e);
    }

    // Wrapperr uses this function to
    // get error messages.
    public bool GetError(out string outError)
    {
        // Out promises initialization
        return errorQueue.TryDequeue(out outError);
    }

TCP Receiving

// State object for reading client data asynchronously  
public class TcpStateObject
{
    // 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 TcpThreadedServer
{
    // Queue for sharing messages with Unity main thread.
    private ConcurrentQueue<NetworkMessage> messageQueue = new ConcurrentQueue<NetworkMessage>();
    // For returning error messages to main Unity thread
    private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();

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

    // For reading new messages from Queue
    // without exposing Queue to other classes.
    public bool GetMessage(out NetworkMessage message)
    {
        if(messageQueue.TryDequeue(out message))
        {
            return true;
        }
        return false;
    }

    // Start the network listening thread.
    public void StartListening(int port)
    {
        // uses Task, but with LongRunning option it will create
        // new thread, not use one from ThreadPool.
        var listenTask = new Task(() => ListenTask(port), TaskCreationOptions.LongRunning);
        listenTask.Start();
    }

    private void ListenTask(int port)
    {
        // Establish the local endpoint for the socket.
        IPAddress ipAddress = IPAddress.Any;
        IPEndPoint localEndPoint = new IPEndPoint(ipAddress, port);

        // Create a TCP/IP socket.  
        Socket listener = new Socket(ipAddress.AddressFamily,
            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.
                listener.BeginAccept(
                    new AsyncCallback(AcceptCallback),
                    listener);

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

        }
        catch (Exception e)
        {
            PrintError("Error opening TCP socket: " + e.ToString());
        }

    }

    private 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.  
        TcpStateObject state = new TcpStateObject();
        state.workSocket = handler;
        handler.BeginReceive(state.buffer, 0, TcpStateObject.BufferSize, 0,
            new AsyncCallback(ReadCallback), state);
    }

    private void ReadCallback(IAsyncResult ar)
    {
        string content = string.Empty;

        // Retrieve the state object and the handler socket  
        // from the asynchronous state object.  
        TcpStateObject state = (TcpStateObject)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();
            int pos = content.IndexOf("<EOF>");
            if (pos > -1)
            {
                // All the data has been read from the   
                // client. Place message into Queue for 
                // Unity to retrieve.
                NetworkMessage message = new NetworkMessage();
                message.content = content.Remove(pos);
                message.clientAddress = ((IPEndPoint)handler.RemoteEndPoint).Address;
                message.clientPort = ((IPEndPoint)handler.RemoteEndPoint).Port;
                messageQueue.Enqueue(message);
            }
            else
            {
                // Not all data received. Get more.  
                handler.BeginReceive(state.buffer, 0, TcpStateObject.BufferSize, 0,
                new AsyncCallback(ReadCallback), state);
            }
        }
    }

    private void PrintError(string e)
    {
        errorQueue.Enqueue(e);
    }

    // For container to check for networking errors.
    public bool GetError(out string outError)
    {
        // Out promises initialization
        return errorQueue.TryDequeue(out outError);
    }

}

UDP Sending

public class UdpThreadedClient
{
    // For returning error messages to main Unity thread
    private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();

    public void Send(string address, int port, string message)
    {
        var sendTask = new Task(() => SendTask(address, port, message));
        sendTask.Start();
    }

    // Even though UDP sends immediately, taking very little time,
    // it is technically blocking.  There are very rare situations
    // in which the call will take longer than usual, so sending
    // takes place on a separate thread.
    public void SendTask(string address, int port, string message)
    {
        using (UdpClient client = new UdpClient())
        {
            try
            {
                // String address to IPAddress
                IPAddress ip;
                IPAddress.TryParse(address, out ip);
                // Convert message to byte array
                byte[] buffer = Encoding.ASCII.GetBytes(message);
                // Creatte IPEndPoint from the given address and port
                IPEndPoint endPoint = new IPEndPoint(ip, port);
                // Send the contents of buffer
                client.Send(buffer, buffer.Length, endPoint);
            }
            catch (SocketException se)
            {
                PrintError("SocketException sending to " + address + ":" + port + ": " + se.ToString());
            }
            catch (Exception e)
            {
                PrintError("Unexpected exception sending to " + address + ":" + port + ": " + e.ToString());
            }
        }
    }

    private void PrintError(string e)
    {
        errorQueue.Enqueue(e);
    }

    // For container to check for networking errors.
    public bool GetError(out string outError)
    {
        // Out promises initialization
        return errorQueue.TryDequeue(out outError);
    }

}

UDP Receiving

public class UdpThreadedServer
{
    // Queue for sharing messages with Unity thread.
    private ConcurrentQueue<NetworkMessage> messageQueue = new ConcurrentQueue<NetworkMessage>();
    // For returning error messages to main Unity thread
    private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();

    // For reading new messages from Queue
    // without exposing Queue to other classes.
    public bool GetMessage(out NetworkMessage message)
    {
        if (messageQueue.TryDequeue(out message))
        {
            return true;
        }
        return false;
    }

    // Start the network listening thread.
    public void StartListening(int port)
    {
        // Uses Task, but with LongRunning option it will create
        // new thread, not use one from ThreadPool.
        var listenTask = new Task(() => ListenTask(port), TaskCreationOptions.LongRunning);
        listenTask.Start();
    }

    private void ListenTask(int port)
    {

        // Bind the socket to the local endpoint and listen for incoming connections.  
        try
        {
            UdpClient udpClient = new UdpClient(port);
            IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Any, 0);

            while (true)
            {
                Byte[] buffer = udpClient.Receive(ref remoteEndPoint);

                string content = Encoding.ASCII.GetString(buffer);
                NetworkMessage message = new NetworkMessage();
                message.content = content;
                message.clientAddress = remoteEndPoint.Address;
                message.clientPort = remoteEndPoint.Port;
                messageQueue.Enqueue(message);
            }

        }
        catch (Exception e)
        {
            PrintError("Error opening UDP socket: " + e.ToString());
        }

    }

    private void PrintError(string e)
    {
        errorQueue.Enqueue(e);
    }

    // For container to check for networking errors.
    public bool GetError(out string outError)
    {
        // Out promises initialization
        return errorQueue.TryDequeue(out outError);
    }

}

Answer 1

General

From the code that you posted it looks like you are creating a client for every message that you are sending, this isn't very common behavior. Unless there is a specific reason i would expect the connection to be reused multiple times. I.e. at start set up the client and then use the existing connection to write as often as needed.

Creating asynchronous flows is not easy and neither is network programming in general. I am not that versed in .NET network programming, but you are already using the socket with some of its callbacks on the server side. I would probably choose a similar approach on the clients side. Using BeginSend() on the clients socket rather than stream.write() will make the writing asynchrounos without having to create a separate task for it, and also without having to start a new connection for every message written.

Why not use the ReadCallback() in the UDP Server code ? You use in in the TCP server code. But the UDP code uses the blocking Read() method.

Calls that may fail

You have GetError() and GetMessage() that indicate success or failure with a boolean. A good convention is to prefix these with Try (as the queue does) to indicate not only a function that may fail but also to remind the user of your API that they should check the return value of these functions.

Errors as strings

Your mileage may vary but having errors only as strings may make discerning what the error was from software a bit harder than necessary, using an enum for the specific error types might make it easier for the client software to actually deal with errors. Pairing this enum with a string explanation might give you the best of both worlds.

PrintError()

I did a double take, and was wondering why just print out the errors, and then I realized you actually are queuing up the errors, in this case i'd probably just get rid of the function and call enqueue directly when needed.

Superclass

To make it a little bit more convenient for the users you could introduce interfaces that can be used independent of whether the actual Server or Client is UPD or TCP.