manual merge eompi

This commit is contained in:
Johann Dreo 2012-07-10 18:00:25 +02:00
commit ea8da0e965
35 changed files with 2866 additions and 23 deletions

5
eo/.gitignore vendored Normal file
View file

@ -0,0 +1,5 @@
*.swp
debug/
release/
*CMakeFiles*
*Makefile

View file

@ -45,10 +45,12 @@ ENABLE_LANGUAGE(C)
### 2) Include required modules / configuration files
#####################################################################################
FIND_PACKAGE(OpenMP)
IF(OPENMP_FOUND)
IF(WITH_OMP)
FIND_PACKAGE(OpenMP)
IF(OPENMP_FOUND)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${OpenMP_C_FLAGS}")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${OpenMP_CXX_FLAGS}")
ENDIF()
ENDIF()
INCLUDE(CMakeBackwardCompatibilityCXX)

View file

@ -6,3 +6,13 @@ SET(PROJECT_VERSION_PATCH 0)
SET(PROJECT_VERSION_MISC "-edge")
# ADD_DEFINITIONS(-DDEPRECATED_MESSAGES) # disable warning deprecated function messages
# If you plan to use OpenMP, put the following boolean to true :
SET(WITH_OMP FALSE CACHE BOOL "Use OpenMP ?" FORCE)
# If you plan to use MPI, precise here where are the static libraries from
# openmpi and boost::mpi.
SET(WITH_MPI FALSE CACHE BOOL "Use mpi ?" FORCE)
SET(MPI_DIR "/mpi/directory" CACHE PATH "OpenMPI directory" FORCE)
SET(BOOST_DIR "/boost/directory" CACHE PATH "Boost directory" FORCE)

View file

@ -5,7 +5,32 @@
INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR})
######################################################################################
### 2) Define the eo target
### 2) Optional: add MPI and Boost MPI dependencies.
######################################################################################
IF(WITH_MPI)
MESSAGE("[EO] Compilation with MPI and BoostMPI.")
SET(CMAKE_CXX_COMPILER "${MPI_DIR}/bin/mpicxx")
# headers location
INCLUDE_DIRECTORIES(${MPI_DIR}/include)
INCLUDE_DIRECTORIES(${BOOST_DIR}/include)
# lib location
LINK_DIRECTORIES(${MPI_DIR}/lib)
LINK_DIRECTORIES(${BOOST_DIR}/lib)
# for conditional compilation in code
ADD_DEFINITIONS(-DWITH_MPI)
LINK_LIBRARIES(boost_mpi boost_serialization)
ADD_SUBDIRECTORY(mpi)
ENDIF()
######################################################################################
### 3) Define the eo target
######################################################################################
SET(EO_LIB_OUTPUT_PATH ${EO_BINARY_DIR}/lib)
@ -27,14 +52,14 @@ FILE(GLOB HDRS *.h eo)
INSTALL(FILES ${HDRS} DESTINATION include/eo COMPONENT headers)
######################################################################################
### 3) Optionnal: define your target(s)'s version: no effect for windows
### 4) Optionnal: define your target(s)'s version: no effect for windows
######################################################################################
SET(EO_VERSION ${GLOBAL_VERSION})
SET_TARGET_PROPERTIES(eo PROPERTIES VERSION "${EO_VERSION}")
######################################################################################
### 4) Where must cmake go now ?
### 5) Where must cmake go now ?
######################################################################################
ADD_SUBDIRECTORY(do)
@ -43,6 +68,7 @@ ADD_SUBDIRECTORY(ga)
ADD_SUBDIRECTORY(gp)
ADD_SUBDIRECTORY(other)
ADD_SUBDIRECTORY(utils)
ADD_SUBDIRECTORY(serial)
IF(ENABLE_PYEO)
ADD_SUBDIRECTORY(pyeo)

View file

