Motivation

| Home | Libraries | People | FAQ | More

## Motivation

In order to support a broad range of execution control behaviour the coroutine types of coroutine<> can be used to escape-and-reenter loops, to escape-and-reenter recursive computations and for cooperative multitasking helping to solve problems in a much simpler and more elegant way than with only a single flow of control.

#### event-driven model

The event-driven model is a programming paradigm where the flow of a program is determined by events. The events are generated by multiple independent sources and an event-dispatcher, waiting on all external sources, triggers callback functions (event-handlers) whenever one of those events is detected (event-loop). The application is divided into event selection (detection) and event handling.

The resulting applications are highly scalable, flexible, have high responsiveness and the components are loosely coupled. This makes the event-driven model suitable for user interface applications, rule-based productions systems or applications dealing with asynchronous I/O (for instance network servers).

#### event-based asynchronous paradigm

A classic synchronous console program issues an I/O request (e.g. for user input or filesystem data) and blocks until the request is complete.

In contrast, an asynchronous I/O function initiates the physical operation but immediately returns to its caller, even though the operation is not yet complete. A program written to leverage this functionality does not block: it can proceed with other work (including other I/O requests in parallel) while the original operation is still pending. When the operation completes, the program is notified. Because asynchronous applications spend less overall time waiting for operations, they can outperform synchronous programs.

Events are one of the paradigms for asynchronous execution, but not all asynchronous systems use events. Although asynchronous programming can be done using threads, they come with their own costs:

- hard to program (traps for the unwary)
- memory requirements are high
- large overhead with creation and maintenance of thread state
- expensive context switching between threads

The event-based asynchronous model avoids those issues:

- simpler because of the single stream of instructions
- much less expensive context switches

The downside of this paradigm consists in a sub-optimal program structure. An event-driven program is required to split its code into multiple small callback functions, i.e. the code is organized in a sequence of small steps that execute intermittently. An algorithm that would usually be expressed as a hierarchy of functions and loops must be transformed into callbacks. The complete state has to be stored into a data structure while the control flow returns to the event-loop. As a consequence, event-driven applications are often tedious and confusing to write. Each callback introduces a new scope, error callback etc. The sequential nature of the algorithm is split into multiple callstacks, making the application hard to debug. Exception handlers are restricted to local handlers: it is impossible to wrap a sequence of events into a single try-catch block. The use of local variables, while/for loops, recursions etc. together with the event-loop is not possible. The code becomes less expressive.

In the past, code using asio's asynchronous operations was convoluted by callback functions.

    class session
    {
    public:
        session(boost::asio::io_service& io_service) :
              socket_(io_service) // construct a TCP-socket from io_service
        {}

        tcp::socket& socket(){
            return socket_;
        }

        void start(){
            // initiate asynchronous read; handle_read() is callback-function
            socket_.async_read_some(boost::asio::buffer(data_,max_length),
                boost::bind(&session::handle_read,this,
                    boost::asio::placeholders::error,
                    boost::asio::placeholders::bytes_transferred));
        }

    private:
        void handle_read(const boost::system::error_code& error,
                         size_t bytes_transferred){
            if (!error)
                // initiate asynchronous write; handle_write() is callback-function
                boost::asio::async_write(socket_,
                    boost::asio::buffer(data_,bytes_transferred),
                    boost::bind(&session::handle_write,this,
                        boost::asio::placeholders::error));
            else
                delete this;
        }

        void handle_write(const boost::system::error_code& error){
            if (!error)
                // initiate asynchronous read; handle_read() is callback-function
                socket_.async_read_some(boost::asio::buffer(data_,max_length),
                    boost::bind(&session::handle_read,this,
                        boost::asio::placeholders::error,
                        boost::asio::placeholders::bytes_transferred));
            else
                delete this;
        }

        boost::asio::ip::tcp::socket socket_;
        enum { max_length=1024 };
        char data_[max_length];
    };

In this example, a simple echo server, the logic is split into three member functions - local state (such as data buffer) is moved to member variables.

