6
\$\begingroup\$

I need a fully asynchronous single-threaded server written using .Net Core. Searched for options and, surprisingly, did't find any. So I decided to write a simple prototype. Borrowed idea from python's selectors module.

Idea is simple. We have SocketSelector class to whom we register sockets for desired events (Read, Write, Error) and provide callbacks, which selector will call. Selector once every loop iteration (server just runs infinite loop) checks what sockets are ready by calling Socket.Select and then adds callbacks of ready sockets to readyQueue. Server every iteration dequeues all callbacks from this queue and calls them.

Each socket (listener and client sockets) are all set to be non-blocking. Say we accepted some new connection and now want to read a message from it asynchronously. We don't call socket.Receive outright because we don't know whether there are bytes to read or not. First we create TaskCompletionSource, call selector.Register(socket, SocketEvent.Read, tcs.SetResult), and then await tcs.Task. Eventually socket becomes ready to read, so selector will put tcs.SetResult into the readyQueue, and server will call it with null as an argument. This will complete awaited task. At this point we can start actually reading from socket knowing that socket.Receive will not block. Same for sending messages.

Code it straightforward. The only strange thing is ListForPolling. This class is needed to alleviate some of Socket.Select's problems. This method takes 3 lists of sockets (1 for each event) and removes (using RemoveAt) from provided lists all sockets that are not yet ready. That means that on every iteration we have to populate lists again. So instead of actually removing sockets I just mark them as removed. Socket.Select implementation uses only Count, Clear, indexer and RemoveAt, so all other IList methods are not required. Unix implementation internally uses poll instead of select and it iterates over lists from back to front, so code is different a little bit. ListForPolling works with Socket.Select and Socket.Select only, so don't use it for anything else, it will break :)

First, I don't really want to reinvent the wheel if some robust solution already exists. So please provide some links if you have any. I don't need servers that use thread pool behind the scenes, I want a single-threaded asynchronous solution. And I don't want to learn NodeJS :)

Second, is this code viable? I tested it with simple client. It works well. Tested with 150k concurrent clients (not really concurrent of course). I didn't test it on Linux, but it should work even better. Poll is much better then Select after all.

There are problems with current solution of course, after all it is just a prototype, but concept works, I think.

#define WINDOWS

using System;
using System.Net;
using System.Net.Sockets;
using System.Collections;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Text;

class Program {

    private static void Main() {
        var server = new Server(7894);
        server.Run();
    }
}

public class Server {

    private ushort port;
    private Socket listener;
    private SocketSelector selector;
    private int numClients = 0;

    private Queue<Action<object>> readyQueue = new Queue<Action<object>>(50);

    private Dictionary<string, string> requestToResponse =
        new Dictionary<string, string>() {
            ["hello"] = "bye",
            ["how are you?"] = "I am fine",
            ["what's up?"] = "nothing",
            ["what do you think about god?"] = "I don't think about him",
            ["I love GoT's final season"] = "maniac",
            ["Answer is always lasers and/or duck tape"] = "true wisdom"
    };

    public Server(ushort port) {
        this.port = port;

        listener = new Socket(AddressFamily.InterNetworkV6, SocketType.Stream, ProtocolType.Tcp);
        listener.Blocking = false;
        listener.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
        listener.Bind(new IPEndPoint(IPAddress.IPv6Any, port));
        listener.Listen(2000);

        selector = new SocketSelector(readyQueue);
        selector.Register(listener, SocketEvent.Read, _ => acceptConnection());
    }

    private void acceptConnection() {
        var clientSocket = listener.Accept();
        clientSocket.Blocking = false;

        ++numClients;

        serveClient(clientSocket);
    }

    private async void serveClient(Socket socket) {
        Console.WriteLine($"Started serving client { socket.RemoteEndPoint }.");

        byte[] buffer = new byte[1024];
        try {
            int messageLength = await readMessage(socket, buffer);
            var request = Encoding.UTF8.GetString(buffer, 0, messageLength);
            string response = requestToResponse[request];
            messageLength = Encoding.UTF8.GetBytes(response, 0, response.Length, buffer, 0);
            await sendMessage(socket, buffer, messageLength);
        } catch (Exception e) {
            Console.WriteLine(e);
            selector.Unregister(socket);
        }

        Console.WriteLine($"Finished serving client { socket.RemoteEndPoint }.");

        selector.Unregister(socket);
        socket.Shutdown(SocketShutdown.Both);
        socket.Close();

        --numClients;
    }

    private async Task read(Socket socket, byte[] buffer, int bytesToRead) {
        int n = 0, offset = 0;
        while (bytesToRead > 0) {
            var tcs = new TaskCompletionSource<object>();
            if (offset == 0) {
                selector.Register(socket, SocketEvent.Read, tcs.SetResult);
            } else {
                selector.SwitchCallbackForRegisteredEvent(socket, tcs.SetResult);
            }

            await tcs.Task;

            try {
                n = socket.Receive(buffer, offset, bytesToRead, 0);
            } catch (SocketException e) {
                // ...
            }

            if (n == 0 && bytesToRead > 0) {
                throw new Exception("Client disconnected before sending the whole message.");
            }

            bytesToRead -= n;
            offset += n;
        }

        selector.Unregister(socket, SocketEvent.Read);
    }