@ -36,6 +36,12 @@
#include <omp.h>
#endif
# ifdef WITH_MPI
# include <mpi/eoMpi.h>
# include <mpi/eoMultiParallelApply.h>
# include <mpi/eoTerminateJob.h>
# endif // WITH_MPI
/**
Applies a unary function to a std::vector of things.
@ -82,6 +88,21 @@ void apply(eoUF<EOT&, void>& _proc, std::vector<EOT>& _pop)
#endif // !_OPENMP
}
#ifdef WITH_MPI
template<class EOT>
void parallelApply(
std::vector<EOT>& _pop,
eo::mpi::AssignmentAlgorithm& _algo,
int _masterRank,
eo::mpi::ParallelEvalStore<EOT> & _store )
{
_store.data( _pop );
_algo.reinit( _pop.size() );
eo::mpi::ParallelApply<EOT> job( _algo, _masterRank, _store );
job.run();
}
#endif
/**
This is a variant of apply<EOT> which is called in parallel
thanks to OpenMP.

View file

@ -141,6 +141,9 @@
#include <utils/eoRealVectorBounds.h> // includes eoRealBounds.h
#include <utils/eoIntBounds.h> // no eoIntVectorBounds
// Serialization stuff
#include <serial/eoSerial.h>
// aliens
#include <other/external_eo>
#include <eoCounter.h>

View file

@ -102,6 +102,33 @@ template<class EOT> class eoEasyEA: public eoAlgo<EOT>
offspring.reserve(_offspringSize); // This line avoids an incremental resize of offsprings.
}
/**
* @brief Ctor allowing to specify which pop eval function we're going to use.
*
* Ctor taking a breed and merge, an overload of ctor to define an offspring size, and
* the pop eval function used. This allows to precise if we would like to use the
* parallel evaluation, for instance.
*/
eoEasyEA(
eoContinue<EOT>& _continuator,
eoEvalFunc<EOT>& _eval,
eoPopEvalFunc<EOT>& _pop_eval,
eoBreed<EOT>& _breed,
eoReplacement<EOT>& _replace,
unsigned _offspringSize
) : continuator(_continuator),
eval (_eval),
loopEval(_eval),
popEval(_pop_eval),
selectTransform(dummySelect, dummyTransform),
breed(_breed),
mergeReduce(dummyMerge, dummyReduce),
replace(_replace),
isFirstCall(true)
{
offspring.reserve(_offspringSize); // This line avoids an incremental resize of offsprings.
}
/*
eoEasyEA(eoContinue <EOT> & _continuator,
eoPopEvalFunc <EOT> & _pop_eval,
@ -219,6 +246,8 @@ template<class EOT> class eoEasyEA: public eoAlgo<EOT>
/// Apply a few generation of evolution to the population.
virtual void operator()(eoPop<EOT>& _pop)
{
eo::log << "[EasyEA] Call to operator()" << std::endl;
if (isFirstCall)
{
size_t total_capacity = _pop.capacity() + offspring.capacity();
@ -227,22 +256,33 @@ template<class EOT> class eoEasyEA: public eoAlgo<EOT>
isFirstCall = false;
}
// TODO TODOB delete all log traces
std::cout << "[EasyEA] After is first call." << std::endl;
eoPop<EOT> empty_pop;
std::cout << "[EasyEA] After empty_pop." << std::endl;
popEval(empty_pop, _pop); // A first eval of pop.
std::cout << "[EasyEA] After pop_eval." << std::endl;
do
{
try
{
std::cout << "[EasyEA] Beginning try." << std::endl;
unsigned pSize = _pop.size();
std::cout << "[EasyEA] psize determinated." << std::endl;
offspring.clear(); // new offspring
std::cout << "[EasyEA] offspring cleared." << std::endl;
breed(_pop, offspring);
std::cout << "[EasyEA] After breed, evaluating pop." << std::endl;
popEval(_pop, offspring); // eval of parents + offspring if necessary
std::cout << "[EasyEA] After evaluation, replacing pop." << std::endl;
replace(_pop, offspring); // after replace, the new pop. is in _pop
std::cout << "[EasyEA] After replacing, continuator." << std::endl;
if (pSize > _pop.size())
throw std::runtime_error("Population shrinking!");

View file

@ -77,6 +77,72 @@ private:
eoEvalFunc<EOT> & eval;
};
#ifdef WITH_MPI
// TODO TODOB commenter
template<class EOT>
class eoParallelPopLoopEval : public eoPopEvalFunc<EOT>
{
public:
/** Ctor: set value of embedded eoEvalFunc */
eoParallelPopLoopEval(
// Job parameters
eo::mpi::AssignmentAlgorithm& _assignAlgo,
int _masterRank,
// Default parameters for store
eoEvalFunc<EOT> & _eval,
int _packetSize = 1
) :
assignAlgo( _assignAlgo ),
masterRank( _masterRank ),
needToDeleteStore( true )
{
store = new eo::mpi::ParallelEvalStore<EOT>( _eval, _masterRank, _packetSize );
}
eoParallelPopLoopEval(
// Job parameters
eo::mpi::AssignmentAlgorithm& _assignAlgo,
int _masterRank,
eo::mpi::ParallelEvalStore<EOT>* _store
) :
assignAlgo( _assignAlgo ),
masterRank( _masterRank ),
store( _store ),
needToDeleteStore( false )
{
// empty
}
~eoParallelPopLoopEval()
{
if( eo::mpi::Node::comm().rank() == masterRank )
{
eo::mpi::EmptyJob job( assignAlgo, masterRank );
job.run();
}
if( needToDeleteStore )
{
delete store;
}
}
/** Do the job: simple loop over the offspring */
void operator()( eoPop<EOT> & _parents, eoPop<EOT> & _offspring )
{
(void)_parents;
parallelApply<EOT>(_offspring, assignAlgo, masterRank, *store);
}
private:
eo::mpi::AssignmentAlgorithm & assignAlgo;
int masterRank;
eo::mpi::ParallelEvalStore<EOT>* store;
bool needToDeleteStore;
};
#endif
/////////////////////////////////////////////////////////////
// eoTimeVaryingLoopEval
/////////////////////////////////////////////////////////////

