eoParallelApply.h

Go to the documentation of this file.

/*
(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_PARALLEL_APPLY_H__
# define __EO_PARALLEL_APPLY_H__

# include "eoMpi.h"

# include <eoFunctor.h> // eoUF
# include <vector> // std::vector population

namespace eo
{
    namespace mpi
    {
        struct ParallelApplyAssignment
        {
            int index;
            int size;
        };

        template<class EOT>
        struct ParallelApplyData
        {
            ParallelApplyData(
                    eoUF<EOT&, void> & _proc,
                    int _masterRank,
                    int _packetSize,
                    std::vector<EOT> * table = 0
                   ) :
                _table( table ), func( _proc ), index( 0 ), packetSize( _packetSize ), masterRank( _masterRank ), comm( Node::comm() )
            {
                if ( _packetSize <= 0 )
                {
                    throw std::runtime_error("Packet size should not be negative.");
                }

                if( table )
                {
                    size = table->size();
                }
            }

            void init( std::vector<EOT>& table )
            {
                index = 0;
                size = table.size();
                _table = &table;
                assignedTasks.clear();
            }

            std::vector<EOT>& table()
            {
                return *_table;
            }

            // All elements are public since functors will often use them.
            std::vector<EOT> * _table;
            eoUF<EOT&, void> & func;
            int index;
            int size;
            std::map< int /* worker rank */, ParallelApplyAssignment /* last assignment */> assignedTasks;
            int packetSize;
            std::vector<EOT> tempArray;

            int masterRank;
            bmpi::communicator& comm;
        };

        template< class EOT >
        class SendTaskParallelApply : public SendTaskFunction< ParallelApplyData<EOT> >
        {
            public:
            using SendTaskFunction< ParallelApplyData<EOT> >::_data;

            SendTaskParallelApply( SendTaskParallelApply<EOT> * w = 0 ) : SendTaskFunction< ParallelApplyData<EOT> >( w )
            {
                // empty
            }

            void operator()(int wrkRank)
            {
                int futureIndex;

                if( _data->index + _data->packetSize < _data->size )
                {
                    futureIndex = _data->index + _data->packetSize;
                } else {
                    futureIndex = _data->size;
                }

                int sentSize = futureIndex - _data->index ;

                _data->comm.send( wrkRank, 1, sentSize );

                eo::log << eo::progress << "Evaluating individual " << _data->index << std::endl;

                _data->assignedTasks[ wrkRank ].index = _data->index;
                _data->assignedTasks[ wrkRank ].size = sentSize;

                _data->comm.send( wrkRank, 1, & ( (_data->table())[ _data->index ] ) , sentSize );
                _data->index = futureIndex;
            }
        };

        template< class EOT >
        class HandleResponseParallelApply : public HandleResponseFunction< ParallelApplyData<EOT> >
        {
            public:
            using HandleResponseFunction< ParallelApplyData<EOT> >::_data;

            HandleResponseParallelApply( HandleResponseParallelApply<EOT> * w = 0 ) : HandleResponseFunction< ParallelApplyData<EOT> >( w )
            {
                // empty
            }

            void operator()(int wrkRank)
            {
                _data->comm.recv( wrkRank, 1, & (_data->table()[ _data->assignedTasks[wrkRank].index ] ), _data->assignedTasks[wrkRank].size );
            }
        };

        template< class EOT >
        class ProcessTaskParallelApply : public ProcessTaskFunction< ParallelApplyData<EOT> >
        {
            public:
            using ProcessTaskFunction< ParallelApplyData<EOT> >::_data;

            ProcessTaskParallelApply( ProcessTaskParallelApply<EOT> * w = 0 ) : ProcessTaskFunction< ParallelApplyData<EOT> >( w )
            {
                // empty
            }

            void operator()()
            {
                int recvSize;

                _data->comm.recv( _data->masterRank, 1, recvSize );
                _data->tempArray.resize( recvSize );
                _data->comm.recv( _data->masterRank, 1, & _data->tempArray[0] , recvSize );
                timerStat.start("worker_processes");
                for( int i = 0; i < recvSize ; ++i )
                {
                    _data->func( _data->tempArray[ i ] );
                }
                timerStat.stop("worker_processes");
                _data->comm.send( _data->masterRank, 1, & _data->tempArray[0], recvSize );
            }
        };

        template< class EOT >
        class IsFinishedParallelApply : public IsFinishedFunction< ParallelApplyData<EOT> >
        {
            public:
            using IsFinishedFunction< ParallelApplyData<EOT> >::_data;

            IsFinishedParallelApply( IsFinishedParallelApply<EOT> * w = 0 ) : IsFinishedFunction< ParallelApplyData<EOT> >( w )
            {
                // empty
            }

            bool operator()()
            {
                return _data->index == _data->size;
            }
        };

        template< class EOT >
        struct ParallelApplyStore : public JobStore< ParallelApplyData<EOT> >
        {
            using JobStore< ParallelApplyData<EOT> >::_stf;
            using JobStore< ParallelApplyData<EOT> >::_hrf;
            using JobStore< ParallelApplyData<EOT> >::_ptf;
            using JobStore< ParallelApplyData<EOT> >::_iff;

            ParallelApplyStore(
                    eoUF<EOT&, void> & _proc,
                    int _masterRank,
                    int _packetSize = 1,
                    // JobStore functors
                    SendTaskParallelApply<EOT> * stpa = 0,
                    HandleResponseParallelApply<EOT>* hrpa = 0,
                    ProcessTaskParallelApply<EOT>* ptpa = 0,
                    IsFinishedParallelApply<EOT>* ifpa = 0
                   ) :
                _data( _proc, _masterRank, _packetSize )
            {
                if( stpa == 0 ) {
                    stpa = new SendTaskParallelApply<EOT>;
                    stpa->needDelete( true );
                }

                if( hrpa == 0 ) {
                    hrpa = new HandleResponseParallelApply<EOT>;
                    hrpa->needDelete( true );
                }

                if( ptpa == 0 ) {
                    ptpa = new ProcessTaskParallelApply<EOT>;
                    ptpa->needDelete( true );
                }

                if( ifpa == 0 ) {
                    ifpa = new IsFinishedParallelApply<EOT>;
                    ifpa->needDelete( true );
                }

                _stf = stpa;
                _hrf = hrpa;
                _ptf = ptpa;
                _iff = ifpa;
            }

            ParallelApplyData<EOT>* data() { return &_data; }

            void data( std::vector<EOT>& _pop )
            {
                _data.init( _pop );
            }

            virtual ~ParallelApplyStore() // for inheritance purposes only
            {
            }

            protected:
            ParallelApplyData<EOT> _data;
        };

        template< typename EOT >
        class ParallelApply : public MultiJob< ParallelApplyData<EOT> >
        {
            public:

            ParallelApply(
                    AssignmentAlgorithm & algo,
                    int _masterRank,
                    ParallelApplyStore<EOT> & store
                    ) :
                MultiJob< ParallelApplyData<EOT> >( algo, _masterRank, store )
            {
                // empty
            }
        };

    }
}
# endif // __EO_PARALLEL_APPLY_H__