src/gras/Transport/rl_transport.c

   1 /* rl_transport - RL specific functions for transport                       */
   2
   3 /* Copyright (c) 2004, 2005, 2006, 2007, 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/xbt_socket_private.h" /* FIXME */
  11 #include "portable.h"
  12 #include "gras/Transport/transport_private.h"
  13
  14 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(gras_trp);
  15
  16 /* check transport_private.h for an explanation of this variable */
  17 xbt_socket_t _gras_lastly_selected_socket = NULL;
  18
  19 /**
  20  * gras_trp_select:
  21  *
  22  * Returns the next socket to service because it receives a message.
  23  *
  24  * if timeout<0, we ought to implement the adaptative timeout (FIXME)
  25  *
  26  *
  27  * if timeout>0 and no message there, wait at most that amount of time before giving up.
  28  */
  29 xbt_socket_t gras_trp_select(double timeout)
  30 {
  31   xbt_dynar_t sockets =
  32       ((gras_trp_procdata_t)
  33        gras_libdata_by_id(gras_trp_libdata_id))->sockets;
  34   int done = -1;
  35   double wakeup = gras_os_time() + timeout;
  36   double now = 0;
  37   /* nextToService used to make sure socket with high number do not starve */
  38   /*  static int nextToService = 0; */
  39   struct timeval tout, *p_tout;
  40
  41   int max_fds = 0;              /* first arg of select: number of existing sockets */
  42   /* but accept() of winsock returns sockets bigger than the limit, so don't bother
  43      with this tiny optimisation on BillWare */
  44   fd_set FDS;
  45   int ready;                    /* return of select: number of socket ready to be serviced */
  46   static int fd_setsize = -1;   /* FD_SETSIZE not always defined. Get this portably */
  47
  48   xbt_socket_t sock_iter;      /* iterating over all sockets */
  49   unsigned int cursor;          /* iterating over all sockets */
  50
  51   /* Check whether there is more data to read from the socket we selected last time.
  52      This can happen with tcp buffered sockets since we try to get as much data as we can for them */
  53   if (_gras_lastly_selected_socket
  54       && _gras_lastly_selected_socket->moredata) {
  55     XBT_VERB
  56         ("Returning _gras_lastly_selected_socket since there is more data on it");
  57     return _gras_lastly_selected_socket;
  58   }
  59
  60   /* Compute FD_SETSIZE on need */
  61   if (fd_setsize < 0) {
  62 #ifdef HAVE_SYSCONF
  63     fd_setsize = sysconf(_SC_OPEN_MAX);
  64 #else
  65 #  ifdef HAVE_GETDTABLESIZE
  66     fd_setsize = getdtablesize();
  67 #  else
  68     fd_setsize = FD_SETSIZE;
  69 #  endif                        /* !USE_SYSCONF */
  70 #endif
  71   }
  72
  73   while (done == -1) {
  74     if (timeout > 0) {          /* did we timeout already? */
  75       now = gras_os_time();
  76       XBT_DEBUG("wakeup=%f now=%f", wakeup, now);
  77       if (now == -1 || now >= wakeup) {
  78         /* didn't find anything; no need to update _gras_lastly_selected_socket since its moredata is 0 (or we would have returned it directly) */
  79         THROWF(timeout_error, 0,
  80                "Timeout (%f) elapsed with selecting for incomming connexions",
  81                timeout);
  82       }
  83     }
  84
  85     /* construct the set of socket to ear from */
  86     FD_ZERO(&FDS);
  87     max_fds = -1;
  88     xbt_dynar_foreach(sockets, cursor, sock_iter) {
  89       if (!sock_iter->valid)
  90         continue;
  91
  92       if (sock_iter->incoming) {
  93         XBT_DEBUG("Considering socket %d for select", sock_iter->sd);
  94 #ifndef HAVE_WINSOCK_H
  95         if (max_fds < sock_iter->sd)
  96           max_fds = sock_iter->sd;
  97 #else
  98         max_fds = 0;
  99
 100 #endif
 101         FD_SET(sock_iter->sd, &FDS);
 102       } else {
 103         XBT_DEBUG("Not considering socket %d for select", sock_iter->sd);
 104       }
 105     }
 106
 107
 108     if (max_fds == -1) {
 109       if (timeout > 0) {
 110         XBT_DEBUG("No socket to select onto. Sleep %f sec instead.", timeout);
 111         gras_os_sleep(timeout);
 112         THROWF(timeout_error, 0,
 113                "No socket to select onto. Sleep %f sec instead", timeout);
 114       } else {
 115         XBT_DEBUG("No socket to select onto. Return directly.");
 116         THROWF(timeout_error, 0,
 117                "No socket to select onto. Return directly.");
 118       }
 119     }
 120 #ifndef HAVE_WINSOCK_H
 121     /* we cannot have more than FD_SETSIZE sockets
 122        ... but with WINSOCK which returns sockets higher than the limit (killing this optim) */
 123     if (++max_fds > fd_setsize && fd_setsize > 0) {
 124       XBT_WARN("too many open sockets (%d).", max_fds);
 125       done = 0;
 126       break;
 127     }
 128 #else
 129     max_fds = fd_setsize;
 130 #endif
 131
 132     tout.tv_sec = tout.tv_usec = 0;
 133     if (timeout > 0) {
 134       /* set the timeout */
 135       tout.tv_sec = (unsigned long) (wakeup - now);
 136       tout.tv_usec =
 137           ((wakeup - now) - ((unsigned long) (wakeup - now))) * 1000000;
 138       p_tout = &tout;
 139     } else if (timeout == 0) {
 140       /* polling only */
 141       tout.tv_sec = 0;
 142       tout.tv_usec = 0;
 143       p_tout = &tout;
 144       /* we just do one loop around */
 145       done = 0;
 146     } else {
 147       /* no timeout: good luck! */
 148       p_tout = NULL;
 149     }
 150
 151     XBT_DEBUG("Selecting over %d socket(s); timeout=%f", max_fds - 1,
 152            timeout);
 153     ready = select(max_fds, &FDS, NULL, NULL, p_tout);
 154     XBT_DEBUG("select returned %d", ready);
 155     if (ready == -1) {
 156       switch (errno) {
 157       case EINTR:              /* a signal we don't care about occured. we don't care */
 158         /* if we cared, we would have set an handler */
 159         continue;
 160       case EINVAL:             /* invalid value */
 161         THROWF(system_error, EINVAL,
 162                "invalid select: nb fds: %d, timeout: %d.%d", max_fds,
 163                (int) tout.tv_sec, (int) tout.tv_usec);
 164       case ENOMEM:
 165         xbt_die("Malloc error during the select");
 166       default:
 167         THROWF(system_error, errno, "Error during select: %s (%d)",
 168                strerror(errno), errno);
 169       }
 170       THROW_IMPOSSIBLE;
 171     } else if (ready == 0) {
 172       continue;                 /* this was a timeout */
 173     }
 174
 175     xbt_dynar_foreach(sockets, cursor, sock_iter) {
 176       if (!FD_ISSET(sock_iter->sd, &FDS)) {     /* this socket is not ready */
 177         continue;
 178       }
 179
 180       /* Got a socket to serve */
 181       ready--;
 182
 183       if (sock_iter->accepting && sock_iter->plugin->socket_accept) {
 184         /* not a socket but an ear. accept on it and serve next socket */
 185         xbt_socket_t accepted = NULL;
 186
 187         /* release mutex before accept; it will change the sockets dynar, so we have to break the foreach asap */
 188         xbt_dynar_cursor_unlock(sockets);
 189         accepted = (sock_iter->plugin->socket_accept) (sock_iter);
 190
 191         XBT_DEBUG("accepted=%p,&accepted=%p", accepted, &accepted);
 192         accepted->meas = sock_iter->meas;
 193         break;
 194
 195       } else {
 196         /* Make sure the socket is still alive by reading the first byte */
 197         int recvd;
 198
 199         if (sock_iter->recvd) {
 200           /* Socket wasn't used since last time! Don't bother checking whether it's still alive */
 201           recvd = 1;
 202         } else {
 203           recvd = read(sock_iter->sd, &sock_iter->recvd_val, 1);
 204         }
 205
 206         if (recvd < 0) {
 207           XBT_WARN("socket %d failed: %s", sock_iter->sd, strerror(errno));
 208           /* done with this socket; remove it and break the foreach since it will change the dynar */
 209           xbt_dynar_cursor_unlock(sockets);
 210           gras_socket_close(sock_iter);
 211           break;
 212         } else if (recvd == 0) {
 213           /* Connection reset (=closed) by peer. */
 214           XBT_DEBUG("Connection %d reset by peer", sock_iter->sd);
 215           sock_iter->valid = 0; /* don't close it. User may keep references to it */
 216         } else {
 217           /* Got a suited socket ! */
 218           XBT_OUT();
 219           sock_iter->recvd = 1;
 220           XBT_DEBUG("Filled little buffer (%c %x %d)", sock_iter->recvd_val,
 221                  sock_iter->recvd_val, recvd);
 222           _gras_lastly_selected_socket = sock_iter;
 223           /* break sync dynar iteration */
 224           xbt_dynar_cursor_unlock(sockets);
 225           return sock_iter;
 226         }
 227       }
 228
 229       /* if we're here, the socket we found wasn't really ready to be served */
 230       if (ready == 0) {         /* exausted all sockets given by select. Request new ones */
 231
 232         xbt_dynar_cursor_unlock(sockets);
 233         break;
 234       }
 235     }
 236
 237   }
 238
 239   /* No socket found. Maybe we had timeout=0 and nothing to do */
 240   XBT_DEBUG("TIMEOUT");
 241   THROWF(timeout_error, 0, "Timeout");
 242 }
 243
 244 /* to make the linker happy in SG mode */
 245 void gras_trp_sg_setup(xbt_trp_plugin_t plug)
 246 {
 247   THROWF(mismatch_error, 0, "No SG transport on live platforms");
 248 }