32
eo/src/mpi/CMakeLists.txt Normal file
View file

@ -0,0 +1,32 @@
######################################################################################
### 1) Include the sources
######################################################################################
INCLUDE_DIRECTORIES(${EO_SOURCE_DIR}/src)
INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR})
######################################################################################
### 2) Define the eompi target
######################################################################################
SET(EOMPI_LIB_OUTPUT_PATH ${EO_BINARY_DIR}/lib)
SET(LIBRARY_OUTPUT_PATH ${EOMPI_LIB_OUTPUT_PATH})
SET(EOMPI_SOURCES
eoMpi.cpp
)
ADD_LIBRARY(eompi STATIC ${EOMPI_SOURCES})
INSTALL(TARGETS eompi ARCHIVE DESTINATION lib COMPONENT libraries)
FILE(GLOB HDRS *.h)
INSTALL(FILES ${HDRS} DESTINATION include/eo/mpi COMPONENT headers)
######################################################################################
### 3) Optionnal
######################################################################################
SET(EOMPI_VERSION ${GLOBAL_VERSION})
SET_TARGET_PROPERTIES(eompi PROPERTIES VERSION "${EOMPI_VERSION}")
######################################################################################

12
eo/src/mpi/eoMpi.cpp Normal file
View file

@ -0,0 +1,12 @@
# include "eoMpi.h"
// MpiNode* MpiNodeStore::singleton;
namespace eo
{
namespace mpi
{
bmpi::communicator Node::_comm;
eoTimerStat timerStat;
}
}

391
eo/src/mpi/eoMpi.h Normal file
View file

@ -0,0 +1,391 @@
# ifndef __EO_MPI_H__
# define __EO_MPI_H__
# include <vector>
# include <map>
# include <sys/time.h>
# include <sys/resource.h>
# include <utils/eoLogger.h>
# include <utils/eoTimer.h>
# include <eoFunctor.h>
# include <eoExceptions.h>
# include "eoMpiNode.h"
# include "eoMpiAssignmentAlgorithm.h"
// TODO TODOB comment!
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; // TODO commentaire : différence entre finir une tâche et arrêter le worker à expliciter.
const int Kill = 2;
}
const int DEFAULT_MASTER = 0;
template< typename JobData, typename Wrapped >
struct SharedDataFunction
{
SharedDataFunction( Wrapped * w = 0 ) : _wrapped( w )
{
// empty
}
void wrapped( Wrapped * w )
{
_wrapped = w;
}
void data( JobData* _d )
{
d = _d;
if( _wrapped )
{
_wrapped->data( _d );
}
}
protected:
JobData* d;
Wrapped* _wrapped;
};
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() {}
};
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() {}
};
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() {}
};
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() {}
};
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
}
SendTaskFunction<JobData> & sendTask() { return *_stf; }
HandleResponseFunction<JobData> & handleResponse() { return *_hrf; }
ProcessTaskFunction<JobData> & processTask() { return *_ptf; }
IsFinishedFunction<JobData> & isFinished() { return *_iff; }
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;
}
}
// TODO commenter : laissé à la couche d'en dessous car impossible d'initialiser une donnée membre d'une classe mère depuis une classe fille.
virtual JobData* data() = 0;
protected:
// TODO commenter : Utiliser des pointeurs pour éviter d'écraser les fonctions wrappées
SendTaskFunction< JobData >* _stf;
HandleResponseFunction< JobData >* _hrf;
ProcessTaskFunction< JobData >* _ptf;
IsFinishedFunction< JobData >* _iff;
};
template< class JobData >
class Job
{
public:
Job( AssignmentAlgorithm& _algo,
int _masterRank,
JobStore<JobData> & store
) :
assignmentAlgo( _algo ),
masterRank( _masterRank ),
comm( Node::comm() ),
// Functors
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 == Message::Kill )
{
# 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;
bmpi::communicator& comm;
SendTaskFunction<JobData> & sendTask;
HandleResponseFunction<JobData> & handleResponse;
ProcessTaskFunction<JobData> & processTask;
IsFinishedFunction<JobData> & isFinished;
bool _isMaster;
};
}
}
# endif // __EO_MPI_H__

