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