GHC/Conc.lhs

   1 \begin{code}
   2 {-# OPTIONS -fno-implicit-prelude #-}
   3 -----------------------------------------------------------------------------
   4 -- |
   5 -- Module      :  GHC.Conc
   6 -- Copyright   :  (c) The University of Glasgow, 1994-2002
   7 -- License     :  see libraries/base/LICENSE
   8 --
   9 -- Maintainer  :  cvs-ghc@haskell.org
  10 -- Stability   :  internal
  11 -- Portability :  non-portable (GHC extensions)
  12 --
  13 -- Basic concurrency stuff.
  14 --
  15 -----------------------------------------------------------------------------
  16
  17 #include "ghcconfig.h"
  18 module GHC.Conc
  19         ( ThreadId(..)
  20
  21         -- Forking and suchlike
  22         , myThreadId    -- :: IO ThreadId
  23         , killThread    -- :: ThreadId -> IO ()
  24         , throwTo       -- :: ThreadId -> Exception -> IO ()
  25         , par           -- :: a -> b -> b
  26         , pseq          -- :: a -> b -> b
  27         , yield         -- :: IO ()
  28         , labelThread   -- :: ThreadId -> String -> IO ()
  29
  30         -- Waiting
  31         , threadDelay           -- :: Int -> IO ()
  32         , threadWaitRead        -- :: Int -> IO ()
  33         , threadWaitWrite       -- :: Int -> IO ()
  34
  35         -- MVars
  36         , MVar          -- abstract
  37         , newMVar       -- :: a -> IO (MVar a)
  38         , newEmptyMVar  -- :: IO (MVar a)
  39         , takeMVar      -- :: MVar a -> IO a
  40         , putMVar       -- :: MVar a -> a -> IO ()
  41         , tryTakeMVar   -- :: MVar a -> IO (Maybe a)
  42         , tryPutMVar    -- :: MVar a -> a -> IO Bool
  43         , isEmptyMVar   -- :: MVar a -> IO Bool
  44         , addMVarFinalizer -- :: MVar a -> IO () -> IO ()
  45
  46 #ifdef mingw32_TARGET_OS
  47         , asyncRead     -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
  48         , asyncWrite    -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
  49         , asyncDoProc   -- :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
  50
  51         , asyncReadBA   -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
  52         , asyncWriteBA  -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
  53 #endif
  54         ) where
  55
  56 import System.Posix.Types
  57 import System.Posix.Internals
  58 import Foreign
  59 import Foreign.C
  60
  61 import Data.Maybe
  62
  63 import GHC.Base
  64 import GHC.IOBase
  65 import GHC.Num          ( Num(..) )
  66 import GHC.Real         ( fromIntegral, quot )
  67 import GHC.Base         ( Int(..) )
  68 import GHC.Exception    ( Exception(..), AsyncException(..) )
  69 import GHC.Pack         ( packCString# )
  70 import GHC.Ptr          ( Ptr(..), plusPtr, FunPtr(..) )
  71 import GHC.STRef
  72
  73 infixr 0 `par`, `pseq`
  74 \end{code}
  75
  76 %************************************************************************
  77 %*                                                                      *
  78 \subsection{@ThreadId@, @par@, and @fork@}
  79 %*                                                                      *
  80 %************************************************************************
  81
  82 \begin{code}
  83 data ThreadId = ThreadId ThreadId#
  84 -- ToDo: data ThreadId = ThreadId (Weak ThreadId#)
  85 -- But since ThreadId# is unlifted, the Weak type must use open
  86 -- type variables.
  87 {- ^
  88 A 'ThreadId' is an abstract type representing a handle to a thread.
  89 'ThreadId' is an instance of 'Eq', 'Ord' and 'Show', where
  90 the 'Ord' instance implements an arbitrary total ordering over
  91 'ThreadId's. The 'Show' instance lets you convert an arbitrary-valued
  92 'ThreadId' to string form; showing a 'ThreadId' value is occasionally
  93 useful when debugging or diagnosing the behaviour of a concurrent
  94 program.
  95
  96 /Note/: in GHC, if you have a 'ThreadId', you essentially have
  97 a pointer to the thread itself.  This means the thread itself can\'t be
  98 garbage collected until you drop the 'ThreadId'.
  99 This misfeature will hopefully be corrected at a later date.
 100
 101 /Note/: Hugs does not provide any operations on other threads;
 102 it defines 'ThreadId' as a synonym for ().
 103 -}
 104
 105 --forkIO has now been hoisted out into the Concurrent library.
 106
 107 {- | 'killThread' terminates the given thread (GHC only).
 108 Any work already done by the thread isn\'t
 109 lost: the computation is suspended until required by another thread.
 110 The memory used by the thread will be garbage collected if it isn\'t
 111 referenced from anywhere.  The 'killThread' function is defined in
 112 terms of 'throwTo':
 113
 114 > killThread tid = throwTo tid (AsyncException ThreadKilled)
 115
 116 -}
 117 killThread :: ThreadId -> IO ()
 118 killThread tid = throwTo tid (AsyncException ThreadKilled)
 119
 120 {- | 'throwTo' raises an arbitrary exception in the target thread (GHC only).
 121
 122 'throwTo' does not return until the exception has been raised in the
 123 target thread.  The calling thread can thus be certain that the target
 124 thread has received the exception.  This is a useful property to know
 125 when dealing with race conditions: eg. if there are two threads that
 126 can kill each other, it is guaranteed that only one of the threads
 127 will get to kill the other. -}
 128 throwTo :: ThreadId -> Exception -> IO ()
 129 throwTo (ThreadId id) ex = IO $ \ s ->
 130    case (killThread# id ex s) of s1 -> (# s1, () #)
 131
 132 -- | Returns the 'ThreadId' of the calling thread (GHC only).
 133 myThreadId :: IO ThreadId
 134 myThreadId = IO $ \s ->
 135    case (myThreadId# s) of (# s1, id #) -> (# s1, ThreadId id #)
 136
 137
 138 -- |The 'yield' action allows (forces, in a co-operative multitasking
 139 -- implementation) a context-switch to any other currently runnable
 140 -- threads (if any), and is occasionally useful when implementing
 141 -- concurrency abstractions.
 142 yield :: IO ()
 143 yield = IO $ \s ->
 144    case (yield# s) of s1 -> (# s1, () #)
 145
 146 {- | 'labelThread' stores a string as identifier for this thread if
 147 you built a RTS with debugging support. This identifier will be used in
 148 the debugging output to make distinction of different threads easier
 149 (otherwise you only have the thread state object\'s address in the heap).
 150
 151 Other applications like the graphical Concurrent Haskell Debugger
 152 (<http://www.informatik.uni-kiel.de/~fhu/chd/>) may choose to overload
 153 'labelThread' for their purposes as well.
 154 -}
 155
 156 labelThread :: ThreadId -> String -> IO ()
 157 labelThread (ThreadId t) str = IO $ \ s ->
 158    let ps  = packCString# str
 159        adr = byteArrayContents# ps in
 160      case (labelThread# t adr s) of s1 -> (# s1, () #)
 161
 162 --      Nota Bene: 'pseq' used to be 'seq'
 163 --                 but 'seq' is now defined in PrelGHC
 164 --
 165 -- "pseq" is defined a bit weirdly (see below)
 166 --
 167 -- The reason for the strange "lazy" call is that
 168 -- it fools the compiler into thinking that pseq  and par are non-strict in
 169 -- their second argument (even if it inlines pseq at the call site).
 170 -- If it thinks pseq is strict in "y", then it often evaluates
 171 -- "y" before "x", which is totally wrong.
 172
 173 {-# INLINE pseq  #-}
 174 pseq :: a -> b -> b
 175 pseq  x y = x `seq` lazy y
 176
 177 {-# INLINE par  #-}
 178 par :: a -> b -> b
 179 par  x y = case (par# x) of { _ -> lazy y }
 180 \end{code}
 181
 182 %************************************************************************
 183 %*                                                                      *
 184 \subsection[mvars]{M-Structures}
 185 %*                                                                      *
 186 %************************************************************************
 187
 188 M-Vars are rendezvous points for concurrent threads.  They begin
 189 empty, and any attempt to read an empty M-Var blocks.  When an M-Var
 190 is written, a single blocked thread may be freed.  Reading an M-Var
 191 toggles its state from full back to empty.  Therefore, any value
 192 written to an M-Var may only be read once.  Multiple reads and writes
 193 are allowed, but there must be at least one read between any two
 194 writes.
 195
 196 \begin{code}
 197 --Defined in IOBase to avoid cycle: data MVar a = MVar (SynchVar# RealWorld a)
 198
 199 -- |Create an 'MVar' which is initially empty.
 200 newEmptyMVar  :: IO (MVar a)
 201 newEmptyMVar = IO $ \ s# ->
 202     case newMVar# s# of
 203          (# s2#, svar# #) -> (# s2#, MVar svar# #)
 204
 205 -- |Create an 'MVar' which contains the supplied value.
 206 newMVar :: a -> IO (MVar a)
 207 newMVar value =
 208     newEmptyMVar        >>= \ mvar ->
 209     putMVar mvar value  >>
 210     return mvar
 211
 212 -- |Return the contents of the 'MVar'.  If the 'MVar' is currently
 213 -- empty, 'takeMVar' will wait until it is full.  After a 'takeMVar',
 214 -- the 'MVar' is left empty.
 215 --
 216 -- If several threads are competing to take the same 'MVar', one is chosen
 217 -- to continue at random when the 'MVar' becomes full.
 218 takeMVar :: MVar a -> IO a
 219 takeMVar (MVar mvar#) = IO $ \ s# -> takeMVar# mvar# s#
 220
 221 -- |Put a value into an 'MVar'.  If the 'MVar' is currently full,
 222 -- 'putMVar' will wait until it becomes empty.
 223 --
 224 -- If several threads are competing to fill the same 'MVar', one is
 225 -- chosen to continue at random when the 'MVar' becomes empty.
 226 putMVar  :: MVar a -> a -> IO ()
 227 putMVar (MVar mvar#) x = IO $ \ s# ->
 228     case putMVar# mvar# x s# of
 229         s2# -> (# s2#, () #)
 230
 231 -- |A non-blocking version of 'takeMVar'.  The 'tryTakeMVar' function
 232 -- returns immediately, with 'Nothing' if the 'MVar' was empty, or
 233 -- @'Just' a@ if the 'MVar' was full with contents @a@.  After 'tryTakeMVar',
 234 -- the 'MVar' is left empty.
 235 tryTakeMVar :: MVar a -> IO (Maybe a)
 236 tryTakeMVar (MVar m) = IO $ \ s ->
 237     case tryTakeMVar# m s of
 238         (# s, 0#, _ #) -> (# s, Nothing #)      -- MVar is empty
 239         (# s, _,  a #) -> (# s, Just a  #)      -- MVar is full
 240
 241 -- |A non-blocking version of 'putMVar'.  The 'tryPutMVar' function
 242 -- attempts to put the value @a@ into the 'MVar', returning 'True' if
 243 -- it was successful, or 'False' otherwise.
 244 tryPutMVar  :: MVar a -> a -> IO Bool
 245 tryPutMVar (MVar mvar#) x = IO $ \ s# ->
 246     case tryPutMVar# mvar# x s# of
 247         (# s, 0# #) -> (# s, False #)
 248         (# s, _  #) -> (# s, True #)
 249
 250 -- |Check whether a given 'MVar' is empty.
 251 --
 252 -- Notice that the boolean value returned  is just a snapshot of
 253 -- the state of the MVar. By the time you get to react on its result,
 254 -- the MVar may have been filled (or emptied) - so be extremely
 255 -- careful when using this operation.   Use 'tryTakeMVar' instead if possible.
 256 isEmptyMVar :: MVar a -> IO Bool
 257 isEmptyMVar (MVar mv#) = IO $ \ s# ->
 258     case isEmptyMVar# mv# s# of
 259         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 260
 261 -- |Add a finalizer to an 'MVar' (GHC only).  See "Foreign.ForeignPtr" and
 262 -- "System.Mem.Weak" for more about finalizers.
 263 addMVarFinalizer :: MVar a -> IO () -> IO ()
 264 addMVarFinalizer (MVar m) finalizer =
 265   IO $ \s -> case mkWeak# m () finalizer s of { (# s1, w #) -> (# s1, () #) }
 266 \end{code}
 267
 268
 269 %************************************************************************
 270 %*                                                                      *
 271 \subsection{Thread waiting}
 272 %*                                                                      *
 273 %************************************************************************
 274
 275 \begin{code}
 276 #ifdef mingw32_TARGET_OS
 277
 278 -- Note: threadDelay, threadWaitRead and threadWaitWrite aren't really functional
 279 -- on Win32, but left in there because lib code (still) uses them (the manner
 280 -- in which they're used doesn't cause problems on a Win32 platform though.)
 281
 282 asyncRead :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 283 asyncRead  (I# fd) (I# isSock) (I# len) (Ptr buf) = do
 284   (l, rc) <- IO (\s -> case asyncRead# fd isSock len buf s  of
 285                          (# s, len#, err# #) -> (# s, (I# len#, I# err#) #))
 286     -- special handling for Ctrl+C-aborted 'standard input' reads;
 287     -- see rts/win32/ConsoleHandler.c for details.
 288   if (l == 0 && rc == -2)
 289    then asyncRead (I# fd) (I# isSock) (I# len) (Ptr buf)
 290    else return (l,rc)
 291
 292 asyncWrite :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 293 asyncWrite  (I# fd) (I# isSock) (I# len) (Ptr buf) =
 294   IO $ \s -> case asyncWrite# fd isSock len buf s  of
 295                (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 296
 297 asyncDoProc :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
 298 asyncDoProc (FunPtr proc) (Ptr param) =
 299     -- the 'length' value is ignored; simplifies implementation of
 300     -- the async*# primops to have them all return the same result.
 301   IO $ \s -> case asyncDoProc# proc param s  of
 302                (# s, len#, err# #) -> (# s, I# err# #)
 303
 304 -- to aid the use of these primops by the IO Handle implementation,
 305 -- provide the following convenience funs:
 306
 307 -- this better be a pinned byte array!
 308 asyncReadBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 309 asyncReadBA fd isSock len off bufB =
 310   asyncRead fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 311
 312 asyncWriteBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 313 asyncWriteBA fd isSock len off bufB =
 314   asyncWrite fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 315
 316 #endif
 317
 318 -- -----------------------------------------------------------------------------
 319 -- Thread IO API
 320
 321 -- | Block the current thread until data is available to read on the
 322 -- given file descriptor (GHC only).
 323 threadWaitRead :: Fd -> IO ()
 324 threadWaitRead fd
 325 #ifndef mingw32_TARGET_OS
 326   | threaded  = waitForReadEvent fd
 327 #endif
 328   | otherwise = IO $ \s ->
 329         case fromIntegral fd of { I# fd# ->
 330         case waitRead# fd# s of { s -> (# s, () #)
 331         }}
 332
 333 -- | Block the current thread until data can be written to the
 334 -- given file descriptor (GHC only).
 335 threadWaitWrite :: Fd -> IO ()
 336 threadWaitWrite fd
 337 #ifndef mingw32_TARGET_OS
 338   | threaded  = waitForWriteEvent fd
 339 #endif
 340   | otherwise = IO $ \s ->
 341         case fromIntegral fd of { I# fd# ->
 342         case waitWrite# fd# s of { s -> (# s, () #)
 343         }}
 344
 345 -- | Suspends the current thread for a given number of microseconds
 346 -- (GHC only).
 347 --
 348 -- Note that the resolution used by the Haskell runtime system's
 349 -- internal timer is 1\/50 second, and 'threadDelay' will round its
 350 -- argument up to the nearest multiple of this resolution.
 351 --
 352 -- There is no guarantee that the thread will be rescheduled promptly
 353 -- when the delay has expired, but the thread will never continue to
 354 -- run /earlier/ than specified.
 355 --
 356 threadDelay :: Int -> IO ()
 357 threadDelay time
 358 #ifndef mingw32_TARGET_OS
 359   | threaded  = waitForDelayEvent time
 360 #else
 361   | threaded  = c_Sleep (fromIntegral (time `quot` 1000))
 362 #endif
 363   | otherwise = IO $ \s ->
 364         case fromIntegral time of { I# time# ->
 365         case delay# time# s of { s -> (# s, () #)
 366         }}
 367
 368 -- On Windows, we just make a safe call to 'Sleep' to implement threadDelay.
 369 #ifdef mingw32_TARGET_OS
 370 foreign import ccall safe "Sleep" c_Sleep :: CInt -> IO ()
 371 #endif
 372
 373 foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
 374
 375 -- ----------------------------------------------------------------------------
 376 -- Threaded RTS implementation of threadWaitRead, threadWaitWrite, threadDelay
 377
 378 -- In the threaded RTS, we employ a single IO Manager thread to wait
 379 -- for all outstanding IO requests (threadWaitRead,threadWaitWrite)
 380 -- and delays (threadDelay).
 381 --
 382 -- We can do this because in the threaded RTS the IO Manager can make
 383 -- a non-blocking call to select(), so we don't have to do select() in
 384 -- the scheduler as we have to in the non-threaded RTS.  We get performance
 385 -- benefits from doing it this way, because we only have to restart the select()
 386 -- when a new request arrives, rather than doing one select() each time
 387 -- around the scheduler loop.  Furthermore, the scheduler can be simplified
 388 -- by not having to check for completed IO requests.
 389
 390 -- Issues, possible problems:
 391 --
 392 --      - we might want bound threads to just do the blocking
 393 --        operation rather than communicating with the IO manager
 394 --        thread.  This would prevent simgle-threaded programs which do
 395 --        IO from requiring multiple OS threads.  However, it would also
 396 --        prevent bound threads waiting on IO from being killed or sent
 397 --        exceptions.
 398 --
 399 --      - Apprently exec() doesn't work on Linux in a multithreaded program.
 400 --        I couldn't repeat this.
 401 --
 402 --      - How do we handle signal delivery in the multithreaded RTS?
 403 --
 404 --      - forkProcess will kill the IO manager thread.  Let's just
 405 --        hope we don't need to do any blocking IO between fork & exec.
 406
 407 #ifndef mingw32_TARGET_OS
 408
 409 data IOReq
 410   = Read   {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 411   | Write  {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 412
 413 data DelayReq
 414   = Delay  {-# UNPACK #-} !Int {-# UNPACK #-} !(MVar ())
 415
 416 pendingEvents :: IORef [IOReq]
 417 pendingDelays :: IORef [DelayReq]
 418         -- could use a strict list or array here
 419 {-# NOINLINE pendingEvents #-}
 420 {-# NOINLINE pendingDelays #-}
 421 (pendingEvents,pendingDelays) = unsafePerformIO $ do
 422   startIOServiceThread
 423   reqs <- newIORef []
 424   dels <- newIORef []
 425   return (reqs, dels)
 426         -- the first time we schedule an IO request, the service thread
 427         -- will be created (cool, huh?)
 428
 429 startIOServiceThread :: IO ()
 430 startIOServiceThread = do
 431         allocaArray 2 $ \fds -> do
 432         throwErrnoIfMinus1 "startIOServiceThread" (c_pipe fds)
 433         rd_end <- peekElemOff fds 0
 434         wr_end <- peekElemOff fds 1
 435         writeIORef stick (fromIntegral wr_end)
 436         quickForkIO $ do
 437             allocaBytes sizeofFdSet   $ \readfds -> do
 438             allocaBytes sizeofFdSet   $ \writefds -> do
 439             allocaBytes sizeofTimeVal $ \timeval -> do
 440             service_loop (fromIntegral rd_end) readfds writefds timeval [] []
 441         return ()
 442
 443 -- XXX: move real forkIO here from Control.Concurrent?
 444 quickForkIO action = IO $ \s ->
 445    case (fork# action s) of (# s1, id #) -> (# s1, ThreadId id #)
 446
 447 service_loop
 448    :: Fd                -- listen to this for wakeup calls
 449    -> Ptr CFdSet
 450    -> Ptr CFdSet
 451    -> Ptr CTimeVal
 452    -> [IOReq]
 453    -> [DelayReq]
 454    -> IO ()
 455 service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do
 456
 457   -- pick up new IO requests
 458   new_reqs <- atomicModifyIORef pendingEvents (\a -> ([],a))
 459   let reqs = new_reqs ++ old_reqs
 460
 461   -- pick up new delay requests
 462   new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
 463   let  delays = foldr insertDelay old_delays new_delays
 464
 465   -- build the FDSets for select()
 466   fdZero readfds
 467   fdZero writefds
 468   fdSet wakeup readfds
 469   maxfd <- buildFdSets 0 readfds writefds reqs
 470
 471   -- check the current time and wake up any thread in threadDelay whose
 472   -- timeout has expired.  Also find the timeout value for the select() call.
 473   now <- getTicksOfDay
 474   (delays', timeout) <- getDelay now ptimeval delays
 475
 476   -- perform the select()
 477   let do_select = do
 478           res <- c_select ((max wakeup maxfd)+1) readfds writefds
 479                         nullPtr timeout
 480           if (res == -1)
 481              then do
 482                 err <- getErrno
 483                 if err == eINTR
 484                         then do_select
 485                         else return res
 486              else
 487                 return res
 488   res <- do_select
 489   -- ToDo: check result
 490
 491   old <- atomicModifyIORef prodding (\old -> (False,old))
 492   if old
 493         then alloca $ \p -> do c_read (fromIntegral wakeup) p 1; return ()
 494         else return ()
 495
 496   reqs' <- completeRequests reqs readfds writefds []
 497   service_loop wakeup readfds writefds ptimeval reqs' delays'
 498
 499 stick :: IORef Fd
 500 {-# NOINLINE stick #-}
 501 stick = unsafePerformIO (newIORef 0)
 502
 503 prodding :: IORef Bool
 504 {-# NOINLINE prodding #-}
 505 prodding = unsafePerformIO (newIORef False)
 506
 507 prodServiceThread :: IO ()
 508 prodServiceThread = do
 509   b <- atomicModifyIORef prodding (\old -> (True,old)) -- compare & swap!
 510   if (not b)
 511         then do
 512           fd <- readIORef stick
 513           with 42 $ \pbuf -> do c_write (fromIntegral fd) pbuf 1; return ()
 514         else
 515           return ()
 516
 517 -- -----------------------------------------------------------------------------
 518 -- IO requests
 519
 520 buildFdSets maxfd readfds writefds [] = return maxfd
 521 buildFdSets maxfd readfds writefds (Read fd m : reqs) = do
 522   fdSet fd readfds
 523   buildFdSets (max maxfd fd) readfds writefds reqs
 524 buildFdSets maxfd readfds writefds (Write fd m : reqs) = do
 525   fdSet fd writefds
 526   buildFdSets (max maxfd fd) readfds writefds reqs
 527
 528 completeRequests [] _ _ reqs' = return reqs'
 529 completeRequests (Read fd m : reqs) readfds writefds reqs' = do
 530   b <- fdIsSet fd readfds
 531   if b /= 0
 532     then do putMVar m (); completeRequests reqs readfds writefds reqs'
 533     else completeRequests reqs readfds writefds (Read fd m : reqs')
 534 completeRequests (Write fd m : reqs) readfds writefds reqs' = do
 535   b <- fdIsSet fd writefds
 536   if b /= 0
 537     then do putMVar m (); completeRequests reqs readfds writefds reqs'
 538     else completeRequests reqs readfds writefds (Write fd m : reqs')
 539
 540 waitForReadEvent :: Fd -> IO ()
 541 waitForReadEvent fd = do
 542   m <- newEmptyMVar
 543   atomicModifyIORef pendingEvents (\xs -> (Read fd m : xs, ()))
 544   prodServiceThread
 545   takeMVar m
 546
 547 waitForWriteEvent :: Fd -> IO ()
 548 waitForWriteEvent fd = do
 549   m <- newEmptyMVar
 550   atomicModifyIORef pendingEvents (\xs -> (Write fd m : xs, ()))
 551   prodServiceThread
 552   takeMVar m
 553
 554 -- XXX: move into GHC.IOBase from Data.IORef?
 555 atomicModifyIORef :: IORef a -> (a -> (a,b)) -> IO b
 556 atomicModifyIORef (IORef (STRef r#)) f = IO $ \s -> atomicModifyMutVar# r# f s
 557
 558 -- -----------------------------------------------------------------------------
 559 -- Delays
 560
 561 waitForDelayEvent :: Int -> IO ()
 562 waitForDelayEvent usecs = do
 563   m <- newEmptyMVar
 564   now <- getTicksOfDay
 565   let target = now + usecs `quot` tick_usecs
 566   atomicModifyIORef pendingDelays (\xs -> (Delay target m : xs, ()))
 567   prodServiceThread
 568   takeMVar m
 569
 570 -- Walk the queue of pending delays, waking up any that have passed
 571 -- and return the smallest delay to wait for.  The queue of pending
 572 -- delays is kept ordered.
 573 getDelay :: Ticks -> Ptr CTimeVal -> [DelayReq] -> IO ([DelayReq], Ptr CTimeVal)
 574 getDelay now ptimeval [] = return ([],nullPtr)
 575 getDelay now ptimeval all@(Delay time m : rest)
 576   | now >= time = do
 577         putMVar m ()
 578         getDelay now ptimeval rest
 579   | otherwise = do
 580         setTimevalTicks ptimeval (time - now)
 581         return (all,ptimeval)
 582
 583 insertDelay :: DelayReq -> [DelayReq] -> [DelayReq]
 584 insertDelay d@(Delay time m) [] = [d]
 585 insertDelay d1@(Delay time m) ds@(d2@(Delay time' m') : rest)
 586   | time <= time' = d1 : ds
 587   | otherwise     = d2 : insertDelay d1 rest
 588
 589 type Ticks = Int
 590 tick_freq  = 50 :: Ticks  -- accuracy of threadDelay (ticks per sec)
 591 tick_usecs = 1000000 `quot` tick_freq :: Int
 592
 593 newtype CTimeVal = CTimeVal ()
 594
 595 foreign import ccall unsafe "sizeofTimeVal"
 596   sizeofTimeVal :: Int
 597
 598 foreign import ccall unsafe "getTicksOfDay"
 599   getTicksOfDay :: IO Ticks
 600
 601 foreign import ccall unsafe "setTimevalTicks"
 602   setTimevalTicks :: Ptr CTimeVal -> Ticks -> IO ()
 603
 604 -- ----------------------------------------------------------------------------
 605 -- select() interface
 606
 607 -- ToDo: move to System.Posix.Internals?
 608
 609 newtype CFdSet = CFdSet ()
 610
 611 foreign import ccall safe "select"
 612   c_select :: Fd -> Ptr CFdSet -> Ptr CFdSet -> Ptr CFdSet -> Ptr CTimeVal
 613            -> IO CInt
 614
 615 foreign import ccall unsafe "hsFD_CLR"
 616   fdClr :: Fd -> Ptr CFdSet -> IO ()
 617
 618 foreign import ccall unsafe "hsFD_ISSET"
 619   fdIsSet :: Fd -> Ptr CFdSet -> IO CInt
 620
 621 foreign import ccall unsafe "hsFD_SET"
 622   fdSet :: Fd -> Ptr CFdSet -> IO ()
 623
 624 foreign import ccall unsafe "hsFD_ZERO"
 625   fdZero :: Ptr CFdSet -> IO ()
 626
 627 foreign import ccall unsafe "sizeof_fd_set"
 628   sizeofFdSet :: Int
 629
 630 #endif
 631 \end{code}