View file

@ -0,0 +1,219 @@
# ifndef __MPI_ASSIGNMENT_ALGORITHM_H__
# define __MPI_ASSIGNMENT_ALGORITHM_H__
# include <vector>
# include "eoMpiNode.h"
namespace eo
{
namespace mpi
{
const int REST_OF_THE_WORLD = -1;
struct AssignmentAlgorithm
{
virtual int get( ) = 0;
virtual int availableWorkers( ) = 0;
virtual void confirm( int wrkRank ) = 0;
virtual std::vector<int> idles( ) = 0;
virtual void reinit( int runs ) = 0;
};
struct DynamicAssignmentAlgorithm : public AssignmentAlgorithm
{
public:
DynamicAssignmentAlgorithm( )
{
for(int i = 1; i < Node::comm().size(); ++i)
{
availableWrk.push_back( i );
}
}
DynamicAssignmentAlgorithm( int unique )
{
availableWrk.push_back( unique );
}
DynamicAssignmentAlgorithm( const std::vector<int> & workers )
{
availableWrk = workers;
}
DynamicAssignmentAlgorithm( int first, int last )
{
if( last == REST_OF_THE_WORLD )
{
last = Node::comm().size() - 1;
}
for( int i = first; i <= last; ++i)
{
availableWrk.push_back( i );
}
}
virtual int get( )
{
int assignee = -1;
if (! availableWrk.empty() )
{
assignee = availableWrk.back();
availableWrk.pop_back();
}
return assignee;
}
int availableWorkers()
{
return availableWrk.size();
}
void confirm( int rank )
{
availableWrk.push_back( rank );
}
std::vector<int> idles( )
{
return availableWrk;
}
void reinit( int _ )
{
++_;
// nothing to do
}
protected:
std::vector< int > availableWrk;
};
struct StaticAssignmentAlgorithm : public AssignmentAlgorithm
{
public:
StaticAssignmentAlgorithm( std::vector<int>& workers, int runs )
{
init( workers, runs );
}
StaticAssignmentAlgorithm( int first, int last, int runs )
{
std::vector<int> workers;
if( last == REST_OF_THE_WORLD )
{
last = Node::comm().size() - 1;
}
for(int i = first; i <= last; ++i)
{
workers.push_back( i );
}
init( workers, runs );
}
StaticAssignmentAlgorithm( int runs )
{
std::vector<int> workers;
for(int i = 1; i < Node::comm().size(); ++i)
{
workers.push_back( i );
}
init( workers, runs );
}
StaticAssignmentAlgorithm( int unique, int runs )
{
std::vector<int> workers;
workers.push_back( unique );
init( workers, runs );
}
private:
void init( std::vector<int> & workers, int runs )
{
unsigned int nbWorkers = workers.size();
freeWorkers = nbWorkers;
attributions.reserve( nbWorkers );
busy.resize( nbWorkers, false );
// Let be the euclidean division of runs by nbWorkers :
// runs == q * nbWorkers + r, 0 <= r < nbWorkers
// This one liner affects q requests to each worker
for (unsigned int i = 0; i < nbWorkers; attributions[i++] = runs / nbWorkers) ;
// The first line computes r and the one liner affects the remaining
// r requests to workers, in ascending order
unsigned int diff = runs - (runs / nbWorkers) * nbWorkers;
for (unsigned int i = 0; i < diff; ++attributions[i++]);
realRank = workers;
}
public:
int get( )
{
int assignee = -1;
for( unsigned i = 0; i < busy.size(); ++i )
{
if( !busy[i] && attributions[i] > 0 )
{
busy[i] = true;
--freeWorkers;
assignee = realRank[ i ];
break;
}
}
return assignee;
}
int availableWorkers( )
{
return freeWorkers;
}
std::vector<int> idles()
{
std::vector<int> ret;
for(unsigned int i = 0; i < busy.size(); ++i)
{
if( !busy[i] )
{
ret.push_back( realRank[i] );
}
}
return ret;
}
void confirm( int rank )
{
int i = -1; // i is the real index in table
for( unsigned int j = 0; j < realRank.size(); ++j )
{
if( realRank[j] == rank )
{
i = j;
break;
}
}
--attributions[ i ];
busy[ i ] = false;
++freeWorkers;
}
void reinit( int runs )
{
init( realRank, runs );
}
private:
std::vector<int> attributions;
std::vector<int> realRank;
std::vector<bool> busy;
unsigned int freeWorkers;
};
}
}
# endif // __MPI_ASSIGNMENT_ALGORITHM_H__

