src/gras/Transport/rl_transport.c

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