eoMpi.h

/*
(c) Thales group, 2012

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation;
    version 2 of the License.

    This library 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
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
Contact: http://eodev.sourceforge.net

Authors:
    Benjamin Bouvier <benjamin.bouvier@gmail.com>
*/
# ifndef __EO_MPI_H__
# define __EO_MPI_H__

# include <vector>  // std::vector

# include <utils/eoLogger.h>
# include <utils/eoTimer.h>
# include <eoFunctor.h>
# include <eoExceptions.h>

# include "eoMpiNode.h"
# include "eoMpiAssignmentAlgorithm.h"

namespace eo
{
    namespace mpi
    {
        extern eoTimerStat timerStat;

        namespace Channel
        {
            const int Commands = 0;
            const int Messages = 1;
        }

        namespace Message
        {
            const int Continue = 0;
            const int Finish = 1;
            const int Kill = 2;
        }

        const int DEFAULT_MASTER = 0;

        template< typename JobData, typename Wrapped >
        struct SharedDataFunction
        {
            SharedDataFunction( Wrapped * w = 0 ) : _data( 0 ), _wrapped( w ), _needDelete( false )
            {
                // empty
            }

            virtual ~SharedDataFunction()
            {
                if( _wrapped && _wrapped->needDelete() )
                {
                    delete _wrapped;
                }
            }

            void wrapped( Wrapped * w )
            {
                _wrapped = w;
            }

            void data( JobData* d )
            {
                _data = d;
                if( _wrapped )
                {
                    _wrapped->data( d );
                }
            }

            bool needDelete() { return _needDelete; }
            void needDelete( bool b ) { _needDelete = b; }

            protected:
            JobData* _data;
            Wrapped* _wrapped; // Pointer and not a reference so as to be set at any time and to avoid affectation
            bool _needDelete;
        };

        template< typename JobData >
        struct SendTaskFunction : public eoUF<int, void>, public SharedDataFunction< JobData, SendTaskFunction<JobData> >
        {
            public:

            SendTaskFunction( SendTaskFunction<JobData>* w = 0 ) : SharedDataFunction<JobData, SendTaskFunction<JobData> >( w )
            {
                // empty
            }

            virtual ~SendTaskFunction() {} // for inherited classes
        };

        template< typename JobData >
        struct HandleResponseFunction : public eoUF<int, void>, public SharedDataFunction< JobData, HandleResponseFunction<JobData> >
        {
            public:

            HandleResponseFunction( HandleResponseFunction<JobData>* w = 0 ) : SharedDataFunction<JobData, HandleResponseFunction<JobData> >( w )
            {
                // empty
            }

            virtual ~HandleResponseFunction() {} // for inherited classes
        };

        template< typename JobData >
        struct ProcessTaskFunction : public eoF<void>, public SharedDataFunction< JobData, ProcessTaskFunction<JobData> >
        {
            public:

            ProcessTaskFunction( ProcessTaskFunction<JobData>* w = 0 ) : SharedDataFunction<JobData, ProcessTaskFunction<JobData> >( w )
            {
                // empty
            }

            virtual ~ProcessTaskFunction() {} // for inherited classes
        };

        template< typename JobData >
        struct IsFinishedFunction : public eoF<bool>, public SharedDataFunction< JobData, IsFinishedFunction<JobData> >
        {
            public:

            IsFinishedFunction( IsFinishedFunction<JobData>* w = 0 ) : SharedDataFunction<JobData, IsFinishedFunction<JobData> >( w )
            {
                // empty
            }

            virtual ~IsFinishedFunction() {} // for inherited classes
        };

        template< typename JobData >
        struct JobStore
        {
            JobStore(
                SendTaskFunction<JobData>* stf,
                HandleResponseFunction<JobData>* hrf,
                ProcessTaskFunction<JobData>* ptf,
                IsFinishedFunction<JobData>* iff
            ) :
                _stf( stf ), _hrf( hrf ), _ptf( ptf ), _iff( iff )
            {
                // empty
            }

            JobStore()
            {
                // empty
            }

            ~JobStore()
            {
                if( _stf->needDelete() ) delete _stf;
                if( _hrf->needDelete() ) delete _hrf;
                if( _ptf->needDelete() ) delete _ptf;
                if( _iff->needDelete() ) delete _iff;
            }

