src/gras/Transport/transport.c

   1 /* transport - low level communication                                      */
   2
   3 /* Copyright (c) 2004, 2005, 2006, 2007, 2008, 2009, 2010. The SimGrid Team.
   4  * All rights reserved.                                                     */
   5
   6 /* This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the license (GNU LGPL) which comes with this package. */
   8
   9 /***
  10  *** Options
  11  ***/
  12 #ifndef NDEBUG
  13 static int gras_opt_trp_nomoredata_on_close = 0;
  14 #endif
  15
  16 #include "xbt/ex.h"
  17 #include "xbt/peer.h"
  18 #include "xbt/socket.h"
  19 #include "xbt/xbt_socket_private.h" /* FIXME */
  20 #include "portable.h"
  21 #include "gras/Transport/transport_private.h"
  22 #include "gras/Msg/msg_interface.h"
  23
  24 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_trp, gras,
  25                                 "Conveying bytes over the network");
  26
  27 /**
  28  * @brief Opens a server socket and makes it ready to be listened to.
  29  * @param port: port on which you want to listen
  30  * @param buf_size: size of the buffer (in byte) on the socket (for TCP sockets only). If 0, a sain default is used (32k, but may change)
  31  * @param measurement: whether this socket is meant to convey measurement (if you don't know, use 0 to exchange regular messages)
  32  *
  33  * In real life, you'll get a TCP socket.
  34  */
  35 xbt_socket_t
  36 gras_socket_server_ext(unsigned short port,
  37                        unsigned long int buf_size, int measurement)
  38 {
  39   xbt_trp_plugin_t trp;
  40   xbt_socket_t sock;
  41
  42   XBT_DEBUG("Create a server socket from plugin %s on port %d",
  43          gras_if_RL() ? "tcp" : "sg", port);
  44   trp = xbt_trp_plugin_get_by_name(gras_if_SG() ? "sg" : "tcp");
  45
  46   /* defaults settings */
  47   xbt_socket_new_ext(1,
  48                      &sock,
  49                      trp,
  50                      (buf_size > 0 ? buf_size : 32 * 1024),
  51                      measurement);
  52
  53   /* Call plugin socket creation function */
  54   XBT_DEBUG("Prepare socket with plugin (fct=%p)", trp->socket_server);
  55   TRY {
  56     trp->socket_server(trp, port, sock);
  57   }
  58   CATCH_ANONYMOUS {
  59     free(sock);
  60     RETHROW;
  61   }
  62
  63   if (!measurement)
  64     ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport
  65         = port;
  66   xbt_dynar_push(((gras_trp_procdata_t)
  67                   gras_libdata_by_id(gras_trp_libdata_id))->sockets,
  68                  &sock);
  69
  70   gras_msg_listener_awake();
  71   return sock;
  72 }
  73
  74 /**
  75  * @brief Opens a server socket on any port in the given range
  76  *
  77  * @param minport: first port we will try
  78  * @param maxport: last port we will try
  79  * @param buf_size: size of the buffer (in byte) on the socket (for TCP sockets only). If 0, a sain default is used (32k, but may change)
  80  * @param measurement: whether this socket is meant to convey measurement (if you don't know, use 0 to exchange regular messages)
  81  *
  82  * If none of the provided ports works, raises the exception got when trying the last possibility
  83  */
  84 xbt_socket_t
  85 gras_socket_server_range(unsigned short minport, unsigned short maxport,
  86                          unsigned long int buf_size, int measurement)
  87 {
  88
  89   int port;
  90   xbt_socket_t res = NULL;
  91   xbt_ex_t e;
  92
  93   for (port = minport; port < maxport; port++) {
  94     TRY {
  95       res = gras_socket_server_ext(port, buf_size, measurement);
  96     }
  97     CATCH(e) {
  98       if (port == maxport)
  99         RETHROW;
 100       xbt_ex_free(e);
 101     }
 102     if (res)
 103       return res;
 104   }
 105   THROW_IMPOSSIBLE;
 106 }
 107
 108 /**
 109  * @brief Opens a client socket to a remote host.
 110  * @param host: who you want to connect to
 111  * @param port: where you want to connect to on this host
 112  * @param buf_size: size of the buffer (in bytes) on the socket (for TCP sockets only). If 0, a sain default is used (32k, but may change)
 113  * @param measurement: whether this socket is meant to convey measurement (if you don't know, use 0 to exchange regular messages)
 114  *
 115  * In real life, you'll get a TCP socket.
 116  */
 117 xbt_socket_t
 118 gras_socket_client_ext(const char *host,
 119                        unsigned short port,
 120                        unsigned long int buf_size, int measurement)
 121 {
 122   xbt_trp_plugin_t trp;
 123   xbt_socket_t sock;
 124
 125   trp = xbt_trp_plugin_get_by_name(gras_if_SG() ? "sg" : "tcp");
 126
 127   XBT_DEBUG("Create a client socket from plugin %s",
 128          gras_if_RL()? "tcp" : "sg");
 129   /* defaults settings */
 130   xbt_socket_new_ext(0,
 131                      &sock,
 132                      trp,
 133                      (buf_size > 0 ? buf_size : 32 * 1024),
 134                      measurement);
 135
 136   /* plugin-specific */
 137   TRY {
 138     (*trp->socket_client) (trp, host, port, sock);
 139   }
 140   CATCH_ANONYMOUS {
 141     free(sock);
 142     RETHROW;
 143   }
 144   xbt_dynar_push(((gras_trp_procdata_t)
 145                   gras_libdata_by_id(gras_trp_libdata_id))->sockets,
 146                  &sock);
 147   gras_msg_listener_awake();
 148   return sock;
 149 }
 150
 151 /**
 152  * @brief Opens a server socket and make it ready to be listened to.
 153  *
 154  * In real life, you'll get a TCP socket.
 155  */
 156 xbt_socket_t gras_socket_server(unsigned short port)
 157 {
 158   return gras_socket_server_ext(port, 32 * 1024, 0);
 159 }
 160
 161 /** @brief Opens a client socket to a remote host */
 162 xbt_socket_t gras_socket_client(const char *host, unsigned short port)
 163 {
 164   return gras_socket_client_ext(host, port, 0, 0);
 165 }
 166
 167 /** @brief Opens a client socket to a remote host specified as '\a host:\a port' */
 168 xbt_socket_t gras_socket_client_from_string(const char *host)
 169 {
 170   xbt_peer_t p = xbt_peer_from_string(host);
 171   xbt_socket_t res = gras_socket_client_ext(p->name, p->port, 0, 0);
 172   xbt_peer_free(p);
 173   return res;
 174 }
 175
 176 void gras_socket_close_voidp(void *sock)
 177 {
 178   gras_socket_close((xbt_socket_t) sock);
 179 }
 180
 181 /** \brief Close socket */
 182 void gras_socket_close(xbt_socket_t sock)
 183 {
 184   if (--sock->refcount)
 185     return;
 186
 187   xbt_dynar_t sockets =
 188       ((gras_trp_procdata_t)
 189        gras_libdata_by_id(gras_trp_libdata_id))->sockets;
 190   xbt_socket_t sock_iter = NULL;
 191   unsigned int cursor;
 192
 193   XBT_IN();
 194   XBT_VERB("Close %p", sock);
 195   if (sock == _gras_lastly_selected_socket) {
 196     xbt_assert(!gras_opt_trp_nomoredata_on_close || !sock->moredata,
 197                 "Closing a socket having more data in buffer while the nomoredata_on_close option is activated");
 198
 199     if (sock->moredata)
 200       XBT_CRITICAL
 201           ("Closing a socket having more data in buffer. Option nomoredata_on_close disabled, so continuing.");
 202     _gras_lastly_selected_socket = NULL;
 203   }
 204
 205   /* FIXME: Issue an event when the socket is closed */
 206   XBT_DEBUG("sockets pointer before %p", sockets);
 207   if (sock) {
 208     /* FIXME: Cannot get the dynar mutex, because it can be already locked */
 209 //              _xbt_dynar_foreach(sockets,cursor,sock_iter) {
 210     for (cursor = 0; cursor < xbt_dynar_length(sockets); cursor++) {
 211       _xbt_dynar_cursor_get(sockets, cursor, &sock_iter);
 212       if (sock == sock_iter) {
 213         XBT_DEBUG("remove sock cursor %u dize %lu\n", cursor,
 214                xbt_dynar_length(sockets));
 215         xbt_dynar_cursor_rm(sockets, &cursor);
 216         if (sock->plugin->socket_close)
 217           (*sock->plugin->socket_close) (sock);
 218
 219         /* free the memory */
 220         free(sock);
 221         XBT_OUT();
 222         return;
 223       }
 224     }
 225     XBT_WARN
 226         ("Ignoring request to free an unknown socket (%p). Execution stack:",
 227          sock);
 228     xbt_backtrace_display_current();
 229   }
 230   XBT_OUT();
 231 }
 232
 233 /*
 234  * Creating procdata for this module
 235  */
 236 static void *gras_trp_procdata_new(void)
 237 {
 238   gras_trp_procdata_t res = xbt_new(s_gras_trp_procdata_t, 1);
 239
 240   res->name = xbt_strdup("gras_trp");
 241   res->name_len = 0;
 242   res->sockets = xbt_dynar_new_sync(sizeof(xbt_socket_t *), NULL);
 243   res->myport = 0;
 244
 245   return (void *) res;
 246 }
 247
 248 /*
 249  * Freeing procdata for this module
 250  */
 251 static void gras_trp_procdata_free(void *data)
 252 {
 253   gras_trp_procdata_t res = (gras_trp_procdata_t) data;
 254
 255   xbt_dynar_free(&(res->sockets));
 256   free(res->name);
 257   free(res);
 258 }
 259
 260 void gras_trp_socketset_dump(const char *name)
 261 {
 262   gras_trp_procdata_t procdata =
 263       (gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id);
 264
 265   unsigned int it;
 266   xbt_socket_t s;
 267
 268   XBT_INFO("** Dump the socket set %s", name);
 269   xbt_dynar_foreach(procdata->sockets, it, s) {
 270     XBT_INFO("  %p -> %s:%d %s",
 271           s, xbt_socket_peer_name(s), xbt_socket_peer_port(s),
 272           s->valid ? "(valid)" : "(peer dead)");
 273   }
 274   XBT_INFO("** End of socket set %s", name);
 275 }
 276
 277 /*
 278  * Module registration
 279  */
 280 int gras_trp_libdata_id;
 281 void gras_trp_register()
 282 {
 283   gras_trp_libdata_id =
 284       gras_procdata_add("gras_trp", gras_trp_procdata_new,
 285                         gras_trp_procdata_free);
 286 }
 287
 288 int gras_os_myport(void)
 289 {
 290   return ((gras_trp_procdata_t)
 291           gras_libdata_by_id(gras_trp_libdata_id))->myport;
 292 }