ceph/src/boost/libs/coroutine/doc/motivation.qbk

   1 [/
   2           Copyright Oliver Kowalke 2009.
   3  Distributed under the Boost Software License, Version 1.0.
   4     (See accompanying file LICENSE_1_0.txt or copy at
   5           http://www.boost.org/LICENSE_1_0.txt
   6 ]
   7
   8 [section:motivation Motivation]
   9
  10 In order to support a broad range of execution control behaviour the coroutine
  11 types of __scoro__ and __acoro__ can be used to ['escape-and-reenter] loops, to
  12 ['escape-and-reenter] recursive computations and for ['cooperative] multitasking
  13 helping to solve problems in a much simpler and more elegant way than with only
  14 a single flow of control.
  15
  16
  17 [heading event-driven model]
  18
  19 The event-driven model is a programming paradigm where the flow of a program is
  20 determined by events. The events are generated by multiple independent sources
  21 and an event-dispatcher, waiting on all external sources, triggers callback
  22 functions (event-handlers) whenever one of those events is detected (event-loop).
  23 The application is divided into event selection (detection) and event handling.
  24
  25 [$../../../../libs/coroutine/doc/images/event_model.png [align center]]
  26
  27 The resulting applications are highly scalable, flexible, have high
  28 responsiveness and the components are loosely coupled. This makes the event-driven
  29 model suitable for user interface applications, rule-based productions systems
  30 or applications dealing with asynchronous I/O (for instance network servers).
  31
  32
  33 [heading event-based asynchronous paradigm]
  34
  35 A classic synchronous console program issues an I/O request (e.g. for user
  36 input or filesystem data) and blocks until the request is complete.
  37
  38 In contrast, an asynchronous I/O function initiates the physical operation but
  39 immediately returns to its caller, even though the operation is not yet
  40 complete. A program written to leverage this functionality does not block: it
  41 can proceed with other work (including other I/O requests in parallel) while
  42 the original operation is still pending. When the operation completes, the
  43 program is notified. Because asynchronous applications spend less overall time
  44 waiting for operations, they can outperform synchronous programs.
  45
  46 Events are one of the paradigms for asynchronous execution, but
  47 not all asynchronous systems use events.
  48 Although asynchronous programming can be done using threads, they come with
  49 their own costs:
  50
  51 * hard to program (traps for the unwary)
  52 * memory requirements are high
  53 * large overhead with creation and maintenance of thread state
  54 * expensive context switching between threads
  55
  56 The event-based asynchronous model avoids those issues:
  57
  58 * simpler because of the single stream of instructions
  59 * much less expensive context switches
  60
  61 The downside of this paradigm consists in a sub-optimal program
  62 structure. An event-driven program is required to split its code into
  63 multiple small callback functions, i.e. the code is organized in a sequence of
  64 small steps that execute intermittently. An algorithm that would usually be expressed
  65 as a hierarchy of functions and loops must be transformed into callbacks. The
  66 complete state has to be stored into a data structure while the control flow
  67 returns to the event-loop.
  68 As a consequence, event-driven applications are often tedious and confusing to
  69 write. Each callback introduces a new scope, error callback etc. The
  70 sequential nature of the algorithm is split into multiple callstacks,
  71 making the application hard to debug. Exception handlers are restricted to
  72 local handlers: it is impossible to wrap a sequence of events into a single
  73 try-catch block.
  74 The use of local variables, while/for loops, recursions etc. together with the
  75 event-loop is not possible. The code becomes less expressive.
  76
  77 In the past, code using asio's ['asynchronous operations] was convoluted by
  78 callback functions.
  79
  80         class session
  81         {
  82         public:
  83             session(boost::asio::io_service& io_service) :
  84                   socket_(io_service) // construct a TCP-socket from io_service
  85             {}
  86
  87             tcp::socket& socket(){
  88                 return socket_;
  89             }
  90
  91             void start(){
  92                 // initiate asynchronous read; handle_read() is callback-function
  93                 socket_.async_read_some(boost::asio::buffer(data_,max_length),
  94                     boost::bind(&session::handle_read,this,
  95                         boost::asio::placeholders::error,
  96                         boost::asio::placeholders::bytes_transferred));
  97             }
  98
  99         private:
 100             void handle_read(const boost::system::error_code& error,
 101                              size_t bytes_transferred){
 102                 if (!error)
 103                     // initiate asynchronous write; handle_write() is callback-function
 104                     boost::asio::async_write(socket_,
 105                         boost::asio::buffer(data_,bytes_transferred),
 106                         boost::bind(&session::handle_write,this,
 107                             boost::asio::placeholders::error));
 108                 else
 109                     delete this;
 110             }
 111
 112             void handle_write(const boost::system::error_code& error){
 113                 if (!error)
 114                     // initiate asynchronous read; handle_read() is callback-function
 115                     socket_.async_read_some(boost::asio::buffer(data_,max_length),
 116                         boost::bind(&session::handle_read,this,
 117                             boost::asio::placeholders::error,
 118                             boost::asio::placeholders::bytes_transferred));
 119                 else
 120                     delete this;
 121             }
 122
 123             boost::asio::ip::tcp::socket socket_;
 124             enum { max_length=1024 };
 125             char data_[max_length];
 126         };
 127
 128 In this example, a simple echo server, the logic is split into three member
 129 functions - local state (such as data buffer) is moved to member variables.
 130
 131 __boost_asio__ provides with its new ['asynchronous result] feature a new
 132 framework combining event-driven model and coroutines, hiding the complexity
 133 of event-driven programming and permitting the style of classic sequential code.
 134 The application is not required to pass callback functions to asynchronous
 135 operations and local state is kept as local variables. Therefore the code
 136 is much easier to read and understand.
 137 [footnote Christopher Kohlhoff,
 138 [@ http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2014/n3964.pdf
 139 N3964 - Library Foundations for Asynchronous Operations, Revision 1]].
 140 __yield_context__ internally uses __boost_coroutine__:
 141
 142         void session(boost::asio::io_service& io_service){
 143             // construct TCP-socket from io_service
 144             boost::asio::ip::tcp::socket socket(io_service);
 145
 146             try{
 147                 for(;;){
 148                     // local data-buffer
 149                     char data[max_length];
 150
 151                     boost::system::error_code ec;
 152
 153                     // read asynchronous data from socket
 154                     // execution context will be suspended until
 155                     // some bytes are read from socket
 156                     std::size_t length=socket.async_read_some(
 157                             boost::asio::buffer(data),
 158                             boost::asio::yield[ec]);
 159                     if (ec==boost::asio::error::eof)
 160                         break; //connection closed cleanly by peer
 161                     else if(ec)
 162                         throw boost::system::system_error(ec); //some other error
 163
 164                     // write some bytes asynchronously
 165                     boost::asio::async_write(
 166                             socket,
 167                             boost::asio::buffer(data,length),
 168                             boost::asio::yield[ec]);
 169                     if (ec==boost::asio::error::eof)
 170                         break; //connection closed cleanly by peer
 171                     else if(ec)
 172                         throw boost::system::system_error(ec); //some other error
 173                 }
 174             } catch(std::exception const& e){
 175                 std::cerr<<"Exception: "<<e.what()<<"\n";
 176             }
 177         }
 178
 179 In contrast to the previous example this one gives the impression of sequential
 180 code and local data (['data]) while using asynchronous operations
 181 (['async_read()], ['async_write()]). The algorithm is implemented in one
 182 function and error handling is done by one try-catch block.
 183
 184 [heading recursive SAX parsing]
 185 To someone who knows SAX, the phrase "recursive SAX parsing" might sound
 186 nonsensical. You get callbacks from SAX; you have to manage the element stack
 187 yourself. If you want recursive XML processing, you must first read the entire
 188 DOM into memory, then walk the tree.
 189
 190 But coroutines let you invert the flow of control so you can ask for SAX
 191 events. Once you can do that, you can process them recursively.
 192
 193         // Represent a subset of interesting SAX events
 194         struct BaseEvent{
 195            BaseEvent(const BaseEvent&)=delete;
 196            BaseEvent& operator=(const BaseEvent&)=delete;
 197         };
 198
 199         // End of document or element
 200         struct CloseEvent: public BaseEvent{
 201            // CloseEvent binds (without copying) the TagType reference.
 202            CloseEvent(const xml::sax::Parser::TagType& name):
 203                mName(name)
 204            {}
 205
 206            const xml::sax::Parser::TagType& mName;
 207         };
 208
 209         // Start of document or element
 210         struct OpenEvent: public CloseEvent{
 211            // In addition to CloseEvent's TagType, OpenEvent binds AttributeIterator.
 212            OpenEvent(const xml::sax::Parser::TagType& name,
 213                      xml::sax::AttributeIterator& attrs):
 214                CloseEvent(name),
 215                mAttrs(attrs)
 216            {}
 217
 218            xml::sax::AttributeIterator& mAttrs;
 219         };
 220
 221         // text within an element
 222         struct TextEvent: public BaseEvent{
 223            // TextEvent binds the CharIterator.
 224            TextEvent(xml::sax::CharIterator& text):
 225                mText(text)
 226            {}
 227
 228            xml::sax::CharIterator& mText;
 229         };
 230
 231         // The parsing coroutine instantiates BaseEvent subclass instances and
 232         // successively shows them to the main program. It passes a reference so we
 233         // don't slice the BaseEvent subclass.
 234         typedef boost::coroutines::asymmetric_coroutine<const BaseEvent&> coro_t;
 235
 236         void parser(coro_t::push_type& sink,std::istream& in){
 237            xml::sax::Parser xparser;
 238            // startDocument() will send OpenEvent
 239            xparser.startDocument([&sink](const xml::sax::Parser::TagType& name,
 240                                          xml::sax::AttributeIterator& attrs)
 241                                  {
 242                                      sink(OpenEvent(name,attrs));
 243                                  });
 244            // startTag() will likewise send OpenEvent
 245            xparser.startTag([&sink](const xml::sax::Parser::TagType& name,
 246                                     xml::sax::AttributeIterator& attrs)
 247                             {
 248                                 sink(OpenEvent(name,attrs));
 249                             });
 250            // endTag() will send CloseEvent
 251            xparser.endTag([&sink](const xml::sax::Parser::TagType& name)
 252                           {
 253                               sink(CloseEvent(name));
 254                           });
 255            // endDocument() will likewise send CloseEvent
 256            xparser.endDocument([&sink](const xml::sax::Parser::TagType& name)
 257                                {
 258                                    sink(CloseEvent(name));
 259                                });
 260            // characters() will send TextEvent
 261            xparser.characters([&sink](xml::sax::CharIterator& text)
 262                               {
 263                                   sink(TextEvent(text));
 264                               });
 265            try
 266            {
 267                // parse the document, firing all the above
 268                xparser.parse(in);
 269            }
 270            catch (xml::Exception e)
 271            {
 272                // xml::sax::Parser throws xml::Exception. Helpfully translate the
 273                // name and provide it as the what() string.
 274                throw std::runtime_error(exception_name(e));
 275            }
 276         }
 277
 278         // Recursively traverse the incoming XML document on the fly, pulling
 279         // BaseEvent& references from 'events'.
 280         // 'indent' illustrates the level of recursion.
 281         // Each time we're called, we've just retrieved an OpenEvent from 'events';
 282         // accept that as a param.
 283         // Return the CloseEvent that ends this element.
 284         const CloseEvent& process(coro_t::pull_type& events,const OpenEvent& context,
 285                                   const std::string& indent=""){
 286            // Capture OpenEvent's tag name: as soon as we advance the parser, the
 287            // TagType& reference bound in this OpenEvent will be invalidated.
 288            xml::sax::Parser::TagType tagName = context.mName;
 289            // Since the OpenEvent is still the current value from 'events', pass
 290            // control back to 'events' until the next event. Of course, each time we
 291            // come back we must check for the end of the results stream.
 292            while(events()){
 293                // Another event is pending; retrieve it.
 294                const BaseEvent& event=events.get();
 295                const OpenEvent* oe;
 296                const CloseEvent* ce;
 297                const TextEvent* te;
 298                if((oe=dynamic_cast<const OpenEvent*>(&event))){
 299                    // When we see OpenEvent, recursively process it.
 300                    process(events,*oe,indent+"    ");
 301                }
 302                else if((ce=dynamic_cast<const CloseEvent*>(&event))){
 303                    // When we see CloseEvent, validate its tag name and then return
 304                    // it. (This assert is really a check on xml::sax::Parser, since
 305                    // it already validates matching open/close tags.)
 306                    assert(ce->mName == tagName);
 307                    return *ce;
 308                }
 309                else if((te=dynamic_cast<const TextEvent*>(&event))){
 310                    // When we see TextEvent, just report its text, along with
 311                    // indentation indicating recursion level.
 312                    std::cout<<indent<<"text: '"<<te->mText.getText()<<"'\n";
 313                }
 314            }
 315         }
 316
 317         // pretend we have an XML file of arbitrary size
 318         std::istringstream in(doc);
 319         try
 320         {
 321            coro_t::pull_type events(std::bind(parser,_1,std::ref(in)));
 322            // We fully expect at least ONE event.
 323            assert(events);
 324            // This dynamic_cast<&> is itself an assertion that the first event is an
 325            // OpenEvent.
 326            const OpenEvent& context=dynamic_cast<const OpenEvent&>(events.get());
 327            process(events, context);
 328         }
 329         catch (std::exception& e)
 330         {
 331            std::cout << "Parsing error: " << e.what() << '\n';
 332         }
 333
 334 This problem does not map at all well to communicating between independent
 335 threads. It makes no sense for either side to proceed independently of the
 336 other. You want them to pass control back and forth.
 337
 338 The solution involves a small polymorphic class event hierarchy, to which
 339 we're passing references. The actual instances are temporaries on the
 340 coroutine's stack; the coroutine passes each reference in turn to the main
 341 logic. Copying them as base-class values would slice them.
 342
 343 If we were trying to let the SAX parser proceed independently of the consuming
 344 logic, one could imagine allocating event-subclass instances on the heap,
 345 passing them along on a thread-safe queue of pointers. But that doesn't work
 346 either, because these event classes bind references passed by the SAX parser.
 347 The moment the parser moves on, those references become invalid.
 348
 349 Instead of binding a ['TagType&] reference, we could store a copy of
 350 the ['TagType] in ['CloseEvent]. But that doesn't solve the whole
 351 problem. For attributes, we get an ['AttributeIterator&]; for text we get
 352 a ['CharIterator&]. Storing a copy of those iterators is pointless: once
 353 the parser moves on, those iterators are invalidated. You must process the
 354 attribute iterator (or character iterator) during the SAX callback for that
 355 event.
 356
 357 Naturally we could retrieve and store a copy of every attribute and its value;
 358 we could store a copy of every chunk of text. That would effectively be all
 359 the text in the document -- a heavy price to pay, if the reason we're using
 360 SAX is concern about fitting the entire DOM into memory.
 361
 362 There's yet another advantage to using coroutines. This SAX parser throws an
 363 exception when parsing fails. With a coroutine implementation, you need only
 364 wrap the calling code in try/catch.
 365
 366 With communicating threads, you would have to arrange to catch the exception
 367 and pass along the exception pointer on the same queue you're using to deliver
 368 the other events. You would then have to rethrow the exception to unwind the
 369 recursive document processing.
 370
 371 The coroutine solution maps very naturally to the problem space.
 372
 373
 374 [heading 'same fringe' problem]
 375
 376 The advantages of suspending at an arbitrary call depth can be seen
 377 particularly clearly with the use of a recursive function, such as traversal
 378 of trees.
 379 If traversing two different trees in the same deterministic order produces the
 380 same list of leaf nodes, then both trees have the same fringe.
 381
 382 [$../../../../libs/coroutine/doc/images/same_fringe.png [align center]]
 383
 384 Both trees in the picture have the same fringe even though the structure of the
 385 trees is different.
 386
 387 The same fringe problem could be solved using coroutines by iterating over the
 388 leaf nodes and comparing this sequence via ['std::equal()]. The range of data
 389 values is generated by function ['traverse()] which recursively traverses the
 390 tree and passes each node's data value to its __push_coro__.
 391 __push_coro__ suspends the recursive computation and transfers the data value to
 392 the main execution context.
 393 __pull_coro_it__, created from __pull_coro__, steps over those data values and
 394 delivers them to  ['std::equal()] for comparison. Each increment of
 395 __pull_coro_it__ resumes ['traverse()]. Upon return from
 396 ['iterator::operator++()], either a new data value is available, or tree
 397 traversal is finished (iterator is invalidated).
 398
 399 In effect, the coroutine iterator presents a flattened view of the recursive
 400 data structure.
 401
 402         struct node{
 403             typedef boost::shared_ptr<node> ptr_t;
 404
 405             // Each tree node has an optional left subtree,
 406             // an optional right subtree and a value of its own.
 407             // The value is considered to be between the left
 408             // subtree and the right.
 409             ptr_t       left,right;
 410             std::string value;
 411
 412             // construct leaf
 413             node(const std::string& v):
 414                 left(),right(),value(v)
 415             {}
 416             // construct nonleaf
 417             node(ptr_t l,const std::string& v,ptr_t r):
 418                 left(l),right(r),value(v)
 419             {}
 420
 421             static ptr_t create(const std::string& v){
 422                 return ptr_t(new node(v));
 423             }
 424
 425             static ptr_t create(ptr_t l,const std::string& v,ptr_t r){
 426                 return ptr_t(new node(l,v,r));
 427             }
 428         };
 429
 430         node::ptr_t create_left_tree_from(const std::string& root){
 431             /* --------
 432                  root
 433                  / \
 434                 b   e
 435                / \
 436               a   c
 437              -------- */
 438             return node::create(
 439                     node::create(
 440                         node::create("a"),
 441                         "b",
 442                         node::create("c")),
 443                     root,
 444                     node::create("e"));
 445         }
 446
 447         node::ptr_t create_right_tree_from(const std::string& root){
 448             /* --------
 449                  root
 450                  / \
 451                 a   d
 452                    / \
 453                   c   e
 454                -------- */
 455             return node::create(
 456                     node::create("a"),
 457                     root,
 458                     node::create(
 459                         node::create("c"),
 460                         "d",
 461                         node::create("e")));
 462         }
 463
 464         // recursively walk the tree, delivering values in order
 465         void traverse(node::ptr_t n,
 466                       boost::coroutines::asymmetric_coroutine<std::string>::push_type& out){
 467             if(n->left) traverse(n->left,out);
 468             out(n->value);
 469             if(n->right) traverse(n->right,out);
 470         }
 471
 472         // evaluation
 473         {
 474             node::ptr_t left_d(create_left_tree_from("d"));
 475             boost::coroutines::asymmetric_coroutine<std::string>::pull_type left_d_reader(
 476                 [&]( boost::coroutines::asymmetric_coroutine<std::string>::push_type & out){
 477                     traverse(left_d,out);
 478                 });
 479
 480             node::ptr_t right_b(create_right_tree_from("b"));
 481             boost::coroutines::asymmetric_coroutine<std::string>::pull_type right_b_reader(
 482                 [&]( boost::coroutines::asymmetric_coroutine<std::string>::push_type & out){
 483                     traverse(right_b,out);
 484                 });
 485
 486             std::cout << "left tree from d == right tree from b? "
 487                       << std::boolalpha
 488                       << std::equal(boost::begin(left_d_reader),
 489                                     boost::end(left_d_reader),
 490                                     boost::begin(right_b_reader))
 491                       << std::endl;
 492         }
 493         {
 494             node::ptr_t left_d(create_left_tree_from("d"));
 495             boost::coroutines::asymmetric_coroutine<std::string>::pull_type left_d_reader(
 496                 [&]( boost::coroutines::asymmetric_coroutine<std::string>::push_type & out){
 497                     traverse(left_d,out);
 498                 });
 499
 500             node::ptr_t right_x(create_right_tree_from("x"));
 501             boost::coroutines::asymmetric_coroutine<std::string>::pull_type right_x_reader(
 502                 [&]( boost::coroutines::asymmetric_coroutine<std::string>::push_type & out){
 503                     traverse(right_x,out);
 504                 });
 505
 506             std::cout << "left tree from d == right tree from x? "
 507                       << std::boolalpha
 508                       << std::equal(boost::begin(left_d_reader),
 509                                     boost::end(left_d_reader),
 510                                     boost::begin(right_x_reader))
 511                       << std::endl;
 512         }
 513         std::cout << "Done" << std::endl;
 514
 515         output:
 516         left tree from d == right tree from b? true
 517         left tree from d == right tree from x? false
 518         Done
 519
 520
 521 [heading merging two sorted arrays]
 522
 523 This example demonstrates how symmetric coroutines merge two sorted arrays.
 524
 525         std::vector<int> merge(const std::vector<int>& a,const std::vector<int>& b){
 526             std::vector<int> c;
 527             std::size_t idx_a=0,idx_b=0;
 528             boost::coroutines::symmetric_coroutine<void>::call_type *other_a=0,*other_b=0;
 529
 530             boost::coroutines::symmetric_coroutine<void>::call_type coro_a(
 531                 [&](boost::coroutines::symmetric_coroutine<void>::yield_type& yield){
 532                     while(idx_a<a.size()){
 533                         if(b[idx_b]<a[idx_a])    // test if element in array b is less than in array a
 534                             yield(*other_b);     // yield to coroutine coro_b
 535                         c.push_back(a[idx_a++]); // add element to final array
 536                     }
 537                     // add remaining elements of array b
 538                     while(idx_b<b.size())
 539                         c.push_back(b[idx_b++]);
 540                 });
 541
 542             boost::coroutines::symmetric_coroutine<void>::call_type coro_b(
 543                 [&](boost::coroutines::symmetric_coroutine<void>::yield_type& yield){
 544                     while(idx_b<b.size()){
 545                         if(a[idx_a]<b[idx_b])    // test if element in array a is less than in array b
 546                             yield(*other_a);     // yield to coroutine coro_a
 547                         c.push_back(b[idx_b++]); // add element to final array
 548                     }
 549                     // add remaining elements of array a
 550                     while(idx_a<a.size())
 551                         c.push_back(a[idx_a++]);
 552                 });
 553
 554
 555             other_a=&coro_a;
 556             other_b=&coro_b;
 557
 558             coro_a(); // enter coroutine-fn of coro_a
 559
 560             return c;
 561         }
 562
 563
 564 [heading chaining coroutines]
 565
 566 This code shows how coroutines could be chained.
 567
 568         typedef boost::coroutines::asymmetric_coroutine<std::string> coro_t;
 569
 570         // deliver each line of input stream to sink as a separate string
 571         void readlines(coro_t::push_type& sink,std::istream& in){
 572             std::string line;
 573             while(std::getline(in,line))
 574                 sink(line);
 575         }
 576
 577         void tokenize(coro_t::push_type& sink, coro_t::pull_type& source){
 578             // This tokenizer doesn't happen to be stateful: you could reasonably
 579             // implement it with a single call to push each new token downstream. But
 580             // I've worked with stateful tokenizers, in which the meaning of input
 581             // characters depends in part on their position within the input line.
 582             BOOST_FOREACH(std::string line,source){
 583                 std::string::size_type pos=0;
 584                 while(pos<line.length()){
 585                     if(line[pos]=='"'){
 586                         std::string token;
 587                         ++pos;              // skip open quote
 588                         while(pos<line.length()&&line[pos]!='"')
 589                             token+=line[pos++];
 590                         ++pos;              // skip close quote
 591                         sink(token);        // pass token downstream
 592                     } else if (std::isspace(line[pos])){
 593                         ++pos;              // outside quotes, ignore whitespace
 594                     } else if (std::isalpha(line[pos])){
 595                         std::string token;
 596                         while (pos < line.length() && std::isalpha(line[pos]))
 597                             token += line[pos++];
 598                         sink(token);        // pass token downstream
 599                     } else {                // punctuation
 600                         sink(std::string(1,line[pos++]));
 601                     }
 602                 }
 603             }
 604         }
 605
 606         void only_words(coro_t::push_type& sink,coro_t::pull_type& source){
 607             BOOST_FOREACH(std::string token,source){
 608                 if (!token.empty() && std::isalpha(token[0]))
 609                     sink(token);
 610             }
 611         }
 612
 613         void trace(coro_t::push_type& sink, coro_t::pull_type& source){
 614             BOOST_FOREACH(std::string token,source){
 615                 std::cout << "trace: '" << token << "'\n";
 616                 sink(token);
 617             }
 618         }
 619
 620         struct FinalEOL{
 621             ~FinalEOL(){
 622                 std::cout << std::endl;
 623             }
 624         };
 625
 626         void layout(coro_t::pull_type& source,int num,int width){
 627             // Finish the last line when we leave by whatever means
 628             FinalEOL eol;
 629
 630             // Pull values from upstream, lay them out 'num' to a line
 631             for (;;){
 632                 for (int i = 0; i < num; ++i){
 633                     // when we exhaust the input, stop
 634                     if (!source) return;
 635
 636                     std::cout << std::setw(width) << source.get();
 637                     // now that we've handled this item, advance to next
 638                     source();
 639                 }
 640                 // after 'num' items, line break
 641                 std::cout << std::endl;
 642             }
 643         }
 644
 645         // For example purposes, instead of having a separate text file in the
 646         // local filesystem, construct an istringstream to read.
 647         std::string data(
 648             "This is the first line.\n"
 649             "This, the second.\n"
 650             "The third has \"a phrase\"!\n"
 651             );
 652
 653         {
 654             std::cout << "\nfilter:\n";
 655             std::istringstream infile(data);
 656             coro_t::pull_type reader(boost::bind(readlines, _1, boost::ref(infile)));
 657             coro_t::pull_type tokenizer(boost::bind(tokenize, _1, boost::ref(reader)));
 658             coro_t::pull_type filter(boost::bind(only_words, _1, boost::ref(tokenizer)));
 659             coro_t::pull_type tracer(boost::bind(trace, _1, boost::ref(filter)));
 660             BOOST_FOREACH(std::string token,tracer){
 661                 // just iterate, we're already pulling through tracer
 662             }
 663         }
 664
 665         {
 666             std::cout << "\nlayout() as coroutine::push_type:\n";
 667             std::istringstream infile(data);
 668             coro_t::pull_type reader(boost::bind(readlines, _1, boost::ref(infile)));
 669             coro_t::pull_type tokenizer(boost::bind(tokenize, _1, boost::ref(reader)));
 670             coro_t::pull_type filter(boost::bind(only_words, _1, boost::ref(tokenizer)));
 671             coro_t::push_type writer(boost::bind(layout, _1, 5, 15));
 672             BOOST_FOREACH(std::string token,filter){
 673                 writer(token);
 674             }
 675         }
 676
 677         {
 678             std::cout << "\nfiltering output:\n";
 679             std::istringstream infile(data);
 680             coro_t::pull_type reader(boost::bind(readlines,_1,boost::ref(infile)));
 681             coro_t::pull_type tokenizer(boost::bind(tokenize,_1,boost::ref(reader)));
 682             coro_t::push_type writer(boost::bind(layout,_1,5,15));
 683             // Because of the symmetry of the API, we can use any of these
 684             // chaining functions in a push_type coroutine chain as well.
 685             coro_t::push_type filter(boost::bind(only_words,boost::ref(writer),_1));
 686             BOOST_FOREACH(std::string token,tokenizer){
 687                 filter(token);
 688             }
 689         }
 690
 691 [endsect]