{"id":2001,"date":"2014-02-14T13:00:10","date_gmt":"2014-02-14T13:00:10","guid":{"rendered":"http:\/\/writeasync.net\/?p=2001"},"modified":"2014-02-14T06:57:22","modified_gmt":"2014-02-14T06:57:22","slug":"native-inputqueue-thread-safety","status":"publish","type":"post","link":"http:\/\/writeasync.net\/?p=2001","title":{"rendered":"Native InputQueue thread-safety"},"content":{"rendered":"

In the previous post<\/a>, I introduced my port of InputQueue to C++<\/a>. As usual, the unit tests drove the creation of a correct single-threaded implementation but slightly more needed to be done to make the code thread-safe.<\/p>\n

The go-to construct for providing a cross-thread mutual exclusion mechanism in Windows is the critical section<\/a>. Perhaps C programmers<\/a> would be satisfied with InitializeCriticalSection<\/code><\/a>, EnterCriticalSection<\/code><\/a>, and LeaveCriticalSection<\/code><\/a> but this is the modern era<\/a>! For exception-safe<\/a>, less error-prone code, we should be using RAII techniques<\/a>. Conveniently, the PPL provides just the thing — a wrapper class critical_section<\/code><\/a> and its associated critical_section::scoped_lock<\/code><\/a>. This will do nicely as a C++ analog to System.Threading.Monitor<\/code><\/a> and the lock<\/code> statement<\/a>. Here’s an example of it in action, in the InputQueue::Enqueue<\/code> method:<\/p>\n

\r\nvoid Enqueue(T item)\r\n{\r\n    std::unique_ptr<concurrency::task_completion_event<T>> current;\r\n    {\r\n        concurrency::critical_section::scoped_lock lock(syncRoot_);\r\n        if (!pending_)\r\n        {\r\n            items_.push(item);\r\n        }\r\n        else\r\n        {\r\n            current = std::move(pending_);\r\n        }\r\n    }\r\n\r\n    if (current)\r\n    {\r\n        current->set(item);\r\n    }\r\n}\r\n<\/pre>\n
Note the use of scoping braces<\/a> to ensure that the lock is held only as long as necessary and then released (upon destruction) before attempting to set the task completion event.<\/p>\n
To gain confidence that the implementation was truly thread-safe, I ported the integration test app from the managed project to native code. The app is supposed to continuously, concurrently, and asynchronously loop, enqueuing and dequeuing items from the queue. Since we don’t have async<\/code>\/await<\/code> in C++, achieving this is relatively complicated. Thanks to a bit of help from the PPL team, however, the job is made much easier. Using create_iterative_task<\/code><\/a>, writing an asynchronous loop is no longer a chore. Compare the two implementations of the enqueue loop, first in C# and then in C++, and note the similarities:<\/p>\n
\r\nprivate static async Task EnqueueLoopAsync(IProducerQueue<int> queue, CancellationToken token)\r\n{\r\n    await Task.Yield();\r\n\r\n    int i = 0;\r\n    while (!token.IsCancellationRequested)\r\n    {\r\n        ++i;\r\n        queue.Enqueue(i);\r\n        await Task.Delay(1);\r\n    }\r\n}\r\n<\/pre>\n
\r\ntemplate<typename TProducerQueue>\r\nconcurrency::task<void> EnqueueLoopAsync(TProducerQueue & queue, concurrency::cancellation_token token)\r\n{\r\n    return concurrency::extras::create_delayed_task(std::chrono::milliseconds(1), [&queue, token]\r\n    {\r\n        shared_ptr<int> i = std::make_shared<int>();\r\n        return concurrency::extras::create_iterative_task([&queue, token, i]()\r\n        {\r\n            if (token.is_canceled())\r\n            {\r\n                return concurrency::task_from_result(false);\r\n            }\r\n\r\n            int next = ++*i;\r\n            queue.Enqueue(next);\r\n            return concurrency::extras::create_delayed_task(std::chrono::milliseconds(1), []() { return true; });\r\n        });\r\n    });\r\n}\r\n<\/pre>\n
With C++, you have to be more explicit about how the memory is managed, using shared_ptr<\/code><\/a> and passing the queue by reference<\/a>. But structurally the code is similar. To “break” out of the loop, note that we need to return a task<\/code> with a false<\/code> result. In this app, create_delayed_task<\/code> does double duty as an initial “yield” mechanism and between every loop iteration as an async rate-limiting delay.<\/p>\n
Overall, the PPL is a great tool to have if you want to write correct and efficient asynchronous native code.<\/p>\n","protected":false},"excerpt":{"rendered":"
In the previous post, I introduced my port of InputQueue to C++. As usual, the unit tests drove the creation of a correct single-threaded implementation but slightly more needed to be done to make the code thread-safe. The go-to construct… <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[21,61,101],"tags":[],"class_list":["post-2001","post","type-post","status-publish","format-standard","hentry","category-async","category-concurrency","category-native"],"_links":{"self":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/2001","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=2001"}],"version-history":[{"count":0,"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/2001\/revisions"}],"wp:attachment":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=2001"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=2001"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=2001"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}