ghc/rts/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
  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 "Timer.h"
  18 #include "Itimer.h"
  19 #include "Signals.h"
  20 #include "Capability.h"
  21
  22 # ifdef HAVE_SYS_TYPES_H
  23 #  include <sys/types.h>
  24 # endif
  25
  26 # ifdef HAVE_SYS_TIME_H
  27 #  include <sys/time.h>
  28 # endif
  29
  30 #include <errno.h>
  31 #include <string.h>
  32
  33 #ifdef HAVE_UNISTD_H
  34 #include <unistd.h>
  35 #endif
  36
  37 /* last timestamp */
  38 nat timestamp = 0;
  39
  40 /* There's a clever trick here to avoid problems when the time wraps
  41  * around.  Since our maximum delay is smaller than 31 bits of ticks
  42  * (it's actually 31 bits of microseconds), we can safely check
  43  * whether a timer has expired even if our timer will wrap around
  44  * before the target is reached, using the following formula:
  45  *
  46  *        (int)((uint)current_time - (uint)target_time) < 0
  47  *
  48  * if this is true, then our time has expired.
  49  * (idea due to Andy Gill).
  50  */
  51 rtsBool
  52 wakeUpSleepingThreads(nat ticks)
  53 {
  54     StgTSO *tso;
  55     rtsBool flag = rtsFalse;
  56
  57     while (sleeping_queue != END_TSO_QUEUE &&
  58            (int)(ticks - sleeping_queue->block_info.target) > 0) {
  59         tso = sleeping_queue;
  60         sleeping_queue = tso->link;
  61         tso->why_blocked = NotBlocked;
  62         tso->link = END_TSO_QUEUE;
  63         IF_DEBUG(scheduler,debugBelch("Waking up sleeping thread %d\n", tso->id));
  64         PUSH_ON_RUN_QUEUE(tso);
  65         flag = rtsTrue;
  66     }
  67     return flag;
  68 }
  69
  70 /* Argument 'wait' says whether to wait for I/O to become available,
  71  * or whether to just check and return immediately.  If there are
  72  * other threads ready to run, we normally do the non-waiting variety,
  73  * otherwise we wait (see Schedule.c).
  74  *
  75  * SMP note: must be called with sched_mutex locked.
  76  *
  77  * Windows: select only works on sockets, so this doesn't really work,
  78  * though it makes things better than before. MsgWaitForMultipleObjects
  79  * should really be used, though it only seems to work for read handles,
  80  * not write handles.
  81  *
  82  */
  83 void
  84 awaitEvent(rtsBool wait)
  85 {
  86     StgTSO *tso, *prev, *next;
  87     rtsBool ready;
  88     fd_set rfd,wfd;
  89     int numFound;
  90     int maxfd = -1;
  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              debugBelch("scheduler: checking for threads blocked on I/O");
 101              if (wait) {
 102                  debugBelch(" (waiting)");
 103              }
 104              debugBelch("\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       /*
 129        * Collect all of the fd's that we're interested in
 130        */
 131       FD_ZERO(&rfd);
 132       FD_ZERO(&wfd);
 133
 134       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 135         next = tso->link;
 136
 137         switch (tso->why_blocked) {
 138         case BlockedOnRead:
 139           {
 140             int fd = tso->block_info.fd;
 141             maxfd = (fd > maxfd) ? fd : maxfd;
 142             FD_SET(fd, &rfd);
 143             continue;
 144           }
 145
 146         case BlockedOnWrite:
 147           {
 148             int fd = tso->block_info.fd;
 149             maxfd = (fd > maxfd) ? fd : maxfd;
 150             FD_SET(fd, &wfd);
 151             continue;
 152           }
 153
 154         default:
 155           barf("AwaitEvent");
 156         }
 157       }
 158
 159       /* Check for any interesting events */
 160
 161       tv.tv_sec  = min / 1000000;
 162       tv.tv_usec = min % 1000000;
 163
 164       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 165           if (errno != EINTR) {
 166             /* Handle bad file descriptors by unblocking all the
 167                waiting threads. Why? Because a thread might have been
 168                a bit naughty and closed a file descriptor while another
 169                was blocked waiting. This is less-than-good programming
 170                practice, but having the RTS as a result fall over isn't
 171                acceptable, so we simply unblock all the waiting threads
 172                should we see a bad file descriptor & give the threads
 173                a chance to clean up their act.
 174
 175                Note: assume here that threads becoming unblocked
 176                will try to read/write the file descriptor before trying
 177                to issue a threadWaitRead/threadWaitWrite again (==> an
 178                IOError will result for the thread that's got the bad
 179                file descriptor.) Hence, there's no danger of a bad
 180                file descriptor being repeatedly select()'ed on, so
 181                the RTS won't loop.
 182             */
 183             if ( errno == EBADF ) {
 184               unblock_all = rtsTrue;
 185               break;
 186             } else {
 187               perror("select");
 188               barf("select failed");
 189             }
 190           }
 191
 192           /* We got a signal; could be one of ours.  If so, we need
 193            * to start up the signal handler straight away, otherwise
 194            * we could block for a long time before the signal is
 195            * serviced.
 196            */
 197 #if defined(RTS_USER_SIGNALS)
 198           if (signals_pending()) {
 199               startSignalHandlers();
 200               return; /* still hold the lock */
 201           }
 202 #endif
 203
 204           /* we were interrupted, return to the scheduler immediately.
 205            */
 206           if (interrupted) {
 207               return; /* still hold the lock */
 208           }
 209
 210           /* check for threads that need waking up
 211            */
 212           wakeUpSleepingThreads(getourtimeofday());
 213
 214           /* If new runnable threads have arrived, stop waiting for
 215            * I/O and run them.
 216            */
 217           if (run_queue_hd != END_TSO_QUEUE) {
 218               return; /* still hold the lock */
 219           }
 220       }
 221
 222       /* Step through the waiting queue, unblocking every thread that now has
 223        * a file descriptor in a ready state.
 224        */
 225
 226       prev = NULL;
 227       if (select_succeeded || unblock_all) {
 228           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 229               next = tso->link;
 230               switch (tso->why_blocked) {
 231               case BlockedOnRead:
 232                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &rfd);
 233                   break;
 234               case BlockedOnWrite:
 235                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &wfd);
 236                   break;
 237               default:
 238                   barf("awaitEvent");
 239               }
 240
 241               if (ready) {
 242                   IF_DEBUG(scheduler,debugBelch("Waking up blocked thread %d\n", tso->id));
 243                   tso->why_blocked = NotBlocked;
 244                   tso->link = END_TSO_QUEUE;
 245                   PUSH_ON_RUN_QUEUE(tso);
 246               } else {
 247                   if (prev == NULL)
 248                       blocked_queue_hd = tso;
 249                   else
 250                       prev->link = tso;
 251                   prev = tso;
 252               }
 253           }
 254
 255           if (prev == NULL)
 256               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 257           else {
 258               prev->link = END_TSO_QUEUE;
 259               blocked_queue_tl = prev;
 260           }
 261       }
 262
 263     } while (wait && !interrupted && run_queue_hd == END_TSO_QUEUE);
 264 }