Boost.Asio provides with its new asynchronous result feature a new framework combining event-driven model and coroutines, hiding the complexity of event-driven programming and permitting the style of classic sequential code. The application is not required to pass callback functions to asynchronous operations and local state is kept as local variables. Therefore the code is much easier to read and understand. [4].

    void session(boost::asio::io_service& io_service){
        // construct TCP-socket from io_service
        boost::asio::ip::tcp::socket socket(io_service);

        try{
            for(;;){
                // local data-buffer
                char data[max_length];

                boost::system::error_code ec;

                // read asynchronous data from socket
                // execution context will be suspended until
                // some bytes are read from socket
                std::size_t length=socket.async_read_some(
                        boost::asio::buffer(data),
                        boost::asio::yield[ec]);
                if (ec==boost::asio::error::eof)
                    break; //connection closed cleanly by peer
                else if(ec)
                    throw boost::system::system_error(ec); //some other error

                // write some bytes asynchronously
                boost::asio::async_write(
                        socket,
                        boost::asio::buffer(data,length),
                        boost::asio::yield[ec]);
                if (ec==boost::asio::error::eof)
                    break; //connection closed cleanly by peer
                else if(ec)
                    throw boost::system::system_error(ec); //some other error
            }
        } catch(std::exception const& e){
            std::cerr<<"Exception: "<<e.what()<<"\n";
        }
    }

In contrast to the previous example this one gives the impression of sequential code and local data (data) while using asynchronous operations (async_read(), async_write()). The algorithm is implemented in one function and error handling is done by one try-catch block.

#### recursive descent parsing

Coroutines let you invert the flow of control so you can ask a recursive descent parser for parsed symbols.

    class Parser{
       char next;
       std::istream& is;
       std::function<void(char)> cb;

       char pull(){
            return std::char_traits<char>::to_char_type(is.get());
       }

       void scan(){
           do{
               next=pull();
           }
           while(isspace(next));
       }

    public:
       Parser(std::istream& is_,std::function<void(char)> cb_) :
          next(), is(is_), cb(cb_)
        {}

       void run() {
          scan();
          E();
       }

    private:
       void E(){
          T();
          while (next=='+'||next=='-'){
             cb(next);
             scan();
             T();
          }
       }

       void T(){
          S();
          while (next=='*'||next=='/'){
             cb(next);
             scan();
             S();
          }
       }

       void S(){
          if (std::isdigit(next)){
             cb(next);
             scan();
          }
          else if(next=='('){
             cb(next);
             scan();
             E();
             if (next==')'){
                 cb(next);
                 scan();
             }else{
                 throw parser_error();
             }
          }
          else{
             throw parser_error();
          }
       }
    };

    typedef boost::coroutines2::coroutine< char > coro_t;

    int main() {
        std::istringstream is("1+1");
        // invert control flow
        coro_t::pull_type seq(
                boost::coroutines2::fixedsize_stack(),
                [&is](coro_t::push_type & yield) {
                    // create parser with callback function
                    Parser p( is,
                              [&yield](char ch){
                                // resume user-code
                                yield(ch);
                              });
                    // start recursive parsing
                    p.run();
                });

        // user-code pulls parsed data from parser
        // invert control flow
        for(char c:seq){
            printf("Parsed: %c\n",c);
        }
    }

This problem does not map at all well to communicating between independent threads. It makes no sense for either side to proceed independently of the other. You want them to pass control back and forth.

There's yet another advantage to using coroutines. This recursive descent parser throws an exception when parsing fails. With a coroutine implementation, you need only wrap the calling code in try/catch.

With communicating threads, you would have to arrange to catch the exception and pass along the exception pointer on the same queue you're using to deliver the other events. You would then have to rethrow the exception to unwind the recursive document processing.

The coroutine solution maps very naturally to the problem space.

#### 'same fringe' problem

The advantages of suspending at an arbitrary call depth can be seen particularly clearly with the use of a recursive function, such as traversal of trees. If traversing two different trees in the same deterministic order produces the same list of leaf nodes, then both trees have the same fringe.

