[boost] Futures revisited

From: Slawomir Lisznianski (slisznianski_at_[hidden])
Date: 2004-03-05 11:52:04

Hello,

During discussions on possible extensions to Boost.Threads, concept of
Futures came up rather frequently. Below, I’d like to introduce a
particular implementation available at www.rhapsodia.org of which
requirements were defined as follows:

1. Provide means to register exception types with a "Future Registry".

Rationale: I find it imperative that users should not have to define a
list of exception types on each invocation yielding a Future object.
Instead, provide the ability to register exception types with a Future
Registry and bundle it with Future objects. I feel that compile time
registration should be sufficient.

2. Work under the assumption that asynchronous invocations returning
Futures and system threads are not coupled. In particular, Futures
implementation should not relay on joining with a particular thread to
perform synchronization for returning results.

Rationale: Although most prevalent, the technique of joining with a
particular thread to synchronize on returning result is too limiting. It
implies that each task be run on a dedicated thread, specially created to
handle it and destroyed to meet with a caller using a Future. Although
this approach may be adequate under certain conditions, it generally
tends to be expensive on system resources and disorganizes the
application. For example, it’s impossible to control the number of
concurrently running threads. It may also perform poorly if executed code
happens to run shortly as the overhead of creating a thread and cleaning
up after it may be significant. More robust approach is to use a queue
and a thread pool for running tasks. This adds complexity on the
synchronization side, as we can no longer rely on a lifetime of a thread
to join with Future’s result.

3. Ability to utilize Futures with arbitrary code without explicit
support for them.

Rationale: For implementation to be widely adopted, it should be possible
to turn just about any function call into a Future-armored asynchronous
invocation. No explicit thread synchronization primitives should be used
as well.

4. Implement using existing Boost.Threads facilities only.

Rationale: Boost.Threads library provides enough support to implement
Futures paradigm that meets requirements listed here.

5. STL and Boost compatible. For example, storing Futures in an STL
container. Be strongly typed.

Futures implementation and usage patterns must naturally coexist with STL
and Boost code. Additionally, compile-time type validation is key to
safer code.

Attached is a sample file from Rhapsodia library showing the use of
Futures.

Regareds,
Slawomir Lisznianski [ www.rhapsodia.org ]


/*
 * Copyright (C) 2003-2004
 * Slawomir Lisznianski <slisznianski_at_[hidden]>
 *
 * Permission to use, copy, modify, distribute and sell this software
 * and its documentation for any purpose is hereby granted without fee,
 * provided that the above copyright notice appear in all copies and
 * that both that copyright notice and this permission notice appear
 * in supporting documentation. Slawomir Lisznianski makes no
 * representations about the suitability of this software for any
 * purpose. It is provided "as is" without express or implied warranty.
 *
 */

#include <iostream>
#include <algorithm>
#include <boost/bind.hpp>

#include <BasicScheduler.hpp>
#include <Futures.hpp>

/**
* We define various task simulators, with different return types;
* some will throw exception.
*/

void doThrowRuntime()
{
  std::cout << "Executing: doThrowRuntime()" << std::endl;
  throw std::runtime_error("a runtime_error exception");
}

int dontThrow(int p)
{
  std::cout << "Executing: dontThrow() with param: " << p << std::endl;
  return p;
}

struct A
{
  int fun_a(int a, int b)
  {
    return (a+b);
  }
};

std::string greeting()
{
  return "hello, world";
}

int main()
{
  try
  {
    // Create a scheduler used for queuing tasks.
    //
    Rha::BasicScheduler sch;

    // Define a Futures Registry; list exception types that can be
    // thrown during task execution.
    //
    typedef Rha::FutureRegistry<std::exception, std::runtime_error, std::logic_error> FuRe;

    /**
    * Test 1)
    * Lets start with a simple example. The task will return a
    * greeting message when "joined" with operator*.
    */

    FuRe::Future<std::string> f1(sch, greeting);

    std::cout << "Gretting: " << *f1 << std::endl;

    /**
    * Test 2)
    * This test will schedule two concurrent tasks (f1 and f2), each returning a number.
    * The returned numbers are then passed as parameters to a third, concurrently
    * run task - class A member method.
    */
    FuRe::Future<int> f2(sch, boost::bind(dontThrow, 1));
    FuRe::Future<int> f3(sch, boost::bind(dontThrow, 2));

    A a;
    FuRe::Future<int> f4(sch, boost::bind(&A::fun_a, a, *f2, *f3));

    std::cout << "3 == " << *f4 << std::endl;

    /**
    * Test 3)
    * This test demonstrates usage of Futures with STL containers
    */
    std::vector<FuRe::Future<int> > v;

    // Schedule tasks and put Futures into a vector.
    //
    for (int i = 0; i < 100; ++i)
      v.push_back(FuRe::Future<int>(sch, boost::bind(dontThrow, i)));

    // Print results of each task.
    //
    std::transform(v.begin(), v.end(),
                   std::ostream_iterator<int>(std::cout, " "),
                   boost::mem_fn(&FuRe::Future<int>::join));

    /**
    * Test 4)
    * We finish with a task that will throw an exception.
    * The exception should be intercepted and re-thrown
    * when future is joined.
    */
    FuRe::Future<void> f5(sch, doThrowRuntime);

    f5.join();

  }
  catch (std::exception& e)
  {
    std::cout << "Exception caught: " << e.what() << std::endl;
  }
  return 0;
}