examples/gras/p2p/chord/chord.c

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