Both trees in the picture have the same fringe even though the structure of the trees is different.

The same fringe problem could be solved using coroutines by iterating over the leaf nodes and comparing this sequence via std::equal(). The range of data values is generated by function traverse() which recursively traverses the tree and passes each node's data value to its coroutine<>::push_type. coroutine<>::push_type suspends the recursive computation and transfers the data value to the main execution context. coroutine<>::pull_type::iterator, created from coroutine<>::pull_type, steps over those data values and delivers them to std::equal() for comparison. Each increment of coroutine<>::pull_type::iterator resumes traverse(). Upon return from iterator::operator++(), either a new data value is available, or tree traversal is finished (iterator is invalidated).

In effect, the coroutine iterator presents a flattened view of the recursive data structure.

    struct node{
        typedef std::shared_ptr<node> ptr_t;

        // Each tree node has an optional left subtree,
        // an optional right subtree and a value of its own.
        // The value is considered to be between the left
        // subtree and the right.
        ptr_t       left,right;
        std::string value;

        // construct leaf
        node(const std::string& v):
            left(),right(),value(v)
        {}
        // construct nonleaf
        node(ptr_t l,const std::string& v,ptr_t r):
            left(l),right(r),value(v)
        {}

        static ptr_t create(const std::string& v){
            return ptr_t(new node(v));
        }

        static ptr_t create(ptr_t l,const std::string& v,ptr_t r){
            return ptr_t(new node(l,v,r));
        }
    };

    node::ptr_t create_left_tree_from(const std::string& root){
        /* --------
             root
             / \
            b   e
           / \
          a   c
         -------- */
        return node::create(
                node::create(
                    node::create("a"),
                    "b",
                    node::create("c")),
                root,
                node::create("e"));
    }

    node::ptr_t create_right_tree_from(const std::string& root){
        /* --------
             root
             / \
            a   d
               / \
              c   e
           -------- */
        return node::create(
                node::create("a"),
                root,
                node::create(
                    node::create("c"),
                    "d",
                    node::create("e")));
    }

    typedef boost::coroutines2::coroutine<std::string>   coro_t;

    // recursively walk the tree, delivering values in order
    void traverse(node::ptr_t n,
                  coro_t::push_type& out){
        if(n->left) traverse(n->left,out);
        out(n->value);
        if(n->right) traverse(n->right,out);
    }

    // evaluation
    {
        node::ptr_t left_d(create_left_tree_from("d"));
        coro_t::pull_type left_d_reader([&](coro_t::push_type & out){
                                            traverse(left_d,out);
                                        });

        node::ptr_t right_b(create_right_tree_from("b"));
        coro_t::pull_type right_b_reader([&](coro_t::push_type & out){
                                            traverse(right_b,out);
                                         });

        std::cout << "left tree from d == right tree from b? "
                  << std::boolalpha
                  << std::equal(begin(left_d_reader),
                                end(left_d_reader),
                                begin(right_b_reader))
                  << std::endl;
    }
    {
        node::ptr_t left_d(create_left_tree_from("d"));
        coro_t::pull_type left_d_reader([&](coro_t::push_type & out){
                                            traverse(left_d,out);
                                        });

        node::ptr_t right_x(create_right_tree_from("x"));
        coro_t::pull_type right_x_reader([&](coro_t::push_type & out){
                                             traverse(right_x,out);
                                         });

        std::cout << "left tree from d == right tree from x? "
                  << std::boolalpha
                  << std::equal(begin(left_d_reader),
                                end(left_d_reader),
                                begin(right_x_reader))
                  << std::endl;
    }
    std::cout << "Done" << std::endl;

    output:
    left tree from d == right tree from b? true
    left tree from d == right tree from x? false
    Done

#### chaining coroutines

