ghc/rts/Select.c

   1 /* -----------------------------------------------------------------------------
   2  * $Id: Select.c,v 1.9 2000/03/16 17:24:08 simonmar Exp $
   3  *
   4  * (c) The GHC Team 1995-1999
   5  *
   6  * Support for concurrent non-blocking I/O and thread waiting.
   7  *
   8  * ---------------------------------------------------------------------------*/
   9
  10 /* we're outside the realms of POSIX here... */
  11 #define NON_POSIX_SOURCE
  12
  13 #include "Rts.h"
  14 #include "Schedule.h"
  15 #include "RtsUtils.h"
  16 #include "RtsFlags.h"
  17 #include "Itimer.h"
  18 #include "Signals.h"
  19
  20 # if defined(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 nat ticks_since_select = 0;
  29
  30 /* Argument 'wait' says whether to wait for I/O to become available,
  31  * or whether to just check and return immediately.  If there are
  32  * other threads ready to run, we normally do the non-waiting variety,
  33  * otherwise we wait (see Schedule.c).
  34  *
  35  * SMP note: must be called with sched_mutex locked.
  36  */
  37 void
  38 awaitEvent(rtsBool wait)
  39 {
  40 #ifdef mingw32_TARGET_OS
  41 /*
  42  * Win32 doesn't support select(). ToDo: use MsgWaitForMultipleObjects()
  43  * to achieve (similar) effect.
  44  *
  45  */
  46     return;
  47 #else
  48
  49     StgTSO *tso, *prev, *next;
  50     rtsBool ready;
  51     fd_set rfd,wfd;
  52     int numFound;
  53     nat min, delta;
  54     int maxfd = -1;
  55
  56     struct timeval tv;
  57 #ifndef linux_TARGET_OS
  58     struct timeval tv_before,tv_after;
  59 #endif
  60
  61     IF_DEBUG(scheduler,belch("Checking for threads blocked on I/O...\n"));
  62
  63     /* loop until we've woken up some threads.  This loop is needed
  64      * because the select timing isn't accurate, we sometimes sleep
  65      * for a while but not long enough to wake up a thread in
  66      * a threadDelay.
  67      */
  68     do {
  69
  70       /* see how long it's been since we last checked the blocked queue.
  71        * ToDo: make this check atomic, so we don't lose any ticks.
  72        */
  73       delta = ticks_since_select;
  74       ticks_since_select = 0;
  75       delta = delta * TICK_MILLISECS * 1000;
  76
  77       min = wait == rtsTrue ? 0x7fffffff : 0;
  78
  79       /*
  80        * Collect all of the fd's that we're interested in, and capture
  81        * the minimum waiting time (in microseconds) for the delayed threads.
  82        */
  83       FD_ZERO(&rfd);
  84       FD_ZERO(&wfd);
  85
  86       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
  87         next = tso->link;
  88
  89         switch (tso->why_blocked) {
  90         case BlockedOnRead:
  91           {
  92             int fd = tso->block_info.fd;
  93             maxfd = (fd > maxfd) ? fd : maxfd;
  94             FD_SET(fd, &rfd);
  95             continue;
  96           }
  97
  98         case BlockedOnWrite:
  99           {
 100             int fd = tso->block_info.fd;
 101             maxfd = (fd > maxfd) ? fd : maxfd;
 102             FD_SET(fd, &wfd);
 103             continue;
 104           }
 105
 106         case BlockedOnDelay:
 107           {
 108             if (tso->block_info.delay < min)
 109               min = tso->block_info.delay;
 110             continue;
 111           }
 112
 113         default:
 114           barf("AwaitEvent");
 115         }
 116       }
 117
 118       /* Release the scheduler lock while we do the poll.
 119        * this means that someone might muck with the blocked_queue
 120        * while we do this, but it shouldn't matter:
 121        *
 122        *   - another task might poll for I/O and remove one
 123        *     or more threads from the blocked_queue.
 124        *   - more I/O threads may be added to blocked_queue.
 125        *   - more delayed threads may be added to blocked_queue. We'll
 126        *     just subtract delta from their delays after the poll.
 127        *
 128        * I believe none of these cases lead to trouble --SDM.
 129        */
 130       RELEASE_LOCK(&sched_mutex);
 131
 132       /* Check for any interesting events */
 133
 134       tv.tv_sec = min / 1000000;
 135       tv.tv_usec = min % 1000000;
 136
 137 #ifndef linux_TARGET_OS
 138       gettimeofday(&tv_before, (struct timezone *) NULL);
 139 #endif
 140
 141       while (!interrupted &&
 142              (numFound = select(maxfd+1, &rfd, &wfd, NULL, &tv)) < 0) {
 143         if (errno != EINTR) {
 144           /* fflush(stdout); */
 145           perror("select");
 146           barf("select failed");
 147         }
 148         ACQUIRE_LOCK(&sched_mutex);
 149
 150         /* We got a signal; could be one of ours.  If so, we need
 151          * to start up the signal handler straight away, otherwise
 152          * we could block for a long time before the signal is
 153          * serviced.
 154          */
 155         if (signals_pending()) {
 156           RELEASE_LOCK(&sched_mutex);
 157           start_signal_handlers();
 158           break;
 159         }
 160
 161         /* If new runnable threads have arrived, stop waiting for
 162          * I/O and run them.
 163          */
 164         if (run_queue_hd != END_TSO_QUEUE) {
 165           RELEASE_LOCK(&sched_mutex);
 166           break;
 167         }
 168
 169         RELEASE_LOCK(&sched_mutex);
 170       }
 171
 172 #ifdef linux_TARGET_OS
 173       /* on Linux, tv is set to indicate the amount of time not
 174        * slept, so we don't need to gettimeofday() to find out.
 175        */
 176       delta += min - (tv.tv_sec * 1000000 + tv.tv_usec);
 177 #else
 178       gettimeofday(&tv_after, (struct timezone *) NULL);
 179       delta += (tv_after.tv_sec - tv_before.tv_sec) * 1000000 +
 180         tv_after.tv_usec - tv_before.tv_usec;
 181 #endif
 182
 183 #if 0
 184       if (delta != 0) { fprintf(stderr,"waited: %d %d %d\n", min, delta,
 185                                 interrupted); }
 186 #endif
 187
 188       ACQUIRE_LOCK(&sched_mutex);
 189
 190       /* Step through the waiting queue, unblocking every thread that now has
 191        * a file descriptor in a ready state.
 192
 193        * For the delayed threads, decrement the number of microsecs
 194        * we've been blocked for. Unblock the threads that have thusly expired.
 195        */
 196
 197       prev = NULL;
 198       for(tso = blocked_queue_hd; tso != END_TSO_QUEUE; tso = next) {
 199         next = tso->link;
 200         switch (tso->why_blocked) {
 201         case BlockedOnRead:
 202           ready = FD_ISSET(tso->block_info.fd, &rfd);
 203           break;
 204
 205         case BlockedOnWrite:
 206           ready = FD_ISSET(tso->block_info.fd, &wfd);
 207           break;
 208
 209         case BlockedOnDelay:
 210           if (tso->block_info.delay > delta) {
 211             tso->block_info.delay -= delta;
 212             ready = 0;
 213           } else {
 214             tso->block_info.delay = 0;
 215             ready = 1;
 216           }
 217           break;
 218
 219         default:
 220           barf("awaitEvent");
 221         }
 222
 223         if (ready) {
 224           IF_DEBUG(scheduler,belch("Waking up thread %d\n", tso->id));
 225           tso->why_blocked = NotBlocked;
 226           tso->link = END_TSO_QUEUE;
 227           PUSH_ON_RUN_QUEUE(tso);
 228         } else {
 229           if (prev == NULL)
 230             blocked_queue_hd = tso;
 231           else
 232             prev->link = tso;
 233           prev = tso;
 234         }
 235       }
 236
 237       if (prev == NULL)
 238         blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 239       else {
 240         prev->link = END_TSO_QUEUE;
 241         blocked_queue_tl = prev;
 242       }
 243
 244     } while (wait && run_queue_hd == END_TSO_QUEUE);
 245 #endif
 246 }