Moved mpi related files into mpi/ and compiling it as an extern library.

eo/src/CMakeLists.txt

@@ -5,7 +5,7 @@
 INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR})
 
 ######################################################################################
-### 2) Optionnal: add MPI and Boost MPI dependencies.
+### 2) Optional: add MPI and Boost MPI dependencies.
 ######################################################################################
 
 IF(WITH_MPI)
@@ -25,6 +25,8 @@ IF(WITH_MPI)
     ADD_DEFINITIONS(-DWITH_MPI)
 
     LINK_LIBRARIES(boost_mpi boost_serialization)
+
+    ADD_SUBDIRECTORY(mpi)
 ENDIF()
 
 ######################################################################################
@@ -41,7 +43,6 @@ SET(EO_SOURCES
   eoCtrlCContinue.cpp
   eoScalarFitnessAssembled.cpp
   eoSIGContinue.cpp
-  eompi.cpp
   )
 
 ADD_LIBRARY(eo STATIC ${EO_SOURCES})

eo/src/apply.h

@@ -34,8 +34,8 @@
 #include <omp.h>
 
 # ifdef WITH_MPI
-# include <eompi.h>
+# include <mpi/eompi.h>
-# include <eoParallelApply.h>
+# include <mpi/eoParallelApply.h>
 # endif // WITH_MPI
 
 /**

eo/src/mpi/CMakeLists.txt

@@ -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}")
+
+######################################################################################

eo/src/mpi/eoParallelApply.h

@@ -0,0 +1,45 @@
+# ifndef __EO_PARALLEL_APPLY_H__
+# define __EO_PARALLEL_APPLY_H__
+
+# include "eompi.h"
+
+# include <eoFunctor.h>
+# include <vector>
+
+template< typename EOT >
+class ParallelApply : public MpiJob< EOT >
+{
+    public:
+
+        ParallelApply( eoUF<EOT&, void> & _proc, std::vector<EOT>& _pop ) :
+            MpiJob<EOT>( _pop ),
+            func( _proc )
+        {
+            // empty
+        }
+
+        virtual void sendTask( int wrkRank, int index )
+        {
+            MpiJob<EOT>::comm.send( wrkRank, 1, MpiJob<EOT>::data[ index ] );
+        }
+
+        virtual void handleResponse( int wrkRank, int index )
+        {
+            MpiJob<EOT>::comm.recv( wrkRank, 1, MpiJob<EOT>::data[ index ] );
+        }
+
+        virtual void processTask( )
+        {
+            EOT ind;
+            MpiJob<EOT>::comm.recv( 0, 1, ind );
+            func( ind );
+            MpiJob<EOT>::comm.send( 0, 1, ind );
+        }
+
+    protected:
+        eoUF<EOT&, void>& func;
+};
+
+# endif // __EO_PARALLEL_APPLY_H__
+
+

eo/src/mpi/eompi.cpp

@@ -0,0 +1,4 @@
+# include "eompi.h"
+
+MpiNode* MpiNodeStore::singleton;
+

eo/src/mpi/eompi.h

@@ -0,0 +1,287 @@
+# ifndef __EO_MPI_H__
+# define __EO_MPI_H__
+
+# include <vector>
+# include <map>
+
+# include <boost/mpi.hpp>
+namespace mpi = boost::mpi;
+
+# include <iostream>
+using namespace std;
+// TODO TODOB comment!
+
+namespace EoMpi
+{
+    namespace Channel
+    {
+        const int Commands = 0;
+    }
+
+    namespace Message
+    {
+        const int Continue = 0;
+        const int Finish = 1;
+    }
+}
+
+class MpiNode;
+
+class MpiNodeStore
+{
+    public:
+
+    static void instance( MpiNode* _instance )
+    {
+        singleton = _instance;
+    }
+
+    static MpiNode* instance()
+    {
+        return singleton;
+    }
+
+    protected:
+
+    static MpiNode* singleton;
+};
+
+class MpiNode
+{
+protected:
+    mpi::environment& env;
+    mpi::communicator& _comm;
+
+    int rank;
+    int size;
+
+    int argc;
+    char** argv;
+
+public:
+    MpiNode( mpi::environment& _env, mpi::communicator& __comm ) :
+        env(_env),
+        _comm(__comm),
+        rank(__comm.rank()),
+        size(__comm.size())
+    {
+        // empty
+    }
+
+    virtual ~MpiNode()
+    {
+        // empty
+    }
+
+    mpi::communicator& comm()
+    {
+        return _comm;
+    }
+};
+
+struct AssignmentAlgorithm
+{
+    virtual int get( ) = 0;
+    virtual void size( int s ) = 0;
+    virtual int size( ) = 0;
+    virtual void confirm( int wrkRank ) = 0;
+};
+
+struct DynamicAssignmentAlgorithm : public AssignmentAlgorithm
+{
+    public:
+        virtual int get( )
+        {
+            int assignee = -1;
+            if (! availableWrk.empty() )
+            {
+                assignee = availableWrk.back();
+                availableWrk.pop_back();
+            }
+            return assignee;
+        }
+;
+        void size( int s )
+        {
+            for( int i = 1; i < s ; ++i )
+            {
+                availableWrk.push_back( i );
+            }
+        }
+
+        int size()
+        {
+            return availableWrk.size();
+        }
+
+        void confirm( int rank )
+        {
+            availableWrk.push_back( rank );
+        }
+
+    protected:
+        std::vector< int > availableWrk;
+};
+
+template< typename EOT >
+class MpiJob
+{
+    public:
+
+    MpiJob( std::vector< EOT > & _data ) :
+        data( _data ),
+        comm( MpiNodeStore::instance()->comm() )
+    {
+        // empty
+    }
+
+    // master
+    virtual void sendTask( int wrkRank, int index ) = 0;
+    virtual void handleResponse( int wrkRank, int index ) = 0;
+    // worker
+    virtual void processTask( ) = 0;
+
+    void master( AssignmentAlgorithm & assignmentAlgorithm )
+    {
+        for( int i = 0, size = data.size();
+                i < size;
+                ++i)
+        {
+            cout << "Beginning loop for i = " << i << endl;
+            int assignee = assignmentAlgorithm.get( );
+            cout << "Assignee : " << assignee << endl;
+            while( assignee <= 0 )
+            {
+                cout << "Waitin' for node..." << endl;
+                mpi::status status = comm.probe( mpi::any_source, mpi::any_tag );
+                int wrkRank = status.source();
+                cout << "Node " << wrkRank << " just terminated." << endl;
+                handleResponse( wrkRank, assignedTasks[ wrkRank ] );
+                assignmentAlgorithm.confirm( wrkRank );
+                assignee = assignmentAlgorithm.get( );
+            }
+            cout << "Assignee found : " << assignee << endl;
+            assignedTasks[ assignee ] = i;
+            comm.send( assignee, EoMpi::Channel::Commands, EoMpi::Message::Continue );
+            sendTask( assignee, i );
+        }
+
+        // frees all the idle workers
+        int idle;
+        vector<int> idles;
+        while ( ( idle = assignmentAlgorithm.get( ) ) > 0 )
+        {
+            comm.send( idle, EoMpi::Channel::Commands, EoMpi::Message::Finish );
+            idles.push_back( idle );
+        }
+        for (int i = 0; i < idles.size(); ++i)
+        {
+            assignmentAlgorithm.confirm( idles[i] );
+        }
+
+        // wait for all responses
+        int wrkNb = comm.size() - 1;
+        while( assignmentAlgorithm.size() != wrkNb )
+        {
+            mpi::status status = comm.probe( mpi::any_source, mpi::any_tag );
+            int wrkRank = status.source();
+            handleResponse( wrkRank, assignedTasks[ wrkRank ] );
+            comm.send( wrkRank, EoMpi::Channel::Commands, EoMpi::Message::Finish );
+            assignmentAlgorithm.confirm( wrkRank );
+        }
+    }
+
+    void worker( )
+    {
+        int order;
+        while( true )
+        {
+            comm.recv( 0, EoMpi::Channel::Commands, order );
+            if ( order == EoMpi::Message::Finish )
+            {
+                return;
+            } else
+            {
+                processTask( );
+            }
+        }
+    }
+
+protected:
+
+    std::vector<EOT> & data;
+    std::map< int /* worker rank */, int /* index in vector */> assignedTasks;
+
+    mpi::communicator& comm;
+};
+
+class MasterNode : public MpiNode
+{
+public:
+    MasterNode( int _argc, char** _argv,
+            mpi::environment& _env,
+            mpi::communicator& _comm
+            ) :
+        MpiNode(_env, _comm )
+    {
+        // empty
+    }
+
+    void setAssignmentAlgorithm( AssignmentAlgorithm* assignmentAlgo )
+    {
+        _assignmentAlgo = assignmentAlgo;
+        _assignmentAlgo->size( _comm.size() );
+    }
+
+    template< typename EOT >
+    void run( MpiJob< EOT > & job )
+    {
+        job.master( *_assignmentAlgo );
+    }
+
+protected:
+    AssignmentAlgorithm* _assignmentAlgo;
+};
+
+class WorkerNode : public MpiNode
+{
+    public:
+
+        WorkerNode(
+                int _argc, char** _argv,
+                mpi::environment& _env,
+                mpi::communicator& _comm ) :
+            MpiNode( _env, _comm )
+        {
+            // empty
+        }
+
+        template< typename EOT >
+        void run( MpiJob<EOT> & job )
+        {
+            job.worker( );
+        }
+};
+
+class MpiSingletonFactory
+{
+    public:
+
+    static void init( int argc, char** argv )
+    {
+        MpiNode* singleton;
+        //mpi::environment* env = new mpi::environment ( argc, argv );
+        //mpi::communicator* world = new mpi::communicator; // TODO clean
+        static mpi::environment env( argc, argv );
+        static mpi::communicator world;
+        if ( world.rank() == 0 )
+        {
+            singleton = new MasterNode( argc, argv, env, world );
+        } else
+        {
+            singleton = new WorkerNode( argc, argv, env, world );
+        }
+        MpiNodeStore::instance( singleton );
+    }
+};
+# endif // __EO_MPI_H__

eo/test/t-eoMpiParallelApply.cpp

@@ -1,5 +1,5 @@
-# include <eompi.h>
+# include <mpi/eompi.h>
-# include <eoParallelApply.h>
+# include <mpi/eoParallelApply.h>
 # include <iostream>
 
 # include <vector>