{"id":311,"date":"2013-12-11T13:00:47","date_gmt":"2013-12-11T13:00:47","guid":{"rendered":"http:\/\/writeasync.net\/?p=311"},"modified":"2013-11-26T17:20:40","modified_gmt":"2013-11-26T17:20:40","slug":"memorychannel-and-concurrency","status":"publish","type":"post","link":"http:\/\/writeasync.net\/?p=311","title":{"rendered":"MemoryChannel and concurrency"},"content":{"rendered":"

In a previous post<\/a>, I described MemoryChannel<\/code> and how I used TDD to implement it. After the 25th commit<\/a>, 20 unit tests later, I had a fully functional but single-threaded<\/em> implementation.<\/p>\n

Simply by inspection, I knew that this code would fail almost immediately if a sender and receiver were ever executed on concurrent threads. For example, Send<\/code> and ReceiveAsync<\/code> can potentially modify the excessBuffers<\/code> LinkedList<\/a> concurrently which may corrupt its internal state.<\/p>\n

I figured that a thread-safe MemoryChannel<\/code> would be generally more useful, if a bit more complex. Thus I began devising an integration test application, knowing that my unit tests could not be of much help here.<\/p>\n

I started by writing a simple skeleton test app with a Receiver<\/code> class (async receive loop) and a Sender<\/code> class (background thread send loop). The test app started out like this:<\/p>\n

\r\nLogger logger = new Logger();\r\nMemoryChannel channel = new MemoryChannel();\r\n\r\nReceiver receiver = new Receiver(channel, logger, 16);\r\nSender sender = new Sender(channel, logger, 16, 1);\r\n\r\nTask receiverTask = receiver.RunAsync();\r\nTask senderTask = sender.RunAsync();\r\n\r\nTask.WaitAll(receiverTask, senderTask);\r\n\r\nchannel.Dispose();\r\n\r\nlogger.WriteLine("Done.");\r\n<\/pre>\n
At this point<\/a>, I had only implemented the receive loop — the Sender<\/code> was just a single Send operation running on a background thread. But this was enough to consistently<\/em> trigger an invalid state:<\/p>\n
System.InvalidOperationException: A receive operation is already in progress.
\nat CommSample.MemoryChannel.ReceiveAsync(Byte[] buffer) in CommSample\\source\\CommSample.Core\\MemoryChannel.cs:line 39
\nat CommSample.Receiver.<RunAsync>d__0.MoveNext() in CommSample\\source\\CommSample.App\\Receiver.cs:line 34
\n--- End of inner exception stack trace ---
\nat System.Threading.Tasks.Task.WaitAll(Task[] tasks, Int32 millisecondsTimeout, CancellationToken cancellationToken)
\nat System.Threading.Tasks.Task.WaitAll(Task[] tasks, Int32 millisecondsTimeout)
\nat System.Threading.Tasks.Task.WaitAll(Task[] tasks)
\nat CommSample.Program.Main(String[] args) in CommSample\\source\\CommSample.App\\Program.cs:line 24<\/code><\/p>\n
Using the debugger and setting a breakpoint on the exception condition in MemoryChannel.ReceiveAsync<\/code>, we can get a more useful stack trace that illustrates the problem:<\/p>\n