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 "gras/Transport/transport_private.h"
  11
  12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(transport,gras,"Conveying bytes over the network");
  13 XBT_LOG_NEW_SUBCATEGORY(raw_trp,transport,"Conveying bytes over the network without formating");
  14
  15 static xbt_dict_t _gras_trp_plugins;      /* All registered plugins */
  16 static void gras_trp_plugin_free(void *p); /* free one of the plugins */
  17
  18 static void
  19 gras_trp_plugin_new(const char *name, gras_trp_setup_t setup) {
  20   xbt_error_t errcode;
  21
  22   gras_trp_plugin_t *plug = xbt_new0(gras_trp_plugin_t, 1);
  23
  24   DEBUG1("Create plugin %s",name);
  25
  26   plug->name=xbt_strdup(name);
  27
  28   errcode = setup(plug);
  29   switch (errcode) {
  30   case mismatch_error:
  31     /* SG plugin return mismatch when in RL mode (and vice versa) */
  32     xbt_free(plug->name);
  33     xbt_free(plug);
  34     break;
  35
  36   case no_error:
  37     xbt_dict_set(_gras_trp_plugins,
  38                   name, plug, gras_trp_plugin_free);
  39     break;
  40
  41   default:
  42     DIE_IMPOSSIBLE;
  43   }
  44
  45 }
  46
  47 void gras_trp_init(void){
  48   /* make room for all plugins */
  49   _gras_trp_plugins=xbt_dict_new();
  50
  51   /* Add them */
  52   gras_trp_plugin_new("tcp", gras_trp_tcp_setup);
  53   gras_trp_plugin_new("file",gras_trp_file_setup);
  54   gras_trp_plugin_new("sg",gras_trp_sg_setup);
  55
  56   /* buf is composed, so it must come after the others */
  57   gras_trp_plugin_new("buf", gras_trp_buf_setup);
  58
  59 }
  60
  61 void
  62 gras_trp_exit(void){
  63   xbt_dict_free(&_gras_trp_plugins);
  64 }
  65
  66
  67 void gras_trp_plugin_free(void *p) {
  68   gras_trp_plugin_t *plug = p;
  69
  70   if (plug) {
  71     if (plug->exit) {
  72       plug->exit(plug);
  73     } else if (plug->data) {
  74       DEBUG1("Plugin %s lacks exit(). Free data anyway.",plug->name);
  75       xbt_free(plug->data);
  76     }
  77
  78     xbt_free(plug->name);
  79     xbt_free(plug);
  80   }
  81 }
  82
  83
  84 /**
  85  * gras_trp_socket_new:
  86  *
  87  * Malloc a new socket, and initialize it with defaults
  88  */
  89 void gras_trp_socket_new(int incoming,
  90                          gras_socket_t *dst) {
  91
  92   gras_socket_t sock=xbt_new0(s_gras_socket_t,1);
  93
  94   DEBUG1("Create a new socket (%p)", (void*)sock);
  95
  96   sock->plugin = NULL;
  97   sock->sd     = -1;
  98   sock->data   = NULL;
  99
 100   sock->incoming  = incoming ? 1:0;
 101   sock->outgoing  = incoming ? 0:1;
 102   sock->accepting = incoming ? 1:0;
 103
 104   sock->port      = -1;
 105   sock->peer_port = -1;
 106   sock->peer_name = NULL;
 107   sock->raw = 0;
 108
 109   *dst = sock;
 110
 111   xbt_dynar_push(gras_socketset_get(),dst);
 112 }
 113
 114
 115 /**
 116  * gras_socket_server_ext:
 117  *
 118  * Opens a server socket and make it ready to be listened to.
 119  * In real life, you'll get a TCP socket.
 120  */
 121 xbt_error_t
 122 gras_socket_server_ext(unsigned short port,
 123
 124                        unsigned long int bufSize,
 125                        int raw,
 126
 127                        /* OUT */ gras_socket_t *dst) {
 128
 129   xbt_error_t errcode;
 130   gras_trp_plugin_t *trp;
 131   gras_socket_t sock;
 132
 133   *dst = NULL;
 134
 135   DEBUG2("Create a server socket from plugin %s on port %d",
 136          gras_if_RL() ? "tcp" : "sg",
 137          port);
 138   TRY(gras_trp_plugin_get_by_name("buf",&trp));
 139
 140   /* defaults settings */
 141   gras_trp_socket_new(1,&sock);
 142   sock->plugin= trp;
 143   sock->port=port;
 144   sock->bufSize = bufSize;
 145   sock->raw = raw;
 146
 147   /* Call plugin socket creation function */
 148   errcode = trp->socket_server(trp, sock);
 149   DEBUG3("in=%c out=%c accept=%c",
 150          sock->incoming?'y':'n',
 151          sock->outgoing?'y':'n',
 152          sock->accepting?'y':'n');
 153
 154   if (errcode != no_error) {
 155     xbt_free(sock);
 156     return errcode;
 157   }
 158
 159   *dst = sock;
 160
 161   return no_error;
 162 }
 163
 164 /**
 165  * gras_socket_client_ext:
 166  *
 167  * Opens a client socket to a remote host.
 168  * In real life, you'll get a TCP socket.
 169  */
 170 xbt_error_t
 171 gras_socket_client_ext(const char *host,
 172                        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   TRY(gras_trp_plugin_get_by_name("buf",&trp));
 186
 187   DEBUG1("Create a client socket from plugin %s",gras_if_RL() ? "tcp" : "sg");
 188   /* defaults settings */
 189   gras_trp_socket_new(0,&sock);
 190   sock->plugin= trp;
 191   sock->peer_port = port;
 192   sock->peer_name = (char*)strdup(host?host:"localhost");
 193   sock->bufSize = bufSize;
 194   sock->raw = raw;
 195
 196   /* plugin-specific */
 197   errcode= (*trp->socket_client)(trp, sock);
 198   DEBUG3("in=%c out=%c accept=%c",
 199          sock->incoming?'y':'n',
 200          sock->outgoing?'y':'n',
 201          sock->accepting?'y':'n');
 202
 203   if (errcode != no_error) {
 204     xbt_free(sock);
 205     return errcode;
 206   }
 207
 208   *dst = sock;
 209
 210   return no_error;
 211 }
 212
 213 /**
 214  * gras_socket_server:
 215  *
 216  * Opens a server socket and make it ready to be listened to.
 217  * In real life, you'll get a TCP socket.
 218  */
 219 xbt_error_t
 220 gras_socket_server(unsigned short port,
 221                    /* OUT */ gras_socket_t *dst) {
 222    return gras_socket_server_ext(port,32,0,dst);
 223 }
 224
 225 /**
 226  * gras_socket_client:
 227  *
 228  * Opens a client socket to a remote host.
 229  * In real life, you'll get a TCP socket.
 230  */
 231 xbt_error_t
 232 gras_socket_client(const char *host,
 233                    unsigned short port,
 234                    /* OUT */ gras_socket_t *dst) {
 235    return gras_socket_client_ext(host,port,32,0,dst);
 236 }
 237
 238
 239 void gras_socket_close(gras_socket_t sock) {
 240   xbt_dynar_t sockets = gras_socketset_get();
 241   gras_socket_t sock_iter;
 242   int cursor;
 243
 244   /* FIXME: Issue an event when the socket is closed */
 245   if (sock) {
 246     xbt_dynar_foreach(sockets,cursor,sock_iter) {
 247       if (sock == sock_iter) {
 248         xbt_dynar_cursor_rm(sockets,&cursor);
 249         if ( sock->plugin->socket_close)
 250           (* sock->plugin->socket_close)(sock);
 251
 252         /* free the memory */
 253         if (sock->peer_name)
 254           xbt_free(sock->peer_name);
 255         xbt_free(sock);
 256         return;
 257       }
 258     }
 259     WARN0("Ignoring request to free an unknown socket");
 260   }
 261 }
 262
 263 /**
 264  * gras_trp_chunk_send:
 265  *
 266  * Send a bunch of bytes from on socket
 267  */
 268 xbt_error_t
 269 gras_trp_chunk_send(gras_socket_t sd,
 270                     char *data,
 271                     long int size) {
 272   xbt_assert1(sd->outgoing,
 273                "Socket not suited for data send (outgoing=%c)",
 274                sd->outgoing?'y':'n');
 275   xbt_assert1(sd->plugin->chunk_send,
 276                "No function chunk_send on transport plugin %s",
 277                sd->plugin->name);
 278   return (*sd->plugin->chunk_send)(sd,data,size);
 279 }
 280 /**
 281  * gras_trp_chunk_recv:
 282  *
 283  * Receive a bunch of bytes from a socket
 284  */
 285 xbt_error_t
 286 gras_trp_chunk_recv(gras_socket_t sd,
 287                     char *data,
 288                     long int size) {
 289   xbt_assert0(sd->incoming,
 290                "Socket not suited for data receive");
 291   xbt_assert1(sd->plugin->chunk_recv,
 292                "No function chunk_recv on transport plugin %s",
 293                sd->plugin->name);
 294   return (sd->plugin->chunk_recv)(sd,data,size);
 295 }
 296
 297 /**
 298  * gras_trp_flush:
 299  *
 300  * Make sure all pending communications are done
 301  */
 302 xbt_error_t
 303 gras_trp_flush(gras_socket_t sd) {
 304   return (sd->plugin->flush)(sd);
 305 }
 306
 307 xbt_error_t
 308 gras_trp_plugin_get_by_name(const char *name,
 309                             gras_trp_plugin_t **dst){
 310
 311   return xbt_dict_get(_gras_trp_plugins,name,(void**)dst);
 312 }
 313
 314 int   gras_socket_my_port  (gras_socket_t sock) {
 315   return sock->port;
 316 }
 317 int   gras_socket_peer_port(gras_socket_t sock) {
 318   return sock->peer_port;
 319 }
 320 char *gras_socket_peer_name(gras_socket_t sock) {
 321   return sock->peer_name;
 322 }
 323
 324 xbt_error_t gras_socket_raw_send(gras_socket_t peer,
 325                                   unsigned int timeout,
 326                                   unsigned long int exp_size,
 327                                   unsigned long int msg_size) {
 328   xbt_error_t errcode;
 329   char *chunk = xbt_malloc(msg_size);
 330   int exp_sofar;
 331
 332   xbt_assert0(peer->raw,"Asked to send raw data on a regular socket");
 333   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
 334      CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
 335              exp_sofar,exp_size,msg_size,
 336              gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 337      TRY(gras_trp_chunk_send(peer,chunk,msg_size));
 338   }
 339   CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
 340           exp_sofar,exp_size,msg_size,
 341           gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 342
 343   xbt_free(chunk);
 344   return no_error;/* gras_socket_raw_exchange(peer,1,timeout,expSize,msgSize);    */
 345 }
 346
 347 xbt_error_t gras_socket_raw_recv(gras_socket_t peer,
 348                                   unsigned int timeout,
 349                                   unsigned long int exp_size,
 350                                   unsigned long int msg_size){
 351
 352   xbt_error_t errcode;
 353   char *chunk = xbt_malloc(msg_size);
 354   int exp_sofar;
 355
 356   xbt_assert0(peer->raw,"Asked to recveive raw data on a regular socket\n");
 357   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
 358      CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
 359              exp_sofar,exp_size,msg_size,
 360              gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 361      TRY(gras_trp_chunk_recv(peer,chunk,msg_size));
 362   }
 363   CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
 364           exp_sofar,exp_size,msg_size,
 365           gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 366
 367   xbt_free(chunk);
 368   return no_error;/* gras_socket_raw_exchange(peer,0,timeout,expSize,msgSize);    */
 369 }