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 #include "PosixSource.h"
  10 #include "Rts.h"
  11
  12 #include "Signals.h"
  13 #include "Schedule.h"
  14 #include "RtsUtils.h"
  15 #include "Itimer.h"
  16 #include "Capability.h"
  17 #include "Select.h"
  18 #include "AwaitEvent.h"
  19
  20 # ifdef HAVE_SYS_TYPES_H
  21 #  include <sys/types.h>
  22 # endif
  23
  24 # ifdef HAVE_SYS_TIME_H
  25 #  include <sys/time.h>
  26 # endif
  27
  28 #include <errno.h>
  29 #include <string.h>
  30
  31 #ifdef HAVE_UNISTD_H
  32 #include <unistd.h>
  33 #endif
  34
  35 #if !defined(THREADED_RTS)
  36 /* last timestamp */
  37 lnat timestamp = 0;
  38
  39 /*
  40  * The threaded RTS uses an IO-manager thread in Haskell instead (see GHC.Conc)
  41  */
  42
  43 /* There's a clever trick here to avoid problems when the time wraps
  44  * around.  Since our maximum delay is smaller than 31 bits of ticks
  45  * (it's actually 31 bits of microseconds), we can safely check
  46  * whether a timer has expired even if our timer will wrap around
  47  * before the target is reached, using the following formula:
  48  *
  49  *        (int)((uint)current_time - (uint)target_time) < 0
  50  *
  51  * if this is true, then our time has expired.
  52  * (idea due to Andy Gill).
  53  */
  54 static rtsBool
  55 wakeUpSleepingThreads(lnat ticks)
  56 {
  57     StgTSO *tso;
  58     rtsBool flag = rtsFalse;
  59
  60     while (sleeping_queue != END_TSO_QUEUE) {
  61         tso = sleeping_queue;
  62         if (tso->what_next == ThreadRelocated) {
  63             sleeping_queue = tso->_link;
  64             continue;
  65         }
  66         if (((long)ticks - (long)tso->block_info.target) < 0) {
  67             break;
  68         }
  69         sleeping_queue = tso->_link;
  70         tso->why_blocked = NotBlocked;
  71         tso->_link = END_TSO_QUEUE;
  72         IF_DEBUG(scheduler,debugBelch("Waking up sleeping thread %lu\n", (unsigned long)tso->id));
  73         // MainCapability: this code is !THREADED_RTS
  74         pushOnRunQueue(&MainCapability,tso);
  75         flag = rtsTrue;
  76     }
  77     return flag;
  78 }
  79
  80 /* Argument 'wait' says whether to wait for I/O to become available,
  81  * or whether to just check and return immediately.  If there are
  82  * other threads ready to run, we normally do the non-waiting variety,
  83  * otherwise we wait (see Schedule.c).
  84  *
  85  * SMP note: must be called with sched_mutex locked.
  86  *
  87  * Windows: select only works on sockets, so this doesn't really work,
  88  * though it makes things better than before. MsgWaitForMultipleObjects
  89  * should really be used, though it only seems to work for read handles,
  90  * not write handles.
  91  *
  92  */
  93 void
  94 awaitEvent(rtsBool wait)
  95 {
  96     StgTSO *tso, *prev, *next;
  97     rtsBool ready;
  98     fd_set rfd,wfd;
  99     int numFound;
 100     int maxfd = -1;
 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              debugBelch("scheduler: checking for threads blocked on I/O");
 111              if (wait) {
 112                  debugBelch(" (waiting)");
 113              }
 114              debugBelch("\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               * RtsFlags.MiscFlags.tickInterval * 1000;
 134       } else {
 135           min = 0x7ffffff;
 136       }
 137
 138       /*
 139        * Collect all of the fd's that we're interested in
 140        */
 141       FD_ZERO(&rfd);
 142       FD_ZERO(&wfd);
 143
 144       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 145         next = tso->_link;
 146
 147       /* On FreeBSD FD_SETSIZE is unsigned. Cast it to signed int
 148        * in order to switch off the 'comparison between signed and
 149        * unsigned error message
 150        */
 151         switch (tso->why_blocked) {
 152         case BlockedOnRead:
 153           {
 154             int fd = tso->block_info.fd;
 155             if (fd >= (int)FD_SETSIZE) {
 156                 barf("awaitEvent: descriptor out of range");
 157             }
 158             maxfd = (fd > maxfd) ? fd : maxfd;
 159             FD_SET(fd, &rfd);
 160             continue;
 161           }
 162
 163         case BlockedOnWrite:
 164           {
 165             int fd = tso->block_info.fd;
 166             if (fd >= (int)FD_SETSIZE) {
 167                 barf("awaitEvent: descriptor out of range");
 168             }
 169             maxfd = (fd > maxfd) ? fd : maxfd;
 170             FD_SET(fd, &wfd);
 171             continue;
 172           }
 173
 174         default:
 175           barf("AwaitEvent");
 176         }
 177       }
 178
 179       /* Check for any interesting events */
 180
 181       tv.tv_sec  = min / 1000000;
 182       tv.tv_usec = min % 1000000;
 183
 184       while ((numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 185           if (errno != EINTR) {
 186             /* Handle bad file descriptors by unblocking all the
 187                waiting threads. Why? Because a thread might have been
 188                a bit naughty and closed a file descriptor while another
 189                was blocked waiting. This is less-than-good programming
 190                practice, but having the RTS as a result fall over isn't
 191                acceptable, so we simply unblock all the waiting threads
 192                should we see a bad file descriptor & give the threads
 193                a chance to clean up their act.
 194
 195                Note: assume here that threads becoming unblocked
 196                will try to read/write the file descriptor before trying
 197                to issue a threadWaitRead/threadWaitWrite again (==> an
 198                IOError will result for the thread that's got the bad
 199                file descriptor.) Hence, there's no danger of a bad
 200                file descriptor being repeatedly select()'ed on, so
 201                the RTS won't loop.
 202             */
 203             if ( errno == EBADF ) {
 204               unblock_all = rtsTrue;
 205               break;
 206             } else {
 207               perror("select");
 208               barf("select failed");
 209             }
 210           }
 211
 212           /* We got a signal; could be one of ours.  If so, we need
 213            * to start up the signal handler straight away, otherwise
 214            * we could block for a long time before the signal is
 215            * serviced.
 216            */
 217 #if defined(RTS_USER_SIGNALS)
 218           if (RtsFlags.MiscFlags.install_signal_handlers && signals_pending()) {
 219               startSignalHandlers(&MainCapability);
 220               return; /* still hold the lock */
 221           }
 222 #endif
 223
 224           /* we were interrupted, return to the scheduler immediately.
 225            */
 226           if (sched_state >= SCHED_INTERRUPTING) {
 227               return; /* still hold the lock */
 228           }
 229
 230           /* check for threads that need waking up
 231            */
 232           wakeUpSleepingThreads(getourtimeofday());
 233
 234           /* If new runnable threads have arrived, stop waiting for
 235            * I/O and run them.
 236            */
 237           if (!emptyRunQueue(&MainCapability)) {
 238               return; /* still hold the lock */
 239           }
 240       }
 241
 242       /* Step through the waiting queue, unblocking every thread that now has
 243        * a file descriptor in a ready state.
 244        */
 245
 246       prev = NULL;
 247       if (select_succeeded || unblock_all) {
 248           for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 249               next = tso->_link;
 250
 251               if (tso->what_next == ThreadRelocated) {
 252                   continue;
 253               }
 254
 255               switch (tso->why_blocked) {
 256               case BlockedOnRead:
 257                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &rfd);
 258                   break;
 259               case BlockedOnWrite:
 260                   ready = unblock_all || FD_ISSET(tso->block_info.fd, &wfd);
 261                   break;
 262               default:
 263                   barf("awaitEvent");
 264               }
 265
 266               if (ready) {
 267                 IF_DEBUG(scheduler,debugBelch("Waking up blocked thread %lu\n", (unsigned long)tso->id));
 268                   tso->why_blocked = NotBlocked;
 269                   tso->_link = END_TSO_QUEUE;
 270                   pushOnRunQueue(&MainCapability,tso);
 271               } else {
 272                   if (prev == NULL)
 273                       blocked_queue_hd = tso;
 274                   else
 275                       setTSOLink(&MainCapability, prev, tso);
 276                   prev = tso;
 277               }
 278           }
 279
 280           if (prev == NULL)
 281               blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 282           else {
 283               prev->_link = END_TSO_QUEUE;
 284               blocked_queue_tl = prev;
 285           }
 286       }
 287
 288     } while (wait && sched_state == SCHED_RUNNING
 289              && emptyRunQueue(&MainCapability));
 290 }
 291
 292 #endif /* THREADED_RTS */