ghc/rts/Select.c

   1 /* -----------------------------------------------------------------------------
   2  * $Id: Select.c,v 1.23 2003/01/25 15:54:50 wolfgang 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 /* 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 #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 # ifdef mingw32_TARGET_OS
  30 #  include <windows.h>
  31 # endif
  32
  33 #include <errno.h>
  34 #include <string.h>
  35
  36 /* last timestamp */
  37 nat timestamp = 0;
  38
  39 #ifdef RTS_SUPPORTS_THREADS
  40 static rtsBool isWorkerBlockedInAwaitEvent = rtsFalse;
  41 static rtsBool workerWakeupPending = rtsFalse;
  42 #ifndef mingw32_TARGET_OS
  43 static int workerWakeupPipe[2];
  44 static rtsBool workerWakeupInited = rtsFalse;
  45 #endif
  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,belch("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 #ifndef mingw32_TARGET_OS
  98     int numFound;
  99     int maxfd = -1;
 100 #endif
 101     rtsBool select_succeeded = rtsTrue;
 102     rtsBool unblock_all = rtsFalse;
 103     struct timeval tv;
 104     lnat min, ticks;
 105
 106     tv.tv_sec  = 0;
 107     tv.tv_usec = 0;
 108
 109     IF_DEBUG(scheduler,
 110              belch("scheduler: checking for threads blocked on I/O");
 111              if (wait) {
 112                  belch(" (waiting)");
 113              }
 114              belch("\n");
 115              );
 116
 117     /* loop until we've woken up some threads.  This loop is needed
 118      * because the select timing isn't accurate, we sometimes sleep
 119      * for a while but not long enough to wake up a thread in
 120      * a threadDelay.
 121      */
 122     do {
 123
 124       ticks = timestamp = getourtimeofday();
 125       if (wakeUpSleepingThreads(ticks)) {
 126           return;
 127       }
 128
 129       if (!wait) {
 130           min = 0;
 131       } else if (sleeping_queue != END_TSO_QUEUE) {
 132           min = (sleeping_queue->block_info.target - ticks)
 133               * TICK_MILLISECS * 1000;
 134       } else {
 135           min = 0x7ffffff;
 136       }
 137
 138 #ifndef mingw32_TARGET_OS
 139       /*
 140        * Collect all of the fd's that we're interested in
 141        */
 142       FD_ZERO(&rfd);
 143       FD_ZERO(&wfd);
 144
 145       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 146         next = tso->link;
 147
 148         switch (tso->why_blocked) {
 149         case BlockedOnRead:
 150           {
 151             int fd = tso->block_info.fd;
 152             maxfd = (fd > maxfd) ? fd : maxfd;
 153             FD_SET(fd, &rfd);
 154             continue;
 155           }
 156
 157         case BlockedOnWrite:
 158           {
 159             int fd = tso->block_info.fd;
 160             maxfd = (fd > maxfd) ? fd : maxfd;
 161             FD_SET(fd, &wfd);
 162             continue;
 163           }
 164
 165         default:
 166           barf("AwaitEvent");
 167         }
 168       }
 169
 170 #ifdef RTS_SUPPORTS_THREADS
 171       if(!workerWakeupInited) {
 172           pipe(workerWakeupPipe);
 173           workerWakeupInited = rtsTrue;
 174       }
 175       FD_SET(workerWakeupPipe[0], &rfd);
 176       maxfd = workerWakeupPipe[0] > maxfd ? workerWakeupPipe[0] : maxfd;
 177 #endif
 178
 179       /* Release the scheduler lock while we do the poll.
 180        * this means that someone might muck with the blocked_queue
 181        * while we do this, but it shouldn't matter:
 182        *
 183        *   - another task might poll for I/O and remove one
 184        *     or more threads from the blocked_queue.
 185        *   - more I/O threads may be added to blocked_queue.
 186        *   - more delayed threads may be added to blocked_queue. We'll
 187        *     just subtract delta from their delays after the poll.
 188        *
 189        * I believe none of these cases lead to trouble --SDM.
 190        */
 191
 192 #ifdef RTS_SUPPORTS_THREADS
 193       isWorkerBlockedInAwaitEvent = rtsTrue;
 194       workerWakeupPending = rtsFalse;
 195 #endif
 196       RELEASE_LOCK(&sched_mutex);
 197
 198       /* Check for any interesting events */
 199
 200       tv.tv_sec  = min / 1000000;
 201       tv.tv_usec = min % 1000000;
 202
 203       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 204           if (errno != EINTR) {
 205             /* Handle bad file descriptors by unblocking all the
 206                waiting threads. Why? Because a thread might have been
 207                a bit naughty and closed a file descriptor while another
 208                was blocked waiting. This is less-than-good programming
 209                practice, but having the RTS as a result fall over isn't
 210                acceptable, so we simply unblock all the waiting threads
 211                should we see a bad file descriptor & give the threads
 212                a chance to clean up their act.
 213
 214                Note: assume here that threads becoming unblocked
 215                will try to read/write the file descriptor before trying
 216                to issue a threadWaitRead/threadWaitWrite again (==> an
 217                IOError will result for the thread that's got the bad
 218                file descriptor.) Hence, there's no danger of a bad
 219                file descriptor being repeatedly select()'ed on, so
 220                the RTS won't loop.
 221             */
 222             if ( errno == EBADF ) {
 223               unblock_all = rtsTrue;
 224               break;
 225             } else {
 226               fprintf(stderr,"%d\n", errno);
 227               fflush(stderr);
 228               perror("select");
 229               barf("select failed");
 230             }
 231           }
 232 #else /* on mingwin */
 233 #ifdef RTS_SUPPORTS_THREADS
 234       isWorkerBlockedInAwaitEvent = rtsTrue;
 235 #endif
 236       RELEASE_LOCK(&sched_mutex);
 237       while (1) {
 238           Sleep(0); /* don't busy wait */
 239 #endif /* mingw32_TARGET_OS */
 240           ACQUIRE_LOCK(&sched_mutex);
 241 #ifdef RTS_SUPPORTS_THREADS
 242           isWorkerBlockedInAwaitEvent = rtsFalse;
 243 #endif
 244
 245 #ifndef mingw32_TARGET_OS
 246           /* We got a signal; could be one of ours.  If so, we need
 247            * to start up the signal handler straight away, otherwise
 248            * we could block for a long time before the signal is
 249            * serviced.
 250            */
 251           if (signals_pending()) {
 252               RELEASE_LOCK(&sched_mutex); /* ToDo: kill */
 253               startSignalHandlers();
 254               ACQUIRE_LOCK(&sched_mutex);
 255               return; /* still hold the lock */
 256           }
 257 #endif
 258
 259           /* we were interrupted, return to the scheduler immediately.
 260            */
 261           if (interrupted) {
 262               return; /* still hold the lock */
 263           }
 264
 265           /* check for threads that need waking up
 266            */
 267           wakeUpSleepingThreads(getourtimeofday());
 268
 269           /* If new runnable threads have arrived, stop waiting for
 270            * I/O and run them.
 271            */
 272           if (run_queue_hd != END_TSO_QUEUE) {
 273               return; /* still hold the lock */
 274           }
 275
 276 #ifdef RTS_SUPPORTS_THREADS
 277           /* If another worker thread wants to take over,
 278            * return to the scheduler
 279            */
 280           if (needToYieldToReturningWorker()) {
 281               return; /* still hold the lock */
 282           }
 283 #endif
 284
 285 #ifdef RTS_SUPPORTS_THREADS
 286           isWorkerBlockedInAwaitEvent = rtsTrue;
 287 #endif
 288           RELEASE_LOCK(&sched_mutex);
 289       }
 290
 291       ACQUIRE_LOCK(&sched_mutex);
 292
 293       /* Step through the waiting queue, unblocking every thread that now has
 294        * a file descriptor in a ready state.
 295        */
 296
 297       prev = NULL;
 298       if (select_succeeded || unblock_all) {
 299           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 300               next = tso->link;
 301               switch (tso->why_blocked) {
 302               case BlockedOnRead:
 303                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &rfd);
 304                   break;
 305               case BlockedOnWrite:
 306                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &wfd);
 307                   break;
 308               default:
 309                   barf("awaitEvent");
 310               }
 311
 312               if (ready) {
 313                   IF_DEBUG(scheduler,belch("Waking up blocked thread %d\n", tso->id));
 314                   tso->why_blocked = NotBlocked;
 315                   tso->link = END_TSO_QUEUE;
 316                   PUSH_ON_RUN_QUEUE(tso);
 317               } else {
 318                   if (prev == NULL)
 319                       blocked_queue_hd = tso;
 320                   else
 321                       prev->link = tso;
 322                   prev = tso;
 323               }
 324           }
 325
 326           if (prev == NULL)
 327               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 328           else {
 329               prev->link = END_TSO_QUEUE;
 330               blocked_queue_tl = prev;
 331           }
 332       }
 333
 334 #if defined(RTS_SUPPORTS_THREADS) && !defined(mingw32_TARGET_OS)
 335         // if we were woken up by wakeBlockedWorkerThread,
 336         // read the dummy byte from the pipe
 337       if(select_succeeded && FD_ISSET(workerWakeupPipe[0], &rfd)) {
 338           unsigned char dummy;
 339           wait = rtsFalse;
 340           read(workerWakeupPipe[0],&dummy,1);
 341       }
 342 #endif
 343     } while (wait && !interrupted && run_queue_hd == END_TSO_QUEUE);
 344 }
 345
 346
 347 #ifdef RTS_SUPPORTS_THREADS
 348 /* wakeBlockedWorkerThread
 349  *
 350  * If a worker thread is currently blocked within awaitEvent,
 351  * wake it.
 352  * Must be called with sched_mutex held.
 353  */
 354
 355 void
 356 wakeBlockedWorkerThread()
 357 {
 358 #ifndef mingw32_TARGET_OS
 359     if(isWorkerBlockedInAwaitEvent && !workerWakeupPending) {
 360         unsigned char dummy = 42;       // Any value will do here
 361
 362                         // write something so that select() wakes up
 363         write(workerWakeupPipe[1],&dummy,1);
 364         workerWakeupPending = rtsTrue;
 365     }
 366 #else
 367         // The Win32 implementation currently uses a polling loop,
 368         // so there is no need to explicitly wake it
 369 #endif
 370 }
 371
 372 #endif