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