ghc/rts/Select.c

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