ghc/rts/Select.c

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