parallel.hpp

Go to the documentation of this file.

/*
 * Copyright 2008,2009,2010,2011,2012 Stephan Ossowski, Korbinian Schneeberger,
 * Felix Ott, Joerg Hagmann, Alf Scotland, Sebastian Bender
 * 
 * This file is part of SHORE.
 * 
 * SHORE is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * SHORE is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with SHORE.  If not, see <http://www.gnu.org/licenses/>.
 */


#ifndef SHORE_PARALLEL_PARALLEL_HPP__
#define SHORE_PARALLEL_PARALLEL_HPP__

#include <iostream>
#include <deque>
#include <stdexcept>
#include <map>
#include <vector>

#include <boost/thread/thread.hpp>
#include <boost/thread/recursive_mutex.hpp>
#include <boost/thread/condition.hpp>

#ifdef WITH_MPI
  #include <boost/serialization/deque.hpp>
#endif // WITH_MPI


#include "shore/processing/feed.hpp"
#include "shore/processing/pipeline.hpp"
#include "shore/parallel/message_passing.hpp"
#include "shore/parallel/message_passing_proxy.hpp"
#include "shore/parallel/thread.hpp"
#include "shore/parallel/thread_index.hpp"
#include "shore/base/memops.hpp"

namespace shore {

class parallelization_core_access;

class parallel_connection_init
{
 public:

        virtual ~parallel_connection_init() {}
        virtual void init()=0;
};

class parallelization_base
{
 protected:

        virtual void parallelization_setup()=0;

        int m_parallelization_numthreads;

 private:

        friend class parallelization_core_access;

        std::vector<parallel_connection_init *> m_pci;

        shore::signal<int> m_signumthreads;

        shore::slot<int> m_slotnumthreads;


        void slotfun_nthreads(const int & nthreads)
        {
                if(shore::mpi::comm_size()>1)
                        m_parallelization_numthreads=std::max(1,nthreads);
                else
                        m_parallelization_numthreads=std::max(0,nthreads);

                parallelization_setup();
                m_signumthreads.emit(m_parallelization_numthreads);

                for(size_t i=0;i<m_pci.size();++i)
                        m_pci[i]->init();
        }

        parallelization_base(const parallelization_base &);
        parallelization_base & operator=(const parallelization_base &);

 public:

        parallelization_base()
        :m_slotnumthreads(boost::bind(&parallelization_base::slotfun_nthreads,this,_1))
        {}

        virtual ~parallelization_base()
        {
                while(!m_pci.empty())
                {
                        delete m_pci.back();
                        m_pci.pop_back();
                }
        }