This code shows how coroutines could be chained.

    typedef boost::coroutines2::coroutine<std::string> coro_t;

    // deliver each line of input stream to sink as a separate string
    void readlines(coro_t::push_type& sink,std::istream& in){
        std::string line;
        while(std::getline(in,line))
            sink(line);
    }

    void tokenize(coro_t::push_type& sink, coro_t::pull_type& source){
        // This tokenizer doesn't happen to be stateful: you could reasonably
        // implement it with a single call to push each new token downstream. But
        // I've worked with stateful tokenizers, in which the meaning of input
        // characters depends in part on their position within the input line.
        for(std::string line:source){
            std::string::size_type pos=0;
            while(pos<line.length()){
                if(line[pos]=='"'){
                    std::string token;
                    ++pos;              // skip open quote
                    while(pos<line.length()&&line[pos]!='"')
                        token+=line[pos++];
                    ++pos;              // skip close quote
                    sink(token);        // pass token downstream
                } else if (std::isspace(line[pos])){
                    ++pos;              // outside quotes, ignore whitespace
                } else if (std::isalpha(line[pos])){
                    std::string token;
                    while (pos < line.length() && std::isalpha(line[pos]))
                        token += line[pos++];
                    sink(token);        // pass token downstream
                } else {                // punctuation
                    sink(std::string(1,line[pos++]));
                }
            }
        }
    }

    void only_words(coro_t::push_type& sink,coro_t::pull_type& source){
        for(std::string token:source){
            if (!token.empty() && std::isalpha(token[0]))
                sink(token);
        }
    }

    void trace(coro_t::push_type& sink, coro_t::pull_type& source){
        for(std::string token:source){
            std::cout << "trace: '" << token << "'\n";
            sink(token);
        }
    }

    struct FinalEOL{
        ~FinalEOL(){
            std::cout << std::endl;
        }
    };

    void layout(coro_t::pull_type& source,int num,int width){
        // Finish the last line when we leave by whatever means
        FinalEOL eol;

        // Pull values from upstream, lay them out 'num' to a line
        for (;;){
            for (int i = 0; i < num; ++i){
                // when we exhaust the input, stop
                if (!source) return;

                std::cout << std::setw(width) << source.get();
                // now that we've handled this item, advance to next
                source();
            }
            // after 'num' items, line break
            std::cout << std::endl;
        }
    }

    // For example purposes, instead of having a separate text file in the
    // local filesystem, construct an istringstream to read.
    std::string data(
        "This is the first line.\n"
        "This, the second.\n"
        "The third has \"a phrase\"!\n"
        );

    {
        std::cout << "\nfilter:\n";
        std::istringstream infile(data);
        coro_t::pull_type reader(std::bind(readlines, _1, std::ref(infile)));
        coro_t::pull_type tokenizer(std::bind(tokenize, _1, std::ref(reader)));
        coro_t::pull_type filter(std::bind(only_words, _1, std::ref(tokenizer)));
        coro_t::pull_type tracer(std::bind(trace, _1, std::ref(filter)));
        for(std::string token:tracer){
            // just iterate, we're already pulling through tracer
        }
    }

    {
        std::cout << "\nlayout() as coroutine::push_type:\n";
        std::istringstream infile(data);
        coro_t::pull_type reader(std::bind(readlines, _1, std::ref(infile)));
        coro_t::pull_type tokenizer(std::bind(tokenize, _1, std::ref(reader)));
        coro_t::pull_type filter(std::bind(only_words, _1, std::ref(tokenizer)));
        coro_t::push_type writer(std::bind(layout, _1, 5, 15));
        for(std::string token:filter){
            writer(token);
        }
    }

    {
        std::cout << "\nfiltering output:\n";
        std::istringstream infile(data);
        coro_t::pull_type reader(std::bind(readlines,_1,std::ref(infile)));
        coro_t::pull_type tokenizer(std::bind(tokenize,_1,std::ref(reader)));
        coro_t::push_type writer(std::bind(layout,_1,5,15));
        // Because of the symmetry of the API, we can use any of these
        // chaining functions in a push_type coroutine chain as well.
        coro_t::push_type filter(std::bind(only_words,std::ref(writer),_1));
        for(std::string token:tokenizer){
            filter(token);
        }
    }

[4] Christopher Kohlhoff, N3964 - Library Foundations for Asynchronous Operations, Revision 1