ghc/rts/Select.c

   1 /* -----------------------------------------------------------------------------
   2  * $Id: Select.c,v 1.33 2004/09/03 15:28:53 simonmar Exp $
   3  *
   4  * (c) The GHC Team 1995-2002
   5  *
   6  * Support for concurrent non-blocking I/O and thread waiting.
   7  *
   8  * ---------------------------------------------------------------------------*/
   9
  10
  11 /* we're outside the realms of POSIX here... */
  12 /* #include "PosixSource.h" */
  13
  14 #include "Rts.h"
  15 #include "Schedule.h"
  16 #include "RtsUtils.h"
  17 #include "RtsFlags.h"
  18 #include "Timer.h"
  19 #include "Itimer.h"
  20 #include "Signals.h"
  21 #include "Capability.h"
  22
  23 # ifdef HAVE_SYS_TYPES_H
  24 #  include <sys/types.h>
  25 # endif
  26
  27 # ifdef HAVE_SYS_TIME_H
  28 #  include <sys/time.h>
  29 # endif
  30
  31 #include <errno.h>
  32 #include <string.h>
  33
  34 #ifdef HAVE_UNISTD_H
  35 #include <unistd.h>
  36 #endif
  37
  38 /* last timestamp */
  39 nat timestamp = 0;
  40
  41 #ifdef RTS_SUPPORTS_THREADS
  42 static rtsBool isWorkerBlockedInAwaitEvent = rtsFalse;
  43 static rtsBool workerWakeupPending = rtsFalse;
  44 static int workerWakeupPipe[2];
  45 static rtsBool workerWakeupInited = rtsFalse;
  46 #endif
  47
  48 /* There's a clever trick here to avoid problems when the time wraps
  49  * around.  Since our maximum delay is smaller than 31 bits of ticks
  50  * (it's actually 31 bits of microseconds), we can safely check
  51  * whether a timer has expired even if our timer will wrap around
  52  * before the target is reached, using the following formula:
  53  *
  54  *        (int)((uint)current_time - (uint)target_time) < 0
  55  *
  56  * if this is true, then our time has expired.
  57  * (idea due to Andy Gill).
  58  */
  59 rtsBool
  60 wakeUpSleepingThreads(nat ticks)
  61 {
  62     StgTSO *tso;
  63     rtsBool flag = rtsFalse;
  64
  65     while (sleeping_queue != END_TSO_QUEUE &&
  66            (int)(ticks - sleeping_queue->block_info.target) > 0) {
  67         tso = sleeping_queue;
  68         sleeping_queue = tso->link;
  69         tso->why_blocked = NotBlocked;
  70         tso->link = END_TSO_QUEUE;
  71         IF_DEBUG(scheduler,debugBelch("Waking up sleeping thread %d\n", tso->id));
  72         PUSH_ON_RUN_QUEUE(tso);
  73         flag = rtsTrue;
  74     }
  75     return flag;
  76 }
  77
  78 /* Argument 'wait' says whether to wait for I/O to become available,
  79  * or whether to just check and return immediately.  If there are
  80  * other threads ready to run, we normally do the non-waiting variety,
  81  * otherwise we wait (see Schedule.c).
  82  *
  83  * SMP note: must be called with sched_mutex locked.
  84  *
  85  * Windows: select only works on sockets, so this doesn't really work,
  86  * though it makes things better than before. MsgWaitForMultipleObjects
  87  * should really be used, though it only seems to work for read handles,
  88  * not write handles.
  89  *
  90  */
  91 void
  92 awaitEvent(rtsBool wait)
  93 {
  94     StgTSO *tso, *prev, *next;
  95     rtsBool ready;
  96     fd_set rfd,wfd;
  97     int numFound;
  98     int maxfd = -1;
  99     rtsBool select_succeeded = rtsTrue;
 100     rtsBool unblock_all = rtsFalse;
 101     struct timeval tv;
 102     lnat min, ticks;
 103
 104     tv.tv_sec  = 0;
 105     tv.tv_usec = 0;
 106
 107     IF_DEBUG(scheduler,
 108              debugBelch("scheduler: checking for threads blocked on I/O");
 109              if (wait) {
 110                  debugBelch(" (waiting)");
 111              }
 112              debugBelch("\n");
 113              );
 114
 115     /* loop until we've woken up some threads.  This loop is needed
 116      * because the select timing isn't accurate, we sometimes sleep
 117      * for a while but not long enough to wake up a thread in
 118      * a threadDelay.
 119      */
 120     do {
 121
 122       ticks = timestamp = getourtimeofday();
 123       if (wakeUpSleepingThreads(ticks)) {
 124           return;
 125       }
 126
 127       if (!wait) {
 128           min = 0;
 129       } else if (sleeping_queue != END_TSO_QUEUE) {
 130           min = (sleeping_queue->block_info.target - ticks)
 131               * TICK_MILLISECS * 1000;
 132       } else {
 133           min = 0x7ffffff;
 134       }
 135
 136       /*
 137        * Collect all of the fd's that we're interested in
 138        */
 139       FD_ZERO(&rfd);
 140       FD_ZERO(&wfd);
 141
 142       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 143         next = tso->link;
 144
 145         switch (tso->why_blocked) {
 146         case BlockedOnRead:
 147           {
 148             int fd = tso->block_info.fd;
 149             maxfd = (fd > maxfd) ? fd : maxfd;
 150             FD_SET(fd, &rfd);
 151             continue;
 152           }
 153
 154         case BlockedOnWrite:
 155           {
 156             int fd = tso->block_info.fd;
 157             maxfd = (fd > maxfd) ? fd : maxfd;
 158             FD_SET(fd, &wfd);
 159             continue;
 160           }
 161
 162         default:
 163           barf("AwaitEvent");
 164         }
 165       }
 166
 167 #ifdef RTS_SUPPORTS_THREADS
 168       if(!workerWakeupInited) {
 169           pipe(workerWakeupPipe);
 170           workerWakeupInited = rtsTrue;
 171       }
 172       FD_SET(workerWakeupPipe[0], &rfd);
 173       maxfd = workerWakeupPipe[0] > maxfd ? workerWakeupPipe[0] : maxfd;
 174 #endif
 175
 176       /* Release the scheduler lock while we do the poll.
 177        * this means that someone might muck with the blocked_queue
 178        * while we do this, but it shouldn't matter:
 179        *
 180        *   - another task might poll for I/O and remove one
 181        *     or more threads from the blocked_queue.
 182        *   - more I/O threads may be added to blocked_queue.
 183        *   - more delayed threads may be added to blocked_queue. We'll
 184        *     just subtract delta from their delays after the poll.
 185        *
 186        * I believe none of these cases lead to trouble --SDM.
 187        */
 188
 189 #ifdef RTS_SUPPORTS_THREADS
 190       isWorkerBlockedInAwaitEvent = rtsTrue;
 191       workerWakeupPending = rtsFalse;
 192 #endif
 193       RELEASE_LOCK(&sched_mutex);
 194
 195       /* Check for any interesting events */
 196
 197       tv.tv_sec  = min / 1000000;
 198       tv.tv_usec = min % 1000000;
 199
 200       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 201           if (errno != EINTR) {
 202             /* Handle bad file descriptors by unblocking all the
 203                waiting threads. Why? Because a thread might have been
 204                a bit naughty and closed a file descriptor while another
 205                was blocked waiting. This is less-than-good programming
 206                practice, but having the RTS as a result fall over isn't
 207                acceptable, so we simply unblock all the waiting threads
 208                should we see a bad file descriptor & give the threads
 209                a chance to clean up their act.
 210
 211                Note: assume here that threads becoming unblocked
 212                will try to read/write the file descriptor before trying
 213                to issue a threadWaitRead/threadWaitWrite again (==> an
 214                IOError will result for the thread that's got the bad
 215                file descriptor.) Hence, there's no danger of a bad
 216                file descriptor being repeatedly select()'ed on, so
 217                the RTS won't loop.
 218             */
 219             if ( errno == EBADF ) {
 220               unblock_all = rtsTrue;
 221               break;
 222             } else {
 223               perror("select");
 224               barf("select failed");
 225             }
 226           }
 227           ACQUIRE_LOCK(&sched_mutex);
 228 #ifdef RTS_SUPPORTS_THREADS
 229           isWorkerBlockedInAwaitEvent = rtsFalse;
 230 #endif
 231
 232           /* We got a signal; could be one of ours.  If so, we need
 233            * to start up the signal handler straight away, otherwise
 234            * we could block for a long time before the signal is
 235            * serviced.
 236            */
 237 #if defined(RTS_USER_SIGNALS)
 238           if (signals_pending()) {
 239               RELEASE_LOCK(&sched_mutex); /* ToDo: kill */
 240               startSignalHandlers();
 241               ACQUIRE_LOCK(&sched_mutex);
 242               return; /* still hold the lock */
 243           }
 244 #endif
 245
 246           /* we were interrupted, return to the scheduler immediately.
 247            */
 248           if (interrupted) {
 249               return; /* still hold the lock */
 250           }
 251
 252           /* check for threads that need waking up
 253            */
 254           wakeUpSleepingThreads(getourtimeofday());
 255
 256           /* If new runnable threads have arrived, stop waiting for
 257            * I/O and run them.
 258            */
 259           if (run_queue_hd != END_TSO_QUEUE) {
 260               return; /* still hold the lock */
 261           }
 262
 263 #ifdef RTS_SUPPORTS_THREADS
 264           /* If another worker thread wants to take over,
 265            * return to the scheduler
 266            */
 267           if (needToYieldToReturningWorker()) {
 268               return; /* still hold the lock */
 269           }
 270 #endif
 271
 272 #ifdef RTS_SUPPORTS_THREADS
 273           isWorkerBlockedInAwaitEvent = rtsTrue;
 274 #endif
 275           RELEASE_LOCK(&sched_mutex);
 276       }
 277
 278       ACQUIRE_LOCK(&sched_mutex);
 279
 280       /* Step through the waiting queue, unblocking every thread that now has
 281        * a file descriptor in a ready state.
 282        */
 283
 284       prev = NULL;
 285       if (select_succeeded || unblock_all) {
 286           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 287               next = tso->link;
 288               switch (tso->why_blocked) {
 289               case BlockedOnRead:
 290                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &rfd);
 291                   break;
 292               case BlockedOnWrite:
 293                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &wfd);
 294                   break;
 295               default:
 296                   barf("awaitEvent");
 297               }
 298
 299               if (ready) {
 300                   IF_DEBUG(scheduler,debugBelch("Waking up blocked thread %d\n", tso->id));
 301                   tso->why_blocked = NotBlocked;
 302                   tso->link = END_TSO_QUEUE;
 303                   PUSH_ON_RUN_QUEUE(tso);
 304               } else {
 305                   if (prev == NULL)
 306                       blocked_queue_hd = tso;
 307                   else
 308                       prev->link = tso;
 309                   prev = tso;
 310               }
 311           }
 312
 313           if (prev == NULL)
 314               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 315           else {
 316               prev->link = END_TSO_QUEUE;
 317               blocked_queue_tl = prev;
 318           }
 319       }
 320
 321 #if defined(RTS_SUPPORTS_THREADS)
 322         // if we were woken up by wakeBlockedWorkerThread,
 323         // read the dummy byte from the pipe
 324       if(select_succeeded && FD_ISSET(workerWakeupPipe[0], &rfd)) {
 325           unsigned char dummy;
 326           wait = rtsFalse;
 327           read(workerWakeupPipe[0],&dummy,1);
 328       }
 329 #endif
 330     } while (wait && !interrupted && run_queue_hd == END_TSO_QUEUE);
 331 }
 332
 333
 334 #ifdef RTS_SUPPORTS_THREADS
 335 /* wakeBlockedWorkerThread
 336  *
 337  * If a worker thread is currently blocked within awaitEvent,
 338  * wake it.
 339  * Must be called with sched_mutex held.
 340  */
 341 void
 342 wakeBlockedWorkerThread()
 343 {
 344     if(isWorkerBlockedInAwaitEvent && !workerWakeupPending) {
 345         unsigned char dummy = 42;       // Any value will do here
 346
 347                         // write something so that select() wakes up
 348         write(workerWakeupPipe[1],&dummy,1);
 349         workerWakeupPending = rtsTrue;
 350     }
 351 }
 352
 353 /* resetWorkerWakeupPipeAfterFork
 354  *
 355  * To be called right after a fork().
 356  * After the fork(), the worker wakeup pipe will be shared
 357  * with the parent process, and that's something we don't want.
 358  */
 359 void
 360 resetWorkerWakeupPipeAfterFork()
 361 {
 362     if(workerWakeupInited) {
 363         close(workerWakeupPipe[0]);
 364         close(workerWakeupPipe[1]);
 365     }
 366     workerWakeupInited = rtsFalse;
 367 }
 368 #endif