examples/gras/pmm/pmm.c

   1 /* pmm - paralel matrix multiplication "double diffusion"                       */
   2
   3 /* Copyright (c) 2006- Ahmed Harbaoui. All rights reserved.                  */
   4
   5 /* This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the license (GNU LGPL) which comes with this package. */
   7
   8 #include "gras.h"
   9 #define MATRIX_SIZE 3
  10 #define SLAVE_COUNT 9
  11
  12 XBT_LOG_NEW_DEFAULT_CATEGORY(pmm,"Parallel Matrix Multiplication");
  13
  14 GRAS_DEFINE_TYPE(s_matrix,struct s_matrix {
  15   int rows;
  16   int cols;
  17   double *data GRAS_ANNOTE(size, rows*cols);
  18 };)
  19      typedef struct s_matrix matrix_t;
  20
  21      /* struct for recovering results */
  22      GRAS_DEFINE_TYPE(s_result,struct s_result {
  23        int i;
  24        int j;
  25        double value;
  26      });
  27 typedef struct s_result result_t;
  28
  29 /* struct to send initial data to slave */
  30 GRAS_DEFINE_TYPE(s_init_data,struct s_init_data {
  31   int myrow;
  32   int mycol;
  33   double a;
  34   double b;
  35 });
  36 typedef struct s_init_data init_data_t;
  37
  38 /* register messages which may be sent (common to client and server) */
  39 static void register_messages(void) {
  40   gras_datadesc_type_t result_type;
  41   gras_datadesc_type_t init_data_type;
  42   result_type=gras_datadesc_by_symbol(s_result);
  43   init_data_type=gras_datadesc_by_symbol(s_init_data);
  44
  45   gras_msgtype_declare("result", result_type);  // recieve a final result from slave
  46   gras_msgtype_declare("init_data", init_data_type);  // send from master to slave to initialize data bA,bB
  47
  48   gras_msgtype_declare("ask_result", gras_datadesc_by_name("int")); // send from master to slave to ask a final result
  49   gras_msgtype_declare("step", gras_datadesc_by_name("int"));// send from master to slave to indicate the begining of step
  50   gras_msgtype_declare("step_ack", gras_datadesc_by_name("int"));//send from slave to master to indicate the end of the current step
  51   gras_msgtype_declare("dataA", gras_datadesc_by_name("double"));// send data between slave
  52   gras_msgtype_declare("dataB", gras_datadesc_by_name("double"));// send data between slave
  53 }
  54
  55 /* Function prototypes */
  56 int slave (int argc,char *argv[]);
  57 int master (int argc,char *argv[]);
  58
  59
  60 /* **********************************************************************
  61  * master code
  62  * **********************************************************************/
  63
  64 /* Global private data */
  65 typedef struct {
  66   int nbr_col,nbr_row;
  67   int remaining_step;
  68   int remaining_ack;
  69 } master_data_t;
  70
  71
  72 /***  Function initilaze matrixs ***/
  73
  74 static void initmatrix(matrix_t *X){
  75   int i;
  76
  77   for(i=0 ; i<(X->rows)*(X->cols); i++)
  78     X->data[i]=1.0;//*rand()/(RAND_MAX+1.0);
  79 } /* end_of_initmatrixs */
  80
  81 /***  Function Scatter Sequentiel ***/
  82
  83 static void scatter(){
  84
  85 }/* end_of_Scatter */
  86
  87 /***  Function: Scatter // ***/
  88
  89 static void scatter_parl(){
  90
  91 }/* end_of_Scatter // */
  92
  93 /***  Function: multiplication ***/
  94
  95 static void multiplication(){
  96
  97 }/* end_of_multiplication */
  98
  99 /***  Function: gather ***/
 100
 101 static void gather(){
 102
 103 }/* end_of_gather */
 104
 105 /***  Function: Display Matrix ***/
 106
 107 static void display(matrix_t X){
 108
 109   int i,j,t=0;
 110
 111   printf("      ");
 112   for(j=0;j<X.cols;j++)
 113     printf("%.3d ",j);
 114   printf("\n");
 115   printf("    __");
 116   for(j=0;j<X.cols;j++)
 117     printf("____");
 118   printf("_\n");
 119
 120   for(i=0;i<X.rows;i++){
 121     printf("%.3d | ",i);
 122     for(j=0;j<X.cols;j++)
 123       printf("%.3g ",X.data[t++]);
 124     printf("|\n");
 125   }
 126   printf("    --");
 127   for(j=0;j<X.cols;j++)
 128     printf("----");
 129   printf("-\n");
 130
 131 }/* end_of_display */
 132
 133 int master (int argc,char *argv[]) {
 134
 135   xbt_ex_t e;
 136
 137   int i,port,ask_result,step;
 138
 139   matrix_t A,B,C;
 140   result_t result;
 141
 142   gras_socket_t from;
 143
 144   /*  Init the GRAS's infrastructure */
 145   gras_init(&argc, argv);
 146
 147   gras_socket_t socket[MATRIX_SIZE*MATRIX_SIZE]; /* sockets for brodcast to other slave */
 148
 149   /*  Initialize Matrixs */
 150
 151   A.rows=A.cols=MATRIX_SIZE;
 152   B.rows=B.cols=MATRIX_SIZE;
 153   C.rows=C.cols=MATRIX_SIZE;
 154
 155   A.data=xbt_malloc0(sizeof(double)*MATRIX_SIZE*MATRIX_SIZE);
 156   B.data=xbt_malloc0(sizeof(double)*MATRIX_SIZE*MATRIX_SIZE);
 157   C.data=xbt_malloc0(sizeof(double)*MATRIX_SIZE*MATRIX_SIZE);
 158
 159   initmatrix(&A);
 160   initmatrix(&B);
 161
 162   /*  Get arguments and create sockets */
 163   port=atoi(argv[1]);
 164   //scatter();
 165   //scatter_parl();
 166   //multiplication();
 167   //gather();
 168   //display(A);
 169   /****************************** Init Data Send *********************************/
 170   int step_ack,j=0;
 171   init_data_t mydata;
 172   gras_os_sleep(60);                                                // MODIFIER LES TEMPS D'ATTENTE 60 c trop normalement
 173   for( i=2;i< argc;i+=3){
 174     TRY {
 175       socket[j]=gras_socket_client(argv[i],port);
 176     } CATCH(e) {
 177       RETHROW0("Unable to connect to the server: %s");
 178     }
 179     INFO2("Connected to %s:%d.",argv[i],port);
 180
 181     mydata.myrow=atoi(argv[i+1]);  // My row
 182     mydata.mycol=atoi(argv[i+2]);  // My column
 183
 184     mydata.a=A.data[(mydata.myrow-1)*MATRIX_SIZE+(mydata.mycol-1)];
 185     mydata.b=B.data[(mydata.myrow-1)*MATRIX_SIZE+(mydata.mycol-1)];;
 186
 187     gras_msg_send(socket[j],gras_msgtype_by_name("init_data"),&mydata);
 188     INFO3("Send Init Data to %s : data A= %.3g & data B= %.3g",gras_socket_peer_name(socket[j]),mydata.a,mydata.b);
 189     j++;
 190   } // end init Data Send
 191
 192   /******************************* multiplication ********************************/
 193   INFO0("begin Multiplication");
 194
 195   for (step=1; step <= MATRIX_SIZE; step++){
 196     gras_os_sleep(50);
 197     for (i=0; i< SLAVE_COUNT; i++){
 198       TRY {
 199         gras_msg_send(socket[i], gras_msgtype_by_name("step"), &step);  /* initialize Mycol, MyRow, mydataA,mydataB*/
 200       } CATCH(e) {
 201         gras_socket_close(socket[i]);
 202         RETHROW0("Unable to send the msg : %s");
 203       }
 204     }
 205     INFO1("send to slave to begin a %d th step",step);
 206     /* wait for computing and slave messages exchange */
 207     i=0;
 208
 209     while  ( i< SLAVE_COUNT){
 210       TRY {
 211         gras_msg_wait(1300,gras_msgtype_by_name("step_ack"),&from,&step_ack);
 212       } CATCH(e) {
 213         RETHROW0("I Can't get a Ack step message from slave : %s");
 214       }
 215       i++;
 216       INFO3("Receive Ack step ack from %s (got %d of %d)",
 217             gras_socket_peer_name(from),
 218             i, SLAVE_COUNT);
 219     }
 220   }
 221   /*********************************  gather ***************************************/
 222
 223   ask_result=0;
 224   for( i=1;i< argc;i++){
 225     gras_msg_send(socket[i],gras_msgtype_by_name("ask_result"),&ask_result);
 226     INFO1("Send (Ask Result) message to %s",gras_socket_peer_name(socket[i]));
 227   }
 228   /* wait for results */
 229   for( i=1;i< argc;i++){
 230     gras_msg_wait(600,gras_msgtype_by_name("result"),&from,&result);
 231     C.data[(result.i-1)*MATRIX_SIZE+(result.j-1)]=result.value;
 232   }
 233   /*    end of gather   */
 234   INFO0 ("The Result of Multiplication is :");
 235   display(C);
 236
 237   return 0;
 238 } /* end_of_master */
 239
 240 /* **********************************************************************
 241  * slave code
 242  * **********************************************************************/
 243
 244 int slave(int argc,char *argv[]) {
 245
 246   xbt_ex_t e;
 247
 248   int step,port,l,result_ack=0;
 249   double bA,bB;
 250
 251   int myrow,mycol;
 252   double mydataA,mydataB;
 253   double bC=0;
 254
 255   //  static end_step;
 256
 257   result_t result;
 258
 259   gras_socket_t from,sock;  /* to recive from server for steps */
 260
 261   gras_socket_t socket_row[MATRIX_SIZE-1],socket_column[MATRIX_SIZE-1]; /* sockets for brodcast to other slave */
 262
 263   /* Init the GRAS's infrastructure */
 264
 265   gras_init(&argc, argv);
 266
 267   /* Get arguments and create sockets */
 268
 269   port=atoi(argv[1]);
 270
 271   /*  Create my master socket */
 272   sock = gras_socket_server(port);
 273   INFO2("Launch %s (port=%d)",argv[0],port);
 274   gras_os_sleep(1); //wait to start all slaves
 275
 276   int i;
 277   for (i=1;i<MATRIX_SIZE;i++){
 278     socket_row[i-1]=gras_socket_client(argv[i+1],port);
 279     socket_column[i-1]=gras_socket_client(argv[i+MATRIX_SIZE],port);
 280   }
 281
 282   /*  Register the known messages */
 283   register_messages();
 284
 285   /* Recover my initialized Data and My Position*/
 286   init_data_t mydata;
 287   INFO0("wait for init Data");
 288   TRY {
 289     gras_msg_wait(600,gras_msgtype_by_name("init_data"),&from,&mydata);
 290   } CATCH(e) {
 291     RETHROW0("I Can't get a init Data message from master : %s");
 292   }
 293   myrow=mydata.myrow;
 294   mycol=mydata.mycol;
 295   mydataA=mydata.a;
 296   mydataB=mydata.b;
 297   INFO4("Recive MY POSITION (%d,%d) and MY INIT DATA ( A=%.3g | B=%.3g )",
 298         myrow,mycol,mydataA,mydataB);
 299   step=1;
 300
 301   do {  //repeat until compute Cb
 302     step=MATRIX_SIZE+1;  // just intilization for loop
 303
 304     TRY {
 305       gras_msg_wait(200,gras_msgtype_by_name("step"),&from,&step);
 306     } CATCH(e) {
 307       RETHROW0("I Can't get a Next Step message from master : %s");
 308     }
 309     INFO1("Recive a step message from master: step = %d ",step);
 310
 311     if (step < MATRIX_SIZE ){
 312       /* a row brodcast */
 313       gras_os_sleep(3);  // IL FAUT EXPRIMER LE TEMPS D'ATTENTE EN FONCTION DE "SLAVE_COUNT"
 314       if(myrow==step){
 315         INFO2("step(%d) = Myrow(%d)",step,myrow);
 316         for (l=1;l < MATRIX_SIZE ;l++){
 317           gras_msg_send(socket_column[l-1], gras_msgtype_by_name("dataB"), &mydataB);
 318           bB=mydataB;
 319           INFO1("send my data B (%.3g) to my (vertical) neighbors",bB);
 320         }
 321       }
 322       if(myrow != step){
 323         INFO2("step(%d) <> Myrow(%d)",step,myrow);
 324         TRY {
 325           gras_msg_wait(600,gras_msgtype_by_name("dataB"),
 326                         &from,&bB);
 327         } CATCH(e) {
 328           RETHROW0("I Can't get a data message from row : %s");
 329         }
 330         INFO2("Recive data B (%.3g) from my neighbor: %s",bB,gras_socket_peer_name(from));
 331       }
 332       /* a column brodcast */
 333       if(mycol==step){
 334         for (l=1;l < MATRIX_SIZE ;l++){
 335           gras_msg_send(socket_row[l-1],gras_msgtype_by_name("dataA"), &mydataA);
 336           bA=mydataA;
 337           INFO1("send my data A (%.3g) to my (horizontal) neighbors",bA);
 338         }
 339       }
 340
 341       if(mycol != step){
 342         TRY {
 343           gras_msg_wait(1200,gras_msgtype_by_name("dataA"),
 344                         &from,&bA);
 345         } CATCH(e) {
 346           RETHROW0("I Can't get a data message from column : %s");
 347         }
 348         INFO2("Recive data A (%.3g) from my neighbor : %s ",bA,gras_socket_peer_name(from));
 349       }
 350       bC+=bA*bB;
 351       INFO1(">>>>>>>> My BC = %.3g",bC);
 352
 353       /* send a ack msg to master */
 354
 355       gras_msg_send(from,gras_msgtype_by_name("step_ack"),&step);
 356
 357       INFO1("Send ack to master for to end %d th step",step);
 358     }
 359     if(step==MATRIX_SIZE-1) break;
 360
 361   } while (step < MATRIX_SIZE);
 362   /*  wait Message from master to send the result */
 363
 364   result.value=bC;
 365   result.i=myrow;
 366   result.j=mycol;
 367
 368   TRY {
 369     gras_msg_wait(600,gras_msgtype_by_name("ask_result"),
 370                   &from,&result_ack);
 371   } CATCH(e) {
 372     RETHROW0("I Can't get a data message from row : %s");
 373   }
 374   /* send Result to master */
 375   TRY {
 376     gras_msg_send(from, gras_msgtype_by_name("result"),&result);
 377   } CATCH(e) {
 378     // gras_socket_close(from);
 379     RETHROW0("Failed to send PING to server: %s");
 380   }
 381   INFO3(">>>>>>>> Result: %.3f sent to %s:%d <<<<<<<<",
 382         bC,
 383         gras_socket_peer_name(from),gras_socket_peer_port(from));
 384   /*  Free the allocated resources, and shut GRAS down */
 385   gras_socket_close(from);
 386   gras_exit();
 387   INFO0("Done.");
 388   return 0;
 389 } /* end_of_slave */