examples/gras/alltoall/alltoall.c

   1 /* $Id$ */
   2
   3 /* ALLTOALL - alltoall of GRAS features                         */
   4
   5 /* Copyright (c) 2006 Ahmed Harbaoui. All rights reserved.      */
   6
   7  /* This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the license (GNU LGPL) which comes with this package. */
   9
  10 #include "gras.h"
  11 #include "xbt/ex.h"
  12
  13 XBT_LOG_NEW_DEFAULT_CATEGORY(alltoall,"Messages specific to this example");
  14
  15 /* register data which may be sent (common to client and server) */
  16 static void register_messages(void) {
  17   gras_msgtype_declare("data", gras_datadesc_by_name("int"));
  18 }
  19
  20 /* Function prototypes */
  21 int node (int argc,char *argv[]);
  22
  23
  24 /* **********************************************************************
  25  * node code
  26  * **********************************************************************/
  27
  28 /* Global private data */
  29 typedef struct {
  30   gras_socket_t sock;
  31   int done;
  32 } node_data_t;
  33
  34 static void free_host(void *d){
  35   xbt_host_t h=*(xbt_host_t*)d;
  36   free(h->name);
  37   free(h);
  38 }
  39
  40 static void kill_buddy(char *name,int port){
  41         gras_socket_t sock=gras_socket_client(name,port);
  42         gras_msg_send(sock,gras_msgtype_by_name("kill"),NULL);
  43         gras_socket_close(sock);
  44 }
  45
  46 static void kill_buddy_dynar(void *b) {
  47         xbt_host_t buddy=*(xbt_host_t*)b;
  48         kill_buddy(buddy->name,buddy->port);
  49 }
  50
  51 static int node_cb_data_handler(gras_msg_cb_ctx_t ctx,
  52                                   void          *payload_data) {
  53
  54   /* Get the payload into the msg variable */
  55   int data=*(int*)payload_data;
  56
  57   gras_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
  58
  59   /* Retrieve the server's state (globals) */
  60
  61   node_data_t *globals=(node_data_t*)gras_userdata_get();
  62   globals->done = 0;
  63
  64   /* Log which client connected */
  65   INFO3(">>>>>>>> Got Data(%d) from %s:%d <<<<<<<<",
  66         data,
  67         gras_socket_peer_name(expeditor), gras_socket_peer_port(expeditor));
  68
  69   /* Set the done boolean to true (and make sure the server stops after receiving it). */
  70   globals->done = 1;
  71
  72   /* Make sure we don't leak sockets */
  73   //gras_socket_close(expeditor);
  74
  75   /* Tell GRAS that we consummed this message */
  76   return 1;
  77 } /* end_of_server_cb_ping_handler */
  78
  79 int node (int argc,char *argv[]) {
  80
  81   xbt_ex_t e;
  82
  83   int port,nb_hosts,data,
  84   i,done;
  85
  86   xbt_host_t h1;
  87
  88   gras_socket_t peer;  /* socket to node */
  89
  90   node_data_t *globals;
  91
  92   /* xbt_dynar for hosts */
  93   xbt_dynar_t hosts = xbt_dynar_new(sizeof(xbt_host_t),&free_host);
  94
  95   /* Init the GRAS infrastructure and declare my globals */
  96   gras_init(&argc,argv);
  97
  98   globals=gras_userdata_new(node_data_t *);
  99
 100   /* Get the port I should listen on from the command line, if specified */
 101   if (argc > 2) {
 102     port=atoi(argv[1]);
 103   }
 104
 105   /* Get the node location from argc/argv */
 106   for (i=2; i<argc; i++){
 107     xbt_host_t host=xbt_new(s_xbt_host_t,1);
 108     host->name=strdup(argv[i]);
 109     host->port=atoi(argv[1]);
 110     INFO2("New node : %s:%d",host->name,host->port);
 111     xbt_dynar_push(hosts,&host);
 112   }
 113   nb_hosts = xbt_dynar_length(hosts);
 114
 115   INFO1("Launch current node (port=%d)", port);
 116
 117   /* Create my master socket */
 118   globals->sock = gras_socket_server(port);
 119
 120   /* Register the known messages */
 121   register_messages();
 122   register_messages();
 123
 124   /* 3. Wait for others nodesthe startup */
 125   gras_os_sleep(1);
 126
 127
 128   /* Register my callback */
 129   gras_cb_register(gras_msgtype_by_name("data"),&node_cb_data_handler);
 130
 131   INFO1(">>>>>>>> Listening on port %d <<<<<<<<", gras_socket_my_port(globals->sock));
 132   globals->done=0;
 133
 134   data =1000;
 135   xbt_dynar_foreach(hosts,i,h1) {
 136           peer = gras_socket_client(h1->name,h1->port);
 137    done=0;
 138     while (!done){
 139       TRY {
 140         gras_msg_handle(0);
 141       }CATCH(e){
 142         if (e.category != timeout_error)
 143         RETHROW;
 144         xbt_ex_free(e);
 145         done = 1;
 146       }
 147     }
 148
 149     gras_msg_send(peer,gras_msgtype_by_name("data"),&data);
 150     INFO3(">>>>>>>> Send Data (%d) from %s to %s <<<<<<<<",
 151           data,argv[0],h1->name);
 152   }
 153
 154   if (!globals->done)
 155      WARN0("An error occured, the done was not set by the callback");
 156
 157  /* Free the allocated resources, and shut GRAS down */
 158   gras_socket_close(globals->sock);
 159   free(globals);
 160   gras_exit();
 161
 162   //xbt_dynar_map(hosts,kill_buddy_dynar);
 163   //xbt_dynar_free(&hosts);
 164
 165   INFO0("Done.");
 166   return 0;
 167 } /* end_of_node */