[boost] [threadpool] new version with fork/join semantics (context swaps)

Subject: [boost] [threadpool] new version with fork/join semantics (context swaps)
From: k-oli_at_[hidden]
Date: 2008-10-19 10:29:16

Hi,
I've uploaded a new version of threadpool (v8). It supports fibers (context 
swapping). Recursiv algorithms will now not block (see example below) -> you 
could recursivly calculate fibonacci number with one thread.

I could verify that fiber-support works on Win XP Prof. (virtualized on Linux) 
with msvc 9.0 and OpenSolaris/Nexenta (also virtualized) with gcc-4.0.

I get segmentation fault if I run the code on Linux (kernel 2.6.26) with 
gcc-4.3:

#0  0x000000000041009b in 
boost::_bi::list1<boost::_bi::value<boost::tp::detail::pool_with_fibers<boost::tp::unbounded_channel<boost::tp::fifo> 
>::worker::wimpl::fiber_wrapper*> >::operator()<boost::_mfi::mf0<void, 
boost::tp::detail::pool_with_fibers<boost::tp::unbounded_channel<boost::tp::fifo> 
>::worker::wimpl::fiber_wrapper>, boost::_bi::list0> 
(this=0xffffffffd4000c28, f=@0xffffffffd4000c18, a=@0x7f81d400535f)
    at /opt/boost/include/boost-1_36/boost/bind.hpp:232
#1  0x00000000004100f4 in boost::_bi::bind_t<void, boost::_mfi::mf0<void, 
boost::tp::detail::pool_with_fibers<boost::tp::unbounded_channel<boost::tp::fifo> 
>::worker::wimpl::fiber_wrapper>, 
boost::_bi::list1<boost::_bi::value<boost::tp::detail::pool_with_fibers<boost::tp::unbounded_channel<boost::tp::fifo> 
>::worker::wimpl::fiber_wrapper*> > >::operator() (this=0xffffffffd4000c18)
    at /opt/boost/include/boost-1_36/boost/bind/bind_template.hpp:20
#2  0x0000000000416c34 in 
boost::tp::detail::wrapp_functor<boost::tp::detail::fiber_impl_wrapper<boost::_bi::bind_t<void, 
boost::_mfi::mf0<void, 
boost::tp::detail::pool_with_fibers<boost::tp::unbounded_channel<boost::tp::fifo> 
>::worker::wimpl::fiber_wrapper>, 
boost::_bi::list1<boost::_bi::value<boost::tp::detail::pool_with_fibers<boost::tp::unbounded_channel<boost::tp::fifo> 
>::worker::wimpl::fiber_wrapper*> > > > > (
    vp=0xffffffffd4000c10) 
at /home/graemer/Projekte/thread_pool/src/boost/tp/detail/fiber_impl_posix.hpp:29
#3  0x00007f81da2cd7e0 in ?? () from /lib/libc.so.6
#4  0x0000000000000000 in ?? ()

I hope you can help me to fix this error (maybe it's raised by a design bug).

regards,
Oliver

example:

#include <iostream>
#include <cstdlib>
#include <stdexcept>
#include <vector>

#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/interprocess/detail/atomic.hpp>
#include <boost/future/future.hpp>
#include <boost/ref.hpp>

#include "boost/tp/fifo.hpp"
#include "boost/tp/pool.hpp"
#include "boost/tp/poolsize.hpp"
#include "boost/tp/support.hpp"
#include "boost/tp/stacksize.hpp"
#include "boost/tp/unbounded_channel.hpp"

namespace pt = boost::posix_time;
namespace tp = boost::tp;

typedef tp::pool<
        tp::unbounded_channel< tp::fifo >,
        tp::with_fibers
> pool_type;

class fib
{
private:
        pool_type	&	pool_;
        int				offset_;
        volatile boost::uint32_t	&	i_;

        int seq_( int n)
        {
                if ( n <= 1) return n;
                else return seq_( n - 2) + seq_( n - 1);
        }
        
        int par_( int n)
        {
                if ( n <= offset_) return seq_( n);
                else
                {
                        tp::task< int > t1(
                                pool_.submit(
                                        boost::bind(
                                                & fib::par_,
                                                boost::ref( * this),
                                                n - 1) ) );
                        tp::task< int > t2(
                                pool_.submit(
                                        boost::bind(
                                                & fib::par_,
                                                boost::ref( * this),
                                                n - 2) ) );
                        return t1.get() + t2.get(); // dos not block if task is not ready
                }
        }

public:
        fib( pool_type & pool, int offset, volatile boost::uint32_t & i)
        : pool_( pool), offset_( offset), i_( i)
        {}

        int execute( int n)
        {
                boost::interprocess::detail::atomic_inc32( & i_);
                int result( par_( n) );
                printf("%d: fibonacci of %d == %d\n", i_, n, result);
                return result;
        }
};

int main( int argc, char *argv[])
{
        try
        {
                pool_type pool( tp::poolsize( 2) );
                boost::uint32_t i( 0);
                fib f( pool, 1, i);
                for( int i = 0; i<100;++i)
                {
                        tp::task< int > t1(
                                pool.submit(
                                        boost::bind(
                                                & fib::execute,
                                                boost::ref( f),
                                                3) ) );
                }
                pool.shutdown();
        }
        catch ( boost::thread_interrupted const& )
        { std::cerr << "thread_interrupted: thread was interrupted" << std::endl; }
        catch ( std::exception const& e)
        { std::cerr << "exception: " << e.what() << std::endl; }
        catch ( ... )
        { std::cerr << "unhandled" << std::endl; }

        return EXIT_SUCCESS;

}