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 SENSOR_NBR 9
  11
  12 XBT_LOG_NEW_DEFAULT_CATEGORY(pmm,"Messages specific to this example");
  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 sensor */
  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 sensor
  46         gras_msgtype_declare("init_data", init_data_type);  // send from maestro to sensor to initialize data bA,bB
  47
  48         gras_msgtype_declare("ask_result", gras_datadesc_by_name("int")); // send from maestro to sensor to ask a final result
  49         gras_msgtype_declare("step", gras_datadesc_by_name("int"));// send from maestro to sensor to indicate the begining of step
  50         gras_msgtype_declare("step_ack", gras_datadesc_by_name("int"));//send from sensor to maestro to indicate the end of the current step
  51         gras_msgtype_declare("dataA", gras_datadesc_by_name("double"));// send data between sensor
  52         gras_msgtype_declare("dataB", gras_datadesc_by_name("double"));// send data between sensor
  53 }
  54
  55 /* Function prototypes */
  56 int sensor (int argc,char *argv[]);
  57 int maestro (int argc,char *argv[]);
  58
  59
  60 /* **********************************************************************
  61  * Maestro 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 } maestro_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 maestro (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 sensor */
 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< SENSOR_NBR; 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 sensor to begin a %d th step",step);
 206         /* wait for computing and sensor messages exchange */
 207         i=0;
 208
 209         while  ( i< SENSOR_NBR){
 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 sensor : %s");
 214                 }
 215                 i++;
 216                 INFO1("Recive Ack step ack from %s",gras_socket_peer_name(from));
 217         }
 218         }
 219         /*********************************  gather ***************************************/
 220
 221         ask_result=0;
 222         for( i=1;i< argc;i++){
 223                 gras_msg_send(socket[i],gras_msgtype_by_name("ask_result"),&ask_result);
 224                 INFO1("Send (Ask Result) message to %s",gras_socket_peer_name(socket[i]));
 225         }
 226         /* wait for results */
 227         for( i=1;i< argc;i++){
 228                 gras_msg_wait(600,gras_msgtype_by_name("result"),&from,&result);
 229                 C.data[(result.i-1)*MATRIX_SIZE+(result.j-1)]=result.value;
 230         }
 231         /*    end of gather   */
 232         INFO0 ("The Result of Multiplication is :");
 233         display(C);
 234
 235 return 0;
 236 } /* end_of_maestro */
 237
 238 /* **********************************************************************
 239  * Sensor code
 240  * **********************************************************************/
 241
 242 int sensor(int argc,char *argv[]) {
 243
 244   xbt_ex_t e;
 245
 246   int step,port,l,result_ack=0;
 247   double bA,bB;
 248
 249   int myrow,mycol;
 250   double mydataA,mydataB;
 251   double bC=0;
 252
 253   static end_step;
 254
 255   result_t result;
 256
 257   gras_socket_t from,sock;  /* to recive from server for steps */
 258
 259   gras_socket_t socket_row[MATRIX_SIZE-1],socket_column[MATRIX_SIZE-1]; /* sockets for brodcast to other sensor */
 260
 261   /* Init the GRAS's infrastructure */
 262
 263   gras_init(&argc, argv);
 264
 265   /* Get arguments and create sockets */
 266
 267   port=atoi(argv[1]);
 268
 269   /*  Create my master socket */
 270   sock = gras_socket_server(port);
 271   INFO2("Launch %s (port=%d)",argv[0],port);
 272   gras_os_sleep(1); //wait to start all sensor
 273
 274   int i;
 275   for (i=1;i<MATRIX_SIZE;i++){
 276   socket_row[i-1]=gras_socket_client(argv[i+1],port);
 277   socket_column[i-1]=gras_socket_client(argv[i+MATRIX_SIZE],port);
 278   }
 279
 280   /*  Register the known messages */
 281   register_messages();
 282
 283   /* Recover my initialized Data and My Position*/
 284   init_data_t mydata;
 285   INFO0("wait for init Data");
 286   TRY {
 287           gras_msg_wait(600,gras_msgtype_by_name("init_data"),&from,&mydata);
 288   } CATCH(e) {
 289         RETHROW0("I Can't get a init Data message from Maestro : %s");
 290   }
 291   myrow=mydata.myrow;
 292   mycol=mydata.mycol;
 293   mydataA=mydata.a;
 294   mydataB=mydata.b;
 295   INFO4("Recive MY POSITION (%d,%d) and MY INIT DATA ( A=%.3g | B=%.3g )",
 296         myrow,mycol,mydataA,mydataB);
 297   step=1;
 298
 299   do {  //repeat until compute Cb
 300         step=MATRIX_SIZE+1;  // just intilization for loop
 301
 302   TRY {
 303         gras_msg_wait(60,gras_msgtype_by_name("step"),&from,&step);
 304  } CATCH(e) {
 305           RETHROW0("I Can't get a Next Step message from Maestro : %s");
 306  }
 307   INFO1("Recive a step message from maestro: step = %d ",step);
 308
 309   if (step < MATRIX_SIZE ){
 310           /* a row brodcast */
 311           gras_os_sleep(3);  // IL FAUT EXPRIMER LE TEMPS D'ATTENTE EN FONCTION DE "SENSOR_NBR"
 312           if(myrow==step){
 313                 INFO2("step(%d) = Myrow(%d)",step,myrow);
 314                 for (l=1;l < MATRIX_SIZE ;l++){
 315                   gras_msg_send(socket_column[l-1], gras_msgtype_by_name("dataB"), &mydataB);
 316                   bB=mydataB;
 317                 INFO1("send my data B (%.3g) to my (vertical) neighbors",bB);
 318                 }
 319         }
 320         if(myrow != step){
 321                 INFO2("step(%d) <> Myrow(%d)",step,myrow);
 322                 TRY {
 323                   gras_msg_wait(600,gras_msgtype_by_name("dataB"),
 324                                 &from,&bB);
 325                 } CATCH(e) {
 326                   RETHROW0("I Can't get a data message from row : %s");
 327                 }
 328                 INFO2("Recive data B (%.3g) from my neighbor: %s",bB,gras_socket_peer_name(from));
 329           }
 330           /* a column brodcast */
 331           if(mycol==step){
 332                 for (l=1;l < MATRIX_SIZE ;l++){
 333                         gras_msg_send(socket_row[l-1],gras_msgtype_by_name("dataA"), &mydataA);
 334                         bA=mydataA;
 335                         INFO1("send my data A (%.3g) to my (horizontal) neighbors",bA);
 336                   }
 337           }
 338
 339         if(mycol != step){
 340                 TRY {
 341                    gras_msg_wait(1200,gras_msgtype_by_name("dataA"),
 342                                 &from,&bA);
 343                 } CATCH(e) {
 344                   RETHROW0("I Can't get a data message from column : %s");
 345                 }
 346                 INFO2("Recive data A (%.3g) from my neighbor : %s ",bA,gras_socket_peer_name(from));
 347         }
 348           bC+=bA*bB;
 349           INFO1(">>>>>>>> My BC = %.3g",bC);
 350
 351           /* send a ack msg to Maestro */
 352
 353           gras_msg_send(from,gras_msgtype_by_name("step_ack"),&step);
 354
 355           INFO1("Send ack to maestro for to end %d th step",step);
 356         }
 357           if(step==MATRIX_SIZE-1) break;
 358
 359   } while (step < MATRIX_SIZE);
 360     /*  wait Message from maestro to send the result */
 361
 362         result.value=bC;
 363         result.i=myrow;
 364         result.j=mycol;
 365
 366           TRY {
 367                   gras_msg_wait(600,gras_msgtype_by_name("ask_result"),
 368                                 &from,&result_ack);
 369           } CATCH(e) {
 370                   RETHROW0("I Can't get a data message from row : %s");
 371           }
 372           /* send Result to Maestro */
 373           TRY {
 374                   gras_msg_send(from, gras_msgtype_by_name("result"),&result);
 375           } CATCH(e) {
 376                  // gras_socket_close(from);
 377                   RETHROW0("Failed to send PING to server: %s");
 378           }
 379           INFO3(">>>>>>>> Result: %.3f sent to %s:%d <<<<<<<<",
 380                 bC,
 381                 gras_socket_peer_name(from),gras_socket_peer_port(from));
 382     /*  Free the allocated resources, and shut GRAS down */
 383           gras_socket_close(from);
 384           gras_exit();
 385           INFO0("Done.");
 386           return 0;
 387 } /* end_of_sensor */