Do correct & fast TCP-Networking in c#

Question

I'm trying to implement a Solution, where on service schedules some work to a number of Workers. The scheduling itself should happen over a custom tcp-based protocol, because the Workers can run either on the same or on a different machine.

After some research I found this post. After reading it, I found out that there were at least 3 different possible solutions, each with its advantages and disatvantages.

I decided to go for the Begin-End solution, and wrote a little test program to play around with it. My client is just a simple program that sends some data to the server and then exits. This is the Client-Code:

class Client
{
    static void Main(string[] args)
    {
        var ep = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 12345);
        var s = new Socket(ep.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
        s.Connect(ep);
        Console.WriteLine("client Startet, socket connected");
        s.Send(Encoding.ASCII.GetBytes("1234"));
        s.Send(Encoding.ASCII.GetBytes("ABCDEFGH"));
        s.Send(Encoding.ASCII.GetBytes("A1B2C3D4E5F6"));
        Console.ReadKey();
        s.Close();
    }
}

My Server is nearly as simple, following the example provided:

class Program
{
    static void Main(string[] args)
    {
        var server = new BeginEndTcpServer(8, 1, new IPEndPoint(IPAddress.Parse("127.0.0.1"), 12345));
        // var server = new ThreadedTcpServer(8, new IPEndPoint(IPAddress.Parse("127.0.0.1"), 12345));
        //server.ClientConnected += new EventHandler<ClientConnectedEventArgs>(server_ClientConnected);
        server.DataReceived += new EventHandler<DataReceivedEventArgs>(server_DataReceived);
        server.Start();
        Console.WriteLine("Server Started");
        Console.ReadKey();
    }

    static void server_DataReceived(object sender, DataReceivedEventArgs e)
    {
        Console.WriteLine("Receveived Data: " + Encoding.ASCII.GetString(e.Data));
    }
}


using System;
using System.Collections.Generic;
using System.Net; 
using System.Net.Sockets;

namespace TcpServerTest
{
public sealed class BeginEndTcpServer
{
    private class Connection
    {
        public Guid id;
        public byte[] buffer;
        public Socket socket;
    }

    private readonly Dictionary<Guid, Connection> _sockets;
    private Socket _serverSocket;
    private readonly int _bufferSize;
    private readonly int _backlog;
    private readonly IPEndPoint serverEndPoint;

    public BeginEndTcpServer(int bufferSize, int backlog, IPEndPoint endpoint)
    {
        _sockets = new Dictionary<Guid, Connection>();
        serverEndPoint = endpoint;
        _bufferSize = bufferSize;
        _backlog = backlog;
    }

