examples/msg/chord/chord.c

   1 /* Copyright (c) 2010. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include <stdio.h>
   8 #include <math.h>
   9 #include "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
  10 #include "xbt/sysdep.h"         /* calloc, printf */
  11
  12 /* Create a log channel to have nice outputs. */
  13 #include "xbt/log.h"
  14 #include "xbt/asserts.h"
  15 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
  16                              "Messages specific for this msg example");
  17 #define KEY_BITS  6
  18 #define CHORD_NB_KEYS 64
  19 /*
  20 * Finger Element
  21 */
  22 typedef struct{
  23         int id;
  24         const char *host_name;
  25         const char *mailbox;
  26 }finger_elem;
  27
  28 /*
  29  * Node Data
  30  */
  31 typedef struct{
  32         int id; // my id
  33         const char *host_name; //my host name
  34         const char* mailbox; //my mailbox
  35         int fingers_nb; // size of finger list
  36         finger_elem* fingers; // finger list  [ fingers[0] >> Successor
  37         int next; //next finger to fix
  38         int pred_id; // predecessor id
  39         const char* pred_host_name; // predecessor host name
  40         const char* pred_mailbox; // predecessor mailbox
  41 }data_node;
  42
  43 //global;
  44
  45 int get_node_from_key(int key);
  46 int node(int argc, char *argv[]);
  47 int sender(int argc,char *argv[]);
  48 void find_successor_node(data_node *my_data, m_task_t join_task);
  49 finger_elem find_finger_elem(data_node* my_data, int id);
  50 const char* find_closest_preceding(data_node* n_node, int id); //return a mailbox
  51 int get_successor_id(m_host_t);
  52 //need by qsort function
  53 int compare (const void * a, const void * b)
  54 {
  55   return ( *(int*)a - *(int*)b );
  56 }
  57
  58 MSG_error_t test_all(const char *platform_file,
  59                      const char *application_file);
  60
  61 static int is_in_interval(unsigned int id, unsigned int start, unsigned int end) {
  62
  63 id = id % CHORD_NB_KEYS;
  64 start = start % CHORD_NB_KEYS;
  65 end = end % CHORD_NB_KEYS;
  66
  67 /* make sure end >= start and id >= start */
  68 if (end < start) {
  69 end += CHORD_NB_KEYS;
  70 }
  71
  72 if (id < start) {
  73 id += CHORD_NB_KEYS;
  74 }
  75
  76 return id < end;
  77 }
  78
  79
  80 /*
  81  * Node Function
  82  */
  83
  84 /*<process host="host_name" function="node">
  85    <argument value="id"/> <!-- my id -->
  86    <argument value="mailbox"/> <!-- mailbox -->
  87    <!-- optional -->
  88    <argument value="known_host_name" />
  89    <argument value="knwon_host_mailbox" />
  90    >argument value="time_to_sleep"/>
  91   </process>
  92 */
  93 static int cpt = 0;
  94 int node(int argc, char *argv[])
  95 {
  96         m_task_t recv_request = NULL;
  97         int first = 1;
  98         int joined = 0;
  99         int res,create = 0;
 100         if ( argc == 3)   // if no known host are declared >>> first node >>> create chord ring
 101         {
 102                 create = 1;
 103         }
 104
 105         int id = atoi(argv[1]);
 106         int mailbox = atoi(argv[2]);
 107         // init data node
 108         data_node *data = xbt_new(data_node,1);
 109         data->host_name = MSG_host_get_name(MSG_host_self());
 110         data->id =  atoi(argv[1]);
 111         data->mailbox = argv[2];
 112         data->fingers_nb = 1;
 113         data->fingers = xbt_new(finger_elem,KEY_BITS);
 114         data->fingers[0].host_name = data->host_name;
 115         data->fingers[0].id = data->id;
 116         data->fingers[0].mailbox = data->mailbox;
 117         data->next = 0;
 118         data->pred_host_name = NULL;
 119         data->pred_id = -1;
 120         data->pred_mailbox = NULL;
 121
 122 /*
 123  * Ring Point Entry Node
 124  */
 125         if (create) // first ring
 126         {
 127                 INFO0("Create new Chord Ring...");
 128                 joined = 1;
 129                 cpt++;
 130                 //sucessor = n
 131                 data->fingers[0].host_name = data->host_name;
 132                 data->fingers[0].id = data->id;
 133                 data->fingers[0].mailbox = data->mailbox;
 134                 while(cpt < MSG_get_host_number()-1)  //make condition!!!
 135                 {
 136                         recv_request = NULL;
 137                         res = MSG_task_receive(&(recv_request),data->mailbox);
 138                         xbt_assert0(res == MSG_OK, "MSG_receiev failed");
 139                         if (!strcmp(MSG_task_get_name(recv_request), "Join Call"))
 140                         {
 141                                 if(MSG_task_get_data(recv_request)==NULL)
 142                                 {
 143                                         WARN0("Receiving an Empty Data");
 144                                 }
 145                                 data_node *recv_data = (data_node*)MSG_task_get_data(recv_request);
 146                                 INFO1("Receiving a Join Call from %s",recv_data->host_name);
 147                                 if (first)
 148                                 {
 149                                         // predecessor(recv_data) >>>> data
 150                                         recv_data->pred_host_name = data->host_name;
 151                                         recv_data->pred_id = data->id;
 152                                         recv_data->pred_mailbox = data->mailbox;
 153                                         data->fingers_nb = 1;
 154                                         // successor(recv_data) >>> data
 155                                         recv_data->fingers[0].id = data->id;
 156                                         recv_data->fingers[0].host_name = data->host_name;
 157                                         recv_data->fingers[0].mailbox = data->mailbox;
 158                                         //successor(data) >>>> recv_data
 159                                         data->fingers[data->fingers_nb - 1].host_name = recv_data->host_name;
 160                                         data->fingers[data->fingers_nb - 1].id = recv_data->id;
 161                                         data->fingers[data->fingers_nb - 1].mailbox = recv_data->mailbox;
 162                                         INFO1("Sending back a Join Request to %s",recv_data->host_name);
 163                                         MSG_task_set_name(recv_request,"Join Response");
 164                                         MSG_task_send(recv_request,recv_data->mailbox);
 165                                         first = 0;
 166                                 }
 167                                 else{
 168                                         find_successor_node(data,recv_request);
 169                                 }
 170
 171                         }
 172                 }
 173         }
 174 /*
 175  * Joining Node
 176  */
 177         else if(!create)
 178         {
 179                 //Sleep Before Starting
 180                 INFO1("Let's Sleep >>%i",atoi(argv[6]));
 181                 MSG_process_sleep(atoi(argv[5]));
 182                 INFO0("Hey! Let's Send a Join Request");
 183                 //send a join task to the known host via its(known host) mailbox
 184                 const char* known_host_name = argv[3];
 185                 const char* known_mailbox = argv[4];
 186                 int known_id = atoi(argv[5]);
 187                 m_task_t join_request = MSG_task_create("Join Call",10000,2000,data);   // define comp size and comm size (#define ...)
 188                 INFO2("Sending a join request to %s via mailbox %s",known_host_name,known_mailbox);
 189                 MSG_task_send(join_request,known_mailbox);
 190                 //wait for answer on my mailbox
 191                 while(cpt < MSG_get_host_number()-1)
 192                 {
 193                         recv_request = NULL;
 194                         int res = MSG_task_receive(&(recv_request),data->mailbox);
 195                         //check if it's the response for my request
 196                         xbt_assert0(res == MSG_OK, "MSG_receiev failed");
 197                         // get data
 198                         data_node *recv_data = (data_node*)MSG_task_get_data(recv_request);
 199                         // Join Call Message
 200                         if(!strcmp(MSG_task_get_name(recv_request), "Join Call"))
 201                         {
 202
 203                                 INFO1("Receiving Join Call From %s",recv_data->host_name);
 204                                 if(!joined)
 205                                 {
 206                                         INFO1("Sorry %s... I'm not yet joined",recv_data->host_name);
 207                                         //No Treatment
 208                                         MSG_task_set_name(recv_request,"Join Failed");
 209                                         MSG_task_send(recv_request,recv_data->mailbox);
 210                                                 }
 211                                         else
 212                                         {
 213                                                 find_successor_node(data,recv_request);
 214                                         }
 215
 216                         }
 217                         // Join Response
 218                         else if(!strcmp(MSG_task_get_name(recv_request), "Join Response"))
 219                         {
 220                                 INFO0("Receiving Join Response!!!");
 221                                 INFO1("My successor is : %s",data->fingers[0].host_name);
 222                                 INFO1("My Predecessor is : %s",data->pred_host_name);
 223                                 cpt++;
 224                                 joined = 1;
 225                                 INFO1("My finger table size : %i",data->fingers_nb);
 226                                 INFO0("***********************************************************************");
 227
 228                                 /*
 229                                 MSG_task_set_name(recv_request,"Fix Fingers");
 230
 231                                 MSG_task_send(recv_request,data->pred_mailbox);
 232                                 MSG_task_send(recv_request,data->fingers[0].mailbox);
 233                                 */
 234                                 //init_finger_table(data,known_id);
 235
 236                                 //treatment
 237                         }
 238                         // Join Failure Message
 239                         else if(!strcmp(MSG_task_get_name(recv_request), "Join Failed"))
 240                         {
 241                                 INFO0("My Join call has failed... let's Try Again");
 242                                 // send back
 243                                 //MSG_task_send(join_request,known_mailbox);
 244                                 // !!!!!!!!! YVes Jaques Always...???§§§§**************************
 245
 246                         }
 247                         else if(!strcmp(MSG_task_get_name(recv_request), "Fix Fingers"))
 248                         {
 249                                 int i;
 250                                 for(i = KEY_BITS -1 ; i>= 0;i--)
 251                                 {
 252                                         data->fingers[i] = find_finger_elem(data,(data->id)+pow(2,i-1));
 253                                 }
 254                         }
 255                 }
 256         }
 257 }
 258
 259 /*
 260  *
 261  */
 262 void find_successor_node(data_node* n_data,m_task_t join_task)  //use all data
 263 {
 264         //get recv data
 265         data_node *recv_data = (data_node*)MSG_task_get_data(join_task);
 266         INFO3("recv_data->id : %i , n_data->id :%i , successor->id :%i",recv_data->id,n_data->id,n_data->fingers[0].id);
 267         //if ((recv_data->id >= n_data->id) && (recv_data->id <= n_data->fingers[0].id))
 268         if (is_in_interval(recv_data->id,n_data->id,n_data->fingers[0].id))
 269         {
 270                 INFO1("Successor Is %s",n_data->fingers[0].host_name);
 271                 //predecessor(recv_data) >>>> n_data
 272                 recv_data->pred_host_name = n_data->host_name;
 273                 recv_data->pred_id = n_data->id;
 274                 recv_data->pred_mailbox = n_data->pred_mailbox;
 275                 // successor(recv_data) >>>> n_data.finger[0]
 276                 recv_data->fingers_nb = 1;
 277                 recv_data->fingers[0].host_name = n_data->fingers[0].host_name;
 278                 recv_data->fingers[0].id = n_data->fingers[0].id;
 279                 recv_data->fingers[0].mailbox = n_data->fingers[0].mailbox;
 280                 // successor(n_data) >>>> recv_sucessor
 281                 n_data->fingers[0].id = recv_data->id;
 282                 n_data->fingers[0].host_name = recv_data->host_name;
 283                 n_data->fingers[0].mailbox = recv_data->mailbox;
 284                 // Logs
 285                 INFO1("Sending back a Join Request to %s",recv_data->host_name);
 286                 MSG_task_set_name(join_task,"Join Response");
 287                 MSG_task_send(join_task,recv_data->mailbox);
 288         }
 289
 290         else
 291         {
 292                 const char* closest_preceding_mailbox = find_closest_preceding(n_data,recv_data->id);
 293                 INFO1("Forwarding Join Call to mailbox %s",closest_preceding_mailbox);
 294                 MSG_task_send(join_task,closest_preceding_mailbox);
 295         }
 296 }
 297
 298 const char* find_closest_preceding(data_node* n_node,int id)
 299 {
 300         int i;
 301         for(i = n_node->fingers_nb-1; i >= 0 ; i--)
 302         {
 303                         if (n_node->fingers[i].id <= id)
 304                                 return n_node->fingers[i].mailbox;
 305         }
 306
 307         return n_node->mailbox; // !!!!!!!!!!!!!!
 308 }
 309 /*
 310  * Fin successor id : used to fix finger list
 311  */
 312 finger_elem find_finger_elem(data_node* n_data, int id)
 313 {
 314         //if(id >= n_data->id && id <= n_data->fingers[0].id)
 315         if (is_in_interval(id,n_data->id,n_data->fingers[0].id))
 316                 return n_data->fingers[0];
 317         //else
 318                 //return find_finger_elem(...,id);
 319
 320 }
 321
 322
 323 /** Test function */
 324 MSG_error_t test_all(const char *platform_file,
 325                      const char *application_file)
 326 {
 327   MSG_error_t res = MSG_OK;
 328
 329   /* MSG_config("workstation/model","KCCFLN05"); */
 330   {                             /*  Simulation setting */
 331     MSG_set_channel_number(0);
 332     MSG_create_environment(platform_file);
 333
 334   }
 335   {                             /*   Application deployment */
 336         MSG_function_register("node",node);
 337     MSG_launch_application(application_file);
 338   }
 339   res = MSG_main();
 340   INFO1("Simulation time %g", MSG_get_clock());
 341
 342   return res;
 343 }                               /* end_of_test_all */
 344
 345 /** Main function */
 346 int main(int argc, char *argv[])
 347 {
 348   MSG_error_t res = MSG_OK;
 349
 350   MSG_global_init(&argc, argv);
 351   if (argc < 3) {
 352     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 353     printf("example: %s msg_platform.xml msg_deployment.xml\n", argv[0]);
 354     exit(1);
 355   }
 356   res = test_all(argv[1], argv[2]);
 357   MSG_clean();
 358
 359   if (res == MSG_OK)
 360     return 0;
 361   else
 362     return 1;
 363 }                               /* end_of_main */