        int num_threads() const
        {
                return m_parallelization_numthreads;
        }
};

//          wrapped object outside the root process.
template<typename T>
class serial
:public parallelization_base
{
 private:

        friend class parallelization_core_access;

        typedef T object_type;

        class lock_link
        {
         private:

                shore::slot<int> m_slotpostlock;
                shore::slot<int> m_slotpreunlock;
                shore::slot<int> m_slotpostunlock;

                shore::slot<void> m_slotflushthread;

                shore::signal<int> m_sigpostlock;
                shore::signal<int> m_sigpreunlock;
                shore::signal<int> m_sigpostunlock;

                shore::signal<void> m_sigflushthread;

         public:

                lock_link()
                :m_slotpostlock(boost::ref(m_sigpostlock.emit)),
                 m_slotpreunlock(boost::ref(m_sigpreunlock.emit)),
                 m_slotpostunlock(boost::ref(m_sigpostunlock.emit)),
                 m_slotflushthread(boost::ref(m_sigflushthread.emit))
                {}

                shore::slot<int> &slotpostlock()
                {
                        return m_slotpostlock;
                }

                shore::slot<int> &slotpreunlock()
                {
                        return m_slotpreunlock;
                }

                shore::slot<int> &slotpostunlock()
                {
                        return m_slotpostunlock;
                }

                shore::slot<void> &slotflushthread()
                {
                        return m_slotflushthread;
                }

                shore::signal<int> &sigpostlock()
                {
                        return m_sigpostlock;
                }

                shore::signal<int> &sigpreunlock()
                {
                        return m_sigpreunlock;
                }

                shore::signal<int> &sigpostunlock()
                {
                        return m_sigpostunlock;
                }

                shore::signal<void> &sigflushthread()
                {
                        return m_sigflushthread;
                }
        };

        object_type *m_object;
        std::vector<lock_link *> m_locklinks;

        shore::slot<void> m_slotjoin;
        shore::signal<void> m_sigjoin;


        serial(const serial &);
        serial &operator=(const serial &);

 protected:

        virtual void parallelization_setup()
        {
                while(!m_locklinks.empty())
                {
                        delete m_locklinks.back();
                        m_locklinks.pop_back();
                }

                for(int i=0;i<num_threads();++i)
                        m_locklinks.push_back(new lock_link);
        }

 public:

        serial()
        :m_slotjoin(boost::ref(m_sigjoin.emit))
        {
                if(shore::mpi::comm_isroot())
                        m_object=new object_type;
                else
                        m_object=0;
        }

        template<typename A>
        serial(const A &arg)
        :m_slotjoin(boost::ref(m_sigjoin.emit))
        {
                if(shore::mpi::comm_isroot())
                        m_object=new object_type(arg);
                else
                        m_object=0;
        }

        template<typename A1,typename A2>
        serial(const A1 & arg1,const A2 & arg2)
        :m_slotjoin(boost::ref(m_sigjoin.emit))
        {
                if(shore::mpi::comm_isroot())
                        m_object=new object_type(arg1,arg2);
                else
                        m_object=0;
        }

        template<typename A1,typename A2,typename A3>
        serial(const A1 & arg1,const A2 & arg2,const A3 & arg3)
        :m_slotjoin(boost::ref(m_sigjoin.emit))
        {
                if(shore::mpi::comm_isroot())
                        m_object=new object_type(arg1,arg2,arg3);
                else
                        m_object=0;
        }

        ~serial()
        {
                while(!m_locklinks.empty())
                {
                        delete m_locklinks.back();
                        m_locklinks.pop_back();
                }

                delete m_object;
        }

        template<typename R>
        R call(R (object_type::*func)())
        {
                if(m_object)
                        return (m_object->*func)();
                return R();
        }

        template<typename R,typename A>
        R call(R (object_type::*func)(const A &),const A &arg)
        {
                if(m_object)
                        return (m_object->*func)(arg);
                return R();
        }

        template<typename R,typename A>
        R call(R (object_type::*func)(const A),const A &arg)
        {
                if(m_object)
                        return (m_object->*func)(arg);
                return R();
        }
};

template<typename T>
class parallel
:public parallelization_base
{
 private:

        friend class parallelization_core_access;

        typedef T object_type;


        struct factory_base
        {
                virtual ~factory_base() {}
                virtual object_type *create()=0;
        };

        template<typename Obj>
        struct arg0_factory
        :public factory_base
        {
                virtual Obj *create()
                {
                        return new Obj;
                }
        };

        template<typename Obj,typename Arg>
        struct arg1_factory
        :public factory_base
        {
                Arg arg;

                arg1_factory(const Arg & a)
                :arg(a)
                {}

                virtual Obj *create()
                {
                        return new Obj(arg);
                }
        };


        std::vector<object_type *> m_objects;
        factory_base * m_factory;

        shore::slot<void> m_slotjoin;
        shore::signal<void> m_sigjoin;

 protected:

        virtual void parallelization_setup()
        {
                while(!m_objects.empty())
                {
                        delete m_objects.back();
                        m_objects.pop_back();
                }

                m_objects.push_back(m_factory->create());
                for(int i=1;i<num_threads();++i)
                        m_objects.push_back(m_factory->create());
        }


 public:

        parallel()
        :m_slotjoin(boost::ref(m_sigjoin.emit))
        {
                m_factory=new arg0_factory<object_type>;
        }

        template<typename A>
        parallel(const A &arg)
        :m_slotjoin(boost::ref(m_sigjoin.emit))
        {
                m_factory=new arg1_factory<object_type,A>(arg);
        }

        ~parallel()
        {
                while(!m_objects.empty())
                {
                        delete m_objects.back();
                        m_objects.pop_back();
                }
                delete m_factory;
        }
};

template<typename T>
class sync
:public parallelization_base
{
 public:

        typedef T append_type;
        typedef T current_type;

 private:

        friend class parallelization_core_access;

        typedef boost::recursive_mutex mutex_type;
        typedef boost::recursive_mutex::scoped_lock lock_type;
        typedef boost::condition condition_type;

        class track;
    

        mutex_type m_mutex;
        condition_type m_flush_condition;

        shore::slot<void> m_slotjoin;
        shore::signal<void> m_sigjoin;

        std::set<int> m_flushed_threads;
        std::set<int> m_flushed_processes;
        shore::signal<void> m_sigsyncflush;

        std::vector<track *> m_thread_tracks;
        std::vector<track *> m_proxy_tracks;

        void slotfun_join()
        {
         #ifdef WITH_MPI
                if(shore::mpi::comm_isroot())
                {
                        lock_type lock(m_mutex);

                        while(m_flushed_processes.size()<(shore::mpi::comm_size()-1))
                                m_flush_condition.wait(lock);

                        m_data_submitter.join();
                        m_flush_submitter.join();
                }
         #endif // WITH_MPI
                m_sigjoin.emit();
        }

 #ifdef WITH_MPI
        std::vector<shore::feed<append_type> > m_proxy_feeds;

        mpi_proxy::submitter<append_type> m_data_submitter;
        shore::slot<const append_type &> m_slotproxy_data;

        mpi_proxy::submitter<int> m_flush_submitter;
        shore::slot<const int &> m_slotproxy_flush;

        void slotfun_proxyappend(const append_type &d)
        {
                m_proxy_feeds[m_data_submitter.get_source()-1].append(d);
        }

        void slotfun_proxyflush(const int &)
        {
                m_proxy_feeds[m_flush_submitter.get_source()-1].flush();
                m_flush_condition.notify_all();
        }
 #endif // WITH_MPI


        sync(const sync &);
        sync &operator=(const sync &);


        class track
        :public shore::pipe_facade<track,append_type,current_type,true>
        {
         private:
    
                friend class shore::pipeline_core_access;
                friend class parallelization_core_access;

                sync &m_sync;
                lock_type *m_lock;

                int m_proxied_process;

                shore::signal<int> m_sigpostlock;
                shore::signal<int> m_sigpreunlock;
                shore::signal<int> m_sigpostunlock;

                track(const track &);
                track &operator=(const track &);

    
                void acquire_lock()
                {
                        m_lock=new lock_type(m_sync.m_mutex);
                        m_sigpostlock.emit(m_proxied_process);
                }

                void release_lock()
                {
                        m_sigpreunlock.emit(m_proxied_process);

                        delete m_lock;
                        m_lock=0;

                        m_sigpostunlock.emit(m_proxied_process);
                }

                //void next() {}

                //void prepare(const append_type &d) {}

                void append(const append_type &d)
                {
                        acquire_lock();

                        try
                        {

                 #ifdef WITH_MPI
                                if(!shore::mpi::comm_isroot())
                                        m_sync.m_data_submitter.submit(d);
                 #endif // WITH_MPI

                                this->emit(d);
                        }
                        catch(...)
                        {
                                delete m_lock;
                                m_lock=0;
                                throw;
                        }

                        release_lock();
                }

                void flush()
                {
                        if(m_sync.num_threads()==0)
                                return;

                        // lock, check if all threads are flushed;
                        // if so, flush the sync'ed range using this thread
                        acquire_lock();

                        try
                        {
                                if(m_proxied_process==0)
                                        m_sync.m_flushed_threads.insert(thread_index::id());
                         #ifdef WITH_MPI
                                else
                                        m_sync.m_flushed_processes.insert(m_proxied_process);
                         #endif // WITH_MPI

                                const int unflushed_threads=
                                        m_sync.num_threads()-int(m_sync.m_flushed_threads.size());

                                int unflushed_processes=0;

                         #ifdef WITH_MPI
                                // Also need to have the other MPI processes flushed
                                unflushed_processes=
                                        shore::mpi::comm_size()-int(m_sync.m_flushed_processes.size())-1;
                         #endif // WITH_MPI

                                if(shore::mpi::comm_isroot())
                                {
                                        if((unflushed_threads==0)&&(unflushed_processes==0))
                                                m_sync.m_sigsyncflush.emit();
                                }
                         #ifdef WITH_MPI
                                else if(unflushed_threads==0)
                                        m_sync.m_flush_submitter.submit(0);
                         #endif // WITH_MPI

                        }
                        catch(...)
                        {
                                delete m_lock;
                                m_lock=0;
                                throw;
                        }

                        release_lock();

                        // now unlocked and pipe_facade will flush this thread
                }


         public:

                track(sync &s,const int pr)
                :m_sync(s),
                 m_lock(0),
                 m_proxied_process(pr)
                {}
        };

 protected:

        virtual void parallelization_setup()
        {
                while(!m_thread_tracks.empty())
                {
                        delete m_thread_tracks.back();
                        m_thread_tracks.pop_back();
                }

                m_thread_tracks.push_back(new track(*this,0));
                for(int i=1;i<num_threads();++i)
                        m_thread_tracks.push_back(new track(*this,0));
        }

 public:

        sync()
        :m_slotjoin(boost::ref(m_sigjoin.emit))
 #ifdef WITH_MPI
        ,
        m_slotproxy_data(boost::bind(&sync::slotfun_proxyappend,this,_1)),
        m_slotproxy_flush(boost::bind(&sync::slotfun_proxyflush,this,_1))
 #endif // WITH_MPI
        {
         #ifdef WITH_MPI
                if(shore::mpi::comm_isroot())
                {
                        m_data_submitter.sigdata().connect(m_slotproxy_data);
                        m_flush_submitter.sigdata().connect(m_slotproxy_flush);

                        m_proxy_feeds.resize(shore::mpi::comm_size()-1);
                        for(size_t i=1;i<shore::mpi::comm_size();++i)
                        {
                                m_proxy_tracks.push_back(new track(*this,i));
                                m_proxy_feeds[i-1]|(*m_proxy_tracks.back());
                        }
                }
         #endif // WITH_MPI
        }

        ~sync()
        {
                while(!m_thread_tracks.empty())
                {
                        delete m_thread_tracks.back();
                        m_thread_tracks.pop_back();
                }
         #ifdef WITH_MPI
                while(!m_proxy_tracks.empty())
                {
                        delete m_proxy_tracks.back();
                        m_proxy_tracks.pop_back();
                }
                m_data_submitter.unsubscribe();
                m_flush_submitter.unsubscribe();
         #endif // WITH_MPI
        }

        track &operator[](const size_t i)
        {
                return *m_thread_tracks[i];
        }
};

template<typename T>
class desync
:public parallelization_base
{
 private:

        friend class parallelization_core_access;


        typedef T append_type;
        typedef T current_type;

        typedef boost::recursive_mutex mutex_type;
        typedef boost::recursive_mutex::scoped_lock lock_type;


        class lock_buffer;
        class multiplexer;

        shore::slot<void> m_slotjoin;
        shore::signal<void> m_sigjoin;

        mutex_type m_mutex;

        multiplexer m_multiplexer;

        std::vector<lock_buffer *> m_thread_tracks;
        std::vector<lock_buffer *> m_process_tracks;

        std::map<boost::thread::id,int> m_bufmapping;

 #ifdef WITH_MPI
        shore::slot<std::deque<append_type> &> m_slotproxy_getdata;
        mpi_proxy::requester<std::deque<append_type> > m_data_requester;

        void slotfun_proxygetdata(std::deque<append_type> &d)
        {
                const int pr=m_data_requester.get_target();
                lock_buffer &b=get_lockbuffer(pr);
                d.swap(b.buffer());
                b.buffer().clear();
        }
 #endif // WITH_MPI


        lock_buffer &get_lockbuffer(int process)
        {
                lock_type lock(m_mutex);
                if(process>0)
                        return *m_process_tracks[process-1];

                int &thr=m_bufmapping[boost::this_thread::get_id()];
                if(thr==0)
                        thr=m_bufmapping.size();
                return *m_thread_tracks[thr-1];
        }

        class lock_buffer
        :public shore::pipe_facade<lock_buffer,append_type,current_type,true>
        {
         private:
    
                friend class shore::pipeline_core_access;
                friend class parallelization_core_access;
                friend class multiplexer;

                std::deque<append_type> m_buffer;

                void next()
                {
                        m_buffer.pop_front();

                        if(!m_buffer.empty())
                                this->emit(m_buffer.front());
                }

                //void prepare(const append_type &d) {}

                void append(const append_type &d)
                {
                        m_buffer.push_back(d);
                }

                //void flush() {}

         public:

                lock_buffer()
                {}

                std::deque<append_type> &buffer()
                {
                        return m_buffer;
                }

                bool empty() const
                {
                        return m_buffer.empty();
                }

                const append_type &front() const
                {
                        return m_buffer.front();
                }

                void pop_front()
                {
                        m_buffer.pop_front();
                }

                void dump()
                {
                        if(!m_buffer.empty())
                                this->emit(m_buffer.front());
                }
        };


        class multiplexer
        :public shore::pipe_facade<multiplexer,append_type,current_type,true>
        {
         private:

                friend class shore::pipeline_core_access;
                friend class parallelization_core_access;



                desync &m_desync;

                shore::slot<int> m_slotpostlock;
                shore::slot<int> m_slotpreunlock;
                shore::slot<int> m_slotpostunlock;

                void slotfun_postlock(int pr)
                {
                        lock_buffer &b=m_desync.get_lockbuffer(pr);
                        (*this)|b;

                 #ifdef WITH_MPI
                        if(!shore::mpi::comm_isroot())
                                m_desync.m_data_requester.request(b.buffer());
                 #endif // WITH_MPI
                }

                void slotfun_preunlock(int pr)
                {
                        this->disconnect_outputs();
                }

                void slotfun_postunlock(int pr)
                {
                        // if pr!=0 i.e. the data belongs to a different process,
                        // leave it in the buffer to be retrieved remotely by
                        // the process' slotfun_postlock, and subsequently
                        // dumped there.
                        if(pr==0)
                                m_desync.get_lockbuffer(pr).dump();
                }

                //void next() {}

                //void prepare(const append_type &d) {}

                void append(const append_type &d)
                {
                        this->emit(d);
                }

                //void flush() {}

         public:

                multiplexer(desync &d)
                :m_desync(d),
                 m_slotpostlock(boost::bind(&multiplexer::slotfun_postlock,this,_1)),
                 m_slotpreunlock(boost::bind(&multiplexer::slotfun_preunlock,this,_1)),
                 m_slotpostunlock(boost::bind(&multiplexer::slotfun_postunlock,this,_1))
                {}

                shore::slot<int> &slotpostlock()
                {
                        return m_slotpostlock;
                }

                shore::slot<int> &slotpreunlock()
                {
                        return m_slotpreunlock;
                }

                shore::slot<int> &slotpostunlock()
                {
                        return m_slotpostunlock;
                }
        };


        desync(const desync &);
        desync &operator=(const desync &);

 protected:

        virtual void parallelization_setup()
        {
                while(!m_thread_tracks.empty())
                {
                        delete m_thread_tracks.back();
                        m_thread_tracks.pop_back();
                }

                for(int i=0;i<num_threads();++i)
                        m_thread_tracks.push_back(new lock_buffer);
        }

 public:

        desync()
        :m_slotjoin(boost::bind(&desync::slotfun_join,this)),
         m_multiplexer(*this)
 #ifdef WITH_MPI
        ,
         m_slotproxy_getdata(boost::bind(&desync::slotfun_proxygetdata,this,_1))
 #endif // WITH_MPI
        {
                if(shore::mpi::comm_isroot())
                        for(size_t i=1;i<shore::mpi::comm_size();++i)
                                m_process_tracks.push_back(new lock_buffer);
         #ifdef WITH_MPI
                m_data_requester.siggetdata().connect(m_slotproxy_getdata);
         #endif // WITH_MPI
        }

        ~desync()
        {
         #ifdef WITH_MPI
                m_data_requester.join();
                m_data_requester.unsubscribe();
         #endif // WITH_MPI
        }
};

template<typename T>
class parallelizer
:public parallelization_base
{
 private:

        friend class parallelization_core_access;

        typedef T append_type;
        typedef T current_type;

        typedef boost::recursive_mutex mutex_type;
        typedef boost::recursive_mutex::scoped_lock lock_type;
        typedef boost::condition condition_type;

        class multiplexer
        :public shore::pipe_facade<multiplexer,append_type,current_type,true>
        {
         private:

                friend class shore::pipeline_core_access;

                parallelizer &m_parallelizer;


                multiplexer(const multiplexer &);

                //void next() {}

                //void prepare(const append_type &d) {}

                void append(const append_type &d)
                {
                        if(m_parallelizer.m_nthreads==0)
                        {
                                this->emit(d);
                        }
                        else
                        {
                                m_parallelizer.errcheck();

                                lock_type lock(m_parallelizer.m_mutex);

                                if(m_parallelizer.data_ready())
                                        m_parallelizer.m_dataretrieved_condition.wait(lock);

                                m_parallelizer.m_input_buffer.push_back(d);

                                if(m_parallelizer.data_ready())
                                        m_parallelizer.m_dataready_condition.notify_one();
                        }
                }

                void flush()
                {
                        if(m_parallelizer.m_nthreads>0)
                        {
                                {
                                        lock_type lock(m_parallelizer.m_mutex);
                                        m_parallelizer.errcheck();

                                        m_parallelizer.m_batchsize=0;
                                        m_parallelizer.m_dataready_condition.notify_all();
                                }
                                m_parallelizer.join();
                        }
                }

         public:

                multiplexer(parallelizer &p)
                :m_parallelizer(p)
                {}
        };

        class thread
        :public shore::pipe_facade<thread,append_type,current_type,true>,
         public shore::thread
        {
         private:

                friend class shore::pipeline_core_access;

                parallelizer &m_parallelizer;

                std::deque<append_type> m_buffer;

                std::ostringstream m_errlog;

                void next()
                {
                        if(!m_buffer.empty())
                                m_buffer.pop_front();
                        if(!m_buffer.empty())
                                this->emit(m_buffer.front());
                }

                //void prepare(const append_type &d) {}

                void append(const append_type &d)
                {
                        // serial mode: just pass through
                        this->emit(d);
                }

                //void flush() {}

         protected:

                virtual void run()
                {
                        for(;;)
                        {
                                if(shore::mpi::comm_isroot())
                                {
                                        lock_type lock(m_parallelizer.m_mutex);

                                        if(m_parallelizer.m_error_thread!=0)
                                                return;

                                        while(!m_parallelizer.data_ready())
                                                m_parallelizer.m_dataready_condition.wait(lock);

                                        if(m_parallelizer.m_error_thread!=0)
                                                return;

                                        m_buffer.swap(m_parallelizer.m_input_buffer);
                                }
                 #ifdef WITH_MPI
                                else
                                        m_parallelizer.m_data_requester.request(m_buffer);
                 #endif // WITH_MPI

                                m_parallelizer.m_dataretrieved_condition.notify_one();

                                if(m_buffer.empty())
                                        break;

                                this->emit(m_buffer.front());
                        }

                        this->sigflush().emit();
                }

                virtual void onjoin()
                {
                        lock_type lock(m_parallelizer.m_mutex);
                        ++m_parallelizer.m_finished_threads;

                        if(status())
                        {
                                m_parallelizer.m_error_thread=this;
                                m_parallelizer.m_batchsize=0;
                                m_parallelizer.m_dataretrieved_condition.notify_one();
                        }
                }

         public:

                thread(parallelizer &p)
                :shore::thread("parallelizer"),
                 m_parallelizer(p)
                {}
        };

        int m_nthreads;
        multiplexer m_multiplexer;

        boost::thread_group m_thread_g;

        size_t m_batchsize;

        mutex_type m_mutex;
        condition_type m_dataready_condition;
        condition_type m_dataretrieved_condition;

        std::deque<append_type> m_input_buffer;

        std::deque<append_type> *m_process_buffers;

        //mutex_type m_join_mutex;
        //condition_type m_join_condition;

        std::vector<thread *> m_threads;
        size_t m_finished_threads;
        thread *m_error_thread;
        bool m_joined;

        shore::slot<void> m_slotjoin;
        shore::signal<void> m_sigjoin;

        void slotfun_join()
        {
                join();
        }

 #ifdef WITH_MPI
        shore::slot<std::deque<append_type> &> m_slotproxy_getdata;
        mpi_proxy::requester<std::deque<append_type> > m_data_requester;

        void slotfun_proxygetdata(std::deque<append_type> &d)
        {
                lock_type lock(m_mutex);

                while(!data_ready())
                        m_dataready_condition.wait(lock);

                if(m_input_buffer.empty())
                        ++m_finished_threads;

                d.swap(m_input_buffer);
                m_input_buffer.clear();
                m_dataretrieved_condition.notify_one();
        }
 #endif // WITH_MPI

        bool data_ready()
        {
                return m_input_buffer.size()>=m_batchsize;
        }

        void create()
        {
                for(int i=0;i<m_nthreads;++i)
                        m_thread_g.create_thread(boost::ref(*m_threads[i]));

         #ifdef WITH_MPI
                if(shore::mpi::comm_isroot()&&(shore::mpi::comm_size()>1))
                        shore::mpi_proxy::instance().request_thread();
         #endif // WITH_MPI
        }

        void join()
        {
                m_thread_g.join_all();

         #ifdef WITH_MPI
                shore::mpi::comm_barrier();
                m_data_requester.join();
         #endif // WITH_MPI

                m_joined=true;

                if(m_error_thread)
                        throw std::runtime_error(m_error_thread->errors());

                m_sigjoin.emit();
        }

        void errcheck()
        {
                bool err=false;

                {
                        lock_type lock(m_mutex);
                        err=m_error_thread;
                }

                if(err)
                {
                        m_dataready_condition.notify_all();
                        join();
                }
        }

 protected:

        virtual void parallelization_setup()
        {
                m_parallelization_numthreads=m_nthreads;
        }

 public:

        parallelizer(int nthreads,int batchsize=1)
        :m_nthreads(nthreads),
         m_multiplexer(*this),
         m_batchsize(batchsize),
         m_finished_threads(0),
         m_error_thread(0),
         m_joined(false),
         m_slotjoin(boost::bind(&parallelizer::slotfun_join,this))
 #ifdef WITH_MPI
        ,
         m_slotproxy_getdata(boost::bind(&parallelizer::slotfun_proxygetdata,this,_1))
 #endif // WITH_MPI
        {
                m_threads.push_back(new thread(*this));

                for(int i=1;i<m_nthreads;++i)
                        m_threads.push_back(new thread(*this));
         #ifdef WITH_MPI
                m_data_requester.siggetdata().connect(m_slotproxy_getdata);
         #endif // WITH_MPI

                if(shore::mpi::comm_size()>1)
                        m_nthreads=std::max(1,m_nthreads);
                else
                        m_nthreads=std::max(0,m_nthreads);

                if(m_nthreads==0)
                        m_multiplexer|(*m_threads.front());
                else
                        create();
        }

        ~parallelizer()
        {
                // if not joined due to error in the root thread
                if(!m_joined)
                {
                        m_batchsize=0;
                        m_input_buffer.clear();
                        m_dataready_condition.notify_all();
                        join();
                }

                while(!m_threads.empty())
                {
                        delete m_threads.back();
                        m_threads.pop_back();
                }
         #ifdef WITH_MPI
                //m_data_requester.join();
                m_data_requester.unsubscribe();
         #endif // WITH_MPI
        }

        int nthreads() const
        {
                return m_nthreads;
        }
};

template<typename T,typename U>
serial<U> &operator|(serial<T> &ser1,serial<U> &ser2);

template<typename T,typename U>
parallel<U> &operator|(parallel<T> &par1,parallel<U> &par2);

template<typename T,typename U>
parallelizer<U> &operator|(serial<T> &ser,parallelizer<U> &paz);

template<typename T,typename U>
parallel<U> &operator|(parallelizer<T> &paz,parallel<U> &par);

template<typename T,typename U>
sync<U> &operator|(parallel<T> &par,sync<U> &syn);

template<typename T,typename U>
serial<U> &operator|(sync<T> &syn,serial<U> &ser);

template<typename T,typename U>
desync<U> &operator|(serial<T> &ser,desync<U> &dsc);

template<typename T,typename U>
parallel<U> &operator|(desync<T> &dsc,parallel<U> &par);

class parallelization_core_access
{
 private:

