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

git-svn-id: svn://scm.gforge.inria.fr/svnroot/paradiseo@1202 331e1502-861f-0410-8da2-ba01fb791d7f

Browse source
This commit is contained in:
atantar 2008-06-15 16:29:20 +00:00
commit 2622e3f2a2
1 changed files with 124 additions and 139 deletions
Download patch file Download diff file Expand all files Collapse all files

269
trunk/paradiseo-peo/src/peoSyncDataTransfer.h
View file

@ -3,7 +3,7 @@
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2008 * Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2008
* (C) OPAC Team, LIFL, 2002-2008 * (C) OPAC Team, LIFL, 2002-2008
* *
* Clive Canape * Alexandru-Adrian TANTAR
* *
* This software is governed by the CeCILL license under French law and * This software is governed by the CeCILL license under French law and
* abiding by the rules of distribution of free software. You can use, * abiding by the rules of distribution of free software. You can use,
@ -57,6 +57,10 @@
#include "rmc/mpi/synchron.h" #include "rmc/mpi/synchron.h"
extern void wakeUpCommunicator();
extern int getNodeRank();
class peoSyncDataTransfer : public Cooperative, public eoUpdater class peoSyncDataTransfer : public Cooperative, public eoUpdater
{ {
@ -131,28 +135,140 @@ class peoSyncDataTransfer : public Cooperative, public eoUpdater
} }
void operator()(); void operator()()
{
void pack(); standbyTransfer = false;
nbTransfersIn = nbTransfersOut = 0;
topology.setNeighbors( this, in, out );
all = topology;
synchronizeCoopEx();
stop();
// sending data out
sendData();
// synchronizing
sem_wait( &sync );
// receiving data in
receiveData();
synchronizeCoopEx();
stop();
}
void unpack(); void pack()
{
::pack( coop_em.front()->getKey() );
source->packMessage();
coop_em.pop();
}
void packSynchronizeReq(); void unpack()
{
destination->unpackMessage();
}
void notifySending(); void packSynchronizeReq()
{
packSynchronRequest( all );
}
void notifyReceiving(); void notifySending()
{
void notifySendingSyncReq(); nbTransfersOut++;
printDebugMessage( "peoSyncDataTransfer: notified of the completion of a transfer round." );
getOwner()->setActive();
if ( nbTransfersOut == out.size() && nbTransfersIn < in.size() )
{
getOwner()->setPassive();
}
}
void notifySynchronized();
void 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 );
}
}
void notifySendingSyncReq()
{
getOwner()->setPassive();
printDebugMessage( "peoSyncIslandMig: synchronization request sent." );
}
void notifySynchronized()
{
printDebugMessage( "peoSyncIslandMig: cooperators synchronized." );
standbyTransfer = true;
getOwner()->setActive();
resume();
}
private: private:
void sendData(); void sendData()
void receiveData(); {
for ( unsigned i = 0; i < out.size(); i ++ )
{
source->pushMessage();
coop_em.push( out[ i ] );
send( out[ i ]);
printDebugMessage( "peoSyncDataTransfer: sending data." );
}
wakeUpCommunicator();
}
void receiveData()
{
assert( !( destination->empty() ) );
while ( !( destination->empty() ) )
{
printDebugMessage( "peoSyncDataTransfer: received data." );
destination->popMessage();
printDebugMessage( "peoSyncDataTransfer: done extracting received data." );
}
}
Topology& topology; // neighboring topology Topology& topology; // neighboring topology
// source and destination end-points // source and destination end-points
@ -169,135 +285,4 @@ class peoSyncDataTransfer : public Cooperative, public eoUpdater
unsigned nbTransfersIn, nbTransfersOut; unsigned nbTransfersIn, nbTransfersOut;
}; };
void peoSyncDataTransfer :: pack()
{
::pack( coop_em.front()->getKey() );
source->packMessage();
coop_em.pop();
}
void peoSyncDataTransfer :: unpack()
{
destination->unpackMessage();
}
void peoSyncDataTransfer :: packSynchronizeReq()
{
packSynchronRequest( all );
}
extern void wakeUpCommunicator();
extern int getNodeRank();
void peoSyncDataTransfer :: sendData()
{
for ( unsigned i = 0; i < out.size(); i ++ )
{
source->pushMessage();
coop_em.push( out[ i ] );
send( out[ i ]);
printDebugMessage( "peoSyncDataTransfer: sending data." );
}
wakeUpCommunicator();
}
void peoSyncDataTransfer :: receiveData()
{
assert( !( destination->empty() ) );
while ( !( destination->empty() ) )
{
printDebugMessage( "peoSyncDataTransfer: received data." );
destination->popMessage();
printDebugMessage( "peoSyncDataTransfer: done extracting received data." );
}
}
void peoSyncDataTransfer :: operator()()
{
standbyTransfer = false;
nbTransfersIn = nbTransfersOut = 0;
topology.setNeighbors( this, in, out );
all = topology;
synchronizeCoopEx();
stop();
// sending data out
sendData();
// synchronizing
sem_wait( &sync );
// receiving data in
receiveData();
synchronizeCoopEx();
stop();
}
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();
}
}
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 );
}
}
void peoSyncDataTransfer :: notifySendingSyncReq ()
{
getOwner()->setPassive();
printDebugMessage( "peoSyncIslandMig: synchronization request sent." );
}
void peoSyncDataTransfer :: notifySynchronized ()
{
printDebugMessage( "peoSyncIslandMig: cooperators synchronized." );
standbyTransfer = true;
getOwner()->setActive();
resume();
}
#endif #endif