ghc/rts/Select.c

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