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@794 331e1502-861f-0410-8da2-ba01fb791d7f

Browse source
This commit is contained in:
atantar 2007-11-20 13:40:15 +00:00
commit aecc5ba8fc
35 changed files with 498 additions and 262 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -55,7 +55,8 @@ Communicator :: Communicator (int * __argc, char * * * __argv) {
the_thread = this;
initNode (__argc, __argv);
loadRMCParameters (* __argc, * __argv);
loadRMCParameters (* __argc, * __argv);
sem_post (& sem_comm_init);
}
@ -68,15 +69,18 @@ void Communicator :: start () {
sendMessages ();
if (! atLeastOneActiveRunner ())
if (theEnd() || ! atLeastOneActiveRunner ())
break;
receiveMessages ();
}
waitBuffers ();
waitBuffers ();
sem_destroy(& sem_comm_init);
printDebugMessage ("finalizing");
MPI_Finalize ();
synchronizeNodes ();
}
void initCommunication () {

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

@ -54,18 +54,18 @@ static std :: vector <MPI_Request *> act_req; /* Active requests */
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 () ;
}

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

@ -58,4 +58,3 @@ extern void waitMessage ();
extern void synchronizeNodes ();
#endif

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

@ -40,6 +40,52 @@
#include <string>
#include <cassert>
class MPIThreadedEnv {
public:
static void init ( int * __argc, char * * * __argv ) {
static MPIThreadedEnv mpiThreadedEnv( __argc, __argv );
}
static void finalize () {
static bool finalizedEnvironment = false;
if (! finalizedEnvironment ) {
MPI_Finalize ();
finalizedEnvironment = true;
}
}
private:
/* No instance of this class can be created outside its domain */
MPIThreadedEnv ( int * __argc, char * * * __argv ) {
static bool MPIThreadedEnvInitialized = false;
int provided = 1;
if (! MPIThreadedEnvInitialized) {
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 */
MPIThreadedEnvInitialized = true;
}
}
~MPIThreadedEnv() {
finalize ();
}
};
static int rk, sz; /* Rank & size */
static std :: map <std :: string, int> name_to_rk;
@ -57,17 +103,19 @@ int getNumberOfNodes () {
}
int getRankFromName (const std :: string & __name) {
return atoi (__name.c_str ());
}
void initNode (int * __argc, char * * * __argv) {
int 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 */
rk_to_name.clear ();
name_to_rk.clear ();
MPIThreadedEnv :: init ( __argc, __argv );
MPI_Comm_rank (MPI_COMM_WORLD, & rk); /* Who ? */
MPI_Comm_size (MPI_COMM_WORLD, & sz); /* How many ? */
@ -77,10 +125,9 @@ void initNode (int * __argc, char * * * __argv) {
/* 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;
}
}

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

@ -42,12 +42,16 @@
#include "../../core/peo_debug.h"
static std :: vector <pthread_t *> ll_threads; /* Low level threads */
static std :: vector <Worker *> worker_threads; /* Worker threads */
static Communicator* communicator_thread = NULL; /* Communicator thread */
void runRMC () {
/* Worker(s) ? */
for (unsigned i = 0; i < my_node -> num_workers; i ++)
addThread (new Worker, ll_threads);
for (unsigned i = 0; i < my_node -> num_workers; i ++) {
worker_threads.push_back (new Worker);
addThread (worker_threads.back(), ll_threads);
}
wakeUpCommunicator ();
}
@ -56,20 +60,26 @@ void initRMC (int & __argc, char * * & __argv) {
/* Communication */
initCommunication ();
addThread (new Communicator (& __argc, & __argv), ll_threads);
communicator_thread = new Communicator (& __argc, & __argv);
addThread (communicator_thread, ll_threads);
waitNodeInitialization ();
initSending ();
/* Scheduler */
if (isScheduleNode ())
initScheduler ();
///
}
void finalizeRMC () {
printDebugMessage ("before join threads RMC");
joinThreads (ll_threads);
for (unsigned i = 0; i < worker_threads.size(); i++ ) {
delete worker_threads [i];
}
worker_threads.clear ();
delete communicator_thread;
printDebugMessage ("after join threads RMC");
}

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

@ -48,11 +48,15 @@ static std :: queue <SCHED_REQUEST> requests; /* Requests */
static unsigned initNumberOfRes = 0;
void initScheduler () {
resources = std :: queue <SCHED_RESOURCE> ();
requests = std :: queue <SCHED_REQUEST> ();
initNumberOfRes = 0;
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));

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

@ -34,9 +34,8 @@
*
*/
#include <iostream>
#include <set>
#include <assert.h>
#include <cassert>
#include "schema.h"
#include "xml_parser.h"
@ -131,9 +130,9 @@ void loadSchema (const char * __filename) {
name = getNextNode ();
assert (name == "schema");
the_schema.clear();
maxSpecifiedRunnerID = 0;
while (true) {
/* TAG: <group> | </schema> */

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

@ -43,6 +43,7 @@
#include "../../core/runner.h"
typedef int RANK_ID;
struct Node {

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

@ -61,9 +61,24 @@ static std :: queue <SEND_REQUEST> mess;
static sem_t sem_send;
static bool contextInitialized = false;
void initSending () {
sem_init (& sem_send, 0, 1);
static bool initializedSem = false;
mess = std :: queue <SEND_REQUEST> ();
if (! initializedSem) {
sem_init (& sem_send, 0, 1);
initializedSem = true;
}
else {
sem_destroy(& sem_send);
sem_init (& sem_send, 0, 1);
}
contextInitialized = false;
}
void send (Communicable * __comm, int __to, int __tag) {
@ -90,17 +105,15 @@ void sendMessages () {
sem_wait (& sem_send);
static bool contextInitialized = false;
if (! contextInitialized) {
contextInitialized = true;
initializeContext();
}
while (! mess.empty ()) {
SEND_REQUEST req = mess.front ();
Communicable * comm = req.comm;
initMessage ();
@ -131,12 +144,12 @@ void sendMessages () {
dynamic_cast <Worker *> (comm) -> packTaskDone ();
dynamic_cast <Worker *> (comm) -> notifySendingTaskDone ();
break;
default :
break;
};
if (req.to == TO_ALL)
sendMessageToAll (req.tag);
else

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

@ -53,6 +53,6 @@ void Service :: packResourceRequest () {
SCHED_REQUEST req;
req.first = getNodeRank ();
req.second = getKey ();
// printf ("demande de ressource pour %d\n", req.second);
:: pack (req);
}

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

@ -37,18 +37,17 @@
#ifndef __tags_h
#define __tags_h
#define EXECUTION_CONTEXT_TAG 1000
#define RUNNER_STOP_TAG 13
#define COOP_TAG 14
#define SCHED_REQUEST_TAG 16
#define SCHED_RESULT_TAG 17
#define TASK_DATA_TAG 18
#define TASK_DATA_TAG 18
#define TASK_RESULT_TAG 19
#define TASK_DONE_TAG 20
#define EXECUTION_CONTEXT_TAG 1000
#endif

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

@ -53,7 +53,7 @@ Worker * getWorker (WORKER_ID __key) {
Worker :: Worker () {
toto = false;
recvAndCompleted = false;
id = key_to_worker.size ();
key_to_worker.push_back (this);
}
@ -84,7 +84,7 @@ void Worker :: packTaskDone () {
void Worker :: notifySendingResult () {
/* Notifying the scheduler of the termination */
toto = true;
recvAndCompleted = true;
wakeUp ();
}
@ -101,15 +101,15 @@ void Worker :: setSource (int __rank) {
void Worker :: start () {
while (true) {
sleep ();
if (! atLeastOneActiveRunner ())
break;
if (toto) {
if (recvAndCompleted) {
send (this, my_node -> rk_sched, TASK_DONE_TAG);
toto = false;
recvAndCompleted = false;
}
else {
@ -119,3 +119,8 @@ void Worker :: start () {
}
}
}
void initWorkersEnv () {
key_to_worker.resize (1);
}

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

@ -41,6 +41,7 @@
#include "../../core/reac_thread.h"
#include "../../core/service.h"
typedef unsigned WORKER_ID;
class Worker : public Communicable, public ReactiveThread {
@ -70,9 +71,11 @@ private :
Service * serv;
int src;
bool toto;
bool recvAndCompleted;
};
extern void initWorkersEnv ();
extern Worker * getWorker (WORKER_ID __key);
#endif

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

@ -83,7 +83,6 @@ std :: string getNextNode () {
xmlTextReaderRead (reader);
name = xmlTextReaderName (reader);
value = xmlTextReaderValue (reader);
// printf ("value = %s\n", value);
} while (! strcmp ((char *) name, "#text") && isSep (value));
std :: string str;