Comments on data type families
[ghc-hetmet.git] / rts / posix / Select.c
1 /* -----------------------------------------------------------------------------
2  *
3  * (c) The GHC Team 1995-2002
4  *
5  * Support for concurrent non-blocking I/O and thread waiting.
6  *
7  * ---------------------------------------------------------------------------*/
8
9 #include "PosixSource.h"
10 #include "Rts.h"
11
12 #include "Signals.h"
13 #include "Schedule.h"
14 #include "RtsUtils.h"
15 #include "Itimer.h"
16 #include "Capability.h"
17 #include "Select.h"
18 #include "AwaitEvent.h"
19
20 # ifdef HAVE_SYS_SELECT_H
21 #  include <sys/select.h>
22 # endif
23
24 # ifdef HAVE_SYS_TYPES_H
25 #  include <sys/types.h>
26 # endif
27
28 # ifdef HAVE_SYS_TIME_H
29 #  include <sys/time.h>
30 # endif
31
32 #include <errno.h>
33 #include <string.h>
34
35 #ifdef HAVE_UNISTD_H
36 #include <unistd.h>
37 #endif
38
39 #if !defined(THREADED_RTS)
40 /* last timestamp */
41 lnat timestamp = 0;
42
43 /* 
44  * The threaded RTS uses an IO-manager thread in Haskell instead (see GHC.Conc) 
45  */
46
47 /* There's a clever trick here to avoid problems when the time wraps
48  * around.  Since our maximum delay is smaller than 31 bits of ticks
49  * (it's actually 31 bits of microseconds), we can safely check
50  * whether a timer has expired even if our timer will wrap around
51  * before the target is reached, using the following formula:
52  *
53  *        (int)((uint)current_time - (uint)target_time) < 0
54  *
55  * if this is true, then our time has expired.
56  * (idea due to Andy Gill).
57  */
58 static rtsBool
59 wakeUpSleepingThreads(lnat ticks)
60 {
61     StgTSO *tso;
62     rtsBool flag = rtsFalse;
63
64     while (sleeping_queue != END_TSO_QUEUE) {
65         tso = sleeping_queue;
66         if (tso->what_next == ThreadRelocated) {
67             sleeping_queue = tso->_link;
68             continue;
69         }
70         if (((long)ticks - (long)tso->block_info.target) < 0) {
71             break;
72         }
73         sleeping_queue = tso->_link;
74         tso->why_blocked = NotBlocked;
75         tso->_link = END_TSO_QUEUE;
76         IF_DEBUG(scheduler,debugBelch("Waking up sleeping thread %lu\n", (unsigned long)tso->id));
77         // MainCapability: this code is !THREADED_RTS
78         pushOnRunQueue(&MainCapability,tso);
79         flag = rtsTrue;
80     }
81     return flag;
82 }
83
84 /* Argument 'wait' says whether to wait for I/O to become available,
85  * or whether to just check and return immediately.  If there are
86  * other threads ready to run, we normally do the non-waiting variety,
87  * otherwise we wait (see Schedule.c).
88  *
89  * SMP note: must be called with sched_mutex locked.
90  *
91  * Windows: select only works on sockets, so this doesn't really work,
92  * though it makes things better than before. MsgWaitForMultipleObjects
93  * should really be used, though it only seems to work for read handles,
94  * not write handles.
95  *
96  */
97 void
98 awaitEvent(rtsBool wait)
99 {
100     StgTSO *tso, *prev, *next;
101     rtsBool ready;
102     fd_set rfd,wfd;
103     int numFound;
104     int maxfd = -1;
105     rtsBool select_succeeded = rtsTrue;
106     rtsBool unblock_all = rtsFalse;
107     struct timeval tv;
108     lnat min, ticks;
109
110     tv.tv_sec  = 0;
111     tv.tv_usec = 0;
112     
113     IF_DEBUG(scheduler,
114              debugBelch("scheduler: checking for threads blocked on I/O");
115              if (wait) {
116                  debugBelch(" (waiting)");
117              }
118              debugBelch("\n");
119              );
120
121     /* loop until we've woken up some threads.  This loop is needed
122      * because the select timing isn't accurate, we sometimes sleep
123      * for a while but not long enough to wake up a thread in
124      * a threadDelay.
125      */
126     do {
127
128       ticks = timestamp = getourtimeofday();
129       if (wakeUpSleepingThreads(ticks)) { 
130           return;
131       }
132
133       if (!wait) {
134           min = 0;
135       } else if (sleeping_queue != END_TSO_QUEUE) {
136           min = (sleeping_queue->block_info.target - ticks) 
137               * RtsFlags.MiscFlags.tickInterval * 1000;
138       } else {
139           min = 0x7ffffff;
140       }
141
142       /* 
143        * Collect all of the fd's that we're interested in
144        */
145       FD_ZERO(&rfd);
146       FD_ZERO(&wfd);
147
148       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
149         next = tso->_link;
150
151       /* On FreeBSD FD_SETSIZE is unsigned. Cast it to signed int
152        * in order to switch off the 'comparison between signed and
153        * unsigned error message
154        */
155         switch (tso->why_blocked) {
156         case BlockedOnRead:
157           { 
158             int fd = tso->block_info.fd;
159             if (fd >= (int)FD_SETSIZE) {
160                 barf("awaitEvent: descriptor out of range");
161             }
162             maxfd = (fd > maxfd) ? fd : maxfd;
163             FD_SET(fd, &rfd);
164             continue;
165           }
166
167         case BlockedOnWrite:
168           { 
169             int fd = tso->block_info.fd;
170             if (fd >= (int)FD_SETSIZE) {
171                 barf("awaitEvent: descriptor out of range");
172             }
173             maxfd = (fd > maxfd) ? fd : maxfd;
174             FD_SET(fd, &wfd);
175             continue;
176           }
177
178         default:
179           barf("AwaitEvent");
180         }
181       }
182
183       /* Check for any interesting events */
184       
185       tv.tv_sec  = min / 1000000;
186       tv.tv_usec = min % 1000000;
187
188       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
189           if (errno != EINTR) {
190             /* Handle bad file descriptors by unblocking all the
191                waiting threads. Why? Because a thread might have been
192                a bit naughty and closed a file descriptor while another
193                was blocked waiting. This is less-than-good programming
194                practice, but having the RTS as a result fall over isn't
195                acceptable, so we simply unblock all the waiting threads
196                should we see a bad file descriptor & give the threads
197                a chance to clean up their act. 
198                
199                Note: assume here that threads becoming unblocked
200                will try to read/write the file descriptor before trying
201                to issue a threadWaitRead/threadWaitWrite again (==> an
202                IOError will result for the thread that's got the bad
203                file descriptor.) Hence, there's no danger of a bad
204                file descriptor being repeatedly select()'ed on, so
205                the RTS won't loop.
206             */
207             if ( errno == EBADF ) {
208               unblock_all = rtsTrue;
209               break;
210             } else {
211               perror("select");
212               barf("select failed");
213             }
214           }
215
216           /* We got a signal; could be one of ours.  If so, we need
217            * to start up the signal handler straight away, otherwise
218            * we could block for a long time before the signal is
219            * serviced.
220            */
221 #if defined(RTS_USER_SIGNALS)
222           if (RtsFlags.MiscFlags.install_signal_handlers && signals_pending()) {
223               startSignalHandlers(&MainCapability);
224               return; /* still hold the lock */
225           }
226 #endif
227
228           /* we were interrupted, return to the scheduler immediately.
229            */
230           if (sched_state >= SCHED_INTERRUPTING) {
231               return; /* still hold the lock */
232           }
233           
234           /* check for threads that need waking up 
235            */
236           wakeUpSleepingThreads(getourtimeofday());
237           
238           /* If new runnable threads have arrived, stop waiting for
239            * I/O and run them.
240            */
241           if (!emptyRunQueue(&MainCapability)) {
242               return; /* still hold the lock */
243           }
244       }
245
246       /* Step through the waiting queue, unblocking every thread that now has
247        * a file descriptor in a ready state.
248        */
249
250       prev = NULL;
251       if (select_succeeded || unblock_all) {
252           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
253               next = tso->_link;
254
255               if (tso->what_next == ThreadRelocated) {
256                   continue;
257               }
258
259               switch (tso->why_blocked) {
260               case BlockedOnRead:
261                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &rfd);
262                   break;
263               case BlockedOnWrite:
264                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &wfd);
265                   break;
266               default:
267                   barf("awaitEvent");
268               }
269       
270               if (ready) {
271                 IF_DEBUG(scheduler,debugBelch("Waking up blocked thread %lu\n", (unsigned long)tso->id));
272                   tso->why_blocked = NotBlocked;
273                   tso->_link = END_TSO_QUEUE;
274                   pushOnRunQueue(&MainCapability,tso);
275               } else {
276                   if (prev == NULL)
277                       blocked_queue_hd = tso;
278                   else
279                       setTSOLink(&MainCapability, prev, tso);
280                   prev = tso;
281               }
282           }
283
284           if (prev == NULL)
285               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
286           else {
287               prev->_link = END_TSO_QUEUE;
288               blocked_queue_tl = prev;
289           }
290       }
291       
292     } while (wait && sched_state == SCHED_RUNNING
293              && emptyRunQueue(&MainCapability));
294 }
295
296 #endif /* THREADED_RTS */