3 /* rl_transport - RL specific functions for transport */
5 /* Copyright (c) 2004 Martin Quinson. All rights reserved. */
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. */
12 #include "gras/Transport/transport_private.h"
13 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(gras_trp);
15 /* check transport_private.h for an explanation of this variable */
16 gras_socket_t _gras_lastly_selected_socket = NULL;
21 * Returns the next socket to service because it receives a message.
23 * if timeout<0, we ought to implement the adaptative timeout (FIXME)
26 * if timeout>0 and no message there, wait at most that amount of time before giving up.
28 gras_socket_t gras_trp_select(double timeout) {
29 xbt_dynar_t sockets= ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->sockets;
31 double wakeup = gras_os_time() + timeout;
33 /* nextToService used to make sure socket with high number do not starve */
34 /* static int nextToService = 0; */
35 struct timeval tout, *p_tout;
37 int max_fds=0; /* first arg of select: number of existing sockets */
38 /* but accept() of winsock returns sockets bigger than the limit, so don't bother
39 with this tiny optimisation on BillWare */
41 int ready; /* return of select: number of socket ready to be serviced */
42 static int fd_setsize=-1; /* FD_SETSIZE not always defined. Get this portably */
44 gras_socket_t sock_iter; /* iterating over all sockets */
45 unsigned int cursor; /* iterating over all sockets */
47 /* Check whether there is more data to read from the socket we selected last time.
48 This can happen with tcp buffered sockets since we try to get as much data as we can for them */
49 if (_gras_lastly_selected_socket && _gras_lastly_selected_socket->moredata) {
50 VERB0("Returning _gras_lastly_selected_socket since there is more data on it");
51 return _gras_lastly_selected_socket;
54 /* Compute FD_SETSIZE on need */
57 fd_setsize = sysconf( _SC_OPEN_MAX );
59 # ifdef HAVE_GETDTABLESIZE
60 fd_setsize = getdtablesize();
62 fd_setsize = FD_SETSIZE;
63 # endif /* !USE_SYSCONF */
68 if (timeout > 0) { /* did we timeout already? */
70 DEBUG2("wakeup=%f now=%f",wakeup, now);
71 if (now == -1 || now >= wakeup) {
72 /* didn't find anything; no need to update _gras_lastly_selected_socket since its moredata is 0 (or we would have returned it directly) */
73 THROW1(timeout_error,0,
74 "Timeout (%f) elapsed with selecting for incomming connexions",
79 /* construct the set of socket to ear from */
82 xbt_dynar_foreach(sockets,cursor,sock_iter) {
83 if (!sock_iter->valid)
86 if (sock_iter->incoming) {
87 DEBUG1("Considering socket %d for select",sock_iter->sd);
88 #ifndef HAVE_WINSOCK_H
89 if (max_fds < sock_iter->sd)
90 max_fds = sock_iter->sd;
95 FD_SET(sock_iter->sd, &FDS);
97 DEBUG1("Not considering socket %d for select",sock_iter->sd);
104 DEBUG1("No socket to select onto. Sleep %f sec instead.",timeout);
105 gras_os_sleep(timeout);
106 THROW1(timeout_error,0,"No socket to select onto. Sleep %f sec instead",timeout);
108 DEBUG0("No socket to select onto. Return directly.");
109 THROW0(timeout_error,0, "No socket to select onto. Return directly.");
113 #ifndef HAVE_WINSOCK_H
114 /* we cannot have more than FD_SETSIZE sockets
115 ... but with WINSOCK which returns sockets higher than the limit (killing this optim) */
116 if (++max_fds > fd_setsize && fd_setsize > 0) {
117 WARN1("too many open sockets (%d).",max_fds);
122 max_fds = fd_setsize;
126 /* set the timeout */
127 tout.tv_sec = (unsigned long)(wakeup - now);
128 tout.tv_usec = ((wakeup -now) - ((unsigned long)(wakeup - now))) * 1000000;
130 } else if (timeout == 0) {
135 /* we just do one loop around */
138 /* no timeout: good luck! */
142 DEBUG2("Selecting over %d socket(s); timeout=%f", max_fds-1,timeout);
143 ready = select(max_fds, &FDS, NULL, NULL, p_tout);
144 DEBUG1("select returned %d", ready);
147 case EINTR: /* a signal we don't care about occured. we don't care */
148 /* if we cared, we would have set an handler */
150 case EINVAL: /* invalid value */
151 THROW3(system_error,EINVAL,"invalid select: nb fds: %d, timeout: %d.%d",
152 max_fds, (int)tout.tv_sec,(int) tout.tv_usec);
154 xbt_die("Malloc error during the select");
156 THROW2(system_error,errno,"Error during select: %s (%d)",
157 strerror(errno),errno);
160 } else if (ready == 0) {
161 continue; /* this was a timeout */
164 xbt_dynar_foreach(sockets,cursor,sock_iter) {
165 if(!FD_ISSET(sock_iter->sd, &FDS)) { /* this socket is not ready */
169 /* Got a socket to serve */
172 if ( sock_iter->accepting
173 && sock_iter->plugin->socket_accept) {
174 /* not a socket but an ear. accept on it and serve next socket */
175 gras_socket_t accepted=NULL;
177 /* release mutex before accept; it will change the sockets dynar, so we have to break the foreach asap */
178 xbt_dynar_cursor_unlock(sockets);
179 accepted = (sock_iter->plugin->socket_accept)(sock_iter);
181 DEBUG2("accepted=%p,&accepted=%p",accepted,&accepted);
182 accepted->meas = sock_iter->meas;
186 /* Make sure the socket is still alive by reading the first byte */
189 if (sock_iter->recvd) {
190 /* Socket wasn't used since last time! Don't bother checking whether it's still alive */
193 recvd = read(sock_iter->sd, &sock_iter->recvd_val, 1);
197 WARN2("socket %d failed: %s", sock_iter->sd, strerror(errno));
198 /* done with this socket; remove it and break the foreach since it will change the dynar */
199 xbt_dynar_cursor_unlock(sockets);
200 gras_socket_close(sock_iter);
202 } else if (recvd == 0) {
203 /* Connection reset (=closed) by peer. */
204 DEBUG1("Connection %d reset by peer", sock_iter->sd);
205 sock_iter->valid=0; /* don't close it. User may keep references to it */
207 /* Got a suited socket ! */
209 sock_iter->recvd = 1;
210 DEBUG3("Filled little buffer (%c %x %d)", sock_iter->recvd_val, sock_iter->recvd_val, recvd);
211 _gras_lastly_selected_socket = sock_iter;
212 /* break sync dynar iteration */
213 xbt_dynar_cursor_unlock(sockets);
218 /* if we're here, the socket we found wasn't really ready to be served */
219 if (ready == 0) { /* exausted all sockets given by select. Request new ones */
221 xbt_dynar_cursor_unlock(sockets);
228 /* No socket found. Maybe we had timeout=0 and nothing to do */
230 THROW0(timeout_error,0,"Timeout");
233 void gras_trp_sg_setup(gras_trp_plugin_t plug) {
234 THROW0(mismatch_error,0,"No SG transport on live platforms");