examples/gras/rpc/rpc.c

   1 /* $Id$ */
   2
   3 /* rpc - demo of the RPC features in GRAS                                   */
   4
   5 /* Copyright (c) 2006 Martin Quinson. 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
  12 XBT_LOG_NEW_DEFAULT_CATEGORY(Rpc,"Messages specific to this example");
  13
  14 /* register messages which may be sent (common to client and server) */
  15 static void register_messages(void) {
  16   gras_msgtype_declare_rpc("plain ping",
  17                            gras_datadesc_by_name("int"),
  18                            gras_datadesc_by_name("int"));
  19
  20   gras_msgtype_declare_rpc("raise exception", NULL, NULL);
  21   gras_msgtype_declare_rpc("forward exception", NULL, NULL);
  22   gras_msgtype_declare("kill",NULL);
  23 }
  24
  25 /* Function prototypes */
  26 int server (int argc,char *argv[]);
  27 int forwarder (int argc,char *argv[]);
  28 int client (int argc,char *argv[]);
  29
  30 #define exception_msg       "Error for the client"
  31 #define exception_raising() THROW0(unknown_error,42,exception_msg)
  32
  33 static void exception_catching(void) {
  34   int gotit = 0,i;
  35   xbt_ex_t e;
  36
  37   for (i=0; i<5; i++) {
  38     gotit=0;
  39     TRY {
  40       exception_raising();
  41     } CATCH(e) {
  42       gotit = 1;
  43     }
  44     if (!gotit) {
  45       THROW0(unknown_error,0,"Didn't got the remote exception!");
  46     }
  47     xbt_assert2(e.category == unknown_error, "Got wrong category: %d (instead of %d)",
  48                 e.category,unknown_error);
  49     xbt_assert1(e.value == 42, "Got wrong value: %d (!=42)", e.value);
  50     xbt_assert1(!strncmp(e.msg,exception_msg, strlen(exception_msg)),
  51                 "Got wrong message: %s", e.msg);
  52     xbt_ex_free(e);
  53   }
  54 }
  55
  56 /* **********************************************************************
  57  * Client code
  58  * **********************************************************************/
  59
  60
  61 int client(int argc,char *argv[]) {
  62   xbt_ex_t e;
  63   gras_socket_t toserver=NULL; /* peer */
  64   gras_socket_t toforwarder=NULL; /* peer */
  65
  66   int ping, pong, i;
  67   volatile int gotit=0;
  68
  69
  70   const char *host = "127.0.0.1";
  71                 int   port = 4000;
  72
  73   memset(&e,0,sizeof(xbt_ex_t));
  74
  75   /* 1. Init the GRAS's infrastructure */
  76   gras_init(&argc, argv);
  77
  78   /* 2. Get the server's address. The command line override defaults when specified */
  79   if (argc == 5) {
  80     host=argv[1];
  81     port=atoi(argv[2]);
  82   }
  83   INFO2("Launch client (server on %s:%d)",host,port);
  84
  85   exception_catching();
  86
  87   /* 3. Wait for the server & forwarder startup */
  88   gras_os_sleep(2);
  89
  90   /* 4. Create a socket to speak to the server */
  91   TRY {
  92     exception_catching();
  93     toserver=gras_socket_client(host,port);
  94     toforwarder=gras_socket_client(argv[3],atoi(argv[4]));
  95   } CATCH(e) {
  96     RETHROW0("Unable to connect to the server: %s");
  97   }
  98   INFO2("Connected to %s:%d.",host,port);
  99
 100
 101   /* 5. Register the messages.
 102         See, it doesn't have to be done completely at the beginning,
 103         but only before use */
 104   exception_catching();
 105   register_messages();
 106
 107   /* 6. Keep the user informed of what's going on */
 108   INFO2("Connected to server which is on %s:%d",
 109         gras_socket_peer_name(toserver),gras_socket_peer_port(toserver));
 110
 111   /* 7. Prepare and send the ping message to the server */
 112   ping = 1234;
 113   TRY {
 114     exception_catching();
 115     gras_msg_rpccall(toserver, 6000.0, "plain ping", &ping, &pong);
 116   } CATCH(e) {
 117     gras_socket_close(toserver);
 118     RETHROW0("Failed to execute a PING rpc on the server: %s");
 119   }
 120   exception_catching();
 121
 122   /* 8. Keep the user informed of what's going on, again */
 123   INFO4("The answer to PING(%d) on %s:%d is PONG(%d)",
 124         ping,
 125         gras_socket_peer_name(toserver),gras_socket_peer_port(toserver),
 126         pong);
 127
 128   /* 9. Call a RPC which raises an exception (to test exception propagation) */
 129   INFO0("Call the exception raising RPC");
 130   TRY {
 131     gras_msg_rpccall(toserver, 6000.0, "raise exception", NULL, NULL);
 132   } CATCH(e) {
 133     gotit = 1;
 134     xbt_assert2(e.category == unknown_error,
 135                 "Got wrong category: %d (instead of %d)",
 136               e.category,unknown_error);
 137     xbt_assert1(e.value == 42, "Got wrong value: %d (!=42)", e.value);
 138     xbt_assert1(!strncmp(e.msg,exception_msg,strlen(exception_msg)),
 139                 "Got wrong message: %s", e.msg);
 140     INFO0("Got the expected exception when calling the exception raising RPC");
 141     xbt_ex_free(e);
 142   }
 143
 144   if (!gotit)
 145     THROW0(unknown_error,0,"Didn't got the remote exception!");
 146
 147   INFO0("Called the exception raising RPC");
 148   exception_catching();
 149
 150   /* doxygen_ignore */
 151   for (i=0; i<5; i++) {
 152
 153      INFO1("Call the exception raising RPC (i=%d)",i);
 154      TRY {
 155         gras_msg_rpccall(toserver, 6000.0, "raise exception", NULL, NULL);
 156      } CATCH(e) {
 157         gotit = 1;
 158         xbt_ex_free(e);
 159      }
 160      if (!gotit) {
 161         THROW0(unknown_error,0,"Didn't got the remote exception!");
 162      }
 163   }
 164   /* doxygen_resume */
 165
 166   /* 9. Call a RPC which raises an exception (to test that exception propagation works) */
 167   for (i=0;i<5;i++) {
 168     INFO1("Call the exception raising RPC on the forwarder (i=%d)",i);
 169     TRY {
 170       gras_msg_rpccall(toforwarder, 6000.0, "forward exception", NULL, NULL);
 171     } CATCH(e) {
 172       gotit = 1;
 173     }
 174     if (!gotit) {
 175       THROW0(unknown_error,0,"Didn't got the remote exception!");
 176     }
 177     xbt_assert1(e.value == 42, "Got wrong value: %d (!=42)", e.value);
 178     xbt_assert1(!strncmp(e.msg,exception_msg,strlen(exception_msg)),
 179                 "Got wrong message: %s", e.msg);
 180     xbt_assert2(e.category == unknown_error,
 181                 "Got wrong category: %d (instead of %d)",
 182                 e.category,unknown_error);
 183     INFO0("Got the expected exception when calling the exception raising RPC");
 184     xbt_ex_free(e);
 185     exception_catching();
 186   }
 187
 188   INFO2("Ask %s:%d to die",gras_socket_peer_name(toforwarder),gras_socket_peer_port(toforwarder));
 189   gras_msg_send(toforwarder,"kill",NULL);
 190   INFO2("Ask %s:%d to die",gras_socket_peer_name(toserver),gras_socket_peer_port(toserver));
 191   gras_msg_send(toserver,"kill",NULL);
 192
 193   /* 11. Cleanup the place before leaving */
 194   gras_socket_close(toserver);
 195   gras_socket_close(toforwarder);
 196   INFO0("Done.");
 197   gras_exit();
 198   return 0;
 199 } /* end_of_client */
 200
 201
 202 /* **********************************************************************
 203  * Forwarder code
 204  * **********************************************************************/
 205 typedef struct {
 206   gras_socket_t server;
 207   int done;
 208 } s_forward_data_t, *forward_data_t;
 209
 210 static int forwarder_cb_kill(gras_msg_cb_ctx_t ctx,
 211                                  void             *payload_data) {
 212   forward_data_t fdata;
 213   gras_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
 214   INFO2("Asked to die by %s:%d",gras_socket_peer_name(expeditor),gras_socket_peer_port(expeditor));
 215   fdata=gras_userdata_get();
 216   fdata->done = 1;
 217   return 0;
 218 }
 219
 220 static int forwarder_cb_forward_ex(gras_msg_cb_ctx_t ctx,
 221                                    void             *payload_data) {
 222   forward_data_t fdata=gras_userdata_get();
 223
 224   INFO0("Forward a request");
 225   gras_msg_rpccall(fdata->server, 60, "raise exception",NULL,NULL);
 226   return 0;
 227 }
 228
 229 int forwarder (int argc,char *argv[]) {
 230   gras_socket_t mysock;
 231   int port;
 232   forward_data_t fdata;
 233
 234   gras_init(&argc,argv);
 235
 236   xbt_assert(argc == 4);
 237
 238   fdata=gras_userdata_new(s_forward_data_t);
 239   fdata->done = 0;
 240   port=atoi(argv[1]);
 241
 242   INFO1("Launch forwarder (port=%d)", port);
 243   mysock = gras_socket_server(port);
 244
 245   gras_os_sleep(1); /* wait for the server to be ready */
 246   fdata->server=gras_socket_client(argv[2],atoi(argv[3]));
 247
 248   register_messages();
 249   gras_cb_register("forward exception", &forwarder_cb_forward_ex);
 250   gras_cb_register("kill",              &forwarder_cb_kill);
 251
 252   while (!fdata->done) {
 253     gras_msg_handle(600.0);
 254   }
 255
 256   gras_socket_close(mysock);
 257   gras_socket_close(fdata->server);
 258   free(fdata);
 259   INFO0("Done.");
 260   gras_exit();
 261   return 0;
 262 }
 263
 264 /* **********************************************************************
 265  * Server code
 266  * **********************************************************************/
 267 typedef struct {
 268   gras_socket_t server;
 269   int done;
 270 } s_server_data_t, *server_data_t;
 271
 272 static int server_cb_kill(gras_msg_cb_ctx_t ctx,
 273                           void             *payload_data) {
 274   gras_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
 275   server_data_t sdata;
 276
 277   INFO2("Asked to die by %s:%d",gras_socket_peer_name(expeditor),gras_socket_peer_port(expeditor));
 278
 279   sdata=gras_userdata_get();
 280   sdata->done = 1;
 281   return 0;
 282 }
 283
 284 static int server_cb_raise_ex(gras_msg_cb_ctx_t ctx,
 285                               void             *payload_data) {
 286   exception_raising();
 287   return 0;
 288 }
 289
 290 static int server_cb_ping(gras_msg_cb_ctx_t ctx,
 291                           void             *payload_data) {
 292
 293   /* 1. Get the payload into the msg variable, and retrieve who called us */
 294   int msg=*(int*)payload_data;
 295   gras_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
 296
 297   /* 2. Log which client connected */
 298   INFO3("Got message PING(%d) from %s:%d",
 299         msg,
 300         gras_socket_peer_name(expeditor), gras_socket_peer_port(expeditor));
 301
 302   /* 4. Change the value of the msg variable */
 303   msg = 4321;
 304
 305   /* 5. Return as result */
 306   gras_msg_rpcreturn(6000,ctx,&msg);
 307   INFO0("Answered with PONG(4321)");
 308
 309   /* 6. Cleanups, if any */
 310
 311   /* 7. Tell GRAS that we consummed this message */
 312   return 0;
 313 } /* end_of_server_cb_ping */
 314
 315
 316 int server (int argc,char *argv[]) {
 317   gras_socket_t mysock;
 318   server_data_t sdata;
 319
 320   int port = 4000;
 321
 322   /* 1. Init the GRAS infrastructure */
 323   gras_init(&argc,argv);
 324
 325   /* 2. Get the port I should listen on from the command line, if specified */
 326   if (argc == 2)
 327     port=atoi(argv[1]);
 328
 329   sdata=gras_userdata_new(s_server_data_t);
 330   sdata->done = 0;
 331
 332   INFO1("Launch server (port=%d)", port);
 333
 334   /* 3. Create my master socket */
 335   mysock = gras_socket_server(port);
 336
 337   /* 4. Register the known messages and register my callback */
 338   register_messages();
 339   gras_cb_register("plain ping",&server_cb_ping);
 340   gras_cb_register("raise exception",&server_cb_raise_ex);
 341   gras_cb_register("kill",&server_cb_kill);
 342
 343   INFO1("Listening on port %d", gras_socket_my_port(mysock));
 344
 345   /* 5. Wait for the ping incomming messages */
 346
 347   /** \bug if the server is gone before the forwarder tries to connect,
 348      it dies awfully with the following message. The problem stands somewhere
 349      at the interface between the gras_socket_t and the msg mess. There is thus
 350      no way for me to dive into this before this interface is rewritten
 351 ==15875== Invalid read of size 4
 352 ==15875==    at 0x408B805: find_port (transport_plugin_sg.c:68)
 353 ==15875==    by 0x408BD64: gras_trp_sg_socket_client (transport_plugin_sg.c:115)
 354 ==15875==    by 0x404A38B: gras_socket_client_ext (transport.c:255)
 355 ==15875==    by 0x404A605: gras_socket_client (transport.c:288)
 356 ==15875==    by 0x804B49D: forwarder (rpc.c:245)
 357 ==15875==    by 0x80491FB: launch_forwarder (_rpc_simulator.c:52)
 358 ==15875==    by 0x406780B: __context_wrapper (context.c:164)
 359 ==15875==    by 0x41A6CB3: pthread_start_thread (manager.c:310)
 360 ==15875==    by 0x42AA549: clone (clone.S:119)
 361 ==15875==  Address 0x433B49C is 44 bytes inside a block of size 48 free'd
 362 ==15875==    at 0x401CF46: free (vg_replace_malloc.c:235)
 363 ==15875==    by 0x408F1FA: gras_process_exit (sg_process.c:117)
 364 ==15875==    by 0x4049386: gras_exit (gras.c:64)
 365 ==15875==    by 0x804B936: server (rpc.c:345)
 366 ==15875==    by 0x80492B1: launch_server (_rpc_simulator.c:69)
 367 ==15875==    by 0x406780B: __context_wrapper (context.c:164)
 368 ==15875==    by 0x41A6CB3: pthread_start_thread (manager.c:310)
 369 ==15875==    by 0x42AA549: clone (clone.S:119)
 370   */
 371   while (!sdata->done) {
 372     gras_msg_handle(600.0);
 373     exception_catching();
 374   }
 375
 376   /* 8. Free the allocated resources, and shut GRAS down */
 377   free(sdata);
 378   gras_socket_close(mysock);
 379   INFO0("Done.");
 380   gras_exit();
 381
 382   return 0;
 383 } /* end_of_server */
 384