31
eo/src/mpi/eoMpiNode.h Normal file
View file

@ -0,0 +1,31 @@
# ifndef __MPI_NODE_H__
# define __MPI_NODE_H__
# include <boost/mpi.hpp>
namespace bmpi = boost::mpi;
namespace eo
{
namespace mpi
{
class Node
{
public:
static void init( int argc, char** argv )
{
static bmpi::environment env( argc, argv );
}
static bmpi::communicator& comm()
{
return _comm;
}
protected:
static bmpi::communicator _comm;
};
}
}
# endif // __MPI_NODE_H__

View file

@ -0,0 +1,59 @@
# ifndef __EO_MULTI_PARALLEL_APPLY_H__
# define __EO_MULTI_PARALLEL_APPLY_H__
# include "eoParallelApply.h"
namespace eo
{
namespace mpi
{
template< class EOT >
class ProcessTaskParallelEval : public ProcessTaskParallelApply<EOT>
{
public:
using ProcessTaskParallelApply<EOT>::_wrapped;
using ProcessTaskParallelApply<EOT>::d;
void operator()()
{
int order = Message::Continue;
while( order != Message::Finish )
{
_wrapped->operator()();
d->comm.recv( d->masterRank, Channel::Commands, order );
}
}
~ProcessTaskParallelEval()
{
delete _wrapped;
}
};
template< class EOT >
struct ParallelEvalStore : public ParallelApplyStore< EOT >
{
using ParallelApplyStore<EOT>::wrapProcessTask;
ParallelEvalStore(
eoUF<EOT&, void> & _proc,
int _masterRank,
// long _maxTime = 0,
int _packetSize = 1
) :
ParallelApplyStore< EOT >( _proc, *( new std::vector<EOT> ), _masterRank, _packetSize )
// FIXME memory leak because of vector ==> use const correctness
{
wrapProcessTask( new ProcessTaskParallelEval<EOT> );
}
void data( std::vector<EOT>& _pop )
{
ParallelApplyStore<EOT>::_data.init( _pop );
}
};
}
}
# endif // __EO_PARALLEL_APPLY_H__

View file

