nojhan/eodev
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Putting everything in namespace eo::mpi

Browse source
This commit is contained in:
Benjamin Bouvier 2012-06-25 14:11:44 +02:00
commit b291e56e03
12 changed files with 492 additions and 466 deletions
Download patch file Download diff file Expand all files Collapse all files

183
eo/src/mpi/eoParallelApply.h
View file

@ -1,102 +1,107 @@
# ifndef __EO_PARALLEL_APPLY_H__
# define __EO_PARALLEL_APPLY_H__
# include "eompi.h"
# include "eoMpi.h"
# include <eoFunctor.h>
# include <vector>
template< typename EOT >
class ParallelApply : public MpiJob
namespace eo
{
private:
struct ParallelApplyAssignment
namespace mpi
{
template< typename EOT >
class ParallelApply : public Job
{
int index;
int size;
private:
struct ParallelApplyAssignment
{
int index;
int size;
};
public:
ParallelApply(
eoUF<EOT&, void> & _proc,
std::vector<EOT>& _pop,
AssignmentAlgorithm & algo,
int _masterRank,
int _packetSize = 1
) :
Job( algo, _masterRank ),
func( _proc ),
index( 0 ),
size( _pop.size() ),
data( _pop ),
packetSize( _packetSize )
{
if ( _packetSize <= 0 )
{
throw std::runtime_error("Packet size should not be negative.");
}
tempArray = new EOT[ packetSize ];
}
~ParallelApply()
{
delete [] tempArray;
}
virtual void sendTask( int wrkRank )
{
int futureIndex;
if( index + packetSize < size )
{
futureIndex = index + packetSize;
} else {
futureIndex = size;
}
int sentSize = futureIndex - index ;
comm.send( wrkRank, 1, sentSize );
assignedTasks[ wrkRank ].index = index;
assignedTasks[ wrkRank ].size = sentSize;
comm.send( wrkRank, 1, &data[ index ] , sentSize );
index = futureIndex;
}
virtual void handleResponse( int wrkRank )
{
comm.recv( wrkRank, 1, &data[ assignedTasks[wrkRank].index ], assignedTasks[wrkRank].size );
}
virtual void processTask( )
{
int recvSize;
comm.recv( masterRank, 1, recvSize );
comm.recv( masterRank, 1, tempArray, recvSize );
for( int i = 0; i < recvSize ; ++i )
{
func( tempArray[ i ] );
}
comm.send( masterRank, 1, tempArray, recvSize );
}
bool isFinished()
{
return index == size;
}
protected:
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;
};
public:
ParallelApply(
eoUF<EOT&, void> & _proc,
std::vector<EOT>& _pop,
AssignmentAlgorithm & algo,
int _masterRank,
int _packetSize = 1
) :
MpiJob( algo, _masterRank ),
func( _proc ),
index( 0 ),
size( _pop.size() ),
data( _pop ),
packetSize( _packetSize )
{
if ( _packetSize <= 0 )
{
throw std::runtime_error("Packet size should not be negative.");
}
tempArray = new EOT[ packetSize ];
}
~ParallelApply()
{
delete [] tempArray;
}
virtual void sendTask( int wrkRank )
{
int futureIndex;
if( index + packetSize < size )
{
futureIndex = index + packetSize;
} else {
futureIndex = size;
}
int sentSize = futureIndex - index ;
comm.send( wrkRank, 1, sentSize );
assignedTasks[ wrkRank ].index = index;
assignedTasks[ wrkRank ].size = sentSize;
comm.send( wrkRank, 1, &data[ index ] , sentSize );
index = futureIndex;
}
virtual void handleResponse( int wrkRank )
{
comm.recv( wrkRank, 1, &data[ assignedTasks[wrkRank].index ], assignedTasks[wrkRank].size );
}
virtual void processTask( )
{
int recvSize;
comm.recv( masterRank, 1, recvSize );
comm.recv( masterRank, 1, tempArray, recvSize );
for( int i = 0; i < recvSize ; ++i )
{
func( tempArray[ i ] );
}
comm.send( masterRank, 1, tempArray, recvSize );
}
bool isFinished()
{
return index == size;
}
protected:
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;
};
}
}
# endif // __EO_PARALLEL_APPLY_H__