    public bool Start()
    {
        //System.Net.IPHostEntry localhost = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName());
        try
        {
            _serverSocket = new Socket(serverEndPoint.Address.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
        }
        catch (System.Net.Sockets.SocketException e)
        {
            throw new ApplicationException("Could not create socket, check to make sure not duplicating port", e);
        }
        try
        {
            _serverSocket.Bind(serverEndPoint);
            _serverSocket.Listen(_backlog);
        }
        catch (Exception e)
        {
            throw new ApplicationException("Error occured while binding socket, check inner exception", e);
        }
        try
        {
            //warning, only call this once, this is a bug in .net 2.0 that breaks if 
            // you're running multiple asynch accepts, this bug may be fixed, but
            // it was a major pain in the ass previously, so make sure there is only one
            //BeginAccept running
            _serverSocket.BeginAccept(new AsyncCallback(AcceptCallback), _serverSocket);
        }
        catch (Exception e)
        {
            throw new ApplicationException("Error occured starting listeners, check inner exception", e);
        }
        return true;
    }

    public void Stop()
    {
        _serverSocket.Close();
        lock (_sockets)
            foreach (var s in _sockets)
                s.Value.socket.Close();
    }

    private void AcceptCallback(IAsyncResult result)
    {
        Connection conn = new Connection();
        try
        {
            //Finish accepting the connection
            System.Net.Sockets.Socket s = (System.Net.Sockets.Socket)result.AsyncState;
            conn = new Connection();
            conn.id = Guid.NewGuid();
            conn.socket = s.EndAccept(result);
            conn.buffer = new byte[_bufferSize];
            lock (_sockets)
                _sockets.Add(conn.id, conn);
            OnClientConnected(conn.id);
            //Queue recieving of data from the connection
            conn.socket.BeginReceive(conn.buffer, 0, conn.buffer.Length, SocketFlags.None, new AsyncCallback(ReceiveCallback), conn);
            //Queue the accept of the next incomming connection
            _serverSocket.BeginAccept(new AsyncCallback(AcceptCallback), _serverSocket);
        }
        catch (SocketException)
        {
            if (conn.socket != null)
            {
                conn.socket.Close();
                lock (_sockets)
                    _sockets.Remove(conn.id);
            }
            //Queue the next accept, think this should be here, stop attacks based on killing the waiting listeners
            _serverSocket.BeginAccept(new AsyncCallback(AcceptCallback), _serverSocket);
        }
        catch (Exception)
        {
            if (conn.socket != null)
            {
                conn.socket.Close();
                lock (_sockets)
                    _sockets.Remove(conn.id);
            }
            //Queue the next accept, think this should be here, stop attacks based on killing the waiting listeners
            _serverSocket.BeginAccept(new AsyncCallback(AcceptCallback), _serverSocket);
        }
    }

    private void ReceiveCallback(IAsyncResult result)
    {
        //get our connection from the callback
        Connection conn = (Connection)result.AsyncState;
        //catch any errors, we'd better not have any
        try
        {
            //Grab our buffer and count the number of bytes receives
            int bytesRead = conn.socket.EndReceive(result);
            //make sure we've read something, if we haven't it supposadly means that the client disconnected
            if (bytesRead > 0)
            {
                //put whatever you want to do when you receive data here
                conn.socket.Receive(conn.buffer);
                OnDataReceived(conn.id, (byte[])conn.buffer.Clone());
                //Queue the next receive
                conn.socket.BeginReceive(conn.buffer, 0, conn.buffer.Length, SocketFlags.None, new AsyncCallback(ReceiveCallback), conn);
            }
            else
            {
                //Callback run but no data, close the connection
                //supposadly means a disconnect
                //and we still have to close the socket, even though we throw the event later
                conn.socket.Close();
                lock (_sockets)
                    _sockets.Remove(conn.id);
            }
        }
        catch (SocketException)
        {
            //Something went terribly wrong
            //which shouldn't have happened
            if (conn.socket != null)
            {
                conn.socket.Close();
                lock (_sockets)
                    _sockets.Remove(conn.id);
            }
        }
    }

    public bool Send(byte[] message, Guid connectionId)
    {
        Connection conn = null;
        lock (_sockets)
            if (_sockets.ContainsKey(connectionId))
                conn = _sockets[connectionId];
        if (conn != null && conn.socket.Connected)
        {
            lock (conn.socket)
            {
                //we use a blocking mode send, no async on the outgoing
                //since this is primarily a multithreaded application, shouldn't cause problems to send in blocking mode
                conn.socket.Send(message, message.Length, SocketFlags.None);
            }
        }
        else
            return false;
        return true;
    }

    public event EventHandler<ClientConnectedEventArgs> ClientConnected;
    private void OnClientConnected(Guid id)
    {
        if (ClientConnected != null)
            ClientConnected(this, new ClientConnectedEventArgs(id));
    }

    public event EventHandler<DataReceivedEventArgs> DataReceived;
    private void OnDataReceived(Guid id, byte[] data)
    {
        if (DataReceived != null)
            DataReceived(this, new DataReceivedEventArgs(id, data));
    }

    public event EventHandler<ConnectionErrorEventArgs> ConnectionError;

}

public class ClientConnectedEventArgs : EventArgs
{
    private readonly Guid _ConnectionId;
    public Guid ConnectionId { get { return _ConnectionId; } }

    public ClientConnectedEventArgs(Guid id)
    {
        _ConnectionId = id;
    }
}

public class DataReceivedEventArgs : EventArgs
{
    private readonly Guid _ConnectionId;
    public Guid ConnectionId { get { return _ConnectionId; } }

    private readonly byte[] _Data;
    public byte[] Data { get { return _Data; } }

    public DataReceivedEventArgs(Guid id, byte[] data)
    {
        _ConnectionId = id;
        _Data = data;
    }
}

public class ConnectionErrorEventArgs : EventArgs
{
    private readonly Guid _ConnectionId;
    public Guid ConnectionId { get { return _ConnectionId; } }

    private readonly Exception _Error;
    public Exception Error { get { return _Error; } }

    public ConnectionErrorEventArgs(Guid id, Exception ex)
    {
        _ConnectionId = id;
        _Error = ex;
    }
}

}

My problem is: The Server receives only a potion of the data (in the example it receives only 'EFGHA1B2'). Also if I send only 4byte of Data, the server does not receive it until the connection is closed. What am I missing or doing wrong?

The other thing is, at the moment I'm really confused by the different possibilities, is the way I'm doing this a good solution? Or should I try something else?

Any help would be greatly appreciated!

Answer 1

The problem is that immediately after receiving the first couple of bytes, you overwrite them with the next batch:

