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