        template<typename T>
        friend void dump(serial<T> &);

        template<typename T,typename U>
        friend serial<U> &operator|(serial<T> &ser1,serial<U> &ser2);
        template<typename T,typename U>
        friend parallel<U> &operator|(parallel<T> &par1,parallel<U> &par2);
        template<typename T,typename U>
        friend parallelizer<U> &operator|(serial<T> &ser,parallelizer<U> &paz);
        template<typename T,typename U>
        friend parallel<U> &operator|(parallelizer<T> &paz,parallel<U> &par);
        template<typename T,typename U>
        friend sync<U> &operator|(parallel<T> &par,sync<U> &syn);
        template<typename T,typename U>
        friend serial<U> &operator|(sync<T> &syn,serial<U> &ser);
        template<typename T,typename U>
        friend desync<U> &operator|(serial<T> &ser,desync<U> &dsc);
        template<typename T,typename U>
        friend parallel<U> &operator|(desync<T> &dsc,parallel<U> &par);

        parallelization_core_access();


        template<typename T,typename U>
        class pci_serial_serial
        :public parallel_connection_init
        {
         private:

                serial<T> * m_serial1;
                serial<U> * m_serial2;

         public:

                pci_serial_serial(serial<T> & s1,serial<U> & s2)
                :m_serial1(&s1),m_serial2(&s2)
                {}

                virtual void init()
                {
                        if(m_serial1->num_threads()!=m_serial2->num_threads())
                                throw std::logic_error("tried to connect pipeline elements"
                                                       " with differing number of threads");
                        for(int i=0;i<m_serial1->num_threads();++i)
                        {
                                typename serial<T>::lock_link &ll1=*(m_serial1->m_locklinks[i]);
                                typename serial<U>::lock_link &ll2=*(m_serial2->m_locklinks[i]);

                                (ll1.sigpostlock().connect(ll2.slotpostlock()));
                                (ll1.sigpreunlock().connect(ll2.slotpreunlock()));
                                (ll1.sigpostunlock().connect(ll2.slotpostunlock()));
                                (ll1.sigflushthread().connect(ll2.slotflushthread()));
                        }
                }
        };

