src/gras/Transport/transport.c

   1 /* $Id$ */
   2
   3 /* transport - low level communication                                      */
   4
   5 /* Copyright (c) 2004 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 "portable.h"
  11 #include "gras/Transport/transport_private.h"
  12
  13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(transport,gras,"Conveying bytes over the network");
  14 XBT_LOG_NEW_SUBCATEGORY(raw_trp,transport,"Conveying bytes over the network without formating");
  15 static short int _gras_trp_started = 0;
  16
  17 static xbt_dict_t _gras_trp_plugins;      /* All registered plugins */
  18 static void gras_trp_plugin_free(void *p); /* free one of the plugins */
  19
  20 static void
  21 gras_trp_plugin_new(const char *name, gras_trp_setup_t setup) {
  22   xbt_error_t errcode;
  23
  24   gras_trp_plugin_t *plug = xbt_new0(gras_trp_plugin_t, 1);
  25
  26   DEBUG1("Create plugin %s",name);
  27
  28   plug->name=xbt_strdup(name);
  29
  30   errcode = setup(plug);
  31   switch (errcode) {
  32   case mismatch_error:
  33     /* SG plugin return mismatch when in RL mode (and vice versa) */
  34     free(plug->name);
  35     free(plug);
  36     break;
  37
  38   case no_error:
  39     xbt_dict_set(_gras_trp_plugins,
  40                   name, plug, gras_trp_plugin_free);
  41     break;
  42
  43   default:
  44     DIE_IMPOSSIBLE;
  45   }
  46
  47 }
  48
  49 void gras_trp_init(void){
  50   if (!_gras_trp_started) {
  51      /* make room for all plugins */
  52      _gras_trp_plugins=xbt_dict_new();
  53
  54 #ifdef HAVE_WINSOCK2_H
  55      /* initialize the windows mechanism */
  56      {
  57         WORD wVersionRequested;
  58         WSADATA wsaData;
  59
  60         wVersionRequested = MAKEWORD( 2, 0 );
  61         xbt_assert0(WSAStartup( wVersionRequested, &wsaData ) == 0,
  62                     "Cannot find a usable WinSock DLL");
  63
  64         /* Confirm that the WinSock DLL supports 2.0.*/
  65         /* Note that if the DLL supports versions greater    */
  66         /* than 2.0 in addition to 2.0, it will still return */
  67         /* 2.0 in wVersion since that is the version we      */
  68         /* requested.                                        */
  69
  70         xbt_assert0(LOBYTE( wsaData.wVersion ) == 2 &&
  71                     HIBYTE( wsaData.wVersion ) == 0,
  72                     "Cannot find a usable WinSock DLL");
  73         INFO0("Found and initialized winsock2");
  74      }       /* The WinSock DLL is acceptable. Proceed. */
  75 #elif HAVE_WINSOCK_H
  76      {       WSADATA wsaData;
  77         xbt_assert0(WSAStartup( 0x0101, &wsaData ) == 0,
  78                     "Cannot find a usable WinSock DLL");
  79         INFO0("Found and initialized winsock");
  80      }
  81 #endif
  82
  83      /* Add plugins */
  84      gras_trp_plugin_new("tcp", gras_trp_tcp_setup);
  85      gras_trp_plugin_new("file",gras_trp_file_setup);
  86      gras_trp_plugin_new("sg",gras_trp_sg_setup);
  87
  88      /* buf is composed, so it must come after the others */
  89      gras_trp_plugin_new("buf", gras_trp_buf_setup);
  90   }
  91
  92   _gras_trp_started++;
  93 }
  94
  95 void
  96 gras_trp_exit(void){
  97    if (_gras_trp_started == 0) {
  98       return;
  99    }
 100
 101    if ( --_gras_trp_started == 0 ) {
 102 #ifdef HAVE_WINSOCK_H
 103       if ( WSACleanup() == SOCKET_ERROR ) {
 104          if ( WSAGetLastError() == WSAEINPROGRESS ) {
 105             WSACancelBlockingCall();
 106             WSACleanup();
 107          }
 108         }
 109 #endif
 110
 111       xbt_dict_free(&_gras_trp_plugins);
 112    }
 113 }
 114
 115
 116 void gras_trp_plugin_free(void *p) {
 117   gras_trp_plugin_t *plug = p;
 118
 119   if (plug) {
 120     if (plug->exit) {
 121       plug->exit(plug);
 122     } else if (plug->data) {
 123       DEBUG1("Plugin %s lacks exit(). Free data anyway.",plug->name);
 124       free(plug->data);
 125     }
 126
 127     free(plug->name);
 128     free(plug);
 129   }
 130 }
 131
 132
 133 /**
 134  * gras_trp_socket_new:
 135  *
 136  * Malloc a new socket, and initialize it with defaults
 137  */
 138 void gras_trp_socket_new(int incoming,
 139                          gras_socket_t *dst) {
 140
 141   gras_socket_t sock=xbt_new0(s_gras_socket_t,1);
 142
 143   DEBUG1("Create a new socket (%p)", (void*)sock);
 144
 145   sock->plugin = NULL;
 146   sock->sd     = -1;
 147   sock->data   = NULL;
 148
 149   sock->incoming  = incoming ? 1:0;
 150   sock->outgoing  = incoming ? 0:1;
 151   sock->accepting = incoming ? 1:0;
 152
 153   sock->port      = -1;
 154   sock->peer_port = -1;
 155   sock->peer_name = NULL;
 156   sock->raw = 0;
 157
 158   *dst = sock;
 159
 160   xbt_dynar_push(gras_socketset_get(),dst);
 161   XBT_OUT;
 162 }
 163
 164
 165 /**
 166  * gras_socket_server_ext:
 167  *
 168  * Opens a server socket and make it ready to be listened to.
 169  * In real life, you'll get a TCP socket.
 170  */
 171 xbt_error_t
 172 gras_socket_server_ext(unsigned short port,
 173
 174                        unsigned long int bufSize,
 175                        int raw,
 176
 177                        /* OUT */ gras_socket_t *dst) {
 178
 179   xbt_error_t errcode;
 180   gras_trp_plugin_t *trp;
 181   gras_socket_t sock;
 182
 183   *dst = NULL;
 184
 185   DEBUG2("Create a server socket from plugin %s on port %d",
 186          gras_if_RL() ? "tcp" : "sg",
 187          port);
 188   TRY(gras_trp_plugin_get_by_name("buf",&trp));
 189
 190   /* defaults settings */
 191   gras_trp_socket_new(1,&sock);
 192   sock->plugin= trp;
 193   sock->port=port;
 194   sock->bufSize = bufSize;
 195   sock->raw = raw;
 196
 197   /* Call plugin socket creation function */
 198   DEBUG1("Prepare socket with plugin (fct=%p)",trp->socket_server);
 199   errcode = trp->socket_server(trp, sock);
 200   DEBUG3("in=%c out=%c accept=%c",
 201          sock->incoming?'y':'n',
 202          sock->outgoing?'y':'n',
 203          sock->accepting?'y':'n');
 204
 205   if (errcode != no_error) {
 206     free(sock);
 207     return errcode;
 208   }
 209
 210   *dst = sock;
 211
 212   return no_error;
 213 }
 214
 215 /**
 216  * gras_socket_client_ext:
 217  *
 218  * Opens a client socket to a remote host.
 219  * In real life, you'll get a TCP socket.
 220  */
 221 xbt_error_t
 222 gras_socket_client_ext(const char *host,
 223                        unsigned short port,
 224
 225                        unsigned long int bufSize,
 226                        int raw,
 227
 228                        /* OUT */ gras_socket_t *dst) {
 229
 230   xbt_error_t errcode;
 231   gras_trp_plugin_t *trp;
 232   gras_socket_t sock;
 233
 234   *dst = NULL;
 235
 236   TRY(gras_trp_plugin_get_by_name("buf",&trp));
 237
 238   DEBUG1("Create a client socket from plugin %s",gras_if_RL() ? "tcp" : "sg");
 239   /* defaults settings */
 240   gras_trp_socket_new(0,&sock);
 241   sock->plugin= trp;
 242   sock->peer_port = port;
 243   sock->peer_name = (char*)strdup(host?host:"localhost");
 244   sock->bufSize = bufSize;
 245   sock->raw = raw;
 246
 247   /* plugin-specific */
 248   errcode= (*trp->socket_client)(trp, sock);
 249   DEBUG3("in=%c out=%c accept=%c",
 250          sock->incoming?'y':'n',
 251          sock->outgoing?'y':'n',
 252          sock->accepting?'y':'n');
 253
 254   if (errcode != no_error) {
 255     free(sock);
 256     return errcode;
 257   }
 258
 259   *dst = sock;
 260
 261   return no_error;
 262 }
 263
 264 /**
 265  * gras_socket_server:
 266  *
 267  * Opens a server socket and make it ready to be listened to.
 268  * In real life, you'll get a TCP socket.
 269  */
 270 xbt_error_t
 271 gras_socket_server(unsigned short port,
 272                    /* OUT */ gras_socket_t *dst) {
 273    return gras_socket_server_ext(port,32,0,dst);
 274 }
 275
 276 /**
 277  * gras_socket_client:
 278  *
 279  * Opens a client socket to a remote host.
 280  * In real life, you'll get a TCP socket.
 281  */
 282 xbt_error_t
 283 gras_socket_client(const char *host,
 284                    unsigned short port,
 285                    /* OUT */ gras_socket_t *dst) {
 286    return gras_socket_client_ext(host,port,32,0,dst);
 287 }
 288
 289
 290 void gras_socket_close(gras_socket_t sock) {
 291   xbt_dynar_t sockets = gras_socketset_get();
 292   gras_socket_t sock_iter;
 293   int cursor;
 294
 295   XBT_IN;
 296   /* FIXME: Issue an event when the socket is closed */
 297   if (sock) {
 298     xbt_dynar_foreach(sockets,cursor,sock_iter) {
 299       if (sock == sock_iter) {
 300         xbt_dynar_cursor_rm(sockets,&cursor);
 301         if (sock->plugin->socket_close)
 302           (* sock->plugin->socket_close)(sock);
 303
 304         /* free the memory */
 305         if (sock->peer_name)
 306           free(sock->peer_name);
 307         free(sock);
 308         XBT_OUT;
 309         return;
 310       }
 311     }
 312     WARN0("Ignoring request to free an unknown socket");
 313   }
 314   XBT_OUT;
 315 }
 316
 317 /**
 318  * gras_trp_chunk_send:
 319  *
 320  * Send a bunch of bytes from on socket
 321  */
 322 xbt_error_t
 323 gras_trp_chunk_send(gras_socket_t sd,
 324                     char *data,
 325                     long int size) {
 326   xbt_assert1(sd->outgoing,
 327                "Socket not suited for data send (outgoing=%c)",
 328                sd->outgoing?'y':'n');
 329   xbt_assert1(sd->plugin->chunk_send,
 330                "No function chunk_send on transport plugin %s",
 331                sd->plugin->name);
 332   return (*sd->plugin->chunk_send)(sd,data,size);
 333 }
 334 /**
 335  * gras_trp_chunk_recv:
 336  *
 337  * Receive a bunch of bytes from a socket
 338  */
 339 xbt_error_t
 340 gras_trp_chunk_recv(gras_socket_t sd,
 341                     char *data,
 342                     long int size) {
 343   xbt_assert0(sd->incoming,
 344                "Socket not suited for data receive");
 345   xbt_assert1(sd->plugin->chunk_recv,
 346                "No function chunk_recv on transport plugin %s",
 347                sd->plugin->name);
 348   return (sd->plugin->chunk_recv)(sd,data,size);
 349 }
 350
 351 /**
 352  * gras_trp_flush:
 353  *
 354  * Make sure all pending communications are done
 355  */
 356 xbt_error_t
 357 gras_trp_flush(gras_socket_t sd) {
 358   return (sd->plugin->flush)(sd);
 359 }
 360
 361 xbt_error_t
 362 gras_trp_plugin_get_by_name(const char *name,
 363                             gras_trp_plugin_t **dst){
 364
 365   return xbt_dict_get(_gras_trp_plugins,name,(void**)dst);
 366 }
 367
 368 int   gras_socket_my_port  (gras_socket_t sock) {
 369   return sock->port;
 370 }
 371 int   gras_socket_peer_port(gras_socket_t sock) {
 372   return sock->peer_port;
 373 }
 374 char *gras_socket_peer_name(gras_socket_t sock) {
 375   return sock->peer_name;
 376 }
 377
 378 xbt_error_t gras_socket_raw_send(gras_socket_t peer,
 379                                   unsigned int timeout,
 380                                   unsigned long int exp_size,
 381                                   unsigned long int msg_size) {
 382   xbt_error_t errcode;
 383   char *chunk = xbt_malloc(msg_size);
 384   int exp_sofar;
 385
 386   xbt_assert0(peer->raw,"Asked to send raw data on a regular socket");
 387   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
 388      CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
 389              exp_sofar,exp_size,msg_size,
 390              gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 391      TRY(gras_trp_chunk_send(peer,chunk,msg_size));
 392   }
 393   CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
 394           exp_sofar,exp_size,msg_size,
 395           gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 396
 397   free(chunk);
 398   return no_error;/* gras_socket_raw_exchange(peer,1,timeout,expSize,msgSize);    */
 399 }
 400
 401 xbt_error_t gras_socket_raw_recv(gras_socket_t peer,
 402                                   unsigned int timeout,
 403                                   unsigned long int exp_size,
 404                                   unsigned long int msg_size){
 405
 406   xbt_error_t errcode;
 407   char *chunk = xbt_malloc(msg_size);
 408   int exp_sofar;
 409
 410   xbt_assert0(peer->raw,"Asked to recveive raw data on a regular socket\n");
 411   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
 412      CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
 413              exp_sofar,exp_size,msg_size,
 414              gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 415      TRY(gras_trp_chunk_recv(peer,chunk,msg_size));
 416   }
 417   CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
 418           exp_sofar,exp_size,msg_size,
 419           gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 420
 421   free(chunk);
 422   return no_error;/* gras_socket_raw_exchange(peer,0,timeout,expSize,msgSize);    */
 423 }
 424
 425 /*
 426  * Creating procdata for this module
 427  */
 428 static void *gras_trp_procdata_new() {
 429    gras_trp_procdata_t res = xbt_new(s_gras_trp_procdata_t,1);
 430
 431    res->sockets   = xbt_dynar_new(sizeof(gras_socket_t*), NULL);
 432
 433    return (void*)res;
 434 }
 435
 436 /*
 437  * Freeing procdata for this module
 438  */
 439 static void gras_trp_procdata_free(void *data) {
 440    gras_trp_procdata_t res = (gras_trp_procdata_t)data;
 441
 442    xbt_dynar_free(&( res->sockets ));
 443 }
 444
 445 /*
 446  * Module registration
 447  */
 448 void gras_trp_register() {
 449    gras_procdata_add("gras_trp",gras_trp_procdata_new, gras_trp_procdata_free);
 450 }
 451
 452
 453 xbt_dynar_t
 454 gras_socketset_get(void) {
 455    /* FIXME: KILLME */
 456    return ((gras_trp_procdata_t) gras_libdata_get("gras_trp"))->sockets;
 457 }