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