        template<typename T,typename U>
        class pci_parallel_parallel
        :public parallel_connection_init
        {
         private:

                parallel<T> * m_parallel1;
                parallel<U> * m_parallel2;

         public:

                pci_parallel_parallel(parallel<T> & s1,parallel<U> & s2)
                :m_parallel1(&s1),m_parallel2(&s2)
                {}

                virtual void init()
                {
                        if(m_parallel1->num_threads()!=m_parallel2->num_threads())
                                throw std::logic_error("tried to connect pipeline elements"
                                                       " with differing number of threads");
                        for(int i=0;i<m_parallel1->num_threads();++i)
                                (*(m_parallel1->m_objects[i]))|(*(m_parallel2->m_objects[i]));
                }
        };

        template<typename T,typename U>
        class pci_parallelizer_parallel
        :public parallel_connection_init
        {
         private:

                parallelizer<T> * m_parallelizer;
                parallel<U> * m_parallel;

         public:

                pci_parallelizer_parallel(parallelizer<T> & s1,parallel<U> & s2)
                :m_parallelizer(&s1),m_parallel(&s2)
                {}

                virtual void init()
                {
                        if(m_parallelizer->num_threads()!=m_parallel->num_threads())
                                throw std::logic_error("tried to connect pipeline elements"
                                                       " with differing number of threads");
                        (*(m_parallelizer->m_threads[0]))|(*(m_parallel->m_objects[0]));
                        for(int i=1;i<m_parallelizer->num_threads();++i)
                                (*(m_parallelizer->m_threads[i]))|(*(m_parallel->m_objects[i]));
                }
        };

