rts/posix/Select.c

   1 /* -----------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team 1995-2002
   4  *
   5  * Support for concurrent non-blocking I/O and thread waiting.
   6  *
   7  * ---------------------------------------------------------------------------*/
   8
   9 /* we're outside the realms of POSIX here... */
  10 /* #include "PosixSource.h" */
  11
  12 #include "Rts.h"
  13 #include "Schedule.h"
  14 #include "RtsUtils.h"
  15 #include "RtsFlags.h"
  16 #include "Timer.h"
  17 #include "Itimer.h"
  18 #include "Signals.h"
  19 #include "Capability.h"
  20 #include "posix/Select.h"
  21
  22 # ifdef HAVE_SYS_TYPES_H
  23 #  include <sys/types.h>
  24 # endif
  25
  26 # ifdef HAVE_SYS_TIME_H
  27 #  include <sys/time.h>
  28 # endif
  29
  30 #include <errno.h>
  31 #include <string.h>
  32
  33 #ifdef HAVE_UNISTD_H
  34 #include <unistd.h>
  35 #endif
  36
  37 #if !defined(THREADED_RTS)
  38 /* last timestamp */
  39 lnat timestamp = 0;
  40
  41 /*
  42  * The threaded RTS uses an IO-manager thread in Haskell instead (see GHC.Conc)
  43  */
  44
  45 /* There's a clever trick here to avoid problems when the time wraps
  46  * around.  Since our maximum delay is smaller than 31 bits of ticks
  47  * (it's actually 31 bits of microseconds), we can safely check
  48  * whether a timer has expired even if our timer will wrap around
  49  * before the target is reached, using the following formula:
  50  *
  51  *        (int)((uint)current_time - (uint)target_time) < 0
  52  *
  53  * if this is true, then our time has expired.
  54  * (idea due to Andy Gill).
  55  */
  56 static rtsBool
  57 wakeUpSleepingThreads(lnat ticks)
  58 {
  59     StgTSO *tso;
  60     rtsBool flag = rtsFalse;
  61
  62     while (sleeping_queue != END_TSO_QUEUE &&
  63            (int)(ticks - sleeping_queue->block_info.target) > 0) {
  64         tso = sleeping_queue;
  65         sleeping_queue = tso->link;
  66         tso->why_blocked = NotBlocked;
  67         tso->link = END_TSO_QUEUE;
  68         IF_DEBUG(scheduler,debugBelch("Waking up sleeping thread %lu\n", (unsigned long)tso->id));
  69         // MainCapability: this code is !THREADED_RTS
  70         pushOnRunQueue(&MainCapability,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              debugBelch("scheduler: checking for threads blocked on I/O");
 107              if (wait) {
 108                  debugBelch(" (waiting)");
 109              }
 110              debugBelch("\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               * RtsFlags.MiscFlags.tickInterval * 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             if (fd >= FD_SETSIZE) {
 148                 barf("awaitEvent: descriptor out of range");
 149             }
 150             maxfd = (fd > maxfd) ? fd : maxfd;
 151             FD_SET(fd, &rfd);
 152             continue;
 153           }
 154
 155         case BlockedOnWrite:
 156           {
 157             int fd = tso->block_info.fd;
 158             if (fd >= FD_SETSIZE) {
 159                 barf("awaitEvent: descriptor out of range");
 160             }
 161             maxfd = (fd > maxfd) ? fd : maxfd;
 162             FD_SET(fd, &wfd);
 163             continue;
 164           }
 165
 166         default:
 167           barf("AwaitEvent");
 168         }
 169       }
 170
 171       /* Check for any interesting events */
 172
 173       tv.tv_sec  = min / 1000000;
 174       tv.tv_usec = min % 1000000;
 175
 176       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 177           if (errno != EINTR) {
 178             /* Handle bad file descriptors by unblocking all the
 179                waiting threads. Why? Because a thread might have been
 180                a bit naughty and closed a file descriptor while another
 181                was blocked waiting. This is less-than-good programming
 182                practice, but having the RTS as a result fall over isn't
 183                acceptable, so we simply unblock all the waiting threads
 184                should we see a bad file descriptor & give the threads
 185                a chance to clean up their act.
 186
 187                Note: assume here that threads becoming unblocked
 188                will try to read/write the file descriptor before trying
 189                to issue a threadWaitRead/threadWaitWrite again (==> an
 190                IOError will result for the thread that's got the bad
 191                file descriptor.) Hence, there's no danger of a bad
 192                file descriptor being repeatedly select()'ed on, so
 193                the RTS won't loop.
 194             */
 195             if ( errno == EBADF ) {
 196               unblock_all = rtsTrue;
 197               break;
 198             } else {
 199               perror("select");
 200               barf("select failed");
 201             }
 202           }
 203
 204           /* We got a signal; could be one of ours.  If so, we need
 205            * to start up the signal handler straight away, otherwise
 206            * we could block for a long time before the signal is
 207            * serviced.
 208            */
 209 #if defined(RTS_USER_SIGNALS)
 210           if (signals_pending()) {
 211               startSignalHandlers(&MainCapability);
 212               return; /* still hold the lock */
 213           }
 214 #endif
 215
 216           /* we were interrupted, return to the scheduler immediately.
 217            */
 218           if (sched_state >= SCHED_INTERRUPTING) {
 219               return; /* still hold the lock */
 220           }
 221
 222           /* check for threads that need waking up
 223            */
 224           wakeUpSleepingThreads(getourtimeofday());
 225
 226           /* If new runnable threads have arrived, stop waiting for
 227            * I/O and run them.
 228            */
 229           if (!emptyRunQueue(&MainCapability)) {
 230               return; /* still hold the lock */
 231           }
 232       }
 233
 234       /* Step through the waiting queue, unblocking every thread that now has
 235        * a file descriptor in a ready state.
 236        */
 237
 238       prev = NULL;
 239       if (select_succeeded || unblock_all) {
 240           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 241               next = tso->link;
 242               switch (tso->why_blocked) {
 243               case BlockedOnRead:
 244                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &rfd);
 245                   break;
 246               case BlockedOnWrite:
 247                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &wfd);
 248                   break;
 249               default:
 250                   barf("awaitEvent");
 251               }
 252
 253               if (ready) {
 254                 IF_DEBUG(scheduler,debugBelch("Waking up blocked thread %lu\n", (unsigned long)tso->id));
 255                   tso->why_blocked = NotBlocked;
 256                   tso->link = END_TSO_QUEUE;
 257                   pushOnRunQueue(&MainCapability,tso);
 258               } else {
 259                   if (prev == NULL)
 260                       blocked_queue_hd = tso;
 261                   else
 262                       prev->link = tso;
 263                   prev = tso;
 264               }
 265           }
 266
 267           if (prev == NULL)
 268               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 269           else {
 270               prev->link = END_TSO_QUEUE;
 271               blocked_queue_tl = prev;
 272           }
 273       }
 274
 275     } while (wait && sched_state == SCHED_RUNNING
 276              && emptyRunQueue(&MainCapability));
 277 }
 278
 279 #endif /* THREADED_RTS */