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