src/gras/Transport/rl_transport.c

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