1 /* rl_transport - RL specific functions for transport */
3 /* Copyright (c) 2004, 2005, 2006, 2007, 2009, 2010. The SimGrid Team.
4 * All rights reserved. */
6 /* This program is free software; you can redistribute it and/or modify it
7 * under the terms of the license (GNU LGPL) which comes with this package. */
10 #include "xbt/xbt_socket_private.h" /* FIXME */
12 #include "gras/Transport/transport_private.h"
14 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(gras_trp);
16 /* check transport_private.h for an explanation of this variable */
17 xbt_socket_t _gras_lastly_selected_socket = NULL;
22 * Returns the next socket to service because it receives a message.
24 * if timeout<0, we ought to implement the adaptative timeout (FIXME)
27 * if timeout>0 and no message there, wait at most that amount of time before giving up.
29 xbt_socket_t gras_trp_select(double timeout)
32 ((gras_trp_procdata_t)
33 gras_libdata_by_id(gras_trp_libdata_id))->sockets;
35 double wakeup = gras_os_time() + timeout;
37 /* nextToService used to make sure socket with high number do not starve */
38 /* static int nextToService = 0; */
39 struct timeval tout, *p_tout;
41 int max_fds = 0; /* first arg of select: number of existing sockets */
42 /* but accept() of winsock returns sockets bigger than the limit, so don't bother
43 with this tiny optimisation on BillWare */
45 int ready; /* return of select: number of socket ready to be serviced */
46 static int fd_setsize = -1; /* FD_SETSIZE not always defined. Get this portably */
48 xbt_socket_t sock_iter; /* iterating over all sockets */
49 unsigned int cursor; /* iterating over all sockets */
51 /* Check whether there is more data to read from the socket we selected last time.
52 This can happen with tcp buffered sockets since we try to get as much data as we can for them */
53 if (_gras_lastly_selected_socket
54 && _gras_lastly_selected_socket->moredata) {
56 ("Returning _gras_lastly_selected_socket since there is more data on it");
57 return _gras_lastly_selected_socket;
60 /* Compute FD_SETSIZE on need */
63 fd_setsize = sysconf(_SC_OPEN_MAX);
65 # ifdef HAVE_GETDTABLESIZE
66 fd_setsize = getdtablesize();
68 fd_setsize = FD_SETSIZE;
69 # endif /* !USE_SYSCONF */
74 if (timeout > 0) { /* did we timeout already? */
76 XBT_DEBUG("wakeup=%f now=%f", wakeup, now);
77 if (now == -1 || now >= wakeup) {
78 /* didn't find anything; no need to update _gras_lastly_selected_socket since its moredata is 0 (or we would have returned it directly) */
79 THROWF(timeout_error, 0,
80 "Timeout (%f) elapsed with selecting for incoming connections",
85 /* construct the set of socket to ear from */
88 xbt_dynar_foreach(sockets, cursor, sock_iter) {
89 if (!sock_iter->valid)
92 if (sock_iter->incoming) {
93 //XBT_DEBUG("Considering socket %d for select", sock_iter->sd);
94 #ifndef HAVE_WINSOCK_H
95 if (max_fds < sock_iter->sd)
96 max_fds = sock_iter->sd;
101 FD_SET(sock_iter->sd, &FDS);
103 XBT_DEBUG("Not considering socket %d for select", sock_iter->sd);
110 XBT_DEBUG("No socket to select onto. Sleep %f sec instead.", timeout);
111 gras_os_sleep(timeout);
112 THROWF(timeout_error, 0,
113 "No socket to select onto. Sleep %f sec instead", timeout);
115 XBT_DEBUG("No socket to select onto. Return directly.");
116 THROWF(timeout_error, 0,
117 "No socket to select onto. Return directly.");
120 #ifndef HAVE_WINSOCK_H
121 /* we cannot have more than FD_SETSIZE sockets
122 ... but with WINSOCK which returns sockets higher than the limit (killing this optim) */
123 if (++max_fds > fd_setsize && fd_setsize > 0) {
124 XBT_WARN("too many open sockets (%d).", max_fds);
129 max_fds = fd_setsize;
132 tout.tv_sec = tout.tv_usec = 0;
134 /* set the timeout */
135 tout.tv_sec = (unsigned long) (wakeup - now);
137 ((wakeup - now) - ((unsigned long) (wakeup - now))) * 1000000;
139 } else if (timeout == 0) {
144 /* we just do one loop around */
147 /* no timeout: good luck! */
151 XBT_DEBUG("Selecting over %d socket(s); timeout=%f", max_fds - 1,
153 ready = select(max_fds, &FDS, NULL, NULL, p_tout);
154 XBT_DEBUG("select returned %d", ready);
157 case EINTR: /* a signal we don't care about occured. we don't care */
158 /* if we cared, we would have set an handler */
160 case EINVAL: /* invalid value */
161 THROWF(system_error, EINVAL,
162 "invalid select: nb fds: %d, timeout: %d.%d", max_fds,
163 (int) tout.tv_sec, (int) tout.tv_usec);
165 xbt_die("Malloc error during the select");
167 THROWF(system_error, errno, "Error during select: %s (%d)",
168 strerror(errno), errno);
171 } else if (ready == 0) {
172 continue; /* this was a timeout */
175 xbt_dynar_foreach(sockets, cursor, sock_iter) {
176 if (!FD_ISSET(sock_iter->sd, &FDS)) { /* this socket is not ready */
180 /* Got a socket to serve */
183 if (sock_iter->accepting && sock_iter->plugin->socket_accept) {
184 /* not a socket but an ear. accept on it and serve next socket */
185 xbt_socket_t accepted = NULL;
187 /* release mutex before accept; it will change the sockets dynar, so we have to break the foreach asap */
188 xbt_dynar_cursor_unlock(sockets);
189 accepted = (sock_iter->plugin->socket_accept) (sock_iter);
191 XBT_DEBUG("accepted=%p,&accepted=%p", accepted, &accepted);
192 accepted->meas = sock_iter->meas;
196 /* Make sure the socket is still alive by reading the first byte */
199 if (sock_iter->recvd) {
200 /* Socket wasn't used since last time! Don't bother checking whether it's still alive */
203 recvd = read(sock_iter->sd, &sock_iter->recvd_val, 1);
207 XBT_WARN("socket %d failed: %s", sock_iter->sd, strerror(errno));
208 /* done with this socket; remove it and break the foreach since it will change the dynar */
209 xbt_dynar_cursor_unlock(sockets);
210 gras_socket_close(sock_iter);
212 } else if (recvd == 0) {
213 /* Connection reset (=closed) by peer. */
214 XBT_DEBUG("Connection %d reset by peer", sock_iter->sd);
215 sock_iter->valid = 0; /* don't close it. User may keep references to it */
217 /* Got a suited socket ! */
219 sock_iter->recvd = 1;
220 XBT_DEBUG("Filled little buffer (%c %x %d)", sock_iter->recvd_val,
221 sock_iter->recvd_val, recvd);
222 _gras_lastly_selected_socket = sock_iter;
223 /* break sync dynar iteration */
224 xbt_dynar_cursor_unlock(sockets);
229 /* if we're here, the socket we found wasn't really ready to be served */
230 if (ready == 0) { /* exausted all sockets given by select. Request new ones */
232 xbt_dynar_cursor_unlock(sockets);
239 /* No socket found. Maybe we had timeout=0 and nothing to do */
240 XBT_DEBUG("TIMEOUT");
241 THROWF(timeout_error, 0, "Timeout");
244 /* to make the linker happy in SG mode */
245 void gras_trp_sg_setup(xbt_trp_plugin_t plug)
247 THROWF(mismatch_error, 0, "No SG transport on live platforms");