examples/gras/p2p/chord/chord.c

   1 /*
   2  * vim:ts=2:sw=2:expandtab
   3  */
   4
   5 #include <stdio.h>
   6 #include "xbt/sysdep.h"
   7 #include "gras.h"
   8
   9 static int closest_preceding_node(int id);
  10 static void check_predecessor(void);
  11
  12 XBT_LOG_NEW_DEFAULT_CATEGORY(chord,"Messages specific to this example");
  13
  14 typedef enum msg_typus{
  15   PING,
  16   PONG,
  17   GET_PRE,
  18   REP_PRE,
  19   GET_SUC,
  20   REP_SUC,
  21   STD,
  22 }msg_typus;
  23
  24 /*GRAS_DEFINE_TYPE(s_pbio,
  25   struct s_pbio{
  26     msg_typus type;
  27     int dest;
  28     char msg[1024];
  29   };
  30 );
  31 typedef struct s_pbio pbio_t;*/
  32
  33 /*GRAS_DEFINE_TYPE(s_ping,*/
  34   struct s_ping{
  35     int id;
  36   };
  37 /*);*/
  38 typedef struct s_ping ping_t;
  39
  40 /*GRAS_DEFINE_TYPE(s_pong,*/
  41   struct s_pong{
  42     int id;
  43     int failed;
  44   };
  45 /*);*/
  46 typedef struct s_pong pong_t;
  47
  48 GRAS_DEFINE_TYPE(s_notify,
  49   struct s_notify{
  50     int id;
  51     char host[1024];
  52     int port;
  53   };
  54 );
  55 typedef struct s_notify notify_t;
  56
  57 GRAS_DEFINE_TYPE(s_get_suc,
  58   struct s_get_suc{
  59     int id;
  60   };
  61 );
  62 typedef struct s_get_suc get_suc_t;
  63
  64 GRAS_DEFINE_TYPE(s_rep_suc,
  65   struct s_rep_suc{
  66     int id;
  67     char host[1024];
  68     int port;
  69   };
  70 );
  71 typedef struct s_rep_suc rep_suc_t;
  72
  73 typedef struct finger_elem{
  74   int id;
  75   char host[1024];
  76   int port;
  77 }finger_elem;
  78
  79
  80
  81 static void register_messages(){
  82 /*  gras_msgtype_declare("chord",gras_datadesc_by_symbol(s_pbio));*/
  83   gras_msgtype_declare("chord_get_suc",gras_datadesc_by_symbol(s_get_suc));
  84   gras_msgtype_declare("chord_rep_suc",gras_datadesc_by_symbol(s_rep_suc));
  85   gras_msgtype_declare("chord_notify",gras_datadesc_by_symbol(s_notify));
  86 }
  87
  88 /* Global private data */
  89 typedef struct{
  90   gras_socket_t sock; /* server socket on which I'm listening */
  91   int id; /* my id number */
  92   char host[1024]; /* my host name */
  93   int port; /* port on which I'm listening FIXME */
  94   int fingers; /* how many fingers */
  95   finger_elem *finger; /* finger table */
  96   int next_to_fix; /* next in the finger list to be checked */
  97   int pre_id; /* predecessor id */
  98   char pre_host[1024]; /* predecessor host */
  99   int pre_port; /* predecessor port */
 100 }node_data_t;
 101
 102
 103 int node(int argc,char **argv);
 104
 105 /*static int node_cb_chord_handler(gras_socket_t expeditor,void *payload_data){
 106   xbt_ex_t e;
 107   pbio_t pbio_i=*(pbio_t*)payload_data;
 108
 109   node_data_t *globals=(node_data_t*)gras_userdata_get();
 110
 111   INFO4(">>> %d : received %d message from %s to %d <<<",globals->id,pbio_i.type,gras_socket_peer_name(expeditor),pbio_i.dest);
 112
 113
 114
 115 }*/
 116
 117 static int node_cb_get_suc_handler(gras_msg_cb_ctx_t ctx,void *payload_data){
 118   gras_socket_t expeditor=gras_msg_cb_ctx_from(ctx);
 119   xbt_ex_t e;
 120   get_suc_t incoming=*(get_suc_t*)payload_data;
 121   rep_suc_t outgoing;
 122   node_data_t *globals=(node_data_t*)gras_userdata_get();
 123   INFO2("Received a get_successor message from %s for %d",
 124         gras_socket_peer_name(expeditor),incoming.id);
 125   if((globals->id==globals->finger[0].id)||
 126      (incoming.id>globals->id&&incoming.id<=globals->finger[0].id)){
 127     outgoing.id=globals->finger[0].id;
 128     snprintf(outgoing.host,1024,globals->finger[0].host);
 129     outgoing.port=globals->finger[0].port;
 130     INFO0("My successor is his successor!");
 131   }else{
 132     gras_socket_t temp_sock;
 133     int contact=closest_preceding_node(incoming.id);
 134     if(contact==-1){
 135       outgoing.id=globals->finger[0].id;
 136       snprintf(outgoing.host,1024,globals->finger[0].host);
 137       outgoing.port=globals->finger[0].port;
 138       INFO0("My successor is his successor!");
 139     }else{
 140       get_suc_t asking;asking.id=incoming.id;
 141       TRY{
 142         temp_sock=gras_socket_client(globals->finger[contact].host,
 143                                      globals->finger[contact].port);
 144       }CATCH(e){
 145         RETHROW0("Unable to connect!: %s");
 146       }
 147       TRY{
 148         gras_msg_send(temp_sock,"chord_get_suc",&asking);
 149       }CATCH(e){
 150         RETHROW0("Unable to ask!: %s");
 151       }
 152       gras_msg_wait(10.,"chord_rep_suc",&temp_sock, &outgoing);
 153     }
 154   }
 155
 156   TRY{
 157     gras_msg_send(expeditor,"chord_rep_suc",&outgoing);
 158     INFO0("Successor information sent!");
 159   }CATCH(e){
 160     RETHROW2("%s:Timeout sending successor information to %s: %s",
 161              globals->host,gras_socket_peer_name(expeditor));
 162   }
 163   gras_socket_close(expeditor);
 164   return 0;
 165 }
 166
 167 static int closest_preceding_node(int id){
 168   node_data_t *globals=(node_data_t*)gras_userdata_get();
 169   int i;
 170   for(i=globals->fingers-1;i>=0;i--){
 171     if(globals->finger[i].id>globals->id&&globals->finger[i].id<id){
 172       return(i);
 173     }
 174   }
 175
 176   return i;
 177 }
 178
 179 static int node_cb_notify_handler(gras_msg_cb_ctx_t ctx,void *payload_data){
 180   gras_socket_t expeditor=gras_msg_cb_ctx_from(ctx);
 181   /*xbt_ex_t e;*/
 182   notify_t incoming=*(notify_t*)payload_data;
 183   node_data_t *globals=(node_data_t*)gras_userdata_get();
 184   INFO2("Received a notifying message from %s as %d",
 185         gras_socket_peer_name(expeditor),incoming.id);
 186   if(globals->pre_id==-1||
 187      (incoming.id>globals->pre_id&&incoming.id<globals->id)){
 188     globals->pre_id=incoming.id;
 189     snprintf(globals->pre_host,1024,incoming.host);
 190     globals->pre_port=incoming.port;
 191     INFO0("Set as my new predecessor!");
 192   }
 193   return 0;
 194 }
 195
 196 static void fix_fingers(){
 197   xbt_ex_t e;
 198   gras_socket_t temp_sock=NULL;
 199   gras_socket_t temp_sock2=NULL;
 200   node_data_t *globals=(node_data_t*)gras_userdata_get();
 201
 202   TRY{
 203     temp_sock=gras_socket_client(globals->host,globals->port);
 204   }CATCH(e){
 205     RETHROW0("Unable to contact known host: %s");
 206   }
 207   get_suc_t get_suc_msg;get_suc_msg.id=globals->id;
 208   TRY{
 209     gras_msg_send(temp_sock,"chord_get_suc",&get_suc_msg);
 210   }CATCH(e){
 211     gras_socket_close(temp_sock);
 212     RETHROW0("Unable to contact known host to get successor!: %s");
 213   }
 214   rep_suc_t rep_suc_msg;
 215   TRY{
 216     INFO0("Waiting for reply!");
 217     gras_msg_wait(6000,"chord_rep_suc",&temp_sock2, &rep_suc_msg);
 218   }CATCH(e){
 219     RETHROW1("%s: Error waiting for successor:%s",globals->host);
 220   }
 221   globals->finger[0].id=rep_suc_msg.id;
 222   snprintf(globals->finger[0].host,1024,rep_suc_msg.host);
 223   globals->finger[0].port=rep_suc_msg.port;
 224   INFO1("→ Finger %d fixed!",globals->next_to_fix);
 225   gras_socket_close(temp_sock);
 226
 227   globals->next_to_fix=(++globals->next_to_fix==globals->fingers)?
 228                        0:globals->next_to_fix;
 229 }
 230
 231 static void check_predecessor(){
 232   node_data_t *globals = (node_data_t*)gras_userdata_get();
 233   gras_socket_t temp_sock;
 234   xbt_ex_t e;
 235   if (globals->pre_id == -1){
 236     return;
 237   }
 238   TRY{
 239     temp_sock = gras_socket_client( globals->pre_host, globals->pre_port );
 240   }CATCH(e){
 241     globals->pre_id = -1;
 242     globals->pre_host[0] = 0;
 243     globals->pre_port = 0;
 244   }
 245   ping_t ping;
 246   pong_t pong;
 247   ping.id = 0;
 248   TRY{
 249     gras_msg_send( temp_sock, "chord_ping",&ping);
 250   }CATCH(e){
 251     globals->pre_id = -1;
 252     globals->pre_host[0] = 0;
 253     globals->pre_port = 0;
 254   }
 255   TRY{
 256     gras_msg_wait( 60, "chord_pong", &temp_sock, &pong);
 257   }CATCH(e){
 258     globals->pre_id = -1;
 259     globals->pre_host[0] = 0;
 260     globals->pre_port = 0;
 261   }
 262   gras_socket_close(temp_sock);
 263 }
 264
 265 int node(int argc,char **argv){
 266   node_data_t *globals=NULL;
 267   gras_socket_t temp_sock=NULL;
 268   gras_socket_t temp_sock2=NULL;
 269
 270   xbt_ex_t e;
 271
 272   int create=0;
 273   int other_port=-1;
 274   char *other_host;
 275
 276   /* 1. Init the GRAS infrastructure and declare my globals */
 277   gras_init(&argc,argv);
 278
 279   gras_os_sleep((15-gras_os_getpid())*20);
 280
 281   globals=gras_userdata_new(node_data_t);
 282
 283   globals->id=atoi(argv[1]);
 284   globals->port=atoi(argv[2]);
 285   globals->fingers=0;
 286   globals->finger=NULL;
 287   globals->pre_id=-1;
 288   globals->pre_host[0]=0;
 289   globals->pre_port=-1;
 290
 291   snprintf(globals->host,1024,gras_os_myname());
 292
 293   if(argc==3){
 294     create=1;
 295   }else{
 296     asprintf(&other_host,"%s",argv[3]);
 297     other_port=atoi(argv[4]);
 298   }
 299
 300   globals->sock=gras_socket_server(globals->port);
 301   gras_os_sleep(1.0);
 302
 303   register_messages();
 304
 305   globals->finger=(finger_elem*)calloc(1,sizeof(finger_elem));
 306   INFO2("Launching node %s:%d",globals->host,globals->port);
 307   if(create){
 308     INFO0("→Creating ring");
 309     globals->finger[0].id=globals->id;
 310     snprintf(globals->finger[0].host,1024,globals->host);
 311     globals->finger[0].port=globals->port;
 312   }else{
 313     INFO2("→Known node %s:%d",other_host,other_port);
 314     INFO0("→Contacting to determine successor");
 315     TRY{
 316       temp_sock=gras_socket_client(other_host,other_port);
 317     }CATCH(e){
 318       RETHROW0("Unable to contact known host: %s");
 319     }
 320     get_suc_t get_suc_msg;get_suc_msg.id=globals->id;
 321     TRY{
 322       gras_msg_send(temp_sock,"chord_get_suc", &get_suc_msg);
 323     }CATCH(e){
 324       gras_socket_close(temp_sock);
 325       RETHROW0("Unable to contact known host to get successor!: %s");
 326     }
 327     rep_suc_t rep_suc_msg;
 328     TRY{
 329       INFO0("Waiting for reply!");
 330       gras_msg_wait(10.,"chord_rep_suc",&temp_sock2, &rep_suc_msg);
 331     }CATCH(e){
 332       RETHROW1("%s: Error waiting for successor:%s",globals->host);
 333     }
 334     globals->finger[0].id=rep_suc_msg.id;
 335     snprintf(globals->finger[0].host,1024,rep_suc_msg.host);
 336     globals->finger[0].port=rep_suc_msg.port;
 337     INFO3("→Got successor : %d-%s:%d",globals->finger[0].id,
 338           globals->finger[0].host,globals->finger[0].port);
 339     gras_socket_close(temp_sock);
 340     TRY{
 341       temp_sock=gras_socket_client(globals->finger[0].host,
 342                                    globals->finger[0].port);
 343     }CATCH(e){
 344       RETHROW0("Unable to contact successor: %s");
 345     }
 346     notify_t notify_msg;
 347     notify_msg.id=globals->id;
 348     snprintf(notify_msg.host,1024,globals->host);
 349     notify_msg.port=globals->port;
 350     TRY{
 351       gras_msg_send(temp_sock,"chord_notify",&notify_msg);
 352     }CATCH(e){
 353       RETHROW0("Unable to notify successor! %s");
 354     }
 355   }
 356
 357   gras_cb_register("chord_get_suc", &node_cb_get_suc_handler);
 358   gras_cb_register("chord_notify",  &node_cb_notify_handler);
 359   /*gras_cb_register("chord_ping",&node_cb_ping_handler);*/
 360  /* gras_timer_repeat(600.,fix_fingers);*/
 361   /*while(1){*/
 362   int l;
 363   for(l=0;l<50;l++){
 364     TRY{
 365       gras_msg_handle(6000000.0);
 366     }CATCH(e){
 367     }
 368   }
 369   /*}*/
 370
 371   gras_socket_close(globals->sock);
 372   free(globals);
 373   gras_exit();
 374   INFO0("Done");
 375   return(0);
 376 }