        template<typename T,typename U>
        class pci_parallel_sync
        :public parallel_connection_init
        {
         private:

                parallel<T> * m_parallel;
                sync<U> * m_sync;

         public:

                pci_parallel_sync(parallel<T> & s1,sync<U> & s2)
                :m_parallel(&s1),m_sync(&s2)
                {}

                virtual void init()
                {
                        if(m_parallel->num_threads()!=m_sync->num_threads())
                                throw std::logic_error("tried to connect pipeline elements"
                                                       " with differing number of threads");
                        (*(m_parallel->m_objects[0]))|(*(m_sync->m_thread_tracks[0]));
                        for(int i=1;i<m_parallel->num_threads();++i)
                                (*(m_parallel->m_objects[i]))|(*(m_sync->m_thread_tracks[i]));
                }
        };

        template<typename T,typename U>
        class pci_sync_serial
        :public parallel_connection_init
        {
         private:

                sync<T> * m_sync;
                serial<U> * m_serial;

         public:

                pci_sync_serial(sync<T> & s1,serial<U> & s2)
                :m_sync(&s1),m_serial(&s2)
                {}

                virtual void init()
                {
                        if(m_sync->num_threads()!=m_serial->num_threads())
                                throw std::logic_error("tried to connect pipeline elements"
                                                       " with differing number of threads");
                        if(m_sync->num_threads()==0)
                                (*(m_sync->m_thread_tracks[0]))|(*(m_serial->m_object));
                        else
                        {
                                for(int i=0;i<m_sync->num_threads();++i)
                                {
                                        if(m_serial->m_object)
                                                (*(m_sync->m_thread_tracks[i]))|(*(m_serial->m_object));

                                        typename serial<U>::lock_link &ll=*(m_serial->m_locklinks[i]);

                                        m_sync->m_thread_tracks[i]->m_sigpostlock.connect(ll.slotpostlock());
                                        m_sync->m_thread_tracks[i]->m_sigpreunlock.connect(ll.slotpreunlock());
                                        m_sync->m_thread_tracks[i]->m_sigpostunlock.connect(ll.slotpostunlock());
                                        m_sync->m_thread_tracks[i]->sigflush().connect(ll.slotflushthread());
                                }

                                if(m_serial->m_object)
                                {
                                        for(size_t i=1;i<shore::mpi::comm_size();++i)
                                                (*(m_sync->m_proxy_tracks[i-1]))|(*(m_serial->m_object));

                                        m_serial->m_object->slotflush().disconnect_all();

                                        m_sync->m_sigsyncflush.connect(m_serial->m_object->slotflush());
                                }
                        }
                }
        };

