peo style
git-svn-id: svn://scm.gforge.inria.fr/svnroot/paradiseo@1120 331e1502-861f-0410-8da2-ba01fb791d7f
This commit is contained in:
canape
parent
32c90fd740
commit
ceeaa4b533
5 changed files with 677 additions and 575 deletions
|
|
@ -22,131 +22,134 @@
|
|||
|
||||
|
||||
class peoSyncDataTransfer : public Cooperative, public eoUpdater
|
||||
{
|
||||
|
||||
public:
|
||||
|
||||
template< typename EndPointType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
EndPointType& __endPoint,
|
||||
Topology& __topology
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< EndPointType >( __endPoint );
|
||||
destination = new MsgTransferQueue< EndPointType >( __endPoint );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
template< typename EndPointType, typename FunctorType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
EndPointType& __endPoint,
|
||||
Topology& __topology,
|
||||
FunctorType& externalFunctorRef
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< EndPointType >( __endPoint, externalFunctorRef );
|
||||
destination = new MsgTransferQueue< EndPointType >( __endPoint, externalFunctorRef );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
template< typename SourceEndPointType, typename DestinationEndPointType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
SourceEndPointType& __source,
|
||||
DestinationEndPointType& __destination,
|
||||
Topology& __topology
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< SourceEndPointType >( __source );
|
||||
destination = new MsgTransferQueue< DestinationEndPointType >( __destination );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
template< typename SourceEndPointType, typename DestinationEndPointType, typename FunctorType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
SourceEndPointType& __source,
|
||||
DestinationEndPointType& __destination,
|
||||
Topology& __topology,
|
||||
FunctorType& externalFunctorRef
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< SourceEndPointType >( __source, externalFunctorRef );
|
||||
destination = new MsgTransferQueue< DestinationEndPointType >( __destination, externalFunctorRef );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
|
||||
void operator()();
|
||||
|
||||
void pack();
|
||||
|
||||
void unpack();
|
||||
|
||||
void packSynchronizeReq();
|
||||
|
||||
void notifySending();
|
||||
|
||||
void notifyReceiving();
|
||||
|
||||
void notifySendingSyncReq();
|
||||
|
||||
void notifySynchronized();
|
||||
|
||||
|
||||
private:
|
||||
|
||||
void sendData();
|
||||
void receiveData();
|
||||
|
||||
Topology& topology; // neighboring topology
|
||||
|
||||
// source and destination end-points
|
||||
AbstractMsgTransferQueue* source;
|
||||
AbstractMsgTransferQueue* destination;
|
||||
|
||||
std :: queue< Cooperative* > coop_em;
|
||||
|
||||
sem_t sync;
|
||||
|
||||
bool standbyTransfer;
|
||||
|
||||
std :: vector< Cooperative* > in, out, all;
|
||||
unsigned nbTransfersIn, nbTransfersOut;
|
||||
};
|
||||
|
||||
|
||||
void peoSyncDataTransfer :: pack()
|
||||
{
|
||||
|
||||
public:
|
||||
|
||||
template< typename EndPointType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
EndPointType& __endPoint,
|
||||
Topology& __topology
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< EndPointType >( __endPoint );
|
||||
destination = new MsgTransferQueue< EndPointType >( __endPoint );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
template< typename EndPointType, typename FunctorType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
EndPointType& __endPoint,
|
||||
Topology& __topology,
|
||||
FunctorType& externalFunctorRef
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< EndPointType >( __endPoint, externalFunctorRef );
|
||||
destination = new MsgTransferQueue< EndPointType >( __endPoint, externalFunctorRef );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
template< typename SourceEndPointType, typename DestinationEndPointType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
SourceEndPointType& __source,
|
||||
DestinationEndPointType& __destination,
|
||||
Topology& __topology
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< SourceEndPointType >( __source );
|
||||
destination = new MsgTransferQueue< DestinationEndPointType >( __destination );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
template< typename SourceEndPointType, typename DestinationEndPointType, typename FunctorType >
|
||||
peoSyncDataTransfer(
|
||||
|
||||
SourceEndPointType& __source,
|
||||
DestinationEndPointType& __destination,
|
||||
Topology& __topology,
|
||||
FunctorType& externalFunctorRef
|
||||
|
||||
) : topology( __topology )
|
||||
{
|
||||
|
||||
source = new MsgTransferQueue< SourceEndPointType >( __source, externalFunctorRef );
|
||||
destination = new MsgTransferQueue< DestinationEndPointType >( __destination, externalFunctorRef );
|
||||
__topology.add( *this );
|
||||
|
||||
sem_init( &sync, 0, 0 );
|
||||
}
|
||||
|
||||
|
||||
void operator()();
|
||||
|
||||
void pack();
|
||||
|
||||
void unpack();
|
||||
|
||||
void packSynchronizeReq();
|
||||
|
||||
void notifySending();
|
||||
|
||||
void notifyReceiving();
|
||||
|
||||
void notifySendingSyncReq();
|
||||
|
||||
void notifySynchronized();
|
||||
|
||||
|
||||
private:
|
||||
|
||||
void sendData();
|
||||
void receiveData();
|
||||
|
||||
Topology& topology; // neighboring topology
|
||||
|
||||
// source and destination end-points
|
||||
AbstractMsgTransferQueue* source;
|
||||
AbstractMsgTransferQueue* destination;
|
||||
|
||||
std :: queue< Cooperative* > coop_em;
|
||||
|
||||
sem_t sync;
|
||||
|
||||
bool standbyTransfer;
|
||||
|
||||
std :: vector< Cooperative* > in, out, all;
|
||||
unsigned nbTransfersIn, nbTransfersOut;
|
||||
};
|
||||
|
||||
|
||||
void peoSyncDataTransfer :: pack() {
|
||||
|
||||
::pack( coop_em.front()->getKey() );
|
||||
source->packMessage();
|
||||
coop_em.pop();
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: unpack() {
|
||||
void peoSyncDataTransfer :: unpack()
|
||||
{
|
||||
|
||||
destination->unpackMessage();
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: packSynchronizeReq() {
|
||||
void peoSyncDataTransfer :: packSynchronizeReq()
|
||||
{
|
||||
|
||||
packSynchronRequest( all );
|
||||
}
|
||||
|
|
@ -154,92 +157,107 @@ void peoSyncDataTransfer :: packSynchronizeReq() {
|
|||
extern void wakeUpCommunicator();
|
||||
extern int getNodeRank();
|
||||
|
||||
void peoSyncDataTransfer :: sendData() {
|
||||
void peoSyncDataTransfer :: sendData()
|
||||
{
|
||||
|
||||
for ( unsigned i = 0; i < out.size(); i ++ ) {
|
||||
for ( unsigned i = 0; i < out.size(); i ++ )
|
||||
{
|
||||
|
||||
source->pushMessage();
|
||||
|
||||
coop_em.push( out[ i ] );
|
||||
send( out[ i ]);
|
||||
|
||||
printDebugMessage( "peoSyncDataTransfer: sending data." );
|
||||
}
|
||||
|
||||
source->pushMessage();
|
||||
|
||||
coop_em.push( out[ i ] );
|
||||
send( out[ i ]);
|
||||
|
||||
printDebugMessage( "peoSyncDataTransfer: sending data." );
|
||||
}
|
||||
|
||||
wakeUpCommunicator();
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: receiveData() {
|
||||
|
||||
void peoSyncDataTransfer :: receiveData()
|
||||
{
|
||||
|
||||
assert( !( destination->empty() ) );
|
||||
|
||||
while ( !( destination->empty() ) ) {
|
||||
|
||||
printDebugMessage( "peoSyncDataTransfer: received data." );
|
||||
destination->popMessage();
|
||||
printDebugMessage( "peoSyncDataTransfer: done extracting received data." );
|
||||
}
|
||||
while ( !( destination->empty() ) )
|
||||
{
|
||||
|
||||
printDebugMessage( "peoSyncDataTransfer: received data." );
|
||||
destination->popMessage();
|
||||
printDebugMessage( "peoSyncDataTransfer: done extracting received data." );
|
||||
}
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: operator()() {
|
||||
void peoSyncDataTransfer :: operator()()
|
||||
{
|
||||
|
||||
standbyTransfer = false;
|
||||
nbTransfersIn = nbTransfersOut = 0;
|
||||
|
||||
topology.setNeighbors( this, in, out ); all = topology;
|
||||
|
||||
synchronizeCoopEx(); stop();
|
||||
|
||||
|
||||
topology.setNeighbors( this, in, out );
|
||||
all = topology;
|
||||
|
||||
synchronizeCoopEx();
|
||||
stop();
|
||||
|
||||
// sending data out
|
||||
sendData();
|
||||
// synchronizing
|
||||
sem_wait( &sync );
|
||||
// receiving data in
|
||||
receiveData();
|
||||
|
||||
synchronizeCoopEx(); stop();
|
||||
|
||||
synchronizeCoopEx();
|
||||
stop();
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: notifySending() {
|
||||
void peoSyncDataTransfer :: notifySending()
|
||||
{
|
||||
|
||||
nbTransfersOut++;
|
||||
|
||||
|
||||
printDebugMessage( "peoSyncDataTransfer: notified of the completion of a transfer round." );
|
||||
|
||||
|
||||
getOwner()->setActive();
|
||||
if ( nbTransfersOut == out.size() && nbTransfersIn < in.size() ) {
|
||||
getOwner()->setPassive();
|
||||
}
|
||||
if ( nbTransfersOut == out.size() && nbTransfersIn < in.size() )
|
||||
{
|
||||
getOwner()->setPassive();
|
||||
}
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: notifyReceiving() {
|
||||
void peoSyncDataTransfer :: notifyReceiving()
|
||||
{
|
||||
|
||||
nbTransfersIn++;
|
||||
printDebugMessage( "peoSyncIslandMig: notified of incoming data." );
|
||||
|
||||
if ( standbyTransfer ) {
|
||||
getOwner()->setActive();
|
||||
if ( nbTransfersOut == out.size() && nbTransfersIn < in.size() )
|
||||
getOwner()->setPassive();
|
||||
}
|
||||
|
||||
if ( nbTransfersIn == in.size() ) {
|
||||
|
||||
printDebugMessage( "peoSyncIslandMig: finished collecting incoming data." );
|
||||
sem_post( &sync );
|
||||
}
|
||||
|
||||
if ( standbyTransfer )
|
||||
{
|
||||
getOwner()->setActive();
|
||||
if ( nbTransfersOut == out.size() && nbTransfersIn < in.size() )
|
||||
getOwner()->setPassive();
|
||||
}
|
||||
|
||||
if ( nbTransfersIn == in.size() )
|
||||
{
|
||||
|
||||
printDebugMessage( "peoSyncIslandMig: finished collecting incoming data." );
|
||||
sem_post( &sync );
|
||||
}
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: notifySendingSyncReq () {
|
||||
|
||||
void peoSyncDataTransfer :: notifySendingSyncReq ()
|
||||
{
|
||||
|
||||
getOwner()->setPassive();
|
||||
printDebugMessage( "peoSyncIslandMig: synchronization request sent." );
|
||||
}
|
||||
|
||||
void peoSyncDataTransfer :: notifySynchronized () {
|
||||
|
||||
void peoSyncDataTransfer :: notifySynchronized ()
|
||||
{
|
||||
|
||||
printDebugMessage( "peoSyncIslandMig: cooperators synchronized." );
|
||||
|
||||
|
||||
standbyTransfer = true;
|
||||
getOwner()->setActive();
|
||||
resume();
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue