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 /**
  16  * gras_trp_select:
  17  *
  18  * Returns the next socket to service because it receives a message.
  19  *
  20  * if timeout<0, we ought to implement the adaptative timeout (FIXME)
  21  *
  22  * if timeout=0, do not wait for new message, only handle the ones already there.
  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   xbt_dynar_t sockets= ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->sockets;
  28   int done = -1;
  29   double wakeup = gras_os_time() + timeout;
  30   double now = 0;
  31   /* nextToService used to make sure socket with high number do not starve */
  32   /*  static int nextToService = 0; */
  33   struct timeval tout, *p_tout;
  34
  35   int max_fds=0; /* first arg of select: number of existing sockets */
  36   /* but accept() of winsock returns sockets bigger than the limit, so don't bother
  37      with this tiny optimisation on BillWare */
  38   fd_set FDS;
  39   int ready; /* return of select: number of socket ready to be serviced */
  40   int fd_setsize; /* FD_SETSIZE not always defined. Get this portably */
  41
  42   gras_socket_t sock_iter; /* iterating over all sockets */
  43   int cursor;              /* iterating over all sockets */
  44
  45
  46   /* Compute FD_SETSIZE */
  47 #ifdef HAVE_SYSCONF
  48    fd_setsize = sysconf( _SC_OPEN_MAX );
  49 #else
  50 #  ifdef HAVE_GETDTABLESIZE
  51    fd_setsize = getdtablesize();
  52 #  else
  53    fd_setsize = FD_SETSIZE;
  54 #  endif /* !USE_SYSCONF */
  55 #endif
  56
  57   while (done == -1) {
  58     if (timeout > 0) { /* did we timeout already? */
  59       now = gras_os_time();
  60       DEBUG2("wakeup=%f now=%f",wakeup, now);
  61       if (now == -1 || now >= wakeup) {
  62         /* didn't find anything */
  63         THROW1(timeout_error,0,
  64                "Timeout (%f) elapsed with selecting for incomming connexions",
  65                timeout);
  66       }
  67     }
  68
  69     /* construct the set of socket to ear from */
  70     FD_ZERO(&FDS);
  71     max_fds = -1;
  72     xbt_dynar_foreach(sockets,cursor,sock_iter) {
  73       if (!sock_iter->valid)
  74          continue;
  75
  76       if (sock_iter->incoming) {
  77         DEBUG1("Considering socket %d for select",sock_iter->sd);
  78 #ifndef HAVE_WINSOCK_H
  79         if (max_fds < sock_iter->sd)
  80           max_fds = sock_iter->sd;
  81 #endif
  82         FD_SET(sock_iter->sd, &FDS);
  83       } else {
  84         DEBUG1("Not considering socket %d for select",sock_iter->sd);
  85       }
  86     }
  87
  88     if (max_fds == -1) {
  89        if (timeout > 0) {
  90           DEBUG1("No socket to select onto. Sleep %f sec instead.",timeout);
  91           gras_os_sleep(timeout);
  92           THROW1(timeout_error,0,"No socket to select onto. Sleep %f sec instead",timeout);
  93        } else {
  94           DEBUG0("No socket to select onto. Return directly.");
  95           THROW0(timeout_error,0, "No socket to select onto. Return directly.");
  96        }
  97     }
  98
  99 #ifndef HAVE_WINSOCK_H
 100     /* we cannot have more than FD_SETSIZE sockets
 101        ... but with WINSOCK which returns sockets higher than the limit (killing this optim) */
 102     if (++max_fds > fd_setsize && fd_setsize > 0) {
 103       WARN1("too many open sockets (%d).",max_fds);
 104       done = 0;
 105       break;
 106     }
 107 #else
 108     max_fds = fd_setsize;
 109 #endif
 110
 111     if (timeout > 0) {
 112       /* set the timeout */
 113       tout.tv_sec = (unsigned long)(wakeup - now);
 114       tout.tv_usec = ((wakeup -now) - ((unsigned long)(wakeup - now))) * 1000000;
 115       p_tout = &tout;
 116     } else if (timeout == 0) {
 117       /* polling only */
 118       tout.tv_sec = 0;
 119       tout.tv_usec = 0;
 120       p_tout = &tout;
 121       /* we just do one loop around */
 122       done = 0;
 123     } else {
 124       /* no timeout: good luck! */
 125       p_tout = NULL;
 126     }
 127
 128     DEBUG2("Selecting over %d socket(s); timeout=%f", max_fds-1,timeout);
 129     ready = select(max_fds, &FDS, NULL, NULL, p_tout);
 130     DEBUG1("select returned %d", ready);
 131     if (ready == -1) {
 132       switch (errno) {
 133       case  EINTR: /* a signal we don't care about occured. we don't care */
 134         /* if we cared, we would have set an handler */
 135         continue;
 136       case EINVAL: /* invalid value */
 137         THROW3(system_error,EINVAL,"invalid select: nb fds: %d, timeout: %d.%d",
 138                max_fds, (int)tout.tv_sec,(int) tout.tv_usec);
 139       case ENOMEM:
 140         xbt_assert0(0,"Malloc error during the select");
 141       default:
 142         THROW2(system_error,errno,"Error during select: %s (%d)",
 143                strerror(errno),errno);
 144       }
 145       THROW_IMPOSSIBLE;
 146     } else if (ready == 0) {
 147       continue;  /* this was a timeout */
 148     }
 149
 150     xbt_dynar_foreach(sockets,cursor,sock_iter) {
 151        if(!FD_ISSET(sock_iter->sd, &FDS)) { /* this socket is not ready */
 152         continue;
 153        }
 154
 155        /* Got a socket to serve */
 156        ready--;
 157
 158        if (   sock_iter->accepting
 159            && sock_iter->plugin->socket_accept) {
 160          /* not a socket but an ear. accept on it and serve next socket */
 161          gras_socket_t accepted=NULL;
 162
 163          accepted = (sock_iter->plugin->socket_accept)(sock_iter);
 164          DEBUG2("accepted=%p,&accepted=%p",accepted,&accepted);
 165          accepted->meas = sock_iter->meas;
 166
 167        } else if (sock_iter->recv_ok) {
 168          /* Make sure the socket is still alive by reading the first byte */
 169          char lookahead;
 170          int recvd;
 171
 172          recvd = recv(sock_iter->sd, &lookahead, 1, MSG_PEEK);
 173          if (recvd < 0) {
 174            WARN2("socket %d failed: %s", sock_iter->sd, strerror(errno));
 175            /* done with this socket */
 176            gras_socket_close(sock_iter);
 177            cursor--;
 178          } else if (recvd == 0) {
 179            /* Connection reset (=closed) by peer. */
 180            DEBUG1("Connection %d reset by peer", sock_iter->sd);
 181            sock_iter->valid=0; /* don't close it. User may keep references to it */
 182          } else {
 183            /* Got a suited socket ! */
 184            XBT_OUT;
 185            return sock_iter;
 186          }
 187
 188        } else {
 189          /* This is a file socket. Cannot recv() on it, but it must be alive */
 190            XBT_OUT;
 191            return sock_iter;
 192        }
 193
 194
 195        /* if we're here, the socket we found wasn't really ready to be served */
 196        if (ready == 0) /* exausted all sockets given by select. Request new ones */
 197          break;
 198     }
 199
 200   }
 201
 202   /* No socket found. Maybe we had timeout=0 and nothing to do */
 203   DEBUG0("TIMEOUT");
 204   THROW0(timeout_error,0,"Timeout");
 205 }
 206
 207 void gras_trp_sg_setup(gras_trp_plugin_t plug) {
 208   THROW0(mismatch_error,0,"No SG transport on live platforms");
 209 }
 210