            // Getters
            SendTaskFunction<JobData> & sendTask() { return *_stf; }
            HandleResponseFunction<JobData> & handleResponse() { return *_hrf; }
            ProcessTaskFunction<JobData> & processTask() { return *_ptf; }
            IsFinishedFunction<JobData> & isFinished() { return *_iff; }

            // Setters
            void sendTask( SendTaskFunction<JobData>* stf ) { _stf = stf; }
            void handleResponse( HandleResponseFunction<JobData>* hrf ) { _hrf = hrf; }
            void processTask( ProcessTaskFunction<JobData>* ptf ) { _ptf = ptf; }
            void isFinished( IsFinishedFunction<JobData>* iff ) { _iff = iff; }

            void wrapSendTask( SendTaskFunction<JobData>* stf )
            {
                if( stf )
                {
                    stf->wrapped( _stf );
                    _stf = stf;
                }
            }

            void wrapHandleResponse( HandleResponseFunction<JobData>* hrf )
            {
                if( hrf )
                {
                    hrf->wrapped( _hrf );
                    _hrf = hrf;
                }
            }

            void wrapProcessTask( ProcessTaskFunction<JobData>* ptf )
            {
                if( ptf )
                {
                    ptf->wrapped( _ptf );
                    _ptf = ptf;
                }
            }

            void wrapIsFinished( IsFinishedFunction<JobData>* iff )
            {
                if( iff )
                {
                    iff->wrapped( _iff );
                    _iff = iff;
                }
            }

            virtual JobData* data() = 0;

            protected:

            SendTaskFunction< JobData >* _stf;
            HandleResponseFunction< JobData >* _hrf;
            ProcessTaskFunction< JobData >* _ptf;
            IsFinishedFunction< JobData >* _iff;
        };

        template< class JobData >
        class Job
        {
            public:
                Job( AssignmentAlgorithm& _algo,
                     int _masterRank,
                     int _workerStopCondition,
                     JobStore<JobData> & _store
                    ) :
                    assignmentAlgo( _algo ),
                    masterRank( _masterRank ),
                    workerStopCondition( _workerStopCondition ),
                    comm( Node::comm() ),
                    // Functors
                    store( _store ),
                    sendTask( _store.sendTask() ),
                    handleResponse( _store.handleResponse() ),
                    processTask( _store.processTask() ),
                    isFinished( _store.isFinished() )
                {
                    _isMaster = Node::comm().rank() == _masterRank;

                    sendTask.data( _store.data() );
                    handleResponse.data( _store.data() );
                    processTask.data( _store.data() );
                    isFinished.data( _store.data() );
                }

            protected:

                struct FinallyBlock
                {
                    FinallyBlock(
                            int _totalWorkers,
                            AssignmentAlgorithm& _algo,
                            Job< JobData > & _that
                            ) :
                        totalWorkers( _totalWorkers ),
                        assignmentAlgo( _algo ),
                        that( _that ),
                        // global field
                        comm( Node::comm() )
                    {
                        // empty
                    }

                    ~FinallyBlock()
                    {
# ifndef NDEBUG
                        eo::log << eo::debug;
                        eo::log << "[M" << comm.rank() << "] Frees all the idle." << std::endl;
# endif
                        // frees all the idle workers
                        timerStat.start("master_wait_for_idles");
                        std::vector<int> idles = assignmentAlgo.idles();
                        for(unsigned int i = 0; i < idles.size(); ++i)
                        {
                            comm.send( idles[i], Channel::Commands, Message::Finish );
                        }
                        timerStat.stop("master_wait_for_idles");

# ifndef NDEBUG
                        eo::log << "[M" << comm.rank() << "] Waits for all responses." << std::endl;
# endif
                        // wait for all responses
                        timerStat.start("master_wait_for_all_responses");
                        while( assignmentAlgo.availableWorkers() != totalWorkers )
                        {
                            bmpi::status status = comm.probe( bmpi::any_source, bmpi::any_tag );
                            int wrkRank = status.source();
                            that.handleResponse( wrkRank );
                            comm.send( wrkRank, Channel::Commands, Message::Finish );
                            assignmentAlgo.confirm( wrkRank );
                        }
                        timerStat.stop("master_wait_for_all_responses");
# ifndef NDEBUG
                        eo::log << "[M" << comm.rank() << "] Leaving master task." << std::endl;
# endif
                    }

