{"id":5662,"date":"2019-12-27T15:00:53","date_gmt":"2019-12-27T15:00:53","guid":{"rendered":"http:\/\/writeasync.net\/?p=5662"},"modified":"2019-12-26T16:46:36","modified_gmt":"2019-12-26T16:46:36","slug":"high-performance-datagramming-channels","status":"publish","type":"post","link":"http:\/\/writeasync.net\/?p=5662","title":{"rendered":"High performance datagramming: channels?"},"content":{"rendered":"

You may have heard of System.IO.Pipelines<\/a>, a set of classes to aid in writing high performance streaming server code. But what about datagram<\/a> scenarios, e.g. using UDP<\/a>? That would seem like a job for System.Threading.Channels<\/a>. Let’s try it and do some high performance “datagramming”!<\/p>\n

Our scenario today will be a simple datagram client and server, transmitting small string messages of the form “Message nnn<\/em>“. We will use the netcoreapp3.1 target framework<\/a> to take advantage of the improved stability and performance of .NET Core 3.x<\/a>.<\/p>\n

Let’s start with the entry point (which uses the new-ish async main<\/a> and using declaration<\/a> features):<\/p>\n

\r\nusing System;\r\nusing System.Collections.Generic;\r\nusing System.Diagnostics;\r\nusing System.Threading;\r\nusing System.Threading.Tasks;\r\n\r\ninternal sealed class Program\r\n{\r\n    private static async Task Main(string[] args)\r\n    {\r\n        using CancellationTokenSource cts = new CancellationTokenSource();\r\n        List<Task> tasks = new List<Task>();\r\n        const ushort Port = 9999;\r\n        MessageCount count = new MessageCount();\r\n        string arg = (args.Length == 1) ? args[0].ToUpperInvariant() : string.Empty;\r\n        switch (arg)\r\n        {\r\n            default:\r\n                Console.WriteLine("Please specify 'client' or 'server'.");\r\n                return;\r\n            case "CLIENT":\r\n                tasks.AddRange(DatagramClient.Start(count, Port, cts.Token));\r\n                break;\r\n            case "SERVER":\r\n                tasks.AddRange(DatagramServer.Start(count, Port, cts.Token));\r\n                break;\r\n        }\r\n\r\n        tasks.Add(TaskEx.RunAsync(nameof(LogAsync), count, cts.Token, (c, t) => LogAsync(c, t)));\r\n        tasks.Add(Task.Run(() => Prompt()));\r\n\r\n        await Task.WhenAny(tasks);\r\n\r\n        Console.WriteLine("Canceling...");\r\n        cts.Cancel();\r\n\r\n        await Task.WhenAll(tasks);\r\n    }\r\n\r\n    private static void Prompt()\r\n    {\r\n        Console.WriteLine("Press ENTER to quit.");\r\n        Console.ReadLine();\r\n    }\r\n\r\n    private static async Task LogAsync(MessageCount count, CancellationToken token)\r\n    {\r\n        Stopwatch stopwatch = Stopwatch.StartNew();\r\n        while (!token.IsCancellationRequested)\r\n        {\r\n            await Task.Delay(1000, token);\r\n            Console.WriteLine($"{stopwatch.ElapsedMilliseconds},{count.Value}");\r\n        }\r\n    }\r\n}\r\n<\/pre>\n
We’re setting up the basic framework here, defining the client and server as a sequence of background tasks. We have the ability for the user to press ENTER to cancel everything as well as a periodic logging task to track the message count. The supporting class TaskEx<\/code> helps standardize the error handling and cancellation patterns:<\/p>\n
\r\nusing System;\r\nusing System.Threading;\r\nusing System.Threading.Tasks;\r\n\r\ninternal static class TaskEx\r\n{\r\n    public static async Task RunAsync<T>(string name, T input, CancellationToken token, Func<T, CancellationToken, Task> doAsync)\r\n    {\r\n        Console.WriteLine($"{name} starting...");\r\n        try\r\n        {\r\n            await doAsync(input, token);\r\n        }\r\n        catch (OperationCanceledException)\r\n        {\r\n            Console.WriteLine($"{name} canceled.");\r\n        }\r\n        catch (Exception e)\r\n        {\r\n            Console.WriteLine($"{name} failed: {e}");\r\n        }\r\n    }\r\n}\r\n<\/pre>\n
And of course MessageCount<\/code> is just a Java-style reference type wrapper<\/a>:<\/p>\n
\r\ninternal sealed class MessageCount\r\n{\r\n    public long Value { get; private set; }\r\n\r\n    public void Increment() => this.Value++;\r\n}\r\n<\/pre>\n
The client just sends UDP messages in a tight loop, using Task.Yield to prevent blocking the thread forever (also note the gratuitous use of the value tuple<\/a> in the SendAsync<\/code> method):<\/p>\n
\r\nusing System.Collections.Generic;\r\nusing System.Net;\r\nusing System.Net.Sockets;\r\nusing System.Text;\r\nusing System.Threading;\r\nusing System.Threading.Tasks;\r\n\r\ninternal static class DatagramClient\r\n{\r\n    public static IEnumerable<Task> Start(MessageCount count, ushort port, CancellationToken token)\r\n    {\r\n        yield return TaskEx.RunAsync(nameof(SendAsync), (count, port), token, (c, t) => SendAsync(c, t));\r\n    }\r\n\r\n    private static async Task SendAsync((MessageCount, ushort) x, CancellationToken token)\r\n    {\r\n        await Task.Yield();\r\n\r\n        MessageCount count = x.Item1;\r\n        ushort port = x.Item2;\r\n        using UdpClient client = new UdpClient();\r\n        client.Connect(IPAddress.Loopback, port);\r\n        while (!token.IsCancellationRequested)\r\n        {\r\n            count.Increment();\r\n            byte[] dgram = Encoding.UTF8.GetBytes("Message " + count.Value);\r\n            client.Send(dgram, dgram.Length);\r\n        }\r\n    }\r\n}\r\n<\/pre>\n
The boilerplate is done. We can move on to the main event — a datagram server which, as promised, uses a Channel<string><\/a> as a producer\/consumer queue:<\/p>\n
\r\nusing System;\r\nusing System.Collections.Generic;\r\nusing System.Net;\r\nusing System.Net.Sockets;\r\nusing System.Text;\r\nusing System.Threading;\r\nusing System.Threading.Channels;\r\nusing System.Threading.Tasks;\r\n\r\ninternal static class DatagramServer\r\n{\r\n    public static IEnumerable<Task> Start(MessageCount count, ushort port, CancellationToken token)\r\n    {\r\n        Channel<string> channel = Channel.CreateBounded<string>(64);\r\n        yield return TaskEx.RunAsync(nameof(ConsumeAsync), (count, channel.Reader), token, (c, t) => ConsumeAsync(c, t));\r\n        yield return TaskEx.RunAsync(nameof(ProduceAsync), (port, channel.Writer), token, (x, t) => ProduceAsync(x, t));\r\n    }\r\n\r\n    private static async Task ProduceAsync((ushort, ChannelWriter<string>) x, CancellationToken token)\r\n    {\r\n        ushort port = x.Item1;\r\n        ChannelWriter<string> writer = x.Item2;\r\n        using Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);\r\n        socket.Bind(new IPEndPoint(IPAddress.Loopback, port));\r\n\r\n        while (!token.IsCancellationRequested)\r\n        {\r\n            byte[] buffer = new byte[64];\r\n            int size = await socket.ReceiveAsync(new Memory<byte>(buffer), SocketFlags.None, token);\r\n            string message = Encoding.UTF8.GetString(buffer, 0, size);\r\n            while (!writer.TryWrite(message))\r\n            {\r\n                await writer.WaitToWriteAsync(token);\r\n            }\r\n        }\r\n    }\r\n\r\n    private static async Task ConsumeAsync((MessageCount, ChannelReader<string>) x, CancellationToken token)\r\n    {\r\n        MessageCount count = x.Item1;\r\n        ChannelReader<string> reader = x.Item2;\r\n        while (!token.IsCancellationRequested)\r\n        {\r\n            string message;\r\n            while (!reader.TryRead(out message))\r\n            {\r\n                await reader.WaitToReadAsync(token);\r\n            }\r\n\r\n            count.Increment();\r\n            int checksum = Fletcher32(message);\r\n            if (checksum == 0x12345678)\r\n            {\r\n                Console.WriteLine("!");\r\n            }\r\n        }\r\n    }\r\n\r\n    private static int Fletcher32(string data)\r\n    {\r\n        int sum1 = 0;\r\n        int sum2 = 0;\r\n        foreach (char c in data)\r\n        {\r\n            sum1 = (sum1 + c) % 0xFFFF;\r\n            sum2 = (sum2 + sum1) % 0xFFFF;\r\n        }\r\n\r\n        return (sum2 << 16) | sum1;\r\n    }\r\n}\r\n<\/pre>\n
Note that we start the consumer before the producer to potentially avoid long chains of synchronous continuations on the socket receive side; an initial Task.Yield would have also been fine. Although there are a few dozen lines here, there is nothing too exotic. The producer side reads from a socket (using the new ValueTask<\/a>-based ReceiveAsync overload<\/a>) and writes the parsed string to the channel. The consumer reads the string from the channel and, in an effort to do “real work” with the result, calculates Fletcher’s checksum<\/a> for the data.<\/p>\n
This is a pretty standard channel use case so one would hope the performance is decent. Indeed (at least on my machine) I see more than acceptable throughput — nearly 137K messages per second:
\n\"Datagram<\/a><\/p>\n
However, if we stop and think about this for a minute, it is not clear we need a channel at all. This is a single socket with a single producer and a single consumer. Isn’t the socket itself already the “queue”? What we have done seems to be no more than double buffering (and not the good kind<\/a>). Could we perhaps run even faster if we ditch the channel entirely? Let’s find out:<\/p>\n
\r\n    public static IEnumerable<Task> Start(MessageCount count, ushort port, CancellationToken token)\r\n    {\r\n        \/\/ Use a single background task to receive from the socket\r\n        yield return TaskEx.RunAsync(nameof(ReceiveAsync), (count, port), token, (x, t) => ReceiveAsync(x, t));\r\n    }\r\n\r\n    \/\/ Do all the work directly in here\r\n    private static async Task ReceiveAsync((MessageCount, ushort) x, CancellationToken token)\r\n    {\r\n        MessageCount count = x.Item1;\r\n        ushort port = x.Item2;\r\n        using Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);\r\n        socket.Bind(new IPEndPoint(IPAddress.Loopback, port));\r\n\r\n        while (!token.IsCancellationRequested)\r\n        {\r\n            byte[] buffer = new byte[64];\r\n            int size = await socket.ReceiveAsync(new Memory<byte>(buffer), SocketFlags.None, token);\r\n            string message = Encoding.UTF8.GetString(buffer, 0, size);\r\n            count.Increment();\r\n            int checksum = Fletcher32(message);\r\n            if (checksum == 0x12345678)\r\n            {\r\n                Console.WriteLine("!");\r\n            }\r\n        }\r\n    }\r\n<\/pre>\n
The code is simpler. Will the lower complexity be rewarded with higher performance?
\n\"Datagram<\/a><\/p>\n
The results look positive. We’re seeing a rate around 146K\/sec now, nearly 7% faster than before — admittedly not huge<\/em> but certainly big enough to be meaningful.<\/p>\n
So what is the conclusion? Should we not use channels? To paraphrase Alan Page<\/a>, you should use channels in 100% of the cases that can benefit from their use. For example, if you needed to ensure single threaded consumption in an inherently multithreaded pipeline, channels would be your best bet.<\/p>\n
As it turned out, this particular socket problem simply had another more efficient option with a different set of tradeoffs. But we aren’t done yet — check back next time to see how we can get additional easy (and other not-so-easy) performance gains.<\/p>\n","protected":false},"excerpt":{"rendered":"
You may have heard of System.IO.Pipelines, a set of classes to aid in writing high performance streaming server code. But what about datagram scenarios, e.g. using UDP? That would seem like a job for System.Threading.Channels. Let’s try it and do… <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[21,104],"tags":[],"class_list":["post-5662","post","type-post","status-publish","format-standard","hentry","category-async","category-performance"],"_links":{"self":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/5662","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=5662"}],"version-history":[{"count":3,"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/5662\/revisions"}],"predecessor-version":[{"id":5667,"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/5662\/revisions\/5667"}],"wp:attachment":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=5662"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=5662"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=5662"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}