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->incoming) {
  74         DEBUG1("Considering socket %d for select",sock_iter->sd);
  75 #ifndef HAVE_WINSOCK_H
  76         if (max_fds < sock_iter->sd)
  77           max_fds = sock_iter->sd;
  78 #endif
  79         FD_SET(sock_iter->sd, &FDS);
  80       } else {
  81         DEBUG1("Not considering socket %d for select",sock_iter->sd);
  82       }
  83     }
  84
  85     if (max_fds == -1) {
  86        if (timeout > 0) {
  87           DEBUG1("No socket to select onto. Sleep %f sec instead.",timeout);
  88           gras_os_sleep(timeout);
  89           THROW1(timeout_error,0,"No socket to select onto. Sleep %f sec instead",timeout);
  90        } else {
  91           DEBUG0("No socket to select onto. Return directly.");
  92           THROW0(timeout_error,0, "No socket to select onto. Return directly.");
  93        }
  94     }
  95
  96 #ifndef HAVE_WINSOCK_H
  97     /* we cannot have more than FD_SETSIZE sockets
  98        ... but with WINSOCK which returns sockets higher than the limit (killing this optim) */
  99     if (++max_fds > fd_setsize && fd_setsize > 0) {
 100       WARN1("too many open sockets (%d).",max_fds);
 101       done = 0;
 102       break;
 103     }
 104 #else
 105     max_fds = fd_setsize;
 106 #endif
 107
 108     if (timeout > 0) {
 109       /* set the timeout */
 110       tout.tv_sec = (unsigned long)(wakeup - now);
 111       tout.tv_usec = ((wakeup -now) - ((unsigned long)(wakeup - now))) * 1000000;
 112       p_tout = &tout;
 113     } else if (timeout == 0) {
 114       /* polling only */
 115       tout.tv_sec = 0;
 116       tout.tv_usec = 0;
 117       p_tout = &tout;
 118       /* we just do one loop around */
 119       done = 0;
 120     } else {
 121       /* no timeout: good luck! */
 122       p_tout = NULL;
 123     }
 124
 125     DEBUG2("Selecting over %d socket(s); timeout=%f", max_fds-1,timeout);
 126     ready = select(max_fds, &FDS, NULL, NULL, p_tout);
 127     DEBUG1("select returned %d", ready);
 128     if (ready == -1) {
 129       switch (errno) {
 130       case  EINTR: /* a signal we don't care about occured. we don't care */
 131         /* if we cared, we would have set an handler */
 132         continue;
 133       case EINVAL: /* invalid value */
 134         THROW3(system_error,EINVAL,"invalid select: nb fds: %d, timeout: %d.%d",
 135                max_fds, (int)tout.tv_sec,(int) tout.tv_usec);
 136       case ENOMEM:
 137         xbt_assert0(0,"Malloc error during the select");
 138       default:
 139         THROW2(system_error,errno,"Error during select: %s (%d)",
 140                strerror(errno),errno);
 141       }
 142       THROW_IMPOSSIBLE;
 143     } else if (ready == 0) {
 144       continue;  /* this was a timeout */
 145     }
 146
 147     xbt_dynar_foreach(sockets,cursor,sock_iter) {
 148        if(!FD_ISSET(sock_iter->sd, &FDS)) { /* this socket is not ready */
 149         continue;
 150        }
 151
 152        /* Got a socket to serve */
 153        ready--;
 154
 155        if (   sock_iter->accepting
 156            && sock_iter->plugin->socket_accept) {
 157          /* not a socket but an ear. accept on it and serve next socket */
 158          gras_socket_t accepted=NULL;
 159
 160          accepted = (sock_iter->plugin->socket_accept)(sock_iter);
 161          DEBUG2("accepted=%p,&accepted=%p",accepted,&accepted);
 162          accepted->meas = sock_iter->meas;
 163
 164        } else if (sock_iter->recv_ok) {
 165          /* Make sure the socket is still alive by reading the first byte */
 166          char lookahead;
 167          int recvd;
 168
 169          recvd = recv(sock_iter->sd, &lookahead, 1, MSG_PEEK);
 170          if (recvd < 0) {
 171            WARN2("socket %d failed: %s", sock_iter->sd, strerror(errno));
 172            /* done with this socket */
 173            gras_socket_close(sock_iter);
 174            cursor--;
 175          } else if (recvd == 0) {
 176            /* Connection reset (=closed) by peer. */
 177            DEBUG1("Connection %d reset by peer", sock_iter->sd);
 178            gras_socket_close(sock_iter);
 179            cursor--;
 180          } else {
 181            /* Got a suited socket ! */
 182            XBT_OUT;
 183            return sock_iter;
 184          }
 185
 186        } else {
 187          /* This is a file socket. Cannot recv() on it, but it must be alive */
 188            XBT_OUT;
 189            return sock_iter;
 190        }
 191
 192
 193        /* if we're here, the socket we found wasn't really ready to be served */
 194        if (ready == 0) /* exausted all sockets given by select. Request new ones */
 195          break;
 196     }
 197
 198   }
 199
 200   XBT_OUT;
 201   return NULL;
 202 }
 203
 204 void gras_trp_sg_setup(gras_trp_plugin_t plug) {
 205   THROW0(mismatch_error,0,"No SG transport on live platforms");
 206 }
 207