[cpp-threads] Thread API interface tweaks

[Peter Dimov]

Howard Hinnant wrote:

[...]

> However Herb spoke eloquently of his "futures" view of threads where
> one regards the future as merely a reference to the return value of
> the thread, and not to the thread itself.  In this viewpoint, the
> reference is logically copyable with reference semantics.  After an
> extremely productive 45-second conversation between Herb and myself
> we came up with the possibility of a layered approach where the
> client might have the option for both view points using different
> types.

Right. I've also been thinking along the same lines. I never got to a formal 
spec, but here's a quick summary to give you an idea:

template<class R> class future
{
public:

 future();

 bool ready() const;

 void wait();
 bool timed_wait( boost::xtime const & xt );

 R const & operator()() const;

 void set_value( R const & r );

 template<class E> void set_exception(); // throw()
 template<class E> void set_exception( char const * what ); // throw()
};

The semantics are "obvious"; set_value and set_exception switch to a ready 
state, and operator() returns the value or throws the exception. wait blocks 
until ready, operator() also blocks. future<> has reference semantics 
(basically a shared_ptr<future_impl>.)

Note that this class is independent of the underlying threading library and 
is not tied to its design. You can put a result into it, and you can get it 
back. Cancellation complicates matters a bit, and so does "multiple wait" 
support, and I haven't implemented these yet.

This is how a future is used:

class synchronous_executor
{
public:

    template<class Fn> future< typename boost::result_of<Fn()>::type > 
execute( Fn fn )
    {
        typedef typename boost::result_of<Fn()>::type R;

        future<R> ft;
        future_wrapper<R, Fn> fw( fn, ft );

        fw();

        return ft;
    }
};

class threaded_executor
{
public:

    template<class Fn> future< typename boost::result_of<Fn()>::type > 
execute( Fn fn )
    {
        typedef typename boost::result_of<Fn()>::type R;

        future<R> ft;
        future_wrapper<R, Fn> fw( fn, ft );

        boost::thread th( fw );

        return ft;
    }
};

I've omitted future_wrapper for brevity (it's at the end), but a basic 
implementation is trivial. synchronous_executor executes its tasks 
sequentially, threaded_executor spawns a thread per task. The syntax is the 
same:

int main()
{
    typedef pooled_executor Executor;

    Executor ex;

    future<double> a = ex.execute( boost::bind( f, 1000000 ) );
    future<double> b = ex.execute( boost::bind( f, 5000000 ) );

    try
    {
        std::cout << b() << " - " << a() << " = " << b() - a() << std::endl;
    }
    catch( std::exception const & x )
    {
        std::cout << "std::exception: " << x.what() << std::endl;
    }
}

Even though the prototype uses boost::thread for spawning threads, it can be 
adapted to any syntax. Note that futures and executors require no special 
support from the thread API. The thread API only spawns threads. This 
separation of concerns is, IMO, the right approach.

And here's a pooled_executor for reference:

class pooled_executor
{
private:

    pooled_executor( pooled_executor const & );
    pooled_executor & operator=( pooled_executor const & );

    std::list< boost::function< void() > > queue_;

    boost::mutex mtx_;
    boost::condition cnd_;

    boost::thread_group pool_;

public:

    void thread_proc()
    {
        for( ;; )
        {
            boost::mutex::scoped_lock lock( mtx_ );

            while( queue_.empty() )
            {
                cnd_.wait( lock );
            }

            boost::function< void() > fn = queue_.front();
            queue_.pop_front();

            lock.unlock();

            fn();
        }
    }

    pooled_executor()
    {
        for( int i = 0; i < 4; ++i )
        {
            pool_.create_thread( boost::bind( &pooled_executor::thread_proc, 
this ) );
        }
    }

    template<class Fn> future< typename boost::result_of<Fn()>::type > 
execute( Fn fn )
    {
        typedef typename boost::result_of<Fn()>::type R;

        future<R> ft;
        future_wrapper<R, Fn> fw( fn, ft );

        boost::mutex::scoped_lock lock( mtx_ );
        queue_.push_back( fw );
        cnd_.notify_one();

        return ft;
    }
};

Finally, a basic implementation of future_wrapper:

template<class R, class Fn> class future_wrapper
{
private:

    Fn fn_;
    future<R> ft_;

public:

    future_wrapper( Fn fn, future<R> const & ft ): fn_( fn ), ft_( ft )
    {
    }

    void operator()() // throw()
    {
        try
        {
            ft_.set_value( fn_() );
        }
        catch( std::bad_alloc const & )
        {
            ft_.template set_exception<std::bad_alloc>();
        }
        catch( std::logic_error const & x )
        {
            ft_.template set_exception<std::logic_error>( x.what() );
        }
        catch( std::exception const & x )
        {
            ft_.template set_exception<std::runtime_error>( x.what() );
        }
        catch( ... )
        {
            ft_.template set_exception<std::bad_exception>();
        }
    }
};