        template<typename T,typename U>
        class pci_serial_desync
        :public parallel_connection_init
        {
         private:

                serial<T> * m_serial;
                desync<U> * m_desync;

         public:

                pci_serial_desync(serial<T> & s1,desync<U> & s2)
                :m_serial(&s1),m_desync(&s2)
                {}

                virtual void init()
                {
                        if(m_serial->num_threads()!=m_desync->num_threads())
                                throw std::logic_error("tried to connect pipeline elements"
                                                       " with differing number of threads");
                        for(int i=0;i<m_serial->num_threads();++i)
                        {
                                typename serial<T>::lock_link &ll=*(m_serial->m_locklinks[i]);

                                (ll.sigpostlock().connect(m_desync->m_multiplexer.slotpostlock()));
                                (ll.sigpreunlock().connect(m_desync->m_multiplexer.slotpreunlock()));
                                (ll.sigpostunlock().connect(m_desync->m_multiplexer.slotpostunlock()));
                                (ll.sigflushthread().connect(m_desync->m_multiplexer.slotflushthread()));
                        }
                }
        };

        template<typename T,typename U>
        class pci_desync_parallel
        :public parallel_connection_init
        {
         private:

                desync<T> * m_desync;
                parallel<U> * m_parallel;

         public:

                pci_desync_parallel(desync<T> & s1,parallel<U> & s2)
                :m_desync(&s1),m_parallel(&s2)
                {}

                virtual void init()
                {
                        if(m_desync->num_threads()!=m_parallel->num_threads())
                                throw std::logic_error("tried to connect pipeline elements"
                                                       " with differing number of threads");
                        if(m_desync->num_threads()==0)
                                (m_desync->m_multiplexer)|(*(m_parallel->m_objects[0]));
                        for(int i=0;i<m_desync->num_threads();++i)
                                (*(m_desync->m_thread_tracks[i]))|(*(m_parallel->m_objects[i]));
                }
        };


