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