@ -0,0 +1,228 @@
# ifndef __EO_PARALLEL_APPLY_H__
# define __EO_PARALLEL_APPLY_H__
# include "eoMpi.h"
# include <eoFunctor.h>
# include <vector>
namespace eo
{
namespace mpi
{
struct ParallelApplyAssignment
{
int index;
int size;
};
template<class EOT>
struct ParallelApplyData
{
ParallelApplyData(
eoUF<EOT&, void> & _proc,
std::vector<EOT> & _pop,
int _masterRank,
// long _maxTime = 0,
int _packetSize
) :
_data( &_pop ), func( _proc ), index( 0 ), size( _pop.size() ), packetSize( _packetSize ), masterRank( _masterRank ), comm( Node::comm() )
{
if ( _packetSize <= 0 )
{
throw std::runtime_error("Packet size should not be negative.");
}
tempArray = new EOT[ _packetSize ];
}
void init( std::vector<EOT>& _pop )
{
index = 0;
size = _pop.size();
_data = &_pop;
assignedTasks.clear();
}
~ParallelApplyData()
{
delete [] tempArray;
}
std::vector<EOT>& data()
{
return *_data;
}
std::vector<EOT> * _data;
eoUF<EOT&, void> & func;
int index;
int size;
std::map< int /* worker rank */, ParallelApplyAssignment /* min indexes in vector */> assignedTasks;
int packetSize;
EOT* tempArray;
int masterRank;
bmpi::communicator& comm;
};
template< class EOT >
class SendTaskParallelApply : public SendTaskFunction< ParallelApplyData<EOT> >
{
public:
using SendTaskFunction< ParallelApplyData<EOT> >::d;
SendTaskParallelApply( SendTaskParallelApply<EOT> * w = 0 ) : SendTaskFunction< ParallelApplyData<EOT> >( w )
{
// empty
}
void operator()(int wrkRank)
{
int futureIndex;
if( d->index + d->packetSize < d->size )
{
futureIndex = d->index + d->packetSize;
} else {
futureIndex = d->size;
}
int sentSize = futureIndex - d->index ;
d->comm.send( wrkRank, 1, sentSize );
eo::log << eo::progress << "Evaluating individual " << d->index << std::endl;
d->assignedTasks[ wrkRank ].index = d->index;
d->assignedTasks[ wrkRank ].size = sentSize;
d->comm.send( wrkRank, 1, & ( (d->data())[ d->index ] ) , sentSize );
d->index = futureIndex;
}
};
template< class EOT >
class HandleResponseParallelApply : public HandleResponseFunction< ParallelApplyData<EOT> >
{
public:
using HandleResponseFunction< ParallelApplyData<EOT> >::d;
HandleResponseParallelApply( HandleResponseParallelApply<EOT> * w = 0 ) : HandleResponseFunction< ParallelApplyData<EOT> >( w )
{
// empty
}
void operator()(int wrkRank)
{
d->comm.recv( wrkRank, 1, & (d->data()[ d->assignedTasks[wrkRank].index ] ), d->assignedTasks[wrkRank].size );
}
};
template< class EOT >
class ProcessTaskParallelApply : public ProcessTaskFunction< ParallelApplyData<EOT> >
{
public:
using ProcessTaskFunction< ParallelApplyData<EOT> >::d;
ProcessTaskParallelApply( ProcessTaskParallelApply<EOT> * w = 0 ) : ProcessTaskFunction< ParallelApplyData<EOT> >( w )
{
// empty
}
void operator()()
{
int recvSize;
d->comm.recv( d->masterRank, 1, recvSize );
d->comm.recv( d->masterRank, 1, d->tempArray, recvSize );
timerStat.start("worker_processes");
for( int i = 0; i < recvSize ; ++i )
{
d->func( d->tempArray[ i ] );
}
timerStat.stop("worker_processes");
d->comm.send( d->masterRank, 1, d->tempArray, recvSize );
}
};
template< class EOT >
class IsFinishedParallelApply : public IsFinishedFunction< ParallelApplyData<EOT> >
{
public:
using IsFinishedFunction< ParallelApplyData<EOT> >::d;
IsFinishedParallelApply( IsFinishedParallelApply<EOT> * w = 0 ) : IsFinishedFunction< ParallelApplyData<EOT> >( w )
{
// empty
}
bool operator()()
{
return d->index == d->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,
std::vector<EOT>& _pop,
int _masterRank,
// long _maxTime = 0,
int _packetSize = 1,
// JobStore functors
SendTaskParallelApply<EOT> * stpa = new SendTaskParallelApply<EOT>,
HandleResponseParallelApply<EOT>* hrpa = new HandleResponseParallelApply<EOT>,
ProcessTaskParallelApply<EOT>* ptpa = new ProcessTaskParallelApply<EOT>,
IsFinishedParallelApply<EOT>* ifpa = new IsFinishedParallelApply<EOT>
) :
_data( _proc, _pop, _masterRank, _packetSize )
{
_stf = stpa;
_hrf = hrpa;
_ptf = ptpa;
_iff = ifpa;
}
ParallelApplyData<EOT>* data() { return &_data; }
virtual ~ParallelApplyStore()
{
delete _stf;
delete _hrf;
delete _ptf;
delete _iff;
}
protected:
ParallelApplyData<EOT> _data;
};
// TODO commentaire : impossible de faire un typedef sur un template sans passer
// par un traits => complique la tâche de l'utilisateur pour rien.
template< typename EOT >
class ParallelApply : public Job< ParallelApplyData<EOT> >
{
public:
ParallelApply(
AssignmentAlgorithm & algo,
int _masterRank,
ParallelApplyStore<EOT> & store
) :
Job< ParallelApplyData<EOT> >( algo, _masterRank, store )
{
// empty
}
};
}
}
# endif // __EO_PARALLEL_APPLY_H__

