ghc/rts/Select.c

   1 /* -----------------------------------------------------------------------------
   2  * $Id: Select.c,v 1.15 2001/02/27 12:43:45 rrt 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 #define NON_POSIX_SOURCE
  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 <w32api/windows.h>
  30 # endif
  31
  32 /* last timestamp */
  33 nat timestamp = 0;
  34
  35 /* keep track of the number of ticks since we last called
  36  * gettimeofday(), to avoid having to call it every time we need
  37  * a timestamp.
  38  */
  39 nat ticks_since_timestamp = 0;
  40
  41 /* There's a clever trick here to avoid problems when the time wraps
  42  * around.  Since our maximum delay is smaller than 31 bits of ticks
  43  * (it's actually 31 bits of microseconds), we can safely check
  44  * whether a timer has expired even if our timer will wrap around
  45  * before the target is reached, using the following formula:
  46  *
  47  *        (int)((uint)current_time - (uint)target_time) < 0
  48  *
  49  * if this is true, then our time has expired.
  50  * (idea due to Andy Gill).
  51  */
  52 rtsBool
  53 wakeUpSleepingThreads(nat ticks)
  54 {
  55     StgTSO *tso;
  56     rtsBool flag = rtsFalse;
  57
  58     while (sleeping_queue != END_TSO_QUEUE &&
  59            (int)(ticks - sleeping_queue->block_info.target) > 0) {
  60         tso = sleeping_queue;
  61         sleeping_queue = tso->link;
  62         tso->why_blocked = NotBlocked;
  63         tso->link = END_TSO_QUEUE;
  64         IF_DEBUG(scheduler,belch("Waking up sleeping thread %d\n", tso->id));
  65         PUSH_ON_RUN_QUEUE(tso);
  66         flag = rtsTrue;
  67     }
  68     return flag;
  69 }
  70
  71 /* Argument 'wait' says whether to wait for I/O to become available,
  72  * or whether to just check and return immediately.  If there are
  73  * other threads ready to run, we normally do the non-waiting variety,
  74  * otherwise we wait (see Schedule.c).
  75  *
  76  * SMP note: must be called with sched_mutex locked.
  77  *
  78  * Windows: select only works on sockets, so this doesn't really work,
  79  * though it makes things better than before. MsgWaitForMultipleObjects
  80  * should really be used, though it only seems to work for read handles,
  81  * not write handles.
  82  *
  83  */
  84 void
  85 awaitEvent(rtsBool wait)
  86 {
  87     StgTSO *tso, *prev, *next;
  88     rtsBool ready;
  89     fd_set rfd,wfd;
  90 #ifndef mingw32_TARGET_OS
  91     int numFound;
  92     int maxfd = -1;
  93 #endif
  94     rtsBool select_succeeded = rtsTrue;
  95     struct timeval tv;
  96     lnat min, ticks;
  97
  98     tv.tv_sec  = 0;
  99     tv.tv_usec = 0;
 100
 101     IF_DEBUG(scheduler,
 102              belch("scheduler: checking for threads blocked on I/O");
 103              if (wait) {
 104                  belch(" (waiting)");
 105              }
 106              belch("\n");
 107              );
 108
 109     /* loop until we've woken up some threads.  This loop is needed
 110      * because the select timing isn't accurate, we sometimes sleep
 111      * for a while but not long enough to wake up a thread in
 112      * a threadDelay.
 113      */
 114     do {
 115
 116       ticks = timestamp = getourtimeofday();
 117       ticks_since_timestamp = 0;
 118       if (wakeUpSleepingThreads(ticks)) {
 119           return;
 120       }
 121
 122       if (!wait) {
 123           min = 0;
 124       } else if (sleeping_queue != END_TSO_QUEUE) {
 125           min = (sleeping_queue->block_info.target - ticks)
 126               * TICK_MILLISECS * 1000;
 127       } else {
 128           min = 0x7ffffff;
 129       }
 130
 131 #ifndef mingw32_TARGET_OS
 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       /* Release the scheduler lock while we do the poll.
 164        * this means that someone might muck with the blocked_queue
 165        * while we do this, but it shouldn't matter:
 166        *
 167        *   - another task might poll for I/O and remove one
 168        *     or more threads from the blocked_queue.
 169        *   - more I/O threads may be added to blocked_queue.
 170        *   - more delayed threads may be added to blocked_queue. We'll
 171        *     just subtract delta from their delays after the poll.
 172        *
 173        * I believe none of these cases lead to trouble --SDM.
 174        */
 175       RELEASE_LOCK(&sched_mutex);
 176
 177       /* Check for any interesting events */
 178
 179       tv.tv_sec  = min / 1000000;
 180       tv.tv_usec = min % 1000000;
 181
 182       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 183           if (errno != EINTR) {
 184
 185               printf("%d\n", errno);
 186               fflush(stdout);
 187               perror("select");
 188               barf("select failed");
 189           }
 190 #else /* on mingwin */
 191       while (1) {
 192           Sleep(0); /* don't busy wait */
 193 #endif /* mingw32_TARGET_OS */
 194           ACQUIRE_LOCK(&sched_mutex);
 195
 196 #ifndef mingw32_TARGET_OS
 197           /* We got a signal; could be one of ours.  If so, we need
 198            * to start up the signal handler straight away, otherwise
 199            * we could block for a long time before the signal is
 200            * serviced.
 201            */
 202           if (signals_pending()) {
 203               RELEASE_LOCK(&sched_mutex); /* ToDo: kill */
 204               start_signal_handlers();
 205               ACQUIRE_LOCK(&sched_mutex);
 206               return; /* still hold the lock */
 207           }
 208 #endif
 209
 210           /* we were interrupted, return to the scheduler immediately.
 211            */
 212           if (interrupted) {
 213               return; /* still hold the lock */
 214           }
 215
 216           /* check for threads that need waking up
 217            */
 218           wakeUpSleepingThreads(getourtimeofday());
 219
 220           /* If new runnable threads have arrived, stop waiting for
 221            * I/O and run them.
 222            */
 223           if (run_queue_hd != END_TSO_QUEUE) {
 224               return; /* still hold the lock */
 225           }
 226
 227           RELEASE_LOCK(&sched_mutex);
 228       }
 229
 230       ACQUIRE_LOCK(&sched_mutex);
 231
 232       /* Step through the waiting queue, unblocking every thread that now has
 233        * a file descriptor in a ready state.
 234        */
 235
 236       prev = NULL;
 237       if (select_succeeded) {
 238           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 239               next = tso->link;
 240               switch (tso->why_blocked) {
 241               case BlockedOnRead:
 242                   ready = FD_ISSET(tso->block_info.fd, &rfd);
 243                   break;
 244               case BlockedOnWrite:
 245                   ready = FD_ISSET(tso->block_info.fd, &wfd);
 246                   break;
 247               default:
 248                   barf("awaitEvent");
 249               }
 250
 251               if (ready) {
 252                   IF_DEBUG(scheduler,belch("Waking up blocked thread %d\n", tso->id));
 253                   tso->why_blocked = NotBlocked;
 254                   tso->link = END_TSO_QUEUE;
 255                   PUSH_ON_RUN_QUEUE(tso);
 256               } else {
 257                   if (prev == NULL)
 258                       blocked_queue_hd = tso;
 259                   else
 260                       prev->link = tso;
 261                   prev = tso;
 262               }
 263           }
 264
 265           if (prev == NULL)
 266               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 267           else {
 268               prev->link = END_TSO_QUEUE;
 269               blocked_queue_tl = prev;
 270           }
 271       }
 272
 273     } while (wait && !interrupted && run_queue_hd == END_TSO_QUEUE);
 274 }