        template<typename T>
        static void dump(serial<T> & ser)
        {
                ser.m_signumthreads.emit(0);

                if(shore::mpi::comm_isroot())
                        ser.call(&T::dump);
                else
                        // non-root process: join the threads
                        ser.m_sigjoin.emit();
        }

        template<typename T,typename U>
        static serial<U> &connect(serial<T> &ser1,serial<U> &ser2)
        {
                ser1.m_sigjoin.connect(ser2.m_slotjoin);
                ser1.m_signumthreads.connect(ser2.m_slotnumthreads);

                if((ser1.m_object!=0)&&(ser2.m_object!=0))
                        (*ser1.m_object)|(*ser2.m_object);

                ser1.m_pci.push_back(new pci_serial_serial<T,U>(ser1,ser2));

                return ser2;
        }

        template<typename T,typename U>
        static parallel<U> &connect(parallel<T> &par1,parallel<U> &par2)
        {
                par1.m_sigjoin.connect(par2.m_slotjoin);
                par1.m_signumthreads.connect(par2.m_slotnumthreads);

                par1.m_pci.push_back(new pci_parallel_parallel<T,U>(par1,par2));

                return par2;
        }

        template<typename T,typename U>
        static parallelizer<U> &connect(serial<T> &ser,parallelizer<U> &paz)
        {
                ser.m_sigjoin.connect(paz.m_slotjoin);
                ser.m_signumthreads.connect(paz.m_slotnumthreads);

                if(ser.m_object)
                        (*ser.m_object)|paz.m_multiplexer;
                return paz;
        }