121
eo/src/mpi/eoTerminateJob.h Normal file
View file

@ -0,0 +1,121 @@
# ifndef __EO_TERMINATE_H__
# define __EO_TERMINATE_H__
# include "eoMpi.h"
namespace eo
{
namespace mpi
{
struct DummySendTaskFunction : public SendTaskFunction<void>
{
void operator()( int _ )
{
++_;
}
};
struct DummyHandleResponseFunction : public HandleResponseFunction<void>
{
void operator()( int _ )
{
++_;
}
};
struct DummyProcessTaskFunction : public ProcessTaskFunction<void>
{
void operator()()
{
// nothing!
}
};
struct DummyIsFinishedFunction : public IsFinishedFunction<void>
{
bool operator()()
{
return true;
}
};
struct DummyJobStore : public JobStore<void>
{
using JobStore<void>::_stf;
using JobStore<void>::_hrf;
using JobStore<void>::_ptf;
using JobStore<void>::_iff;
DummyJobStore()
{
_stf = new DummySendTaskFunction;
_hrf = new DummyHandleResponseFunction;
_ptf = new DummyProcessTaskFunction;
_iff = new DummyIsFinishedFunction;
}
~DummyJobStore()
{
delete _stf;
delete _hrf;
delete _ptf;
delete _iff;
}
void* data() { return 0; }
};
struct EmptyJob : public Job<void>
{
EmptyJob( AssignmentAlgorithm& algo, int masterRank ) :
Job<void>( algo, masterRank, *(new DummyJobStore) )
// FIXME memory leak => will be corrected by using const correctness
{
// empty
}
~EmptyJob()
{
std::vector< int > idles = assignmentAlgo.idles();
for(unsigned i = 0, size = idles.size(); i < size; ++i)
{
comm.send( idles[i], Channel::Commands, Message::Kill );
}
}
};
/*
class TerminateJob : public Job
{
public:
TerminateJob( AssignmentAlgorithm& algo, int _ )
: Job( algo, _ )
{
// empty
}
void sendTask( int wrkRank )
{
// empty
}
void handleResponse( int wrkRank )
{
// empty
}
void processTask( )
{
// empty
}
bool isFinished()
{
return true;
}
};
*/
}
}
# endif // __EO_TERMINATE_H__

38
eo/src/serial/Array.cpp Normal file
View file

@ -0,0 +1,38 @@
# include "Array.h"
namespace eoserial
{
std::ostream& Array::print( std::ostream& out ) const
{
out << "[";
bool first = true;
for (ArrayChildren::const_iterator it = begin(),
end = this->end();
it != end;
++it)
{
if ( first )
{
first = false;
} else {
out << ", ";
}
(*it)->print( out );
}
out << "]\n";
return out;
}
Array::~Array()
{
for (ArrayChildren::iterator it = begin(),
end = this->end();
it != end;
++it)
{
delete *it;
}
}
} // namespace eoserial

149
eo/src/serial/Array.h Normal file
View file

