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

New style for PEO

Browse source 
git-svn-id: svn://scm.gforge.inria.fr/svnroot/paradiseo@789 331e1502-861f-0410-8da2-ba01fb791d7f
This commit is contained in:
canape 2007-11-16 11:34:20 +00:00
commit 9c87b3b0c0
132 changed files with 3781 additions and 3396 deletions
Download patch file Download diff file Expand all files Collapse all files

44
trunk/paradiseo-peo/src/rmc/mpi/comm.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <comm.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -51,42 +51,48 @@ static sem_t sem_comm_init;
static Communicator * the_thread;
Communicator :: Communicator (int * __argc, char * * * __argv) {
Communicator :: Communicator (int * __argc, char * * * __argv)
{
the_thread = this;
the_thread = this;
initNode (__argc, __argv);
loadRMCParameters (* __argc, * __argv);
loadRMCParameters (* __argc, * __argv);
sem_post (& sem_comm_init);
}
void Communicator :: start () {
void Communicator :: start ()
{
while (true) {
/* Zzz Zzz Zzz :-))) */
sleep ();
sendMessages ();
while (true)
{
if (! atLeastOneActiveRunner ())
break;
receiveMessages ();
}
waitBuffers ();
/* Zzz Zzz Zzz :-))) */
sleep ();
sendMessages ();
if (! atLeastOneActiveRunner ())
break;
receiveMessages ();
}
waitBuffers ();
printDebugMessage ("finalizing");
MPI_Finalize ();
MPI_Finalize ();
}
void initCommunication () {
void initCommunication ()
{
sem_init (& sem_comm_init, 0, 0);
}
void waitNodeInitialization () {
void waitNodeInitialization ()
{
sem_wait (& sem_comm_init);
}
void wakeUpCommunicator () {
void wakeUpCommunicator ()
{
the_thread -> wakeUp ();
}

17
trunk/paradiseo-peo/src/rmc/mpi/comm.h
View file

@ -1,4 +1,4 @@
/*
/*
* <comm.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -40,15 +40,16 @@
#include "../../core/communicable.h"
#include "../../core/reac_thread.h"
class Communicator : public ReactiveThread {
class Communicator : public ReactiveThread
{
public :
/* Ctor */
Communicator (int * __argc, char * * * __argv);
public :
void start ();
};
/* Ctor */
Communicator (int * __argc, char * * * __argv);
void start ();
};
extern void initCommunication ();

19
trunk/paradiseo-peo/src/rmc/mpi/coop.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <coop.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -41,28 +41,33 @@
#include "mess.h"
#include "../../core/peo_debug.h"
Runner * Cooperative :: getOwner () {
Runner * Cooperative :: getOwner ()
{
return owner;
}
void Cooperative :: setOwner (Runner & __runner) {
void Cooperative :: setOwner (Runner & __runner)
{
owner = & __runner;
}
void Cooperative :: send (Cooperative * __coop) {
void Cooperative :: send (Cooperative * __coop)
{
:: send (this, getRankOfRunner (__coop -> getOwner () -> getID ()), COOP_TAG);
:: send (this, getRankOfRunner (__coop -> getOwner () -> getID ()), COOP_TAG);
// stop ();
}
Cooperative * getCooperative (COOP_ID __key) {
Cooperative * getCooperative (COOP_ID __key)
{
return dynamic_cast <Cooperative *> (getCommunicable (__key));
}
void Cooperative :: notifySending () {
void Cooperative :: notifySending ()
{
//getOwner -> setPassive ();
// resume ();

165
trunk/paradiseo-peo/src/rmc/mpi/mess.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <mess.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -42,54 +42,60 @@
#include "node.h"
#define MPI_BUF_SIZE 1024*64
static char mpi_buf [MPI_BUF_SIZE];
static int pos_buf ;
static std :: vector <char *> act_buf; /* Active buffers */
static std :: vector <MPI_Request *> act_req; /* Active requests */
void cleanBuffers () {
void cleanBuffers ()
{
for (unsigned i = 0; i < act_req.size ();) {
MPI_Status stat ;
int flag ;
MPI_Test (act_req [i], & flag, & stat) ;
if (flag) {
delete act_buf [i] ;
delete act_req [i] ;
act_buf [i] = act_buf.back () ;
act_buf.pop_back () ;
act_req [i] = act_req.back () ;
act_req.pop_back () ;
for (unsigned i = 0; i < act_req.size ();)
{
MPI_Status stat ;
int flag ;
MPI_Test (act_req [i], & flag, & stat) ;
if (flag)
{
delete act_buf [i] ;
delete act_req [i] ;
act_buf [i] = act_buf.back () ;
act_buf.pop_back () ;
act_req [i] = act_req.back () ;
act_req.pop_back () ;
}
else
i ++;
}
else
i ++;
}
}
void waitBuffers () {
void waitBuffers ()
{
printDebugMessage ("waiting the termination of the asynchronous operations to complete");
for (unsigned i = 0; i < act_req.size (); i ++) {
MPI_Status stat ;
for (unsigned i = 0; i < act_req.size (); i ++)
{
MPI_Wait (act_req [i], & stat) ;
delete act_buf [i] ;
delete act_req [i] ;
}
MPI_Status stat ;
MPI_Wait (act_req [i], & stat) ;
delete act_buf [i] ;
delete act_req [i] ;
}
}
bool probeMessage (int & __src, int & __tag) {
bool probeMessage (int & __src, int & __tag)
{
int flag;
@ -103,36 +109,41 @@ bool probeMessage (int & __src, int & __tag) {
return flag;
}
void waitMessage () {
void waitMessage ()
{
MPI_Status stat;
MPI_Status stat;
MPI_Probe (MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, & stat);
}
void initMessage () {
void initMessage ()
{
pos_buf = 0;
}
void sendMessage (int __to, int __tag) {
void sendMessage (int __to, int __tag)
{
cleanBuffers ();
cleanBuffers ();
act_buf.push_back (new char [pos_buf]);
act_req.push_back (new MPI_Request);
memcpy (act_buf.back (), mpi_buf, pos_buf);
MPI_Isend (act_buf.back (), pos_buf, MPI_PACKED, __to, __tag, MPI_COMM_WORLD, act_req.back ());
act_req.push_back (new MPI_Request);
memcpy (act_buf.back (), mpi_buf, pos_buf);
MPI_Isend (act_buf.back (), pos_buf, MPI_PACKED, __to, __tag, MPI_COMM_WORLD, act_req.back ());
}
void sendMessageToAll (int __tag) {
void sendMessageToAll (int __tag)
{
for (int i = 0; i < getNumberOfNodes (); i ++)
sendMessage (i, __tag);
}
void receiveMessage (int __from, int __tag) {
MPI_Status stat;
void receiveMessage (int __from, int __tag)
{
MPI_Status stat;
MPI_Request req;
MPI_Irecv (mpi_buf, MPI_BUF_SIZE, MPI_PACKED, __from, __tag, MPI_COMM_WORLD, & req) ;
@ -140,126 +151,146 @@ void receiveMessage (int __from, int __tag) {
}
/* Char */
void pack (const char & __c) {
void pack (const char & __c)
{
MPI_Pack ((void *) & __c, 1, MPI_CHAR, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Float */
void pack (const float & __f, int __nitem) {
void pack (const float & __f, int __nitem)
{
MPI_Pack ((void *) & __f, __nitem, MPI_FLOAT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Double */
void pack (const double & __d, int __nitem) {
void pack (const double & __d, int __nitem)
{
MPI_Pack ((void *) & __d, __nitem, MPI_DOUBLE, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Integer */
void pack (const int & __i, int __nitem) {
void pack (const int & __i, int __nitem)
{
MPI_Pack ((void *) & __i, __nitem, MPI_INT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Unsigned int. */
void pack (const unsigned int & __ui, int __nitem) {
void pack (const unsigned int & __ui, int __nitem)
{
MPI_Pack ((void *) & __ui, __nitem, MPI_UNSIGNED, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Short int. */
void pack (const short & __sh, int __nitem) {
void pack (const short & __sh, int __nitem)
{
MPI_Pack ((void *) & __sh, __nitem, MPI_SHORT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Unsigned short */
void pack (const unsigned short & __ush, int __nitem) {
void pack (const unsigned short & __ush, int __nitem)
{
MPI_Pack ((void *) & __ush, __nitem, MPI_UNSIGNED_SHORT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Long */
void pack (const long & __l, int __nitem) {
void pack (const long & __l, int __nitem)
{
MPI_Pack ((void *) & __l, __nitem, MPI_LONG, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Unsigned long */
void pack (const unsigned long & __ul, int __nitem) {
void pack (const unsigned long & __ul, int __nitem)
{
MPI_Pack ((void *) & __ul, __nitem, MPI_UNSIGNED_LONG, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* String */
void pack (const char * __str) {
void pack (const char * __str)
{
int len = strlen (__str) + 1;
MPI_Pack (& len, 1, MPI_INT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
MPI_Pack ((void *) __str, len, MPI_CHAR, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
}
/* Char */
void unpack (char & __c) {
void unpack (char & __c)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __c, 1, MPI_CHAR, MPI_COMM_WORLD);
}
/* Float */
void unpack (float & __f, int __nitem) {
void unpack (float & __f, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __f, __nitem, MPI_FLOAT, MPI_COMM_WORLD);
}
/* Double */
void unpack (double & __d, int __nitem) {
void unpack (double & __d, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __d, __nitem, MPI_DOUBLE, MPI_COMM_WORLD);
}
/* Integer */
void unpack (int & __i, int __nitem) {
void unpack (int & __i, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __i, __nitem, MPI_INT, MPI_COMM_WORLD);
}
/* Unsigned int. */
void unpack (unsigned int & __ui, int __nitem) {
void unpack (unsigned int & __ui, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __ui, __nitem, MPI_UNSIGNED, MPI_COMM_WORLD);
}
/* Short int. */
void unpack (short & __sh, int __nitem) {
void unpack (short & __sh, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __sh, __nitem, MPI_SHORT, MPI_COMM_WORLD);
}
/* Unsigned short */
void unpack (unsigned short & __ush, int __nitem) {
void unpack (unsigned short & __ush, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __ush, __nitem, MPI_UNSIGNED_SHORT, MPI_COMM_WORLD);
}
/* Long */
void unpack (long & __l, int __nitem) {
void unpack (long & __l, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __l, __nitem, MPI_LONG, MPI_COMM_WORLD);
}
/* Unsigned long */
void unpack (unsigned long & __ul, int __nitem) {
void unpack (unsigned long & __ul, int __nitem)
{
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __ul, __nitem, MPI_UNSIGNED_LONG, MPI_COMM_WORLD);
}
/* String */
void unpack (char * __str) {
void unpack (char * __str)
{
int len;
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & len, 1, MPI_INT, MPI_COMM_WORLD);
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, __str, len, MPI_CHAR, MPI_COMM_WORLD);
MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, __str, len, MPI_CHAR, MPI_COMM_WORLD);
}

2
trunk/paradiseo-peo/src/rmc/mpi/mess.h
View file

@ -1,4 +1,4 @@
/*
/*
* <mess.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007

41
trunk/paradiseo-peo/src/rmc/mpi/node.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <node.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -46,41 +46,46 @@ static std :: map <std :: string, int> name_to_rk;
static std :: vector <std :: string> rk_to_name;
int getNodeRank () {
int getNodeRank ()
{
return rk;
}
int getNumberOfNodes () {
int getNumberOfNodes ()
{
return sz;
}
int getRankFromName (const std :: string & __name) {
return atoi (__name.c_str ());
int getRankFromName (const std :: string & __name)
{
return atoi (__name.c_str ());
}
void initNode (int * __argc, char * * * __argv) {
void initNode (int * __argc, char * * * __argv)
{
int provided;
MPI_Init_thread (__argc, __argv, MPI_THREAD_FUNNELED, & provided);
MPI_Init_thread (__argc, __argv, MPI_THREAD_FUNNELED, & provided);
assert (provided == MPI_THREAD_FUNNELED); /* The MPI implementation must be multi-threaded.
Yet, only one thread performs the comm.
operations */
Yet, only one thread performs the comm.
operations */
MPI_Comm_rank (MPI_COMM_WORLD, & rk); /* Who ? */
MPI_Comm_size (MPI_COMM_WORLD, & sz); /* How many ? */
char names [sz] [MPI_MAX_PROCESSOR_NAME];
int len;
/* Processor names */
MPI_Get_processor_name (names [0], & len); /* Me */
/* Processor names */
MPI_Get_processor_name (names [0], & len); /* Me */
MPI_Allgather (names, MPI_MAX_PROCESSOR_NAME, MPI_CHAR, names, MPI_MAX_PROCESSOR_NAME, MPI_CHAR, MPI_COMM_WORLD); /* Broadcast */
for (int i = 0; i < sz; i ++) {
rk_to_name.push_back (names [i]);
name_to_rk [names [i]] = i;
}
for (int i = 0; i < sz; i ++)
{
rk_to_name.push_back (names [i]);
name_to_rk [names [i]] = i;
}
}

2
trunk/paradiseo-peo/src/rmc/mpi/node.h
View file

@ -1,4 +1,4 @@
/*
/*
* <node.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007

5
trunk/paradiseo-peo/src/rmc/mpi/param.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <param.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -38,7 +38,8 @@
#include "schema.h"
void loadRMCParameters (int & __argc, char * * & __argv) {
void loadRMCParameters (int & __argc, char * * & __argv)
{
eoParser parser (__argc, __argv);

2
trunk/paradiseo-peo/src/rmc/mpi/param.h
View file

@ -1,4 +1,4 @@
/*
/*
* <param.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007

186
trunk/paradiseo-peo/src/rmc/mpi/recv.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <recv.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -44,97 +44,103 @@
#include "../../core/cooperative.h"
#include "../../core/peo_debug.h"
void receiveMessages () {
void receiveMessages ()
{
cleanBuffers ();
do {
if (! atLeastOneActiveThread ()) {
// printDebugMessage ("debut wait");
waitMessage ();
//printDebugMessage ("fin wait");
do
{
if (! atLeastOneActiveThread ())
{
// printDebugMessage ("debut wait");
waitMessage ();
//printDebugMessage ("fin wait");
}
int src, tag;
while (probeMessage (src, tag))
{
receiveMessage (src, tag);
initMessage ();
/*
char b [1000];
sprintf (b, "traitement recv %d\n", tag);
printDebugMessage (b);
*/
switch (tag)
{
case RUNNER_STOP_TAG:
unpackTerminationOfRunner ();
wakeUpCommunicator ();
break;
case COOP_TAG:
// printDebugMessage ("reception de message de cooperation");
COOP_ID coop_id;
unpack (coop_id);
getCooperative (coop_id) -> unpack ();
break;
case SCHED_REQUEST_TAG:
unpackResourceRequest ();
break;
case SCHED_RESULT_TAG:
{
/* Unpacking the resource */
SERVICE_ID serv_id;
unpack (serv_id);
Service * serv = getService (serv_id);
int dest;
unpack (dest);
WORKER_ID worker_id;
unpack (worker_id);
/* Going back ... */
initMessage ();
pack (worker_id);
pack (serv_id);
serv -> packData ();
serv -> notifySendingData ();
sendMessage (dest, TASK_DATA_TAG);
break;
}
case TASK_DATA_TAG:
{
WORKER_ID worker_id;
unpack (worker_id);
Worker * worker = getWorker (worker_id);
worker -> setSource (src);
worker -> unpackData ();
worker -> wakeUp ();
break;
}
case TASK_RESULT_TAG:
{
SERVICE_ID serv_id;
unpack (serv_id);
Service * serv = getService (serv_id);
serv -> unpackResult ();
break;
}
case TASK_DONE_TAG:
unpackTaskDone ();
break;
default:
;
};
}
}
int src, tag;
while (probeMessage (src, tag)) {
receiveMessage (src, tag);
initMessage ();
/*
char b [1000];
sprintf (b, "traitement recv %d\n", tag);
printDebugMessage (b);
*/
switch (tag) {
case RUNNER_STOP_TAG:
unpackTerminationOfRunner ();
wakeUpCommunicator ();
break;
case COOP_TAG:
// printDebugMessage ("reception de message de cooperation");
COOP_ID coop_id;
unpack (coop_id);
getCooperative (coop_id) -> unpack ();
break;
case SCHED_REQUEST_TAG:
unpackResourceRequest ();
break;
case SCHED_RESULT_TAG:
{
/* Unpacking the resource */
SERVICE_ID serv_id;
unpack (serv_id);
Service * serv = getService (serv_id);
int dest;
unpack (dest);
WORKER_ID worker_id;
unpack (worker_id);
/* Going back ... */
initMessage ();
pack (worker_id);
pack (serv_id);
serv -> packData ();
serv -> notifySendingData ();
sendMessage (dest, TASK_DATA_TAG);
break;
}
case TASK_DATA_TAG:
{
WORKER_ID worker_id;
unpack (worker_id);
Worker * worker = getWorker (worker_id);
worker -> setSource (src);
worker -> unpackData ();
worker -> wakeUp ();
break;
}
case TASK_RESULT_TAG:
{
SERVICE_ID serv_id;
unpack (serv_id);
Service * serv = getService (serv_id);
serv -> unpackResult ();
break;
}
case TASK_DONE_TAG:
unpackTaskDone ();
break;
default:
;
};
}
} while (! atLeastOneActiveThread () && atLeastOneActiveRunner () /*&& ! allResourcesFree ()*/);
while (! atLeastOneActiveThread () && atLeastOneActiveRunner () /*&& ! allResourcesFree ()*/);
}

2
trunk/paradiseo-peo/src/rmc/mpi/recv.h
View file

@ -1,4 +1,4 @@
/*
/*
* <recv.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007

13
trunk/paradiseo-peo/src/rmc/mpi/rmc.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <rmc.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -43,16 +43,18 @@
static std :: vector <pthread_t *> ll_threads; /* Low level threads */
void runRMC () {
void runRMC ()
{
/* Worker(s) ? */
for (unsigned i = 0; i < my_node -> num_workers; i ++)
for (unsigned i = 0; i < my_node -> num_workers; i ++)
addThread (new Worker, ll_threads);
wakeUpCommunicator ();
}
void initRMC (int & __argc, char * * & __argv) {
void initRMC (int & __argc, char * * & __argv)
{
/* Communication */
initCommunication ();
@ -67,7 +69,8 @@ void initRMC (int & __argc, char * * & __argv) {
///
}
void finalizeRMC () {
void finalizeRMC ()
{
printDebugMessage ("before join threads RMC");
joinThreads (ll_threads);

13
trunk/paradiseo-peo/src/rmc/mpi/runner.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <runner.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -41,7 +41,8 @@
#include "tags.h"
#include "schema.h"
bool Runner :: isLocal () {
bool Runner :: isLocal ()
{
for (unsigned i = 0; i < my_node -> id_run.size (); i ++)
if (my_node -> id_run [i] == id)
@ -49,13 +50,15 @@ bool Runner :: isLocal () {
return false;
}
void Runner :: packTermination () {
void Runner :: packTermination ()
{
pack (id);
}
void Runner :: terminate () {
void Runner :: terminate ()
{
sendToAll (this, RUNNER_STOP_TAG);
sendToAll (this, RUNNER_STOP_TAG);
}

65
trunk/paradiseo-peo/src/rmc/mpi/scheduler.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <scheduler.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -47,45 +47,51 @@ static std :: queue <SCHED_REQUEST> requests; /* Requests */
static unsigned initNumberOfRes = 0;
void initScheduler () {
for (unsigned i = 0; i < the_schema.size (); i ++) {
const Node & node = the_schema [i];
if (node.rk_sched == my_node -> rk)
for (unsigned j = 0; j < node.num_workers; j ++)
resources.push (std :: pair <RANK_ID, WORKER_ID> (i, j + 1));
}
void initScheduler ()
{
for (unsigned i = 0; i < the_schema.size (); i ++)
{
const Node & node = the_schema [i];
if (node.rk_sched == my_node -> rk)
for (unsigned j = 0; j < node.num_workers; j ++)
resources.push (std :: pair <RANK_ID, WORKER_ID> (i, j + 1));
}
initNumberOfRes = resources.size ();
}
bool allResourcesFree () {
bool allResourcesFree ()
{
return resources.size () == initNumberOfRes;
}
static void update () {
static void update ()
{
unsigned num_alloc = std :: min (resources.size (), requests.size ());
for (unsigned i = 0; i < num_alloc; i ++) {
SCHED_REQUEST req = requests.front ();
requests.pop ();
SCHED_RESOURCE res = resources.front ();
resources.pop ();
printDebugMessage ("allocating a resource.");
initMessage ();
pack (req.second);
pack (res);
sendMessage (req.first, SCHED_RESULT_TAG);
}
for (unsigned i = 0; i < num_alloc; i ++)
{
SCHED_REQUEST req = requests.front ();
requests.pop ();
SCHED_RESOURCE res = resources.front ();
resources.pop ();
printDebugMessage ("allocating a resource.");
initMessage ();
pack (req.second);
pack (res);
sendMessage (req.first, SCHED_RESULT_TAG);
}
}
void unpackResourceRequest () {
void unpackResourceRequest ()
{
printDebugMessage ("queuing a resource request.");
SCHED_REQUEST req;
@ -94,7 +100,8 @@ void unpackResourceRequest () {
update ();
}
void unpackTaskDone () {
void unpackTaskDone ()
{
printDebugMessage ("I'm notified a worker is now idle.");
SCHED_RESOURCE res;

4
trunk/paradiseo-peo/src/rmc/mpi/scheduler.h
View file

@ -1,4 +1,4 @@
/*
/*
* <scheduler.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -53,7 +53,7 @@ extern void initScheduler ();
extern void unpackResourceRequest ();
/* Being known a worker is now idle :-) */
extern void unpackTaskDone ();
extern void unpackTaskDone ();
extern bool allResourcesFree ();

146
trunk/paradiseo-peo/src/rmc/mpi/schema.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <schema.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -47,20 +47,22 @@ std :: vector <Node> the_schema;
Node * my_node;
RANK_ID getRankOfRunner (RUNNER_ID __key) {
RANK_ID getRankOfRunner (RUNNER_ID __key)
{
for (unsigned i = 0; i < the_schema.size (); i ++)
for (unsigned j = 0; j < the_schema [i].id_run.size (); j ++)
if (the_schema [i].id_run [j] == __key)
return the_schema [i].rk;
return the_schema [i].rk;
assert (false);
return 0;
return 0;
}
static void loadNode (int __rk_sched) {
static void loadNode (int __rk_sched)
{
Node node;
node.rk_sched = __rk_sched;
/* ATT: name*/
@ -68,96 +70,106 @@ static void loadNode (int __rk_sched) {
/* ATT: num_workers */
node.num_workers = atoi (getAttributeValue ("num_workers").c_str ());
while (true) {
/* TAG: <runner> | </node> */
std :: string name = getNextNode ();
assert (name == "runner" || name == "node");
if (name == "runner") {
/* TAG: </node> */
node.id_run.push_back (atoi (getNextNode ().c_str ()));
/* TAG: </runner> */
assert (getNextNode () == "runner");
while (true)
{
/* TAG: <runner> | </node> */
std :: string name = getNextNode ();
assert (name == "runner" || name == "node");
if (name == "runner")
{
/* TAG: </node> */
node.id_run.push_back (atoi (getNextNode ().c_str ()));
/* TAG: </runner> */
assert (getNextNode () == "runner");
}
else
{
/* TAG: </node> */
the_schema.push_back (node);
break;
}
}
else {
/* TAG: </node> */
the_schema.push_back (node);
break;
}
}
}
static void loadGroup () {
static void loadGroup ()
{
std :: string name;
/* ATT: scheduler*/
int rk_sched = getRankFromName (getAttributeValue ("scheduler"));
while (true) {
/* TAG: <node> | </group> */
name = getNextNode ();
assert (name == "node" || name == "group");
if (name == "node")
/* TAG: <node> */
loadNode (rk_sched);
else
/* TAG: </group> */
break;
}
while (true)
{
/* TAG: <node> | </group> */
name = getNextNode ();
assert (name == "node" || name == "group");
if (name == "node")
/* TAG: <node> */
loadNode (rk_sched);
else
/* TAG: </group> */
break;
}
}
bool isScheduleNode () {
bool isScheduleNode ()
{
return my_node -> rk == my_node -> rk_sched;
}
void loadSchema (const char * __filename) {
void loadSchema (const char * __filename)
{
openXMLDocument (__filename);
std :: string name;
/* TAG: <schema> */
name = getNextNode ();
assert (name == "schema");
while (true) {
/* TAG: <group> | </schema> */
name = getNextNode ();
assert (name == "group" || name == "schema");
if (name == "group")
/* TAG: <group> */
loadGroup ();
else
/* TAG: </schema> */
break;
}
while (true)
{
/* TAG: <group> | </schema> */
name = getNextNode ();
assert (name == "group" || name == "schema");
if (name == "group")
/* TAG: <group> */
loadGroup ();
else
/* TAG: </schema> */
break;
}
/* Looking for my node */
for (unsigned i = 0; i < the_schema.size (); i ++)
if (the_schema [i].rk == getNodeRank ())
my_node = & (the_schema [i]);
/* About me */
char mess [1000];
sprintf (mess, "my rank is %d", my_node -> rk);
printDebugMessage (mess);
if (isScheduleNode ())
printDebugMessage ("I'am a scheduler");
for (unsigned i = 0; i < my_node -> id_run.size (); i ++) {
sprintf (mess, "I manage the runner %d", my_node -> id_run [i]);
printDebugMessage (mess);
}
if (my_node -> num_workers) {
sprintf (mess, "I manage %d worker(s)", my_node -> num_workers);
printDebugMessage (mess);
}
printDebugMessage ("I'am a scheduler");
for (unsigned i = 0; i < my_node -> id_run.size (); i ++)
{
sprintf (mess, "I manage the runner %d", my_node -> id_run [i]);
printDebugMessage (mess);
}
if (my_node -> num_workers)
{
sprintf (mess, "I manage %d worker(s)", my_node -> num_workers);
printDebugMessage (mess);
}
closeXMLDocument ();
}

19
trunk/paradiseo-peo/src/rmc/mpi/schema.h
View file

@ -1,4 +1,4 @@
/*
/*
* <schema.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -45,14 +45,15 @@
typedef int RANK_ID;
struct Node {
RANK_ID rk; /* Rank */
std :: string name; /* Host name */
unsigned num_workers; /* Number of parallel workers */
int rk_sched; /* rank of the scheduler */
std :: vector <RUNNER_ID> id_run; /* List of runners */
};
struct Node
{
RANK_ID rk; /* Rank */
std :: string name; /* Host name */
unsigned num_workers; /* Number of parallel workers */
int rk_sched; /* rank of the scheduler */
std :: vector <RUNNER_ID> id_run; /* List of runners */
};
extern std :: vector <Node> the_schema;

124
trunk/paradiseo-peo/src/rmc/mpi/send.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <send.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -49,26 +49,30 @@
#define TO_ALL -1
typedef struct {
typedef struct
{
Communicable * comm;
int to;
int tag;
Communicable * comm;
int to;
int tag;
}
SEND_REQUEST;
} SEND_REQUEST;
static std :: queue <SEND_REQUEST> mess;
static sem_t sem_send;
void initSending () {
void initSending ()
{
sem_init (& sem_send, 0, 1);
}
void send (Communicable * __comm, int __to, int __tag) {
void send (Communicable * __comm, int __to, int __tag)
{
SEND_REQUEST req;
SEND_REQUEST req;
req.comm = __comm;
req.to = __to;
req.tag = __tag;
@ -79,66 +83,70 @@ void send (Communicable * __comm, int __to, int __tag) {
wakeUpCommunicator ();
}
void sendToAll (Communicable * __comm, int __tag) {
void sendToAll (Communicable * __comm, int __tag)
{
send (__comm, TO_ALL, __tag);
}
void sendMessages () {
void sendMessages ()
{
sem_wait (& sem_send);
while (! mess.empty ()) {
SEND_REQUEST req = mess.front ();
/*
char b [1000];
sprintf (b, "traitement send %d\n", req.tag);
printDebugMessage (b);
*/
Communicable * comm = req.comm;
while (! mess.empty ())
{
initMessage ();
SEND_REQUEST req = mess.front ();
/*
char b [1000];
sprintf (b, "traitement send %d\n", req.tag);
printDebugMessage (b);
*/
switch (req.tag) {
Communicable * comm = req.comm;
case RUNNER_STOP_TAG:
dynamic_cast <Runner *> (comm) -> packTermination ();
dynamic_cast <Runner *> (comm) -> notifySendingTermination ();
break;
initMessage ();
case COOP_TAG:
dynamic_cast <Cooperative *> (comm) -> pack ();
dynamic_cast <Cooperative *> (comm) -> notifySending ();
break;
case SCHED_REQUEST_TAG:
dynamic_cast <Service *> (comm) -> packResourceRequest ();
dynamic_cast <Service *> (comm) -> notifySendingResourceRequest ();
break;
switch (req.tag)
{
case TASK_RESULT_TAG:
dynamic_cast <Worker *> (comm) -> packResult ();
dynamic_cast <Worker *> (comm) -> notifySendingResult ();
break;
case RUNNER_STOP_TAG:
dynamic_cast <Runner *> (comm) -> packTermination ();
dynamic_cast <Runner *> (comm) -> notifySendingTermination ();
break;
case TASK_DONE_TAG:
dynamic_cast <Worker *> (comm) -> packTaskDone ();
dynamic_cast <Worker *> (comm) -> notifySendingTaskDone ();
break;
default :
break;
case COOP_TAG:
dynamic_cast <Cooperative *> (comm) -> pack ();
dynamic_cast <Cooperative *> (comm) -> notifySending ();
break;
};
if (req.to == TO_ALL)
sendMessageToAll (req.tag);
else
sendMessage (req.to, req.tag);
mess.pop ();
}
case SCHED_REQUEST_TAG:
dynamic_cast <Service *> (comm) -> packResourceRequest ();
dynamic_cast <Service *> (comm) -> notifySendingResourceRequest ();
break;
sem_post (& sem_send);
case TASK_RESULT_TAG:
dynamic_cast <Worker *> (comm) -> packResult ();
dynamic_cast <Worker *> (comm) -> notifySendingResult ();
break;
case TASK_DONE_TAG:
dynamic_cast <Worker *> (comm) -> packTaskDone ();
dynamic_cast <Worker *> (comm) -> notifySendingTaskDone ();
break;
default :
break;
};
if (req.to == TO_ALL)
sendMessageToAll (req.tag);
else
sendMessage (req.to, req.tag);
mess.pop ();
}
sem_post (& sem_send);
}

2
trunk/paradiseo-peo/src/rmc/mpi/send.h
View file

@ -1,4 +1,4 @@
/*
/*
* <send.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007

8
trunk/paradiseo-peo/src/rmc/mpi/service.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <service.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -41,14 +41,16 @@
#include "send.h"
#include "scheduler.h"
void Service :: requestResourceRequest (unsigned __how_many) {
void Service :: requestResourceRequest (unsigned __how_many)
{
num_sent_rr = __how_many;
for (unsigned i = 0; i < __how_many; i ++)
send (this, my_node -> rk_sched, SCHED_REQUEST_TAG);
}
void Service :: packResourceRequest () {
void Service :: packResourceRequest ()
{
SCHED_REQUEST req;
req.first = getNodeRank ();

2
trunk/paradiseo-peo/src/rmc/mpi/tags.h
View file

@ -1,4 +1,4 @@
/*
/*
* <tags.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007

78
trunk/paradiseo-peo/src/rmc/mpi/worker.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <worker.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -46,76 +46,88 @@
static std :: vector <Worker *> key_to_worker (1); /* Vector of registered workers */
Worker * getWorker (WORKER_ID __key) {
Worker * getWorker (WORKER_ID __key)
{
return key_to_worker [__key];
return key_to_worker [__key];
}
Worker :: Worker () {
Worker :: Worker ()
{
toto = false;
id = key_to_worker.size ();
key_to_worker.push_back (this);
}
void Worker :: packResult () {
void Worker :: packResult ()
{
pack (serv_id);
serv -> packResult ();
serv -> packResult ();
}
void Worker :: unpackData () {
void Worker :: unpackData ()
{
printDebugMessage ("unpacking the ID. of the service.");
unpack (serv_id);
serv = getService (serv_id);
serv = getService (serv_id);
printDebugMessage ("found the service.");
serv -> unpackData ();
serv -> unpackData ();
printDebugMessage ("unpacking the data.");
setActive ();
}
void Worker :: packTaskDone () {
void Worker :: packTaskDone ()
{
pack (getNodeRank ());
pack (id);
}
void Worker :: notifySendingResult () {
void Worker :: notifySendingResult ()
{
/* Notifying the scheduler of the termination */
toto = true;
wakeUp ();
}
void Worker :: notifySendingTaskDone () {
void Worker :: notifySendingTaskDone ()
{
setPassive ();
}
void Worker :: setSource (int __rank) {
void Worker :: setSource (int __rank)
{
src = __rank;
}
void Worker :: start () {
void Worker :: start ()
{
while (true) {
sleep ();
while (true)
{
if (! atLeastOneActiveRunner ())
break;
if (toto) {
send (this, my_node -> rk_sched, TASK_DONE_TAG);
toto = false;
sleep ();
if (! atLeastOneActiveRunner ())
break;
if (toto)
{
send (this, my_node -> rk_sched, TASK_DONE_TAG);
toto = false;
}
else
{
printDebugMessage ("executing the task.");
serv -> execute ();
send (this, src, TASK_RESULT_TAG);
}
}
else {
printDebugMessage ("executing the task.");
serv -> execute ();
send (this, src, TASK_RESULT_TAG);
}
}
}

43
trunk/paradiseo-peo/src/rmc/mpi/worker.h
View file

@ -1,4 +1,4 @@
/*
/*
* <worker.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -41,37 +41,38 @@
#include "../../core/reac_thread.h"
#include "../../core/service.h"
typedef unsigned WORKER_ID;
typedef unsigned WORKER_ID;
class Worker : public Communicable, public ReactiveThread {
class Worker : public Communicable, public ReactiveThread
{
public :
public :
Worker ();
Worker ();
void start ();
void start ();
void packResult ();
void packResult ();
void unpackData ();
void unpackData ();
void packTaskDone ();
void packTaskDone ();
void notifySendingResult ();
void notifySendingResult ();
void notifySendingTaskDone ();
void setSource (int __rank);
private :
void notifySendingTaskDone ();
WORKER_ID id;
SERVICE_ID serv_id;
Service * serv;
int src;
void setSource (int __rank);
bool toto;
};
private :
WORKER_ID id;
SERVICE_ID serv_id;
Service * serv;
int src;
bool toto;
};
extern Worker * getWorker (WORKER_ID __key);

62
trunk/paradiseo-peo/src/rmc/mpi/xml_parser.cpp
View file

@ -1,4 +1,4 @@
/*
/*
* <xml_parser.cpp>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007
@ -40,51 +40,59 @@
static xmlTextReaderPtr reader;
void openXMLDocument (const char * __filename) {
void openXMLDocument (const char * __filename)
{
reader = xmlNewTextReaderFilename (__filename);
if (! reader) {
fprintf (stderr, "unable to open '%s'.\n", __filename);
exit (1);
}
if (! reader)
{
fprintf (stderr, "unable to open '%s'.\n", __filename);
exit (1);
}
}
void closeXMLDocument () {
void closeXMLDocument ()
{
xmlFreeTextReader (reader);
}
std :: string getAttributeValue (const std :: string & __attr) {
std :: string getAttributeValue (const std :: string & __attr)
{
xmlChar * value = xmlTextReaderGetAttribute (reader, (const xmlChar *) __attr.c_str ());
std :: string str ((const char *) value);
xmlFree (value);
return str;
}
static bool isSep (const xmlChar * __text) {
static bool isSep (const xmlChar * __text)
{
for (unsigned i = 0; i < strlen ((char *) __text); i ++)
if (__text [i] != ' ' && __text [i] != '\t' && __text [i] != '\n')
return false;
return true;
}
std :: string getNextNode () {
std :: string getNextNode ()
{
xmlChar * name, * value;
do {
xmlTextReaderRead (reader);
name = xmlTextReaderName (reader);
value = xmlTextReaderValue (reader);
// printf ("value = %s\n", value);
} while (! strcmp ((char *) name, "#text") && isSep (value));
do
{
xmlTextReaderRead (reader);
name = xmlTextReaderName (reader);
value = xmlTextReaderValue (reader);
// printf ("value = %s\n", value);
}
while (! strcmp ((char *) name, "#text") && isSep (value));
std :: string str;
@ -92,12 +100,12 @@ std :: string getNextNode () {
str.assign ((char *) name);
else
str.assign ((char *) value);
if (name)
xmlFree (name);
if (value)
xmlFree (value);
return str;
}

2
trunk/paradiseo-peo/src/rmc/mpi/xml_parser.h
View file

@ -1,4 +1,4 @@
/*
/*
* <xml_parser.h>
* Copyright (C) DOLPHIN Project-Team, INRIA Futurs, 2006-2007
* (C) OPAC Team, LIFL, 2002-2007