{"id":8692,"date":"2023-12-18T11:09:57","date_gmt":"2023-12-18T11:09:57","guid":{"rendered":"https:\/\/www.modernescpp.com\/?p=8692"},"modified":"2023-12-18T11:09:57","modified_gmt":"2023-12-18T11:09:57","slug":"a-coroutines-based-single-consumer-single-producer-workflow-by-ljubic-damir","status":"publish","type":"post","link":"https:\/\/www.modernescpp.com\/index.php\/a-coroutines-based-single-consumer-single-producer-workflow-by-ljubic-damir\/","title":{"rendered":"A Coroutines-Based Single Consumer – Single Producer Workflow by Ljubic Damir"},"content":{"rendered":"\n

I’m happy to present in this guest post by Ljubic Damir a typical use-case for coroutines: a producer – consumer workflow.<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

Although this producer-consumer workflow is challenging, it is a nice starting point for your coroutine experiments.<\/p>\n\n\n\n

Intro<\/h2>\n\n\n\n

Coroutines provide a more intuitive and structured way of writing asynchronous code by allowing you to write asynchronous operations in a procedural style. They are a feature introduced in C++20 to simplify asynchronous programming.<\/p>\n\n\n\n

Pre-existing mechanisms like std::async<\/code> tasks, std::packaged_task<\/code>, or events (std::condition_variable<\/code> & std::mutex<\/code>), synchronize two or more threads on the result of the task by establishing a communication channel. This communication channel has two ends:<\/p>\n\n\n\n

– std::promise<\/strong><\/code> that writes into the shared state either the result or the exception, and<\/p>\n\n\n\n

– std::future<\/strong><\/code> (std::shared_future<\/code>) – a receiving end that waits on the result of the task (or the exception). <\/p>\n\n\n\n

Unlike this pre-existing mechanism, coroutines don\u2018t directly involve threads or other OS synchronization primitives. They are a pure software abstraction based on the coroutine’s control object and the state-machine<\/strong> logic built around it.<\/p>\n\n\n\n

Coroutines are stackless<\/em> – which means that the control object has to be created on the heap. Coincidentally, it\u2018s a library wrapper around the promise_type<\/code> (std::coroutine_handle<promise_type><\/code>), but it doesn\u2019t have anything in common with std::promise<\/code>.<\/p>\n\n\n\n

The promise_type<\/code><\/strong> is an interface (a customization point<\/em>) that describes the predefined transition states in a coroutine’s state machine.<\/p>\n\n\n\n

Coroutines are highly versatile and can be used in various scenarios where you must manage asynchronous message flow. One common example is socket-based communication.<\/p>\n\n\n\n

Today, I\u2018ll try to enlighten coroutines through another example: single producer – single consumer<\/strong> workflow.<\/p>\n\n\n\n

Implementation<\/h2>\n\n\n\n

First, we need to define result type<\/strong> for the coroutine<\/p>\n\n\n\n

class<\/span>[[<\/span>nodiscard<\/span>]] AudioDataResult final\n\n{\n\n    public:<\/span>\n\n        class<\/span> promise_type<\/span>;\n\n        using<\/span> handle_type =<\/span> std::<\/span>coroutine_handle<<\/span>promise_type><\/span>;\n\n                \n\n        class<\/span> promise_type<\/span>\n\n        {\n\n            ...\n\n        };\n\n};\n<\/pre><\/div>\n\n\n\n
as a wrapper around the inner: promise_type<\/em><\/code> type. We decorate the enclosing class with [[nodiscard]]<\/code> attribute, since the result type is the control object of the coroutine that we return to the client code – to manage its suspension\/resumption.<\/p>\n\n\n\n
@note The destructor of the class cleans the resources (dynamic memory) in RAII fashion, so strictly speaking, the return type could be ignored if there is no need to manage<\/em><\/strong> the coroutine state<\/em><\/strong>.<\/p>\n\n\n\n
~<\/span>AudioDataResult() { if<\/span>(handle_) { handle_.destroy(); } }\n<\/pre><\/div>\n\n\n\n
The result type is move-only: the copy operations are forbidden – to prevent the control object from being multiplicated.<\/p>\n\n\n\n
\/\/ Make the result type move-only, due to exclusive ownership over the handle<\/span>\n\nAudioDataResult(const<\/span> AudioDataResult&<\/span> ) =<\/span> delete<\/span>;\n\nAudioDataResult&<\/span> operator<\/span>=<\/span> (constAudioDataResult&<\/span> ) =<\/span> delete<\/span>;\n\n \n\nAudioDataResult(AudioDataResult&&<\/span> other) noexcept:<\/span>\n\nhandle_(std::<\/span>exchange(other.handle_, nullptr))\n\n{}\n\n \n\nAudioDataResult&<\/span> operator<\/span> =<\/span> (AudioDataResult&&<\/span> other) noexcept\n\n{\n\n    using<\/span> namespace<\/span> std;\n\n    AudioDataResult tmp =<\/span>std::<\/span>move(other);\n\n    swap(*<\/span>this<\/span>, tmp);\n\n    return<\/span> *<\/span>this<\/span>;\n\n}\n<\/pre><\/div>\n\n\n\n
Let’s define the promise_type<\/code> interface itself:<\/p>\n\n\n\n
\/\/ Predefined interface that has to be specify in order to implement<\/span>\n\n\/\/ coroutine's state-machine transitions<\/span>\n\nclass<\/span> promise_type<\/span>\n\n{\n\n        \n\n    public:<\/span>\n\n                \n\n        using<\/span> value_type =<\/span> std::<\/span>vector<<\/span>int<\/span>><\/span>;\n\n \n\n        AudioDataResult get_return_object<\/span>()\n\n        {\n\n            return<\/span> AudioDataResult{ handle_type::<\/span>from_promise(*<\/span>this<\/span>) };\n\n        }\n\n        std::<\/span>suspend_never initial_suspend() noexcept { return<\/span>{}; }\n\n        std::<\/span>suspend_always final_suspend() noexcept { return<\/span>{}; }\n\n        \n\n        void<\/span> return_void() {}\n\n        void<\/span> unhandled_exception()\n\n        {\n\n            std::<\/span>rethrow_exception(std::<\/span>current_exception());\n\n        }\n\n \n\n        \/\/ Generates the value and suspend the "producer"<\/span>\n\n        template<\/span> <<\/span>typename<\/span> Data><\/span>\n\n        requires std::<\/span>convertible_to<<\/span>std::<\/span>decay_t<\/span><<\/span>Data><\/span>, value_type><\/span>\n\n        std::<\/span>suspend_always yield_value(Data&&<\/span> value)\n\n        {\n\n            data_ =<\/span> std::<\/span>forward<<\/span>Data><\/span>(value);\n\n            data_ready_.store(true<\/span>, std::<\/span>memory_order::<\/span>relaxed);\n\n            return<\/span> {};\n\n        }\n\n \n\n    private:<\/span>\n\n        value_type data_;\n\n        std::<\/span>atomic<<\/span>bool<\/span>><\/span> data_ready_;\n\n};\/\/promise_type interface<\/span>\n<\/pre><\/div>\n\n\n\n
The promise_type<\/code> defines the necessary infrastructure of the coroutine. Additionally, for any coroutines that want to act as a generator – “producer<\/em><\/strong>“, to yield the values: promise_type<\/code> has to be enhanced with the yield_value<\/code> method (co_yield \u2261 co_await promise_.yield_value<\/code>). Also, to resume the producer, at the point when provided data are consumed – we need to expose the appropriate wrapper method resume<\/em>():<\/code><\/p>\n\n\n\n
void<\/span> resume<\/span>() { if<\/span>( not handle_.done()) { handle_.resume();} }\n<\/pre><\/div>\n\n\n\n
Now – we need to extend the coroutine to support the consumer<\/em><\/strong> requirements: to be synchronized on the data readiness. In other words, the consumer will be suspended until the data are signaled as available by the producer. For that, we need to implement the Awaiter interface:<\/p>\n\n\n\n
class<\/span> promise_type<\/span>\n\n{\n\n    \/\/ Awaiter interface: for consumer waiting on data being ready<\/span>\n\n    struct<\/span> AudioDataAwaiter\n\n    {\n\n                                                \n\n        explicit<\/span> AudioDataAwaiter(promise_type&<\/span> promise) noexcept:<\/span> promise_(promise) {}\n\n \n\n        bool<\/span> await_ready() const<\/span>\n\n        {\n\n            return<\/span> promise_.data_ready_.load(std::<\/span>memory_order::<\/span>relaxed);\n\n        }\n\n                                        \n\n        void<\/span> await_suspend(handle_type) const<\/span>\n\n        {\n\n            while<\/span>( not promise_.data_ready_.exchange(false<\/span>))\n\n            {\n\n                std::<\/span>this_thread::<\/span>yield();\n\n            }\n\n        }\n\n       \n\n        \/\/ move assignment at client invocation side:<\/span>\n\n        \/\/        const auto data = co_await audioDataResult;<\/span>\n\n        \/\/ This requires that coroutine's result type provides the co_await unary operator<\/span>\n\n \n\n        value_type&&<\/span> await_resume() const<\/span>\n\n        {\n\n            return<\/span> std::<\/span>move(promise_.data_);\n\n        }\n\n \n\n    private:<\/span>\n\n            promise_type&<\/span> promise_;\n\n \n\n    };\/\/Awaiter interface<\/span>\n\n \n\n};\/\/promise_type<\/span>\n<\/pre><\/div>\n\n\n\n
In its state machine, the await_ready()<\/code><\/em> will be the first transition state: it will be inspected on data readiness. If the data are not ready, as next the <\/em>await_suspend()<\/code> will be called. Here we wait – until the matching flag is being set. Finally, the await_resume()<\/code><\/em> will be called: we “move” the value from the promise_type<\/code>, by unconditionally cast to the rvalue reference. On the client invocation side, this will cause the move assignment operator on the returned value – data to be called.<\/p>\n\n\n\n
const<\/span> auto<\/span> data =<\/span> co_await audioDataResult;\n<\/pre><\/div>\n\n\n\n
For that to work, the result type must provide the co_await <\/code>unary operator, which returns our Awaiter interface.<\/p>\n\n\n\n
class<\/span> AudioDataResult<\/span>\n\n{\n\n    auto<\/span> operator<\/span> co_await() noexcept\n\n    {\n\n        return<\/span> promise_type::<\/span>AudioDataAwaiter{handle_.promise()};\n\n    }\n\n};\n<\/pre><\/div>\n\n\n\n
<Example 1>:  https:\/\/godbolt.org\/z\/MvYfbEP8r<\/a><\/p>\n\n\n\n
The following program producerConsumer.cpp<\/code> shows a simplified version of example 1:<\/p>\n\n\n\n
\/\/ producerConsumer.cpp<\/span>\n\n#include <algorithm><\/span>\n#include <atomic><\/span>\n#include <chrono><\/span>\n#include <coroutine><\/span>\n#include <functional><\/span>\n#include <iostream><\/span>\n#include <iterator><\/span>\n#include <memory><\/span>\n#include <source_location><\/span>\n#include <thread><\/span>\n#include <utility><\/span>\n#include <vector><\/span>\n\n\nvoid<\/span> funcName<\/span>(const<\/span> std::<\/span>source_location location =<\/span> std::<\/span>source_location::<\/span>current()) {\n    std::<\/span>cout <<<\/span> location.function_name() <<<\/span> '\\n'<\/span>;\n}\n\n\ntemplate<\/span> <<\/span>typename<\/span> Container><\/span>\nvoid<\/span> printContainer(const<\/span> Container&<\/span> container)\n{\n    typedef<\/span> typename<\/span> Container::<\/span>value_type value_type;\n    auto<\/span> first =<\/span> std::<\/span>cbegin(container);\n    auto<\/span> last =<\/span> std::<\/span>cend(container);\n\n    std::<\/span>cout <<<\/span> " ["<\/span>;\n    std::<\/span>copy(first, std::<\/span>prev(last), std::<\/span>ostream_iterator<<\/span>value_type><\/span>(std::<\/span>cout, ", "<\/span>));\n    std::<\/span>cout <<<\/span> *<\/span>std::<\/span>prev(last) <<<\/span> "]<\/span>\\n<\/span>"<\/span>;\n}\n\n\n\n\nclass<\/span> [[<\/span>nodiscard<\/span>]] AudioDataResult final\n{\n    public:<\/span>\n        class<\/span> promise_type<\/span>;\n        using<\/span> handle_type =<\/span> std::<\/span>coroutine_handle<<\/span>promise_type><\/span>;\n        \n        \/\/ Predefined interface that has to be specify in order to implement<\/span>\n        \/\/ coroutine's state-machine transitions<\/span>\n        class<\/span> promise_type<\/span> \n        {\n            \n            public:<\/span>\n                \n                using<\/span> value_type =<\/span> std::<\/span>vector<<\/span>int<\/span>><\/span>;\n\n                AudioDataResult get_return_object<\/span>() \n                {\n                    return<\/span> AudioDataResult{handle_type::<\/span>from_promise(*<\/span>this<\/span>)};\n                }\n                std::<\/span>suspend_never initial_suspend() noexcept { return<\/span> {}; }\n                std::<\/span>suspend_always final_suspend() noexcept { return<\/span> {}; }\n                void<\/span> return_void() {}\n                void<\/span> unhandled_exception() \n                {\n                    std::<\/span>rethrow_exception(std::<\/span>current_exception());\n                }\n\n                \/\/ Generates the value and suspend the "producer"<\/span>\n                template<\/span> <<\/span>typename<\/span> Data><\/span>\n                requires std::<\/span>convertible_to<<\/span>std::<\/span>decay_t<\/span><<\/span>Data><\/span>, value_type><\/span>\n                std::<\/span>suspend_always yield_value(Data&&<\/span> value) \n                {\n                    data_ =<\/span> std::<\/span>forward<<\/span>Data><\/span>(value);\n                    data_ready_.store(true<\/span>);\n                    return<\/span> {};\n                }\n\n                \/\/ Awaiter interface: for consumer waiting on data being ready<\/span>\n                struct<\/span> AudioDataAwaiter \n                {\n                    explicit<\/span> AudioDataAwaiter(promise_type&<\/span> promise) noexcept:<\/span> promise_(promise) {}\n\n                    bool<\/span> await_ready() const<\/span> { return<\/span> promise_.data_ready_.load();}\n                    \n                    void<\/span> await_suspend(handle_type) const<\/span>\n                    {\n                        while<\/span>(not promise_.data_ready_.exchange(false<\/span>)) {\n                             std::<\/span>this_thread::<\/span>yield(); \n                        }\n                    }\n                    \/\/ move assignment at client invocation side: const auto data = co_await audioDataResult;<\/span>\n                    \/\/ This requires that coroutine's result type provides the co_await unary operator<\/span>\n                    value_type&&<\/span> await_resume() const<\/span> \n                    {\n                        return<\/span> std::<\/span>move(promise_.data_);\n                    }\n\n                    private:<\/span> \n                        promise_type&<\/span> promise_;\n                };\/\/Awaiter interface<\/span>\n\n        \n            private:<\/span>\n                value_type data_;\n                std::<\/span>atomic<<\/span>bool<\/span>><\/span> data_ready_;\n        }; \/\/promise_type interface<\/span>\n\n        \n        auto<\/span> operator<\/span> co_await() noexcept   \n        {\n            return<\/span> promise_type::<\/span>AudioDataAwaiter{handle_.promise()};\n        }\n\n        \/\/ Make the result type move-only, due to ownership over the handle<\/span>\n        AudioDataResult(const<\/span> AudioDataResult&<\/span>) =<\/span> delete<\/span>;\n        AudioDataResult&<\/span> operator<\/span>=<\/span>(const<\/span> AudioDataResult&<\/span>) =<\/span> delete<\/span>;\n\n        AudioDataResult(AudioDataResult&&<\/span> other) noexcept:<\/span> handle_(std::<\/span>exchange(other.handle_, nullptr)){}\n        AudioDataResult&<\/span> operator<\/span>=<\/span>(AudioDataResult&&<\/span> other) noexcept \n        {\n            using<\/span> namespace<\/span> std;\n            AudioDataResult tmp =<\/span> std::<\/span>move(other);\n            swap(*<\/span>this<\/span>, tmp);\n            return<\/span> *<\/span>this<\/span>;\n        }\n\n        \/\/ d-tor: RAII<\/span>\n        ~<\/span>AudioDataResult() { if<\/span> (handle_) {funcName(); handle_.destroy();}}\n\n        \/\/ For resuming the producer - at the point when the data are consumed<\/span>\n        void<\/span> resume() {if<\/span> (not handle_.done()) { funcName(); handle_.resume();}}\n    \n    private:<\/span>\n        AudioDataResult(handle_type handle) noexcept :<\/span> handle_(handle) {}\n\n    private:<\/span>\n    handle_type handle_;\n};\n\n\nusing<\/span> data_type =<\/span> std::<\/span>vector<<\/span>int<\/span>><\/span>;\nAudioDataResult producer<\/span>(const<\/span> data_type&<\/span> data) \n{\n    for<\/span> (std::<\/span>size_t<\/span> i =<\/span> 0<\/span>; i <<\/span> 5<\/span>; ++<\/span>i) {\n        funcName();\n        co_yield data;\n    }\n    co_yield data_type{}; \/\/ exit criteria<\/span>\n}\n\nAudioDataResult consumer<\/span>(AudioDataResult&<\/span> audioDataResult) \n{\n    while<\/span>(true<\/span>)\n    {\n        funcName();\n        const<\/span> auto<\/span> data =<\/span> co_await audioDataResult;\n        if<\/span> (data.empty()) {std::<\/span>cout <<<\/span> "No data - exit!<\/span>\\n<\/span>"<\/span>; break<\/span>;}\n        std::<\/span>cout <<<\/span> "Data received:"<\/span>;\n        printContainer(data);\n        audioDataResult.resume(); \/\/ resume producer<\/span>\n    }\n}\n\nint<\/span> main<\/span>() \n{\n    {\n        const<\/span> data_type data =<\/span> {1<\/span>, 2<\/span>, 3<\/span>, 4<\/span>};\n        auto<\/span> audioDataProducer =<\/span> producer(data);\n        std::<\/span>thread<\/span> t ([&<\/span>]{auto<\/span> audioRecorded =<\/span> consumer(audioDataProducer);});\n        t.join();\n    }\n\n    std::<\/span>cout <<<\/span> "bye-bye!<\/span>\\n<\/span>"<\/span>;\n}\n<\/pre><\/div>\n\n\n\n
Finally, here is the output of the program:<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
The other way around is to use the promise_type::await_transform()<\/code> – to wait on the value stored in the promise_type<\/code> instance used by the producer.<\/p>\n\n\n\n
class<\/span> promise_type<\/span>\n\n{\n\n    auto<\/span> await_transform(handle_type other)\n\n    {\n\n        \/\/ Awaiter interface: remained the same<\/span>\n\n        struct<\/span> AudioDataAwaiter\n\n        {\n\n            explicit<\/span> AudioDataAwaiter(promise_type&<\/span> promise)noexcept:<\/span> promise_(promise) {}\n\n            ...\n\n        };\n\n    \n\n        return<\/span> AudioDataAwaiter{other.promise()};\n\n    }\n\n};\n<\/pre><\/div>\n\n\n\n
This way, we don’t need to specify the co_await<\/code> unary operator of the result type anymore, but rather (explicit) conversion operator<\/p>\n\n\n\n
explicit<\/span> operator<\/span> handle_type() const<\/span> {return<\/span> handle_;}\n<\/pre><\/div>\n\n\n\n
so that we can pass it at the point when the consumer calls co_await<\/code>, which will internally be translated to the await_transform()<\/code> call.<\/p>\n\n\n\n
const<\/span> auto<\/span> data =<\/span> co_await static_cast<\/span><<\/span>AudioDataResult::<\/span>handle_type><\/span>(audioDataResult);\n<\/pre><\/div>\n\n\n\n
We can illustrate this as:   me.handle_.promise().await_transform(other.handle_)<\/code><\/em><\/p>\n\n\n\n
Conclusion<\/h2>\n\n\n\n
In this simple example, the producer will be suspended without any penalty – since, after being resumed, it will provide the very same – upfront known sequence of data. In a real scenario, that is likely not the case: the producer will probably be some kind of mediator – receiver of asynchronously emitted data that will be retransmitted to the consumer. For that, some queuing logic needs to be implemented at the producer side to avoid the data loss at the point of being suspended, waiting for the consumer to resume it – to compensate for the differences in producer data arrival vs. consumer consumption rate.<\/p>\n\n\n\n
What’s Next? <\/h2>\n\n\n\n
In C++20, you can define or default<\/code> the three-way comparison operator.  This gives you all six comparison operators: ==, !=, <, <=, ><\/code>, and >=<\/code>. Do you know that you can also define or default<\/code> the equal operator (==)<\/code> ?<\/p>\n\n\n\n
A Short Christmas Break<\/h2>\n\n\n\n
My blog makes a short Christmas break. The next post will be published on the 8th January 2024. Have a good time.<\/p>\n","protected":false},"excerpt":{"rendered":"
I’m happy to present in this guest post by Ljubic Damir a typical use-case for coroutines: a producer – consumer workflow. Although this producer-consumer workflow is challenging, it is a nice starting point for your coroutine experiments. Intro Coroutines provide a more intuitive and structured way of writing asynchronous code by allowing you to write […]<\/p>\n","protected":false},"author":21,"featured_media":8607,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[375],"tags":[445],"class_list":["post-8692","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-c-20","tag-coroutines"],"_links":{"self":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/8692","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/users\/21"}],"replies":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/comments?post=8692"}],"version-history":[{"count":26,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/8692\/revisions"}],"predecessor-version":[{"id":8801,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/8692\/revisions\/8801"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/media\/8607"}],"wp:attachment":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/media?parent=8692"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/categories?post=8692"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/tags?post=8692"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}
<Example 2>:  https:\/\/godbolt.org\/z\/57zsK9rEn<\/a><\/p>\n\n\n\n