@ -0,0 +1,149 @@
# ifndef __EOSERIAL_ARRAY_H__
# define __EOSERIAL_ARRAY_H__
# include <vector>
# include <iostream>
# include "Entity.h"
# include "Serializable.h"
# include "Object.h"
# include "String.h"
namespace eoserial
{
// Forward declaration for below declarations.
class Array;
/*
* Declarations of functions present in Utils.h
* These are put here to avoid instead of including the file Utils.h, which would
* cause a circular inclusion.
*/
template< class T >
void unpack( const Array & array, unsigned int index, T & value );
void unpackObject( const Array & array, unsigned int index, Persistent & value );
template< class Container, template<class> class UnpackAlgorithm >
void unpackArray( const Array & array, unsigned int index, Container & container );
/**
* @brief Represents a JSON array.
*
* Wrapper for an array, so as to be used as a JSON object.
*/
class Array : public eoserial::Entity, public std::vector< eoserial::Entity* >
{
protected:
typedef std::vector< eoserial::Entity* > ArrayChildren;
public:
/**
* @brief Adds the serializable object as a JSON object.
* @param obj Object which implemnets JsonSerializable.
*/
void push_back( const eoserial::Printable* obj )
{
ArrayChildren::push_back( obj->pack() );
}
/**
* @brief Proxy for vector::push_back.
*/
void push_back( eoserial::Entity* json )
{
ArrayChildren::push_back( json );
}
/**
* @brief Prints the JSON array into the given stream.
* @param out The stream
*/
virtual std::ostream& print( std::ostream& out ) const;
/**
* @brief Dtor
*/
~Array();
/*
* The following parts allows the user to automatically deserialize an eoserial::Array into a
* standard container, by giving the algorithm which will be used to deserialize contained entities.
*/
/**
* @brief Functor which determines how to retrieve the real value contained in a eoserial::Entity at
* a given place.
*
* It will be applied for each contained variable in the array.
*/
template<class Container>
struct BaseAlgorithm
{
/**
* @brief Main operator.
*
* @param array The eoserial::Array from which we're reading.
* @param i The index of the contained value.
* @param container The standard (STL) container in which we'll push back the read value.
*/
virtual void operator()( const eoserial::Array& array, unsigned int i, Container & container ) const = 0;
};
/**
* @brief BaseAlgorithm for retrieving primitive variables.
*
* This one should be used to retrieve primitive (and types which implement operator>>) variables, for instance
* int, double, std::string, etc...
*/
template<typename C>
struct UnpackAlgorithm : public BaseAlgorithm<C>
{
void operator()( const eoserial::Array& array, unsigned int i, C & container ) const
{
typename C::value_type t;
unpack( array, i, t );
container.push_back( t );
}
};
/**
* @brief BaseAlgorithm for retrieving eoserial::Persistent objects.
*
* This one should be used to retrieve objects which implement eoserial::Persistent.
*/
template<typename C>
struct UnpackObjectAlgorithm : public BaseAlgorithm<C>
{
void operator()( const eoserial::Array& array, unsigned int i, C & container ) const
{
typename C::value_type t;
unpackObject( array, i, t );
container.push_back( t );
}
};
/**
* @brief General algorithm for array deserialization.
*
* Applies the BaseAlgorithm to each contained variable in the eoserial::Array.
*/
template<class Container, template<class T> class UnpackAlgorithm>
inline void deserialize( Container & array )
{
UnpackAlgorithm< Container > algo;
for( unsigned int i = 0, size = this->size();
i < size;
++i)
{
algo( *this, i, array );
}
}
};
} // namespace eoserial
# endif // __EOSERIAL_ARRAY_H__

View file

@ -0,0 +1,35 @@
######################################################################################
### 1) Include the sources
######################################################################################
INCLUDE_DIRECTORIES(${EO_SOURCE_DIR}/src)
INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR})
######################################################################################
### 2) Define the eoserial target
######################################################################################
SET(EOSERIAL_LIB_OUTPUT_PATH ${EO_BINARY_DIR}/lib)
SET(LIBRARY_OUTPUT_PATH ${EOSERIAL_LIB_OUTPUT_PATH})
SET(EOSERIAL_SOURCES
Array.cpp
Object.cpp
Parser.cpp
String.cpp
)
ADD_LIBRARY(eoserial STATIC ${EOSERIAL_SOURCES})
INSTALL(TARGETS eoserial ARCHIVE DESTINATION lib COMPONENT libraries)
FILE(GLOB HDRS *.h)
INSTALL(FILES ${HDRS} DESTINATION include/eo/serial COMPONENT headers)
######################################################################################
### 3) Optionnal
######################################################################################
SET(EOSERIAL_VERSION ${GLOBAL_VERSION})
SET_TARGET_PROPERTIES(eoserial PROPERTIES VERSION "${EOSERIAL_VERSION}")
######################################################################################

34
eo/src/serial/Entity.h Normal file
View file

@ -0,0 +1,34 @@
# ifndef __EOSERIAL_ENTITY_H__
# define __EOSERIAL_ENTITY_H__
# include <iostream>
# include <sstream>
namespace eoserial
{
/**
* @brief JSON entity
*
* This class represents a JSON entity, which can be JSON objects,
* strings or arrays. It is the base class for the JSON hierarchy.
*/
class Entity
{
public:
/**
* Virtual dtor (base class).
*/
virtual ~Entity() { /* empty */ }
/**
* @brief Prints the content of a JSON object into a stream.
* @param out The stream in which we're printing.
*/
virtual std::ostream& print( std::ostream& out ) const = 0;
};
} // namespace eoserial
# endif // __ENTITY_H__

Some files were not shown because too many files have changed in this diff Show more