src/xbt/xbt_socket.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 #include "xbt/ex.h"
  10 #include "xbt/peer.h"
  11 #include "xbt/dict.h"
  12 #include "xbt/socket.h"
  13 #include "xbt_modinter.h"
  14 #include "portable.h"
  15 #include "xbt_socket_private.h"
  16 #include "gras/Msg/msg_interface.h" /* FIXME */
  17
  18 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_trp, xbt,
  19                                 "Conveying bytes over the network");
  20 XBT_LOG_NEW_SUBCATEGORY(xbt_trp_meas, xbt_trp,
  21                         "Conveying bytes over the network without formating for perf measurements");
  22
  23 static short int xbt_trp_started = 0;
  24 static xbt_dict_t xbt_trp_plugins;          /* all registered plugins */
  25 static void xbt_trp_plugin_free(void *p);   /* free one of the plugins */
  26
  27 void xbt_trp_plugin_new(const char *name, xbt_trp_setup_t setup)
  28 {
  29   xbt_trp_plugin_t plug = xbt_new0(s_xbt_trp_plugin_t, 1);
  30
  31   XBT_DEBUG("Create plugin %s", name);
  32
  33   plug->name = xbt_strdup(name);
  34   setup(plug);
  35   xbt_dict_set(xbt_trp_plugins, name, plug, NULL);
  36 }
  37
  38 void xbt_trp_preinit(void)
  39 {
  40   if (!xbt_trp_started) {
  41     /* make room for all plugins */
  42     xbt_trp_plugins = xbt_dict_new_homogeneous(xbt_trp_plugin_free);
  43
  44 #ifdef HAVE_WINSOCK2_H
  45     /* initialize the windows mechanism */
  46     {
  47       WORD wVersionRequested;
  48       WSADATA wsaData;
  49
  50       wVersionRequested = MAKEWORD(2, 0);
  51       int res;
  52       res = WSAStartup(wVersionRequested, &wsaData);
  53       xbt_assert(res == 0, "Cannot find a usable WinSock DLL");
  54
  55       /* Confirm that the WinSock DLL supports 2.0. */
  56       /* Note that if the DLL supports versions greater    */
  57       /* than 2.0 in addition to 2.0, it will still return */
  58       /* 2.0 in wVersion since that is the version we      */
  59       /* requested.                                        */
  60
  61       xbt_assert(LOBYTE(wsaData.wVersion) == 2 &&
  62                   HIBYTE(wsaData.wVersion) == 0,
  63                   "Cannot find a usable WinSock DLL");
  64       XBT_INFO("Found and initialized winsock2");
  65     }                           /* The WinSock DLL is acceptable. Proceed. */
  66 #elif HAVE_WINSOCK_H
  67     {
  68       WSADATA wsaData;
  69       int res;
  70       res = WSAStartup(0x0101, &wsaData);
  71       xbt_assert(res == 0, "Cannot find a usable WinSock DLL");
  72       XBT_INFO("Found and initialized winsock");
  73     }
  74 #endif
  75
  76     /* create the TCP transport plugin */
  77     xbt_trp_plugin_new("tcp", xbt_trp_tcp_setup);
  78   }
  79
  80   xbt_trp_started++;
  81 }
  82
  83 void xbt_trp_postexit(void)
  84 {
  85   XBT_DEBUG("xbt_trp value %d", xbt_trp_started);
  86   if (xbt_trp_started == 0) {
  87     return;
  88   }
  89
  90   if (--xbt_trp_started == 0) {
  91 #ifdef HAVE_WINSOCK_H
  92     if (WSACleanup() == SOCKET_ERROR) {
  93       if (WSAGetLastError() == WSAEINPROGRESS) {
  94         WSACancelBlockingCall();
  95         WSACleanup();
  96       }
  97     }
  98 #endif
  99
 100     /* Delete the plugins */
 101     xbt_dict_free(&xbt_trp_plugins);
 102   }
 103 }
 104
 105 void xbt_trp_plugin_free(void *p)
 106 {
 107   xbt_trp_plugin_t plug = p;
 108
 109   if (plug) {
 110     if (plug->exit) {
 111       plug->exit(plug);
 112     } else if (plug->data) {
 113       XBT_DEBUG("Plugin %s lacks exit(). Free data anyway.", plug->name);
 114       free(plug->data);
 115     }
 116
 117     free(plug->name);
 118     free(plug);
 119   }
 120 }
 121
 122
 123 /**
 124  * xbt_trp_socket:
 125  *
 126  * Malloc a new socket with the TCP transport plugin and default parameters.
 127  */
 128 void xbt_socket_new(int incoming, xbt_socket_t* dst)
 129 {
 130   xbt_socket_new_ext(incoming, dst, xbt_trp_plugin_get_by_name("tcp"), 0, 0);
 131 }
 132
 133 /**
 134  * xbt_trp_socket_new:
 135  *
 136  * Malloc a new socket.
 137  */
 138 void xbt_socket_new_ext(int incoming,
 139                         xbt_socket_t * dst,
 140                         xbt_trp_plugin_t plugin,
 141                         unsigned long int buf_size,
 142                         int measurement)
 143 {
 144   xbt_socket_t sock = xbt_new0(s_xbt_socket_t, 1);
 145
 146   XBT_VERB("Create a new socket (%p)", (void *) sock);
 147
 148   sock->plugin = plugin;
 149
 150   sock->incoming = incoming ? 1 : 0;
 151   sock->outgoing = incoming ? 0 : 1;
 152   sock->accepting = incoming ? 1 : 0;
 153   sock->meas = measurement;
 154   sock->recvd = 0;
 155   sock->valid = 1;
 156   sock->moredata = 0;
 157
 158   sock->refcount = 1;
 159   sock->buf_size = buf_size;
 160   sock->sd = -1;
 161
 162   sock->data = NULL;
 163   sock->bufdata = NULL;
 164
 165   *dst = sock;
 166
 167   XBT_OUT();
 168 }
 169
 170 XBT_INLINE void* xbt_socket_get_data(xbt_socket_t sock) {
 171   return sock->data;
 172 }
 173
 174 XBT_INLINE void xbt_socket_set_data(xbt_socket_t sock, void* data) {
 175   sock->data = data;
 176 }
 177
 178 /**
 179  * xbt_trp_send:
 180  *
 181  * Send a bunch of bytes from on socket
 182  * (stable if we know the storage will keep as is until the next trp_flush)
 183  */
 184 void xbt_trp_send(xbt_socket_t sd, char *data, long int size, int stable)
 185 {
 186   xbt_assert(sd->outgoing, "Socket not suited for data send");
 187   sd->plugin->send(sd, data, size, stable);
 188 }
 189
 190 /**
 191  * xbt_trp_recv:
 192  *
 193  * Receive a bunch of bytes from a socket
 194  */
 195 void xbt_trp_recv(xbt_socket_t sd, char *data, long int size)
 196 {
 197   xbt_assert(sd->incoming, "Socket not suited for data receive");
 198   (sd->plugin->recv) (sd, data, size);
 199 }
 200
 201 /**
 202  * gras_trp_flush:
 203  *
 204  * Make sure all pending communications are done
 205  */
 206 void xbt_trp_flush(xbt_socket_t sd)
 207 {
 208   if (sd->plugin->flush)
 209     (sd->plugin->flush) (sd);
 210 }
 211
 212 xbt_trp_plugin_t xbt_trp_plugin_get_by_name(const char *name)
 213 {
 214   return xbt_dict_get(xbt_trp_plugins, name);
 215 }
 216
 217 int xbt_socket_my_port(xbt_socket_t sock)
 218 {
 219   if (!sock->plugin->my_port)
 220     THROWF(unknown_error, 0, "Function my_port unimplemented in plugin %s",sock->plugin->name);
 221   return sock->plugin->my_port(sock);
 222 }
 223
 224 int xbt_socket_peer_port(xbt_socket_t sock)
 225 {
 226   if (!sock->plugin->peer_port)
 227     THROWF(unknown_error, 0, "Function peer_port unimplemented in plugin %s",sock->plugin->name);
 228   return sock->plugin->peer_port(sock);
 229 }
 230
 231 const char *xbt_socket_peer_name(xbt_socket_t sock)
 232 {
 233   xbt_assert(sock->plugin);
 234   return sock->plugin->peer_name(sock);
 235 }
 236
 237 const char *xbt_socket_peer_proc(xbt_socket_t sock)
 238 {
 239   return sock->plugin->peer_proc(sock);
 240 }
 241
 242 void xbt_socket_peer_proc_set(xbt_socket_t sock, char *peer_proc)
 243 {
 244   return sock->plugin->peer_proc_set(sock,peer_proc);
 245 }
 246
 247 /** \brief Check if the provided socket is a measurement one (or a regular one) */
 248 int xbt_socket_is_meas(xbt_socket_t sock)
 249 {
 250   return sock->meas;
 251 }
 252
 253 /** \brief Send a chunk of (random) data over a measurement socket
 254  *
 255  * @param peer measurement socket to use for the experiment
 256  * @param timeout timeout (in seconds)
 257  * @param msg_size size of each chunk sent over the socket (in bytes).
 258  * @param msg_amount how many of these packets you want to send.
 259  *
 260  * Calls to xbt_socket_meas_send() and xbt_socket_meas_recv() on
 261  * each side of the socket should be paired.
 262  *
 263  * The exchanged data is zeroed to make sure it's initialized, but
 264  * there is no way to control what is sent (ie, you cannot use these
 265  * functions to exchange data out of band).
 266  *
 267  * @warning: in SimGrid version 3.1 and previous, the numerical arguments
 268  *           were the total amount of data to send and the msg_size. This
 269  *           was changed for the fool wanting to send more than MAXINT
 270  *           bytes in a fat pipe.
 271  */
 272 void xbt_socket_meas_send(xbt_socket_t peer,
 273                           unsigned int timeout,
 274                           unsigned long int msg_size,
 275                           unsigned long int msg_amount)
 276 {
 277   char *chunk = NULL;
 278   unsigned long int sent_sofar;
 279
 280   XBT_IN("");
 281   THROWF(unknown_error, 0, "measurement sockets were broken in this release of SimGrid and should be ported back in the future."
 282       "If you depend on it, sorry, you have to use an older version, or wait for the future version using it...");
 283
 284   if (peer->plugin == xbt_trp_plugin_get_by_name("tcp")) {
 285     chunk = xbt_malloc0(msg_size);
 286   }
 287
 288   xbt_assert(peer->meas,
 289               "Asked to send measurement data on a regular socket");
 290   xbt_assert(peer->outgoing,
 291               "Socket not suited for data send (this is a server socket)");
 292
 293   for (sent_sofar = 0; sent_sofar < msg_amount; sent_sofar++) {
 294     XBT_CDEBUG(xbt_trp_meas,
 295             "Sent %lu msgs of %lu (size of each: %ld) to %s:%d",
 296             sent_sofar, msg_amount, msg_size, xbt_socket_peer_name(peer),
 297             xbt_socket_peer_port(peer));
 298     peer->plugin->raw_send(peer, chunk, msg_size);
 299   }
 300   XBT_CDEBUG(xbt_trp_meas,
 301           "Sent %lu msgs of %lu (size of each: %ld) to %s:%d", sent_sofar,
 302           msg_amount, msg_size, xbt_socket_peer_name(peer),
 303           xbt_socket_peer_port(peer));
 304
 305   if (peer->plugin == xbt_trp_plugin_get_by_name("tcp")) {
 306     free(chunk);
 307   }
 308
 309   XBT_OUT();
 310 }
 311
 312 /** \brief Receive a chunk of data over a measurement socket
 313  *
 314  * Calls to xbt_socket_meas_send() and xbt_socket_meas_recv() on
 315  * each side of the socket should be paired.
 316  *
 317  * @warning: in SimGrid version 3.1 and previous, the numerical arguments
 318  *           were the total amount of data to send and the msg_size. This
 319  *           was changed for the fool wanting to send more than MAXINT
 320  *           bytes in a fat pipe.
 321  */
 322 void xbt_socket_meas_recv(xbt_socket_t peer,
 323                            unsigned int timeout,
 324                            unsigned long int msg_size,
 325                            unsigned long int msg_amount)
 326 {
 327
 328   char *chunk = NULL;
 329   unsigned long int got_sofar;
 330
 331   XBT_IN("");
 332   THROWF(unknown_error,0,"measurement sockets were broken in this release of SimGrid and should be ported back in the future."
 333       "If you depend on it, sorry, you have to use an older version, or wait for the future version using it...");
 334
 335   if (peer->plugin == xbt_trp_plugin_get_by_name("tcp")) {
 336     chunk = xbt_malloc(msg_size);
 337   }
 338
 339   xbt_assert(peer->meas,
 340               "Asked to receive measurement data on a regular socket");
 341   xbt_assert(peer->incoming, "Socket not suited for data receive");
 342
 343   for (got_sofar = 0; got_sofar < msg_amount; got_sofar++) {
 344     XBT_CDEBUG(xbt_trp_meas,
 345             "Recvd %ld msgs of %lu (size of each: %ld) from %s:%d",
 346             got_sofar, msg_amount, msg_size, xbt_socket_peer_name(peer),
 347             xbt_socket_peer_port(peer));
 348     (peer->plugin->raw_recv) (peer, chunk, msg_size);
 349   }
 350   XBT_CDEBUG(xbt_trp_meas,
 351           "Recvd %ld msgs of %lu (size of each: %ld) from %s:%d",
 352           got_sofar, msg_amount, msg_size, xbt_socket_peer_name(peer),
 353           xbt_socket_peer_port(peer));
 354
 355   if (peer->plugin == xbt_trp_plugin_get_by_name("tcp")) {
 356     free(chunk);
 357   }
 358
 359   XBT_OUT();
 360 }
 361
 362 /**
 363  * \brief Something similar to the good old accept system call.
 364  *
 365  * Make sure that there is someone speaking to the provided server socket.
 366  * In RL, it does an accept(2) and return the result as last argument.
 367  * In SG, as accepts are useless, it returns the provided argument as result.
 368  * You should thus test whether (peer != accepted) before closing both of them.
 369  *
 370  * You should only call this on measurement sockets. It is automatically
 371  * done for regular sockets, but you usually want more control about
 372  * what's going on with measurement sockets.
 373  */
 374 xbt_socket_t xbt_socket_meas_accept(xbt_socket_t peer)
 375 {
 376   xbt_socket_t res;
 377   THROWF(unknown_error,0,"measurement sockets were broken in this release of SimGrid and should be ported back in the future."
 378       "If you depend on it, sorry, you have to use an older version, or wait for the future version using it...");
 379
 380   xbt_assert(peer->meas,
 381               "No need to accept on non-measurement sockets (it's automatic)");
 382
 383   if (!peer->accepting) {
 384     /* nothing to accept here (must be in SG) */
 385     /* BUG: FIXME: this is BAD! it makes tricky to free the accepted socket */
 386     return peer;
 387   }
 388
 389   res = (peer->plugin->socket_accept) (peer);
 390   res->meas = peer->meas;
 391   XBT_CDEBUG(xbt_trp_meas, "meas_accepted onto %d", res->sd);
 392
 393   return res;
 394 }