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   volatile int port;
  89   xbt_socket_t res = NULL;
  90   xbt_ex_t e;
  91
  92   for (port = minport; port < maxport; port++) {
  93     TRY {
  94       res = gras_socket_server_ext(port, buf_size, measurement);
  95     }
  96     CATCH(e) {
  97       if (port == maxport)
  98         RETHROW;
  99       xbt_ex_free(e);
 100     }
 101     if (res)
 102       return res;
 103   }
 104   THROW_IMPOSSIBLE;
 105 }
 106
 107 /**
 108  * @brief Opens a client socket to a remote host.
 109  * @param host: who you want to connect to
 110  * @param port: where you want to connect to on this host
 111  * @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)
 112  * @param measurement: whether this socket is meant to convey measurement (if you don't know, use 0 to exchange regular messages)
 113  *
 114  * In real life, you'll get a TCP socket.
 115  */
 116 xbt_socket_t
 117 gras_socket_client_ext(const char *host,
 118                        unsigned short port,
 119                        unsigned long int buf_size, int measurement)
 120 {
 121   xbt_trp_plugin_t trp;
 122   xbt_socket_t sock;
 123
 124   trp = xbt_trp_plugin_get_by_name(gras_if_SG() ? "sg" : "tcp");
 125
 126   XBT_DEBUG("Create a client socket from plugin %s",
 127          gras_if_RL()? "tcp" : "sg");
 128   /* defaults settings */
 129   xbt_socket_new_ext(0,
 130                      &sock,
 131                      trp,
 132                      (buf_size > 0 ? buf_size : 32 * 1024),
 133                      measurement);
 134
 135   /* plugin-specific */
 136   TRY {
 137     (*trp->socket_client) (trp, host, port, sock);
 138   }
 139   CATCH_ANONYMOUS {
 140     free(sock);
 141     RETHROW;
 142   }
 143   xbt_dynar_push(((gras_trp_procdata_t)
 144                   gras_libdata_by_id(gras_trp_libdata_id))->sockets,
 145                  &sock);
 146   gras_msg_listener_awake();
 147   return sock;
 148 }
 149
 150 /**
 151  * @brief Opens a server socket and make it ready to be listened to.
 152  *
 153  * In real life, you'll get a TCP socket.
 154  */
 155 xbt_socket_t gras_socket_server(unsigned short port)
 156 {
 157   return gras_socket_server_ext(port, 32 * 1024, 0);
 158 }
 159
 160 /** @brief Opens a client socket to a remote host */
 161 xbt_socket_t gras_socket_client(const char *host, unsigned short port)
 162 {
 163   return gras_socket_client_ext(host, port, 0, 0);
 164 }
 165
 166 /** @brief Opens a client socket to a remote host specified as '\a host:\a port' */
 167 xbt_socket_t gras_socket_client_from_string(const char *host)
 168 {
 169   xbt_peer_t p = xbt_peer_from_string(host);
 170   xbt_socket_t res = gras_socket_client_ext(p->name, p->port, 0, 0);
 171   xbt_peer_free(p);
 172   return res;
 173 }
 174
 175 void gras_socket_close_voidp(void *sock)
 176 {
 177   gras_socket_close((xbt_socket_t) sock);
 178 }
 179
 180 /** \brief Close socket */
 181 void gras_socket_close(xbt_socket_t sock)
 182 {
 183   if (--sock->refcount)
 184     return;
 185
 186   xbt_dynar_t sockets =
 187       ((gras_trp_procdata_t)
 188        gras_libdata_by_id(gras_trp_libdata_id))->sockets;
 189   xbt_socket_t sock_iter = NULL;
 190   unsigned int cursor;
 191
 192   XBT_IN();
 193   XBT_VERB("Close %p", sock);
 194   if (sock == _gras_lastly_selected_socket) {
 195     xbt_assert(!gras_opt_trp_nomoredata_on_close || !sock->moredata,
 196                 "Closing a socket having more data in buffer while the nomoredata_on_close option is activated");
 197
 198     if (sock->moredata)
 199       XBT_CRITICAL
 200           ("Closing a socket having more data in buffer. Option nomoredata_on_close disabled, so continuing.");
 201     _gras_lastly_selected_socket = NULL;
 202   }
 203
 204   /* FIXME: Issue an event when the socket is closed */
 205   XBT_DEBUG("sockets pointer before %p", sockets);
 206   if (sock) {
 207     /* FIXME: Cannot get the dynar mutex, because it can be already locked */
 208 //              _xbt_dynar_foreach(sockets,cursor,sock_iter) {
 209     for (cursor = 0; cursor < xbt_dynar_length(sockets); cursor++) {
 210       _xbt_dynar_cursor_get(sockets, cursor, &sock_iter);
 211       if (sock == sock_iter) {
 212         XBT_DEBUG("remove sock cursor %u dize %lu\n", cursor,
 213                xbt_dynar_length(sockets));
 214         xbt_dynar_cursor_rm(sockets, &cursor);
 215         if (sock->plugin->socket_close)
 216           (*sock->plugin->socket_close) (sock);
 217
 218         /* free the memory */
 219         free(sock);
 220         XBT_OUT();
 221         return;
 222       }
 223     }
 224     XBT_WARN
 225         ("Ignoring request to free an unknown socket (%p). Execution stack:",
 226          sock);
 227     xbt_backtrace_display_current();
 228   }
 229   XBT_OUT();
 230 }
 231
 232 /*
 233  * Creating procdata for this module
 234  */
 235 static void *gras_trp_procdata_new(void)
 236 {
 237   gras_trp_procdata_t res = xbt_new(s_gras_trp_procdata_t, 1);
 238
 239   res->name = xbt_strdup("gras_trp");
 240   res->name_len = 0;
 241   res->sockets = xbt_dynar_new_sync(sizeof(xbt_socket_t *), NULL);
 242   res->myport = 0;
 243
 244   return (void *) res;
 245 }
 246
 247 /*
 248  * Freeing procdata for this module
 249  */
 250 static void gras_trp_procdata_free(void *data)
 251 {
 252   gras_trp_procdata_t res = (gras_trp_procdata_t) data;
 253
 254   xbt_dynar_free(&(res->sockets));
 255   free(res->name);
 256   free(res);
 257 }
 258
 259 void gras_trp_socketset_dump(const char *name)
 260 {
 261   gras_trp_procdata_t procdata =
 262       (gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id);
 263
 264   unsigned int it;
 265   xbt_socket_t s;
 266
 267   XBT_INFO("** Dump the socket set %s", name);
 268   xbt_dynar_foreach(procdata->sockets, it, s) {
 269     XBT_INFO("  %p -> %s:%d %s",
 270           s, xbt_socket_peer_name(s), xbt_socket_peer_port(s),
 271           s->valid ? "(valid)" : "(peer dead)");
 272   }
 273   XBT_INFO("** End of socket set %s", name);
 274 }
 275
 276 /*
 277  * Module registration
 278  */
 279 int gras_trp_libdata_id;
 280 void gras_trp_register()
 281 {
 282   gras_trp_libdata_id =
 283       gras_procdata_add("gras_trp", gras_trp_procdata_new,
 284                         gras_trp_procdata_free);
 285 }
 286
 287 int gras_os_myport(void)
 288 {
 289   return ((gras_trp_procdata_t)
 290           gras_libdata_by_id(gras_trp_libdata_id))->myport;
 291 }