        template<typename T,typename U>
        static parallel<U> &connect(parallelizer<T> &paz,parallel<U> &par)
        {
                paz.m_sigjoin.connect(par.m_slotjoin);
                paz.m_signumthreads.connect(par.m_slotnumthreads);

                paz.m_pci.push_back(new pci_parallelizer_parallel<T,U>(paz,par));

                return par;
        }

        template<typename T,typename U>
        static sync<U> &connect(parallel<T> &par,sync<U> &syn)
        {
                par.m_sigjoin.connect(syn.m_slotjoin);
                par.m_signumthreads.connect(syn.m_slotnumthreads);

                par.m_pci.push_back(new pci_parallel_sync<T,U>(par,syn));

                return syn;
        }

        template<typename T,typename U>
        static serial<U> &connect(sync<T> &syn,serial<U> &ser)
        {
                syn.m_sigjoin.connect(ser.m_slotjoin);
                syn.m_signumthreads.connect(ser.m_slotnumthreads);

                syn.m_pci.push_back(new pci_sync_serial<T,U>(syn,ser));

                return ser;
        }

        template<typename T,typename U>
        static desync<U> &connect(serial<T> &ser,desync<U> &dsc)
        {
                ser.m_sigjoin.connect(dsc.m_slotjoin);
                ser.m_signumthreads.connect(dsc.m_slotnumthreads);

                (*ser.m_object)|dsc.m_multiplexer;

                ser.m_pci.push_back(new pci_serial_desync<T,U>(ser,dsc));

                return dsc;
        }

        template<typename T,typename U>
        static parallel<U> &connect(desync<T> &dsc,parallel<U> &par)
        {
                dsc.m_sigjoin.connect(par.m_slotjoin);
                dsc.m_signumthreads.connect(par.m_slotnumthreads);
                dsc.m_pci.push_back(new pci_desync_parallel<T,U>(dsc,par));

                return par;
        }
};

template<typename T>
void dump(serial<T> & ser)
{
        parallelization_core_access::dump(ser);
}

template<typename T,typename U>
serial<U> &operator|(serial<T> &ser1,serial<U> &ser2)
{
    return parallelization_core_access::connect(ser1,ser2);
}

template<typename T,typename U>
parallel<U> &operator|(parallel<T> &par1,parallel<U> &par2)
{
    return parallelization_core_access::connect(par1,par2);
}

template<typename T,typename U>
parallelizer<U> &operator|(serial<T> &ser,parallelizer<U> &paz)
{
    return parallelization_core_access::connect(ser,paz);
}

template<typename T,typename U>
parallel<U> &operator|(parallelizer<T> &paz,parallel<U> &par)
{
    return parallelization_core_access::connect(paz,par);
}

template<typename T,typename U>
sync<U> &operator|(parallel<T> &par,sync<U> &syn)
{
    return parallelization_core_access::connect(par,syn);
}

template<typename T,typename U>
serial<U> &operator|(sync<T> &syn,serial<U> &ser)
{
    return parallelization_core_access::connect(syn,ser);
}

template<typename T,typename U>
desync<U> &operator|(serial<T> &ser,desync<U> &dsc)
{
    return parallelization_core_access::connect(ser,dsc);
}

template<typename T,typename U>
parallel<U> &operator|(desync<T> &dsc,parallel<U> &par)
{
    return parallelization_core_access::connect(dsc,par);
}

} // namespace

#endif // SHORE_PARALLEL_PARALLEL_HPP__