        int bytesRead = conn.socket.EndReceive(result);
        if (bytesRead > 0)
        {
            //** The line below reads the next batch of data
            conn.socket.Receive(conn.buffer);

            OnDataReceived(conn.id, (byte[])conn.buffer.Clone());

EndReceive will put received data in your buffer, so there is no need to call Receive afterwards. The reason why you get only the middle 8 bytes is that after receiving the last batch the code blocks in Receive, waiting for more data. When the client closes the connection, Receive returns 0, does not modify the buffer and your callback is invoked with whatever EndReceive had received.

Answer 2

I would suspect that the socket is only calling your callback method when the number of bytes received equals the length that you specify in the call to BeginReceive or the socket is closed.

I think the question you need to ask yourself is what the nature of the data that you will be sending over this connection is. Is the data truly an unbroken stream of bytes, such as the content of a file? If so, then perhaps you could reasonably expect the client to send all of the bytes and then close the connection immediately. As you've said, if the client closes the connection, then your callback method will be called for the remaining bytes.

If the connection will remain open and be used to send individual messages, then you may need to devise a simple protocol for that communication. Will the client be sending multiple arbitrarily long strings, and does the server need to handle each string separately? If so, then one approach could be to prefix each string with an integer value (let's say 4 bytes) that indicates the length of the string that you will be sending. Then your server could first call BeginReceive specifying a length of 4 bytes, use those 4 bytes to determine the length of the incoming string, then call Receive specifying the length for the incoming string. If the length of the incoming string exceeds the size of your buffer, then you'll need to handle that case as well.

1. Call BeginReceive to wait for 4 bytes to be received.
2. Calculate the length of the incoming string from those 4 bytes.
3. Call Receive to wait for bytes to be received for the incoming string.
4. Repeat steps 1-3 for the next string.

I've created a small example server and client application that implements this approach. I have the buffer hard-coded as 2048 bytes, but it still works if you specify a buffer as small as 4 bytes.

In my approach, if the incoming data exceeds the size of my existing buffer, then I create a separate byte array to store that incoming data. That may not be the best way to handle that situation, but I think how you handle that situation depends on what you're actually doing with the data.

Server

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;

namespace SocketServer
{
    class Connection
    {
        public Socket Socket { get; private set; }
        public byte[] Buffer { get; private set; }
        public Encoding Encoding { get; private set; }

        public Connection(Socket socket)
        {
            this.Socket = socket;
            this.Buffer = new byte[2048];
            this.Encoding = Encoding.UTF8;
        }

        public void WaitForNextString(AsyncCallback callback)
        {
            this.Socket.BeginReceive(this.Buffer, 0, 4, SocketFlags.None, callback, this);
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            Connection connection;
            using (Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                listener.Bind(new IPEndPoint(IPAddress.Loopback, 6767));
                listener.Listen(1);
                Console.WriteLine("Listening for a connection.  Press any key to end the session.");
                connection = new Connection(listener.Accept());
                Console.WriteLine("Connection established.");
            }

            connection.WaitForNextString(ReceivedString);

            Console.ReadKey();
        }

        static void ReceivedString(IAsyncResult asyncResult)
        {
            Connection connection = (Connection)asyncResult.AsyncState;

            int bytesReceived = connection.Socket.EndReceive(asyncResult);

            if (bytesReceived > 0)
            {
                int length = BitConverter.ToInt32(connection.Buffer, 0);

                byte[] buffer;
                if (length > connection.Buffer.Length)
                    buffer = new byte[length];
                else
                    buffer = connection.Buffer;

                int index = 0;
                int remainingLength = length;
                do
                {
                    bytesReceived = connection.Socket.Receive(buffer, index, remainingLength, SocketFlags.None);
                    index += bytesReceived;
                    remainingLength -= bytesReceived;
                }
                while (bytesReceived > 0 && remainingLength > 0);

                if (remainingLength > 0)
                {
                    Console.WriteLine("Connection was closed before entire string could be received");
                }
                else
                {
                    Console.WriteLine(connection.Encoding.GetString(buffer, 0, length));
                }

                connection.WaitForNextString(ReceivedString);
            }
        }
    }
}

Client

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;

namespace SocketClient
{
    class Program
    {
        static void Main(string[] args)
        {
            var encoding = Encoding.UTF8;
            using (var connector = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                connector.Connect(new IPEndPoint(IPAddress.Loopback, 6767));

                string value;

                value = "1234";
                Console.WriteLine("Press any key to send \"" + value + "\".");
                Console.ReadKey();
                SendString(connector, encoding, value);

                value = "ABCDEFGH";
                Console.WriteLine("Press any key to send \"" + value + "\".");
                Console.ReadKey();
                SendString(connector, encoding, value);

                value = "A1B2C3D4E5F6";
                Console.WriteLine("Press any key to send \"" + value + "\".");
                Console.ReadKey();
                SendString(connector, encoding, value);

                Console.WriteLine("Press any key to exit.");
                Console.ReadKey();

                connector.Close();
            }
        }

        static void SendString(Socket socket, Encoding encoding, string value)
        {
            socket.Send(BitConverter.GetBytes(encoding.GetByteCount(value)));
            socket.Send(encoding.GetBytes(value));
        }
    }
}