ghc/rts/Select.c

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