ghc/rts/Select.c

   1 /* -----------------------------------------------------------------------------
   2  * $Id: Select.c,v 1.19 2001/11/13 13:38:02 simonmar 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     struct timeval tv;
  90     lnat min, ticks;
  91
  92     tv.tv_sec  = 0;
  93     tv.tv_usec = 0;
  94
  95     IF_DEBUG(scheduler,
  96              belch("scheduler: checking for threads blocked on I/O");
  97              if (wait) {
  98                  belch(" (waiting)");
  99              }
 100              belch("\n");
 101              );
 102
 103     /* loop until we've woken up some threads.  This loop is needed
 104      * because the select timing isn't accurate, we sometimes sleep
 105      * for a while but not long enough to wake up a thread in
 106      * a threadDelay.
 107      */
 108     do {
 109
 110       ticks = timestamp = getourtimeofday();
 111       if (wakeUpSleepingThreads(ticks)) {
 112           return;
 113       }
 114
 115       if (!wait) {
 116           min = 0;
 117       } else if (sleeping_queue != END_TSO_QUEUE) {
 118           min = (sleeping_queue->block_info.target - ticks)
 119               * TICK_MILLISECS * 1000;
 120       } else {
 121           min = 0x7ffffff;
 122       }
 123
 124 #ifndef mingw32_TARGET_OS
 125       /*
 126        * Collect all of the fd's that we're interested in
 127        */
 128       FD_ZERO(&rfd);
 129       FD_ZERO(&wfd);
 130
 131       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 132         next = tso->link;
 133
 134         switch (tso->why_blocked) {
 135         case BlockedOnRead:
 136           {
 137             int fd = tso->block_info.fd;
 138             maxfd = (fd > maxfd) ? fd : maxfd;
 139             FD_SET(fd, &rfd);
 140             continue;
 141           }
 142
 143         case BlockedOnWrite:
 144           {
 145             int fd = tso->block_info.fd;
 146             maxfd = (fd > maxfd) ? fd : maxfd;
 147             FD_SET(fd, &wfd);
 148             continue;
 149           }
 150
 151         default:
 152           barf("AwaitEvent");
 153         }
 154       }
 155
 156       /* Release the scheduler lock while we do the poll.
 157        * this means that someone might muck with the blocked_queue
 158        * while we do this, but it shouldn't matter:
 159        *
 160        *   - another task might poll for I/O and remove one
 161        *     or more threads from the blocked_queue.
 162        *   - more I/O threads may be added to blocked_queue.
 163        *   - more delayed threads may be added to blocked_queue. We'll
 164        *     just subtract delta from their delays after the poll.
 165        *
 166        * I believe none of these cases lead to trouble --SDM.
 167        */
 168       RELEASE_LOCK(&sched_mutex);
 169
 170       /* Check for any interesting events */
 171
 172       tv.tv_sec  = min / 1000000;
 173       tv.tv_usec = min % 1000000;
 174
 175       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 176           if (errno != EINTR) {
 177
 178               printf("%d\n", errno);
 179               fflush(stdout);
 180               perror("select");
 181               barf("select failed");
 182           }
 183 #else /* on mingwin */
 184       while (1) {
 185           Sleep(0); /* don't busy wait */
 186 #endif /* mingw32_TARGET_OS */
 187           ACQUIRE_LOCK(&sched_mutex);
 188
 189 #ifndef mingw32_TARGET_OS
 190           /* We got a signal; could be one of ours.  If so, we need
 191            * to start up the signal handler straight away, otherwise
 192            * we could block for a long time before the signal is
 193            * serviced.
 194            */
 195           if (signals_pending()) {
 196               RELEASE_LOCK(&sched_mutex); /* ToDo: kill */
 197               startSignalHandlers();
 198               ACQUIRE_LOCK(&sched_mutex);
 199               return; /* still hold the lock */
 200           }
 201 #endif
 202
 203           /* we were interrupted, return to the scheduler immediately.
 204            */
 205           if (interrupted) {
 206               return; /* still hold the lock */
 207           }
 208
 209           /* check for threads that need waking up
 210            */
 211           wakeUpSleepingThreads(getourtimeofday());
 212
 213           /* If new runnable threads have arrived, stop waiting for
 214            * I/O and run them.
 215            */
 216           if (run_queue_hd != END_TSO_QUEUE) {
 217               return; /* still hold the lock */
 218           }
 219
 220           RELEASE_LOCK(&sched_mutex);
 221       }
 222
 223       ACQUIRE_LOCK(&sched_mutex);
 224
 225       /* Step through the waiting queue, unblocking every thread that now has
 226        * a file descriptor in a ready state.
 227        */
 228
 229       prev = NULL;
 230       if (select_succeeded) {
 231           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 232               next = tso->link;
 233               switch (tso->why_blocked) {
 234               case BlockedOnRead:
 235                   ready = FD_ISSET(tso->block_info.fd, &rfd);
 236                   break;
 237               case BlockedOnWrite:
 238                   ready = FD_ISSET(tso->block_info.fd, &wfd);
 239                   break;
 240               default:
 241                   barf("awaitEvent");
 242               }
 243
 244               if (ready) {
 245                   IF_DEBUG(scheduler,belch("Waking up blocked thread %d\n", tso->id));
 246                   tso->why_blocked = NotBlocked;
 247                   tso->link = END_TSO_QUEUE;
 248                   PUSH_ON_RUN_QUEUE(tso);
 249               } else {
 250                   if (prev == NULL)
 251                       blocked_queue_hd = tso;
 252                   else
 253                       prev->link = tso;
 254                   prev = tso;
 255               }
 256           }
 257
 258           if (prev == NULL)
 259               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 260           else {
 261               prev->link = END_TSO_QUEUE;
 262               blocked_queue_tl = prev;
 263           }
 264       }
 265
 266     } while (wait && !interrupted && run_queue_hd == END_TSO_QUEUE);
 267 }