    private async Task<int> readMessage(Socket socket, byte[] buffer) {
        await read(socket, buffer, 2);

        var messageLength = Utils.UnpackUshort(buffer, 0);
        if (messageLength > buffer.Length) {
            throw new Exception("Incoming message is too long.");
        }

        await read(socket, buffer, messageLength);

        return messageLength;
    }

    private async Task sendAll(Socket socket, byte[] buffer, int bytesTotal) {
        int bytesSent = 0;
        while (bytesSent < bytesTotal) {
            var tcs = new TaskCompletionSource<object>();
            if (bytesSent == 0)
                selector.Register(socket, SocketEvent.Write, tcs.SetResult);
            else
                selector.SwitchCallbackForRegisteredEvent(socket, tcs.SetResult);

            await tcs.Task;

            try {
                bytesSent += socket.Send(buffer, bytesSent, bytesTotal - bytesSent, 0);
            } catch (SocketException e) {
                // ...
            }
        }

        selector.Unregister(socket, SocketEvent.Write);
    }

    private async Task sendMessage(Socket socket, byte[] buffer, int messageLength) {
        byte b0 = buffer[0];
        byte b1 = buffer[1];

        Utils.PackUshort(buffer, 0, (ushort) messageLength);

        await sendAll(socket, buffer, 2);

        buffer[0] = b0;
        buffer[1] = b1;

        await sendAll(socket, buffer, messageLength);
    }

    public void Run() {
        while (true) {
            selector.Poll();

            while (readyQueue.Count > 0) {
                var action = readyQueue.Dequeue();
                action(null);
            }
        }
    }
}

internal enum SocketEvent {
    Read, Write, Error
}

internal class SocketSelector {

    private ListForPolling[] lists = {
        new ListForPolling(), // read list
        new ListForPolling(), // write list
        new ListForPolling()  // error list
    };

    private Dictionary<IntPtr, (SocketEvent SocketEvent, Action<object> Callback)> handleToSocketEventAndCallback =
        new Dictionary<IntPtr, (SocketEvent SocketEvent, Action<object> Callback)>();

    private Queue<Action<object>> readyQueue;

    public SocketSelector(Queue<Action<object>> readyQueue) {
        this.readyQueue = readyQueue;
    }

    private void addCallbacksToReadyQueue(int index) {
        var lst = lists[index];
        lst.ResetRemovedCount();

        for (int i = 0; i < lst.Count; ++i) {
            if (!lst.IsRemoved(i)) {
                IntPtr handle = ((Socket) lst[i]).Handle;
                var callback = handleToSocketEventAndCallback[handle].Callback;
                if (callback != null)
                    readyQueue.Enqueue(callback);
            }
        }

        lst.ResetRemovedSet();
    }

    public void Poll() {
        if (lists[0].Count > 0 || lists[1].Count > 0 || lists[2].Count > 0) {
            Socket.Select(lists[0], lists[1], lists[2], -1);

            addCallbacksToReadyQueue(0);
            addCallbacksToReadyQueue(1);
            addCallbacksToReadyQueue(2);
        }
    }

    public void SwitchCallbackForRegisteredEvent(Socket socket, Action<object> callback) {
        IntPtr handle = socket.Handle;
        if (handleToSocketEventAndCallback.ContainsKey(handle)) {
            var socketEvent = handleToSocketEventAndCallback[handle].SocketEvent;
            handleToSocketEventAndCallback[handle] = (socketEvent, callback);
        } else {
            throw new Exception("Cannot switch callback for unregistered socket.");
        }
    }

    public void Register(Socket socket, SocketEvent socketEvent, Action<object> callback) {
        IntPtr handle = socket.Handle;
        if (!handleToSocketEventAndCallback.ContainsKey(handle)) {
            lists[(int) socketEvent].Add(socket);
            handleToSocketEventAndCallback[handle] = (socketEvent, callback);
        } else {
            throw new Exception("Socket can be registered to listen to only one event at a time.");
        }
    }

    public void Unregister(Socket socket, SocketEvent socketEvent) {
        IntPtr handle = socket.Handle;
        if (
            handleToSocketEventAndCallback.ContainsKey(handle) &&
            handleToSocketEventAndCallback[handle].SocketEvent == socketEvent
        ) {
            handleToSocketEventAndCallback.Remove(handle);
            lists[(int) socketEvent].Remove(socket);
        } else {
            throw new Exception("Cannot unregister not registered event.");
        }
    }

    public void Unregister(Socket socket) {
        IntPtr handle = socket.Handle;
        if (handleToSocketEventAndCallback.ContainsKey(handle)) {
            Unregister(socket, handleToSocketEventAndCallback[handle].SocketEvent);
        }
    }
}

internal class ListForPolling : IList {

