source: doc/papers/concurrency/examples/ProdCons.cpp @ 33a484f

#include <cassert>
#include <cstdlib>
#include <iostream>
#include <experimental/coroutine>
#include <unistd.h>

int random(int max) {
        return std::rand() % max;
}

struct Prod;
struct Cons;

struct resumable {
        virtual resumable * resume() = 0;
};

struct Prod : public resumable {
        struct local {
                Cons * c;
                int N, money, receipt;
        };

        struct promise_type {
                local _l;
                resumable * next;

                Prod get_return_object() {
                        return Prod(std::experimental::coroutine_handle<promise_type>::from_promise(*this));
                }

                auto initial_suspend() { return std::experimental::suspend_never(); }
                auto final_suspend()   { return std::experimental::suspend_always(); }

                void return_void() {}
                void unhandled_exception() {}
        };

        struct data {
                promise_type * _promise = nullptr;
                bool await_ready() noexcept { return false; }
                void await_suspend(std::experimental::coroutine_handle<promise_type> _coroutine) noexcept {
                        _promise = &_coroutine.promise();
                }
                local & await_resume() noexcept { assert(_promise); return _promise->_l; }
        };

        std::experimental::coroutine_handle<promise_type> _coroutine = nullptr;

        explicit Prod(std::experimental::coroutine_handle<promise_type> coroutine)
                : _coroutine(coroutine)
        {}

        ~Prod() {
                if(_coroutine) { _coroutine.destroy(); }
        }

        Prod() = default;
        Prod(Prod const &) = delete;
        Prod& operator=(Prod const &) = delete;

        Prod(Prod&& other) {
                std::swap(_coroutine, other._coroutine);
        }

        Prod& operator=(Prod&& other) {
                if(&other != this) {
                        _coroutine = other._coroutine;
                        other._coroutine = nullptr;
                }
                return *this;
        }

        static Prod main();

        struct payment_return;

        payment_return payment(int money);

        auto start(int N, Cons & c) {
                _coroutine.promise()._l.c = &c;
                _coroutine.promise()._l.N = N;
                _coroutine.promise()._l.receipt = 0;
        }

        virtual resumable * resume() override final {
                _coroutine.resume();
                return _coroutine.promise().next;
        }
};

struct Cons : public resumable {
        struct local {
                Prod * p;
                int p1, p2, status;
                bool done;
        };

        struct promise_type {
                local _l;
                resumable * next;

                Cons get_return_object() {
                        return Cons(std::experimental::coroutine_handle<promise_type>::from_promise(*this));
                }

                auto initial_suspend() { return std::experimental::suspend_never(); }
                auto final_suspend()   { return std::experimental::suspend_always(); }

                void return_void() {}
                void unhandled_exception() {}
        };

        struct data {
                Prod * _p;
                data(Prod & prod) : _p(&prod) {}
                promise_type * _promise = nullptr;
                bool await_ready() noexcept { return false; }
                void await_suspend(std::experimental::coroutine_handle<promise_type> _coroutine) noexcept {
                        _promise = &_coroutine.promise();
                }
                local & await_resume() noexcept { assert(_promise); _promise->_l.p = _p; return _promise->_l; }
        };

        std::experimental::coroutine_handle<promise_type> _coroutine = nullptr;

        explicit Cons(std::experimental::coroutine_handle<promise_type> coroutine)
                : _coroutine(coroutine)
        {}

        ~Cons() {
                if(_coroutine) { _coroutine.destroy(); }
        }

        Cons() = default;
        Cons(Cons const &) = delete;
        Cons& operator=(Cons const &) = delete;

        Cons(Cons&& other) {
                std::swap(_coroutine, other._coroutine);
        }

        Cons& operator=(Cons&& other) {
                if(&other != this) {
                        _coroutine = other._coroutine;
                        other._coroutine = nullptr;
                }
                return *this;
        }

        static Cons main( Prod & prod );

        auto deliver(int p1, int p2) {
                _coroutine.promise()._l.p1 = p1;
                _coroutine.promise()._l.p2 = p2;

                struct ret {
                        int _status;
                        Cons * c;
                        bool await_ready() { return false; }
                        void await_suspend(std::experimental::coroutine_handle<Prod::promise_type> _coroutine) {
                                _coroutine.promise().next = c;
                        }
                        int await_resume() { return _status; }
                };
                return ret{ _coroutine.promise()._l.status, this };
        }

        auto stop() {
                _coroutine.promise()._l.done = true;
                struct ret {
                        Cons * c;
                        Prod::promise_type * _promise;
                        bool await_ready() { return false; }
                        void await_suspend(std::experimental::coroutine_handle<Prod::promise_type> _coroutine) {
                                _promise = &_coroutine.promise();
                                _promise->next = c;
                        }
                        void await_resume() {
                                _promise->next = nullptr;
                        }
                };
                return ret{this, nullptr};
        }

        virtual resumable * resume() override final {
                _coroutine.resume();
                return _coroutine.promise().next;
        }
};

struct Prod::payment_return {
        int _receipt;
        Prod * p;
        bool await_ready() { return false; }
        void await_suspend(std::experimental::coroutine_handle<Cons::promise_type> _coroutine) {
                _coroutine.promise().next = p;
        }
        int await_resume() { return _receipt; }
};

Prod::payment_return Prod::payment(int money)  {
        _coroutine.promise()._l.money = money;
        return payment_return{ _coroutine.promise()._l.receipt, this };
}

Prod Prod::main() {
        auto & p = co_await Prod::data();
        for(int i = 0; i < p.N; i++) {
                int p1 = random(100), p2 = random(100);
                std::cout << p1 << " " << p2 << std::endl;
                int status = co_await p.c->deliver(p1, p2);
                std::cout << " $" << p.money << std::endl << status << std::endl;
                p.receipt += 1;
        }
        co_await p.c->stop();
        std::cout << "prod stops" << std::endl;
}

Cons Cons::main( Prod & prod ) {
        auto & c = co_await Cons::data( prod );
        int money = 1, receipt;
        for(;!c.done ;) {
                std::cout << c.p1 << " " << c.p2 << std::endl;
                std::cout << " $ " << money << std::endl;
                c.status += 1;
                receipt = co_await c.p->payment( money );
                std::cout << " # " << receipt << std::endl;
                money += 1;
        }
        std::cout << "cons stops" << std::endl;
}

void dispatch(resumable * r) {
        while((r = r->resume()));
}

int main() {
        auto prod = Prod::main();
        auto cons = Cons::main( prod );
        srandom( getpid() );
        prod.start(5, cons);
        dispatch(&prod);
}