                    protected:

                    int totalWorkers;
                    AssignmentAlgorithm& assignmentAlgo;
                    Job< JobData > & that;

                    bmpi::communicator & comm;
                };

                void master( )
                {
                    int totalWorkers = assignmentAlgo.availableWorkers();
# ifndef NDEBUG
                    eo::log << eo::debug;
                    eo::log << "[M" << comm.rank() << "] Have " << totalWorkers << " workers." << std::endl;
# endif
                    try {
                        FinallyBlock finally( totalWorkers, assignmentAlgo, *this );
                        while( ! isFinished() )
                        {
                            timerStat.start("master_wait_for_assignee");
                            int assignee = assignmentAlgo.get( );
                            while( assignee <= 0 )
                            {
# ifndef NDEBUG
                                eo::log << "[M" << comm.rank() << "] Waitin' for node..." << std::endl;
# endif
                                bmpi::status status = comm.probe( bmpi::any_source, bmpi::any_tag );
                                int wrkRank = status.source();
# ifndef NDEBUG
                                eo::log << "[M" << comm.rank() << "] Node " << wrkRank << " just terminated." << std::endl;
# endif
                                handleResponse( wrkRank );
                                assignmentAlgo.confirm( wrkRank );
                                assignee = assignmentAlgo.get( );
                            }
                            timerStat.stop("master_wait_for_assignee");
# ifndef NDEBUG
                            eo::log << "[M" << comm.rank() << "] Assignee : " << assignee << std::endl;
# endif

                            timerStat.start("master_wait_for_send");
                            comm.send( assignee, Channel::Commands, Message::Continue );
                            sendTask( assignee );
                            timerStat.stop("master_wait_for_send");
                        }
                    } catch( const std::exception & e )
                    {
                        std::string s = e.what();
                        s.append( " in eoMpi loop");
                        throw std::runtime_error( s );
                    }
                }

                void worker( )
                {
                    int order;
# ifndef NDEBUG
                    eo::log << eo::debug;
# endif
                    timerStat.start("worker_wait_for_order");
                    comm.recv( masterRank, Channel::Commands, order );
                    timerStat.stop("worker_wait_for_order");

                    while( true )
                    {
# ifndef NDEBUG
                        eo::log << "[W" << comm.rank() << "] Waiting for an order..." << std::endl;
# endif
                        if ( order == workerStopCondition )
                        {
# ifndef NDEBUG
                            eo::log << "[W" << comm.rank() << "] Leaving worker task." << std::endl;
# endif
                            return;
                        } else if( order == Message::Continue )
                        {
# ifndef NDEBUG
                            eo::log << "[W" << comm.rank() << "] Processing task..." << std::endl;
# endif
                            processTask( );
                        }

                        timerStat.start("worker_wait_for_order");
                        comm.recv( masterRank, Channel::Commands, order );
                        timerStat.stop("worker_wait_for_order");
                    }
                }

            public:

                void run( )
                {
                    ( _isMaster ) ? master( ) : worker( );
                }

                bool isMaster( )
                {
                    return _isMaster;
                }

            protected:

                AssignmentAlgorithm& assignmentAlgo;
                int masterRank;
                const int workerStopCondition;
                bmpi::communicator& comm;

                JobStore<JobData>& store;
                SendTaskFunction<JobData> & sendTask;
                HandleResponseFunction<JobData> & handleResponse;
                ProcessTaskFunction<JobData> & processTask;
                IsFinishedFunction<JobData> & isFinished;

                bool _isMaster;
        };

        template< class JobData >
        class OneShotJob : public Job< JobData >
        {
            public:
                OneShotJob( AssignmentAlgorithm& algo,
                            int masterRank,
                            JobStore<JobData> & store )
                    : Job<JobData>( algo, masterRank, Message::Finish, store )
                {
                    // empty
                }
        };

        template< class JobData >
        class MultiJob : public Job< JobData >
        {
            public:
                MultiJob ( AssignmentAlgorithm& algo,
                            int masterRank,
                            JobStore<JobData> & store )
                    : Job<JobData>( algo, masterRank, Message::Kill, store )
                {
                    // empty
                }
        };
    }

}
# endif // __EO_MPI_H__