{"id":1511,"date":"2014-01-29T13:00:43","date_gmt":"2014-01-29T13:00:43","guid":{"rendered":"http:\/\/writeasync.net\/?p=1511"},"modified":"2014-01-27T10:15:37","modified_gmt":"2014-01-27T10:15:37","slug":"non-blockingcollection","status":"publish","type":"post","link":"http:\/\/writeasync.net\/?p=1511","title":{"rendered":"InputQueue, the non-BlockingCollection"},"content":{"rendered":"

The .NET 4.0+ solution to the producer-consumer problem<\/a> is BlockingCollection<\/code><\/a>. A sample app with a single producer and consumer using an ordered queue of integers would look something like this:<\/p>\n

\r\nprivate static async Task EnqueueLoopAsync(BlockingCollection<int> queue, CancellationToken token)\r\n{\r\n    int i = 0;\r\n    while (!token.IsCancellationRequested)\r\n    {\r\n        ++i;\r\n        queue.Add(i);\r\n        await Task.Delay(1);\r\n    }\r\n}\r\n\r\nprivate static void DequeueLoop(BlockingCollection<int> queue, CancellationToken token)\r\n{\r\n    int previous = 0;\r\n    try\r\n    {\r\n        while (!token.IsCancellationRequested)\r\n        {\r\n            int current = queue.Take(token);\r\n            if (current - previous != 1)\r\n            {\r\n                throw GetOutOfOrderError(current, previous);\r\n            }\r\n\r\n            previous = current;\r\n        }\r\n    }\r\n    catch (OperationCanceledException)\r\n    {\r\n    }\r\n}\r\n<\/pre>\n
This is simple enough, but the problem is that DequeueLoop<\/code>. It spends a lot of time (relatively speaking) blocking the thread due to the synchronous Take<\/code><\/a> method. About the best you can do, async-wise, is use TryTake<\/code><\/a> with a small timeout and yield a bit on empty, for example:<\/p>\n
\r\nprivate static async Task DequeueLoopAsync(BlockingCollection<int> queue, CancellationToken token)\r\n{\r\n    int previous = 0;\r\n    try\r\n    {\r\n        while (!token.IsCancellationRequested)\r\n        {\r\n            int current;\r\n            if (queue.TryTake(out current, 1))\r\n            {\r\n                if (current - previous != 1)\r\n                {\r\n                    throw GetOutOfOrderError(current, previous);\r\n                }\r\n\r\n                previous = current;\r\n            }\r\n            else\r\n            {\r\n                await Task.Delay(1);\r\n            }\r\n        }\r\n    }\r\n    catch (OperationCanceledException)\r\n    {\r\n    }\r\n}\r\n<\/pre>\n
Certainly not ideal, but what did we expect from a Blocking<\/em>Collection<\/code> anyway?<\/p>\n
For true async support, we’ll need to come up with a different data structure. And this is exactly what I did in my new QueueSample project<\/a>.<\/p>\n
The main class of interest is InputQueue<\/code><\/a>. (If you’ve seen my earlier blog posts, you might note some similarities to MemoryChannel<\/code><\/a>.) Its interface is as follows:<\/p>\n
\r\npublic class InputQueue<T> : IInputQueue<T>\r\n{\r\n    public InputQueue()\r\n    public Task<T> DequeueAsync();\r\n    public void Enqueue(T item);\r\n    public void Dispose();\r\n}\r\n<\/pre>\n
Some implementation notes:<\/p>\n