ghc/rts/Select.c

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