    private List<object> list;
    private HashSet<int> removed = new HashSet<int>();
    private int removedCount = 0; // always 0 for Unix

    public int Count => list.Count - removedCount;

    public object this[int index] {
        get { return list[index + removedCount]; }
        set { }
    }

    public ListForPolling(int capacity = 100) {
        list = new List<object>(capacity);
    }

    public int Add(object item) {
        list.Add(item);
        return list.Count - 1;
    }

    public void Clear() {
        removedCount = 0;
        for (int i = 0; i < Count; ++i) {
            removed.Add(i);
        }
    }

    public void Remove(object item) => list.Remove(item);

    public void RemoveAt(int index) {
        removed.Add(index + removedCount);
#if WINDOWS
        ++removedCount;
#endif
    }

    public bool IsRemoved(int index) {
        return removed.Contains(index);
    }

    public void ResetRemovedSet() {
        removed.Clear();
    }

    public void ResetRemovedCount() {
        removedCount = 0;
    }

    public bool IsFixedSize { get { throw new NotSupportedException(); } }
    public bool IsReadOnly { get { throw new NotSupportedException(); } }
    public bool IsSynchronized { get { throw new NotSupportedException(); } }
    public object SyncRoot { get { throw new NotSupportedException(); } }

    public bool Contains(object item) { throw new NotSupportedException(); }

    public void CopyTo(Array array, int index) { throw new NotSupportedException(); }

    public IEnumerator GetEnumerator() { throw new NotSupportedException(); }

    public int IndexOf(object item) { throw new NotSupportedException(); }

    public void Insert(int index, object item) { throw new NotSupportedException(); }
}

internal static class Utils {

    public static ushort UnpackUshort(byte[] buffer, int offset) {
        if (BitConverter.IsLittleEndian) {
            byte temp = buffer[offset];
            buffer[offset] = buffer[offset + 1];
            buffer[offset + 1] = temp;
        }

        return BitConverter.ToUInt16(buffer, offset);
    }

    public static void PackUshort(byte[] buffer, int offset, ushort num) {
        unsafe {
            byte* p = (byte*) &num;
            buffer[offset] = *p++;
            buffer[offset + 1] = *p;
        }

        if (BitConverter.IsLittleEndian) {
            byte temp = buffer[offset];
            buffer[offset] = buffer[offset + 1];
            buffer[offset + 1] = temp;
        }
    }
}

Here is simple client.

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

class Client {

    private static int serverPort = 7894;

    private static string[] requests = {
        "hello",
        "how are you?",
        "what's up?",
        "what do you think about god?",
        "I love GoT's final season",
        "Answer is always lasers and/or duck tape"
    };

    private static void packUshort(byte[] buffer, int offset, ushort num) {
        unsafe {
            byte* p = (byte*) &num;
            buffer[offset] = *p++;
            buffer[offset + 1] = *p;
        }

        if (BitConverter.IsLittleEndian) {
            byte temp = buffer[offset];
            buffer[offset] = buffer[offset + 1];
            buffer[offset + 1] = temp;
        }
    }

    private static int readUntilClosedOrBufferFilled(Socket socket, byte[] buffer, int offset = 0) {
        int n = -1, initialOffset = offset;
        int availableRoom = buffer.Length - offset;
        while (availableRoom > 0 && n != 0) {
            n = socket.Receive(buffer, offset, availableRoom, 0);
            availableRoom -= n;
            offset += n;
        }

        return buffer.Length - availableRoom - initialOffset;
    }

    private static void sendMessage(Socket socket, string msg, byte[] buffer) {
        int msgLength = Encoding.UTF8.GetByteCount(msg);

        packUshort(buffer, 0, (ushort) msgLength);

        socket.Send(buffer, 0, 2, 0);

        int n = Encoding.UTF8.GetBytes(msg, 0, msg.Length, buffer, 0);
        socket.Send(buffer, 0, n, 0);
    }

    private static async Task startClient(int cc) {
        var hostEntry = Dns.GetHostEntry("localhost");
        EndPoint remoteEndPoint = new IPEndPoint(hostEntry.AddressList[0], serverPort);


        var tasks = new Task[cc];
        for (int i = 0; i < cc; ++i) {
            int j = i;
            var task = Task.Run(() => {
                var socket = new Socket(remoteEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
                socket.Connect(remoteEndPoint);

                byte[] buffer = new byte[1024];
                sendMessage(socket, requests[j % 6], buffer);

                int n = readUntilClosedOrBufferFilled(socket, buffer);
                Console.WriteLine($"Response { j }: { Encoding.UTF8.GetString(buffer, 2, n - 2) }");

                socket.Close();
            });

            tasks[i] = task;
        }

        await Task.WhenAll(tasks);
    }

    private static async Task Main(string[] args) {
        await startClient(int.Parse(args[0]));
    }
}

Just provide client count as command line argument.

And by the way, communication is performed with very simple protocol: 2 bytes message length and then payload.

Test it please and tell me your thoughts. Don't forget to remove define if you are not on Windows.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.