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