{"id":125,"date":"2016-12-20T12:45:58","date_gmt":"2016-12-20T04:45:58","guid":{"rendered":"https:\/\/blogs.msdn.microsoft.com\/seteplia\/?p=125"},"modified":"2019-06-11T22:44:34","modified_gmt":"2019-06-12T05:44:34","slug":"dissecting-the-actionblock-a-short-story-about-a-nasty-deadlock","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/premier-developer\/dissecting-the-actionblock-a-short-story-about-a-nasty-deadlock\/","title":{"rendered":"Dissecting the ActionBlock: a Short Story About a Nasty Deadlock"},"content":{"rendered":"

I think almost every project in the real world uses some form of producer-consumer queue<\/a>. The idea behind this problem is very simple. Application needs to decouple consumers of some data from the logic that processes it. Consider, for instance, the thread pool from the CLR: application can schedule some work using ThreadPool.QueueUserWorkItem<\/b> and the thread pool will do its best to maximize application throughput by using optimal number of threads that will process the input data.<\/p>\n

But using the thread pool is not always possible and\/or appropriate. Even though you can control minimum and maximum number of threads in the thread pool, this configuration will affect the entire application but not some specific parts.<\/p>\n

There are numerous possible solutions to producer-consumer problem. You can implement entirely custom solution with the application logic intermixed with low level threading aspects. It can be something semi-custom, like a list of tasks that deals with shared BlockingCollection<\/a>. Or it could be a simple solution that is based on an existing component, like the ActionBlock<T><\/a> from TPL Dataflow.<\/p>\n

Today we\u2019re going to explore internals of the ActionBlock<\/b>, discuss design decisions that were made by their authors and learn why you need to know them in order to avoid some weird behavior. Ready? Let\u2019s go.<\/p>\n

In my current project we have several cases where we need to solve producer-consumer problem. Here is one case: we have a parser and custom interpreter for a TypeScript-like language. Without digging too deep into the details, you can assume that we just need to parse a set of files and build what is called a \u2018transitive closure\u2019 of all the dependencies .<\/p>\n

Roughly the logic is following:<\/p>\n

    \n
  1. Parse the file<\/li>\n
  2. Analyze the file to understand what the dependencies are.<\/li>\n
  3. Resolve the dependencies (i.e. resolve which TypeScript files are required by this one by analysig \u2018import * from\u2019, \u2018require\u2019 etc)<\/li>\n
  4. Schedule all the dependencies for parsing.<\/li>\n<\/ol>\n

    Pretty simple, right? Actually it is and here you can see how the logic could be implemented using TPL DataFlow and ActionBlock<T><\/b>:<\/p>\n

    private static Task<ParsedFile> ParseFileAsync(string path)\r\n{\r\n    Console.WriteLine($\"Parsing '{path}'. {{0}}\",\r\n        $\"Thread Id - {Thread.CurrentThread.ManagedThreadId}\");\r\n    Thread.Sleep(10);\r\n\r\n    return Task.FromResult(\r\n        new ParsedFile()\r\n        {\r\n            FileName = path,\r\n            Dependencies = GetFileDependencies(path),\r\n        });\r\n}\r\n\r\nstatic void Main(string[] args)\r\n{\r\n    long numberOfProcessedFiles = 0;\r\n    ActionBlock<string> actionBlock = null;\r\n    Func<string, Task> processFile = async path =>\r\n    {\r\n        Interlocked.Increment(ref numberOfProcessedFiles);\r\n        ParsedFile parsedFile = await ParseFileAsync(path);\r\n\r\n        foreach (var dependency in parsedFile.Dependencies)\r\n        {\r\n            Console.WriteLine($\"Sending '{dependency}' to the queue... {{0}}\",\r\n                $\"Thread Id - {Thread.CurrentThread.ManagedThreadId}\");\r\n            await actionBlock.SendAsync(dependency);\r\n        }\r\n\r\n        if (actionBlock.InputCount == 0)\r\n        {\r\n            \/\/ This is a marker that this is a last file and there \r\n            \/\/ is nothing to process\r\n            actionBlock.Complete();\r\n        }\r\n    };\r\n\r\n    actionBlock = new ActionBlock<string>(processFile);\r\n\r\n    actionBlock.SendAsync(\"FooBar.ts\").GetAwaiter().GetResult();\r\n    Console.WriteLine(\"Waiting for an action block to finish...\");\r\n\r\n    actionBlock.Completion.GetAwaiter().GetResult();\r\n    Console.WriteLine($\"Done. Processed {numberOfProcessedFiles}\");\r\n    Console.ReadLine();<\/pre>\n
    Let\u2019s discuss what is going on here. For the sake of simplicity all the logic resides in the Main<\/b> method. numberOfProcessedFiles<\/b> is used to check that the logic is correct and we\u2019re not missing files because of race conditions. The main part of it resides in the processFile<\/b> action that will be provided to an ActionBlock<\/b> during construction. This delegate acts like \u2018consumer\u2019 and \u2018producer\u2019 at the same time: it gets a path<\/b> from the queue, parses the file, schedules new items by calling actionBlock.SendAsync<\/b> and then, if the input of the queue is empty it signals the queue that all the files were processed by calling actionBlock.Complete()<\/b> (*). Then the Main<\/b> method creates an ActionBlock,<\/b> initiate the process by sending the first file and then waits for the completion.<\/p>\n
    ParseFileAsync<\/b> fakes a parsing process and computes the dependencies by using following logic: file \u2018foo.ts\u2019 depends on \u2018fo.ts\u2019, that depends on \u2018f.ts\u2019 etc. Basically, each file depends on the same file with a shorter name. This is a very na\u00efve logic but it helps to show the process.<\/p>\n
    ActionBlock<\/b> provides an easy to use API and handles concurrency for you. The only caveat here is that the default \u2018degree of parallelism\u2019 is 1, and you should override the default by using ExecutionDataflowBlockOptions<\/a> <\/b>during construction of an ActionBlock<\/b>. If the MaxDegreeOfParallelism<\/a><\/b> property is greater than 1 then the ActionBlock<\/b> will call a callback function from different threads (actually, from different Tasks) to process multiple incoming items at the same time.<\/p>\n
    Post vs. SendAsync or what to use when<\/h3>\n
    Everyone who tried to solve producer-consumer problem faced a dilemma: what to do if an incoming flow exceeds an ability to process it? How to \u2018throttle\u2019 back? Just keep every possible object in memory and grow the queue indefinitely? Throw an exception? Return \u2018false\u2019? Block an \u2018Enqueue\u2019 method while the queue is full?<\/p>\n
    To solve this problem, ActionBlock<\/b> authors decided to use the following well-known pattern:<\/p>\n
      \n
    1. The client of an action block may provide a queue size (in the constructor).<\/li>\n
    2. When a queue is full the Post<\/a><\/b> method returns false<\/b> and SendAsync<\/a><\/b> method \u201cblocks\u201d until the queue will get a free spot (**).<\/li>\n<\/ol>\n
      In our first example, the code doesn\u2019t specify any limits to the queue. This means that if the producer will push really hard the application can end up with OutOfMemoryException<\/b>. Let\u2019s change it. And for a sake of this example let\u2019s set the queue size to some ridiculously low number, like \u2026 1.<\/p>\n
      actionBlock <\/span>=<\/span>\u00a0<\/span>new<\/span>\u00a0<\/span>ActionBlock<\/span><<\/span>string<\/span>>(processFile,\r\n\u00a0\u00a0\u00a0 <\/span>new<\/span>\u00a0<\/span>ExecutionDataflowBlockOptions<\/span>() {BoundedCapacity <\/span>=<\/span>\u00a0<\/span>1<\/span>});<\/span><\/pre>\n
      Now, if when we\u2019ll run this code, we\u2019ll get \u2026 a deadlock!<\/p>\n
      Deadlock<\/h3>\n
      Let\u2019s think about producer-consumer queue from the design point of view. You\u2019re building your own custom implementation of the queue that takes a callback for processing elements. You need to decide should the queue be bounded or not. If the queue is bounded, you will end up with a similar set of methods that ActionBlock<\/b> has: synchronous API for adding elements, that returns false<\/b> when the queue is full and asynchronous version that returns a task. If the queue is full a client of the queue can decide what to do: to handle \u2018overflow\u2019 manually by using synchronous version or \u2018await\u2019 on the task with asynchronous version.<\/p>\n
      Then you need to decide when a given callback should be called. You may end up with a following logic: check the queue size, if the queue is not empty pick the element from it, call the callback, wait for it to finish and then remove the item from the queue. (Real implementation will be way more complicated because it should consider all possible race conditions that could happen here.) You may remove the item from the queue before calling the callback but as we\u2019ll see in a moment it won\u2019t change the possibility of getting deadlock.<\/p>\n
      This design is clean and simple but it can cause a problem very easily. Suppose the queue is full and the queue calling back a function to process the element. And instead of quickly processing the element, this callback tries to schedule another item by awaiting SendAsync<\/b>:<\/p>\n
      \"image\"<\/a><\/p>\n
      The queue is full and the queue can\u2019t accept new element because the callback is in progress. But the callback itself just got stuck on awaiting for SendAsync<\/b> to finish. Classical deadlock!<\/p>\n
      Ok. We\u2019re getting a deadlock because an ActionBlock<\/b> removes elements only *after* the callback is called. Let\u2019s consider an alternative: what if ActionBlock<\/b> will remove an item *before* calling the callback?<\/p>\n
      Actually, deadlock would still be possible. Let\u2019s consider ActionBlock<\/b> with a bound size of 1 and degree of parallelism of 2.<\/p>\n
      \"image\"<\/a><\/p>\n