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  * if timeout=0, do not wait for new message, only handle the ones already there.
  26  *
  27  * if timeout>0 and no message there, wait at most that amount of time before giving up.
  28  */
  29 gras_socket_t gras_trp_select(double timeout) {
  30   xbt_dynar_t sockets= ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->sockets;
  31   int done = -1;
  32   double wakeup = gras_os_time() + timeout;
  33   double now = 0;
  34   /* nextToService used to make sure socket with high number do not starve */
  35   /*  static int nextToService = 0; */
  36   struct timeval tout, *p_tout;
  37
  38   int max_fds=0; /* first arg of select: number of existing sockets */
  39   /* but accept() of winsock returns sockets bigger than the limit, so don't bother
  40      with this tiny optimisation on BillWare */
  41   fd_set FDS;
  42   int ready; /* return of select: number of socket ready to be serviced */
  43   static int fd_setsize=-1; /* FD_SETSIZE not always defined. Get this portably */
  44
  45   gras_socket_t sock_iter; /* iterating over all sockets */
  46   int cursor;              /* iterating over all sockets */
  47
  48   /* Check whether there is more data to read from the socket we selected last time.
  49      This can happen with tcp buffered sockets since we try to get as much data as we can for them */
  50   if (_gras_lastly_selected_socket && _gras_lastly_selected_socket->moredata) {
  51      VERB0("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,"No socket to select onto. Sleep %f sec instead",timeout);
 108        } else {
 109           DEBUG0("No socket to select onto. Return directly.");
 110           THROW0(timeout_error,0, "No socket to select onto. Return directly.");
 111        }
 112     }
 113
 114 #ifndef HAVE_WINSOCK_H
 115     /* we cannot have more than FD_SETSIZE sockets
 116        ... but with WINSOCK which returns sockets higher than the limit (killing this optim) */
 117     if (++max_fds > fd_setsize && fd_setsize > 0) {
 118       WARN1("too many open sockets (%d).",max_fds);
 119       done = 0;
 120       break;
 121     }
 122 #else
 123     max_fds = fd_setsize;
 124 #endif
 125
 126     if (timeout > 0) {
 127       /* set the timeout */
 128       tout.tv_sec = (unsigned long)(wakeup - now);
 129       tout.tv_usec = ((wakeup -now) - ((unsigned long)(wakeup - now))) * 1000000;
 130       p_tout = &tout;
 131     } else if (timeout == 0) {
 132       /* polling only */
 133       tout.tv_sec = 0;
 134       tout.tv_usec = 0;
 135       p_tout = &tout;
 136       /* we just do one loop around */
 137       done = 0;
 138     } else {
 139       /* no timeout: good luck! */
 140       p_tout = NULL;
 141     }
 142
 143     DEBUG2("Selecting over %d socket(s); timeout=%f", max_fds-1,timeout);
 144     ready = select(max_fds, &FDS, NULL, NULL, p_tout);
 145     DEBUG1("select returned %d", ready);
 146     if (ready == -1) {
 147       switch (errno) {
 148       case  EINTR: /* a signal we don't care about occured. we don't care */
 149         /* if we cared, we would have set an handler */
 150         continue;
 151       case EINVAL: /* invalid value */
 152         THROW3(system_error,EINVAL,"invalid select: nb fds: %d, timeout: %d.%d",
 153                max_fds, (int)tout.tv_sec,(int) tout.tv_usec);
 154       case ENOMEM:
 155         xbt_assert0(0,"Malloc error during the select");
 156       default:
 157         THROW2(system_error,errno,"Error during select: %s (%d)",
 158                strerror(errno),errno);
 159       }
 160       THROW_IMPOSSIBLE;
 161     } else if (ready == 0) {
 162       continue;  /* this was a timeout */
 163     }
 164
 165     xbt_dynar_foreach(sockets,cursor,sock_iter) {
 166        if(!FD_ISSET(sock_iter->sd, &FDS)) { /* this socket is not ready */
 167         continue;
 168        }
 169
 170        /* Got a socket to serve */
 171        ready--;
 172
 173        if (   sock_iter->accepting
 174            && sock_iter->plugin->socket_accept) {
 175          /* not a socket but an ear. accept on it and serve next socket */
 176          gras_socket_t accepted=NULL;
 177
 178          /* release mutex before accept; it will change the sockets dynar, so we have to break the foreach asap */
 179          xbt_dynar_cursor_unlock(sockets);
 180          accepted = (sock_iter->plugin->socket_accept)(sock_iter);
 181
 182          DEBUG2("accepted=%p,&accepted=%p",accepted,&accepted);
 183          accepted->meas = sock_iter->meas;
 184          break;
 185
 186        } else if (sock_iter->recv_ok) {
 187          /* Make sure the socket is still alive by reading the first byte */
 188          char lookahead;
 189          int recvd;
 190
 191          recvd = recv(sock_iter->sd, &lookahead, 1, MSG_PEEK);
 192          if (recvd < 0) {
 193            WARN2("socket %d failed: %s", sock_iter->sd, strerror(errno));
 194            /* done with this socket; remove it and break the foreach since it will change the dynar */
 195            xbt_dynar_cursor_unlock(sockets);
 196            gras_socket_close(sock_iter);
 197            break;
 198          } else if (recvd == 0) {
 199            /* Connection reset (=closed) by peer. */
 200            DEBUG1("Connection %d reset by peer", sock_iter->sd);
 201            sock_iter->valid=0; /* don't close it. User may keep references to it */
 202          } else {
 203            /* Got a suited socket ! */
 204            XBT_OUT;
 205            _gras_lastly_selected_socket = sock_iter;
 206             /* break sync dynar iteration */
 207             xbt_dynar_cursor_unlock(sockets);
 208            return sock_iter;
 209          }
 210
 211        } else {
 212          /* This is a file socket. Cannot recv() on it, but it must be alive */
 213            XBT_OUT;
 214            _gras_lastly_selected_socket = sock_iter;
 215           xbt_dynar_cursor_unlock(sockets);
 216            return sock_iter;
 217        }
 218
 219        /* if we're here, the socket we found wasn't really ready to be served */
 220        if (ready == 0) { /* exausted all sockets given by select. Request new ones */
 221
 222           xbt_dynar_cursor_unlock(sockets);
 223           break;
 224        }
 225     }
 226
 227   }
 228
 229   /* No socket found. Maybe we had timeout=0 and nothing to do */
 230   DEBUG0("TIMEOUT");
 231   THROW0(timeout_error,0,"Timeout");
 232 }
 233
 234 void gras_trp_sg_setup(gras_trp_plugin_t plug) {
 235   THROW0(mismatch_error,0,"No SG transport on live platforms");
 236 }
 237