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