GHC/Conc.lhs

   1 \begin{code}
   2 {-# OPTIONS_GHC -fno-implicit-prelude #-}
   3 {-# OPTIONS_HADDOCK not-home #-}
   4 -----------------------------------------------------------------------------
   5 -- |
   6 -- Module      :  GHC.Conc
   7 -- Copyright   :  (c) The University of Glasgow, 1994-2002
   8 -- License     :  see libraries/base/LICENSE
   9 --
  10 -- Maintainer  :  cvs-ghc@haskell.org
  11 -- Stability   :  internal
  12 -- Portability :  non-portable (GHC extensions)
  13 --
  14 -- Basic concurrency stuff.
  15 --
  16 -----------------------------------------------------------------------------
  17
  18 -- No: #hide, because bits of this module are exposed by the stm package.
  19 -- However, we don't want this module to be the home location for the
  20 -- bits it exports, we'd rather have Control.Concurrent and the other
  21 -- higher level modules be the home.  Hence:
  22
  23 #include "Typeable.h"
  24
  25 -- #not-home
  26 module GHC.Conc
  27         ( ThreadId(..)
  28
  29         -- * Forking and suchlike
  30         , forkIO        -- :: IO a -> IO ThreadId
  31         , forkOnIO      -- :: Int -> IO a -> IO ThreadId
  32         , numCapabilities -- :: Int
  33         , childHandler  -- :: Exception -> IO ()
  34         , myThreadId    -- :: IO ThreadId
  35         , killThread    -- :: ThreadId -> IO ()
  36         , throwTo       -- :: ThreadId -> Exception -> IO ()
  37         , par           -- :: a -> b -> b
  38         , pseq          -- :: a -> b -> b
  39         , yield         -- :: IO ()
  40         , labelThread   -- :: ThreadId -> String -> IO ()
  41
  42         -- * Waiting
  43         , threadDelay           -- :: Int -> IO ()
  44         , registerDelay         -- :: Int -> IO (TVar Bool)
  45         , threadWaitRead        -- :: Int -> IO ()
  46         , threadWaitWrite       -- :: Int -> IO ()
  47
  48         -- * MVars
  49         , MVar(..)
  50         , newMVar       -- :: a -> IO (MVar a)
  51         , newEmptyMVar  -- :: IO (MVar a)
  52         , takeMVar      -- :: MVar a -> IO a
  53         , putMVar       -- :: MVar a -> a -> IO ()
  54         , tryTakeMVar   -- :: MVar a -> IO (Maybe a)
  55         , tryPutMVar    -- :: MVar a -> a -> IO Bool
  56         , isEmptyMVar   -- :: MVar a -> IO Bool
  57         , addMVarFinalizer -- :: MVar a -> IO () -> IO ()
  58
  59         -- * TVars
  60         , STM(..)
  61         , atomically    -- :: STM a -> IO a
  62         , retry         -- :: STM a
  63         , orElse        -- :: STM a -> STM a -> STM a
  64         , catchSTM      -- :: STM a -> (Exception -> STM a) -> STM a
  65         , alwaysSucceeds -- :: STM a -> STM ()
  66         , always        -- :: STM Bool -> STM ()
  67         , TVar(..)
  68         , newTVar       -- :: a -> STM (TVar a)
  69         , newTVarIO     -- :: a -> STM (TVar a)
  70         , readTVar      -- :: TVar a -> STM a
  71         , writeTVar     -- :: a -> TVar a -> STM ()
  72         , unsafeIOToSTM -- :: IO a -> STM a
  73
  74         -- * Miscellaneous
  75 #ifdef mingw32_HOST_OS
  76         , asyncRead     -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
  77         , asyncWrite    -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
  78         , asyncDoProc   -- :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
  79
  80         , asyncReadBA   -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
  81         , asyncWriteBA  -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
  82 #endif
  83
  84 #ifndef mingw32_HOST_OS
  85         , signalHandlerLock
  86 #endif
  87
  88         , ensureIOManagerIsRunning
  89
  90 #ifdef mingw32_HOST_OS
  91         , ConsoleEvent(..)
  92         , win32ConsoleHandler
  93         , toWin32ConsoleEvent
  94 #endif
  95         ) where
  96
  97 import System.Posix.Types
  98 #ifndef mingw32_HOST_OS
  99 import System.Posix.Internals
 100 #endif
 101 import Foreign
 102 import Foreign.C
 103
 104 #ifndef __HADDOCK__
 105 import {-# SOURCE #-} GHC.TopHandler ( reportError, reportStackOverflow )
 106 #endif
 107
 108 import Data.Maybe
 109
 110 import GHC.Base
 111 import GHC.IOBase
 112 import GHC.Num          ( Num(..) )
 113 import GHC.Real         ( fromIntegral, div )
 114 #ifndef mingw32_HOST_OS
 115 import GHC.Base         ( Int(..) )
 116 #endif
 117 #ifdef mingw32_HOST_OS
 118 import GHC.Read         ( Read )
 119 import GHC.Enum         ( Enum )
 120 #endif
 121 import GHC.Exception
 122 import GHC.Pack         ( packCString# )
 123 import GHC.Ptr          ( Ptr(..), plusPtr, FunPtr(..) )
 124 import GHC.STRef
 125 import GHC.Show         ( Show(..), showString )
 126 import Data.Typeable
 127
 128 infixr 0 `par`, `pseq`
 129 \end{code}
 130
 131 %************************************************************************
 132 %*                                                                      *
 133 \subsection{@ThreadId@, @par@, and @fork@}
 134 %*                                                                      *
 135 %************************************************************************
 136
 137 \begin{code}
 138 data ThreadId = ThreadId ThreadId# deriving( Typeable )
 139 -- ToDo: data ThreadId = ThreadId (Weak ThreadId#)
 140 -- But since ThreadId# is unlifted, the Weak type must use open
 141 -- type variables.
 142 {- ^
 143 A 'ThreadId' is an abstract type representing a handle to a thread.
 144 'ThreadId' is an instance of 'Eq', 'Ord' and 'Show', where
 145 the 'Ord' instance implements an arbitrary total ordering over
 146 'ThreadId's. The 'Show' instance lets you convert an arbitrary-valued
 147 'ThreadId' to string form; showing a 'ThreadId' value is occasionally
 148 useful when debugging or diagnosing the behaviour of a concurrent
 149 program.
 150
 151 /Note/: in GHC, if you have a 'ThreadId', you essentially have
 152 a pointer to the thread itself.  This means the thread itself can\'t be
 153 garbage collected until you drop the 'ThreadId'.
 154 This misfeature will hopefully be corrected at a later date.
 155
 156 /Note/: Hugs does not provide any operations on other threads;
 157 it defines 'ThreadId' as a synonym for ().
 158 -}
 159
 160 instance Show ThreadId where
 161    showsPrec d t =
 162         showString "ThreadId " .
 163         showsPrec d (getThreadId (id2TSO t))
 164
 165 foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> CInt
 166
 167 id2TSO :: ThreadId -> ThreadId#
 168 id2TSO (ThreadId t) = t
 169
 170 foreign import ccall unsafe "cmp_thread" cmp_thread :: ThreadId# -> ThreadId# -> CInt
 171 -- Returns -1, 0, 1
 172
 173 cmpThread :: ThreadId -> ThreadId -> Ordering
 174 cmpThread t1 t2 =
 175    case cmp_thread (id2TSO t1) (id2TSO t2) of
 176       -1 -> LT
 177       0  -> EQ
 178       _  -> GT -- must be 1
 179
 180 instance Eq ThreadId where
 181    t1 == t2 =
 182       case t1 `cmpThread` t2 of
 183          EQ -> True
 184          _  -> False
 185
 186 instance Ord ThreadId where
 187    compare = cmpThread
 188
 189 {- |
 190 Sparks off a new thread to run the 'IO' computation passed as the
 191 first argument, and returns the 'ThreadId' of the newly created
 192 thread.
 193
 194 The new thread will be a lightweight thread; if you want to use a foreign
 195 library that uses thread-local storage, use 'Control.Concurrent.forkOS' instead.
 196
 197 GHC note: the new thread inherits the /blocked/ state of the parent
 198 (see 'Control.Exception.block').
 199 -}
 200 forkIO :: IO () -> IO ThreadId
 201 forkIO action = IO $ \ s ->
 202    case (fork# action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
 203  where
 204   action_plus = catchException action childHandler
 205
 206 {- |
 207 Like 'forkIO', but lets you specify on which CPU the thread is
 208 created.  Unlike a `forkIO` thread, a thread created by `forkOnIO`
 209 will stay on the same CPU for its entire lifetime (`forkIO` threads
 210 can migrate between CPUs according to the scheduling policy).
 211 `forkOnIO` is useful for overriding the scheduling policy when you
 212 know in advance how best to distribute the threads.
 213
 214 The `Int` argument specifies the CPU number; it is interpreted modulo
 215 'numCapabilities' (note that it actually specifies a capability number
 216 rather than a CPU number, but to a first approximation the two are
 217 equivalent).
 218 -}
 219 forkOnIO :: Int -> IO () -> IO ThreadId
 220 forkOnIO (I# cpu) action = IO $ \ s ->
 221    case (forkOn# cpu action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
 222  where
 223   action_plus = catchException action childHandler
 224
 225 -- | the value passed to the @+RTS -N@ flag.  This is the number of
 226 -- Haskell threads that can run truly simultaneously at any given
 227 -- time, and is typically set to the number of physical CPU cores on
 228 -- the machine.
 229 numCapabilities :: Int
 230 numCapabilities = unsafePerformIO $  do
 231                     n <- peek n_capabilities
 232                     return (fromIntegral n)
 233
 234 foreign import ccall "&n_capabilities" n_capabilities :: Ptr CInt
 235
 236 childHandler :: Exception -> IO ()
 237 childHandler err = catchException (real_handler err) childHandler
 238
 239 real_handler :: Exception -> IO ()
 240 real_handler ex =
 241   case ex of
 242         -- ignore thread GC and killThread exceptions:
 243         BlockedOnDeadMVar            -> return ()
 244         BlockedIndefinitely          -> return ()
 245         AsyncException ThreadKilled  -> return ()
 246
 247         -- report all others:
 248         AsyncException StackOverflow -> reportStackOverflow
 249         other       -> reportError other
 250
 251 {- | 'killThread' terminates the given thread (GHC only).
 252 Any work already done by the thread isn\'t
 253 lost: the computation is suspended until required by another thread.
 254 The memory used by the thread will be garbage collected if it isn\'t
 255 referenced from anywhere.  The 'killThread' function is defined in
 256 terms of 'throwTo':
 257
 258 > killThread tid = throwTo tid (AsyncException ThreadKilled)
 259
 260 -}
 261 killThread :: ThreadId -> IO ()
 262 killThread tid = throwTo tid (AsyncException ThreadKilled)
 263
 264 {- | 'throwTo' raises an arbitrary exception in the target thread (GHC only).
 265
 266 'throwTo' does not return until the exception has been raised in the
 267 target thread.
 268 The calling thread can thus be certain that the target
 269 thread has received the exception.  This is a useful property to know
 270 when dealing with race conditions: eg. if there are two threads that
 271 can kill each other, it is guaranteed that only one of the threads
 272 will get to kill the other.
 273
 274 If the target thread is currently making a foreign call, then the
 275 exception will not be raised (and hence 'throwTo' will not return)
 276 until the call has completed.  This is the case regardless of whether
 277 the call is inside a 'block' or not.
 278
 279 Important note: the behaviour of 'throwTo' differs from that described in
 280 the paper \"Asynchronous exceptions in Haskell\"
 281 (<http://research.microsoft.com/~simonpj/Papers/asynch-exns.htm>).
 282 In the paper, 'throwTo' is non-blocking; but the library implementation adopts
 283 a more synchronous design in which 'throwTo' does not return until the exception
 284 is received by the target thread.  The trade-off is discussed in Section 8 of the paper.
 285 Like any blocking operation, 'throwTo' is therefore interruptible (see Section 4.3 of
 286 the paper).
 287
 288 There is currently no guarantee that the exception delivered by 'throwTo' will be
 289 delivered at the first possible opportunity.  In particular, if a thread may
 290 unblock and then re-block exceptions (using 'unblock' and 'block') without receiving
 291 a pending 'throwTo'.  This is arguably undesirable behaviour.
 292
 293  -}
 294 throwTo :: ThreadId -> Exception -> IO ()
 295 throwTo (ThreadId id) ex = IO $ \ s ->
 296    case (killThread# id ex s) of s1 -> (# s1, () #)
 297
 298 -- | Returns the 'ThreadId' of the calling thread (GHC only).
 299 myThreadId :: IO ThreadId
 300 myThreadId = IO $ \s ->
 301    case (myThreadId# s) of (# s1, id #) -> (# s1, ThreadId id #)
 302
 303
 304 -- |The 'yield' action allows (forces, in a co-operative multitasking
 305 -- implementation) a context-switch to any other currently runnable
 306 -- threads (if any), and is occasionally useful when implementing
 307 -- concurrency abstractions.
 308 yield :: IO ()
 309 yield = IO $ \s ->
 310    case (yield# s) of s1 -> (# s1, () #)
 311
 312 {- | 'labelThread' stores a string as identifier for this thread if
 313 you built a RTS with debugging support. This identifier will be used in
 314 the debugging output to make distinction of different threads easier
 315 (otherwise you only have the thread state object\'s address in the heap).
 316
 317 Other applications like the graphical Concurrent Haskell Debugger
 318 (<http://www.informatik.uni-kiel.de/~fhu/chd/>) may choose to overload
 319 'labelThread' for their purposes as well.
 320 -}
 321
 322 labelThread :: ThreadId -> String -> IO ()
 323 labelThread (ThreadId t) str = IO $ \ s ->
 324    let ps  = packCString# str
 325        adr = byteArrayContents# ps in
 326      case (labelThread# t adr s) of s1 -> (# s1, () #)
 327
 328 --      Nota Bene: 'pseq' used to be 'seq'
 329 --                 but 'seq' is now defined in PrelGHC
 330 --
 331 -- "pseq" is defined a bit weirdly (see below)
 332 --
 333 -- The reason for the strange "lazy" call is that
 334 -- it fools the compiler into thinking that pseq  and par are non-strict in
 335 -- their second argument (even if it inlines pseq at the call site).
 336 -- If it thinks pseq is strict in "y", then it often evaluates
 337 -- "y" before "x", which is totally wrong.
 338
 339 {-# INLINE pseq  #-}
 340 pseq :: a -> b -> b
 341 pseq  x y = x `seq` lazy y
 342
 343 {-# INLINE par  #-}
 344 par :: a -> b -> b
 345 par  x y = case (par# x) of { _ -> lazy y }
 346 \end{code}
 347
 348
 349 %************************************************************************
 350 %*                                                                      *
 351 \subsection[stm]{Transactional heap operations}
 352 %*                                                                      *
 353 %************************************************************************
 354
 355 TVars are shared memory locations which support atomic memory
 356 transactions.
 357
 358 \begin{code}
 359 -- |A monad supporting atomic memory transactions.
 360 newtype STM a = STM (State# RealWorld -> (# State# RealWorld, a #))
 361
 362 unSTM :: STM a -> (State# RealWorld -> (# State# RealWorld, a #))
 363 unSTM (STM a) = a
 364
 365 INSTANCE_TYPEABLE1(STM,stmTc,"STM")
 366
 367 instance  Functor STM where
 368    fmap f x = x >>= (return . f)
 369
 370 instance  Monad STM  where
 371     {-# INLINE return #-}
 372     {-# INLINE (>>)   #-}
 373     {-# INLINE (>>=)  #-}
 374     m >> k      = thenSTM m k
 375     return x    = returnSTM x
 376     m >>= k     = bindSTM m k
 377
 378 bindSTM :: STM a -> (a -> STM b) -> STM b
 379 bindSTM (STM m) k = STM ( \s ->
 380   case m s of
 381     (# new_s, a #) -> unSTM (k a) new_s
 382   )
 383
 384 thenSTM :: STM a -> STM b -> STM b
 385 thenSTM (STM m) k = STM ( \s ->
 386   case m s of
 387     (# new_s, a #) -> unSTM k new_s
 388   )
 389
 390 returnSTM :: a -> STM a
 391 returnSTM x = STM (\s -> (# s, x #))
 392
 393 -- | Unsafely performs IO in the STM monad.
 394 unsafeIOToSTM :: IO a -> STM a
 395 unsafeIOToSTM (IO m) = STM m
 396
 397 -- |Perform a series of STM actions atomically.
 398 --
 399 -- You cannot use 'atomically' inside an 'unsafePerformIO' or 'unsafeInterleaveIO'.
 400 -- Any attempt to do so will result in a runtime error.  (Reason: allowing
 401 -- this would effectively allow a transaction inside a transaction, depending
 402 -- on exactly when the thunk is evaluated.)
 403 --
 404 -- However, see 'newTVarIO', which can be called inside 'unsafePerformIO',
 405 -- and which allows top-level TVars to be allocated.
 406
 407 atomically :: STM a -> IO a
 408 atomically (STM m) = IO (\s -> (atomically# m) s )
 409
 410 -- |Retry execution of the current memory transaction because it has seen
 411 -- values in TVars which mean that it should not continue (e.g. the TVars
 412 -- represent a shared buffer that is now empty).  The implementation may
 413 -- block the thread until one of the TVars that it has read from has been
 414 -- udpated. (GHC only)
 415 retry :: STM a
 416 retry = STM $ \s# -> retry# s#
 417
 418 -- |Compose two alternative STM actions (GHC only).  If the first action
 419 -- completes without retrying then it forms the result of the orElse.
 420 -- Otherwise, if the first action retries, then the second action is
 421 -- tried in its place.  If both actions retry then the orElse as a
 422 -- whole retries.
 423 orElse :: STM a -> STM a -> STM a
 424 orElse (STM m) e = STM $ \s -> catchRetry# m (unSTM e) s
 425
 426 -- |Exception handling within STM actions.
 427 catchSTM :: STM a -> (Exception -> STM a) -> STM a
 428 catchSTM (STM m) k = STM $ \s -> catchSTM# m (\ex -> unSTM (k ex)) s
 429
 430 -- | Low-level primitive on which always and alwaysSucceeds are built.
 431 -- checkInv differs form these in that (i) the invariant is not
 432 -- checked when checkInv is called, only at the end of this and
 433 -- subsequent transcations, (ii) the invariant failure is indicated
 434 -- by raising an exception.
 435 checkInv :: STM a -> STM ()
 436 checkInv (STM m) = STM (\s -> (check# m) s)
 437
 438 -- | alwaysSucceeds adds a new invariant that must be true when passed
 439 -- to alwaysSucceeds, at the end of the current transaction, and at
 440 -- the end of every subsequent transaction.  If it fails at any
 441 -- of those points then the transaction violating it is aborted
 442 -- and the exception raised by the invariant is propagated.
 443 alwaysSucceeds :: STM a -> STM ()
 444 alwaysSucceeds i = do ( do i ; retry ) `orElse` ( return () )
 445                       checkInv i
 446
 447 -- | always is a variant of alwaysSucceeds in which the invariant is
 448 -- expressed as an STM Bool action that must return True.  Returning
 449 -- False or raising an exception are both treated as invariant failures.
 450 always :: STM Bool -> STM ()
 451 always i = alwaysSucceeds ( do v <- i
 452                                if (v) then return () else ( error "Transacional invariant violation" ) )
 453
 454 -- |Shared memory locations that support atomic memory transactions.
 455 data TVar a = TVar (TVar# RealWorld a)
 456
 457 INSTANCE_TYPEABLE1(TVar,tvarTc,"TVar")
 458
 459 instance Eq (TVar a) where
 460         (TVar tvar1#) == (TVar tvar2#) = sameTVar# tvar1# tvar2#
 461
 462 -- |Create a new TVar holding a value supplied
 463 newTVar :: a -> STM (TVar a)
 464 newTVar val = STM $ \s1# ->
 465     case newTVar# val s1# of
 466          (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
 467
 468 -- |@IO@ version of 'newTVar'.  This is useful for creating top-level
 469 -- 'TVar's using 'System.IO.Unsafe.unsafePerformIO', because using
 470 -- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't
 471 -- possible.
 472 newTVarIO :: a -> IO (TVar a)
 473 newTVarIO val = IO $ \s1# ->
 474     case newTVar# val s1# of
 475          (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
 476
 477 -- |Return the current value stored in a TVar
 478 readTVar :: TVar a -> STM a
 479 readTVar (TVar tvar#) = STM $ \s# -> readTVar# tvar# s#
 480
 481 -- |Write the supplied value into a TVar
 482 writeTVar :: TVar a -> a -> STM ()
 483 writeTVar (TVar tvar#) val = STM $ \s1# ->
 484     case writeTVar# tvar# val s1# of
 485          s2# -> (# s2#, () #)
 486
 487 \end{code}
 488
 489 %************************************************************************
 490 %*                                                                      *
 491 \subsection[mvars]{M-Structures}
 492 %*                                                                      *
 493 %************************************************************************
 494
 495 M-Vars are rendezvous points for concurrent threads.  They begin
 496 empty, and any attempt to read an empty M-Var blocks.  When an M-Var
 497 is written, a single blocked thread may be freed.  Reading an M-Var
 498 toggles its state from full back to empty.  Therefore, any value
 499 written to an M-Var may only be read once.  Multiple reads and writes
 500 are allowed, but there must be at least one read between any two
 501 writes.
 502
 503 \begin{code}
 504 --Defined in IOBase to avoid cycle: data MVar a = MVar (SynchVar# RealWorld a)
 505
 506 -- |Create an 'MVar' which is initially empty.
 507 newEmptyMVar  :: IO (MVar a)
 508 newEmptyMVar = IO $ \ s# ->
 509     case newMVar# s# of
 510          (# s2#, svar# #) -> (# s2#, MVar svar# #)
 511
 512 -- |Create an 'MVar' which contains the supplied value.
 513 newMVar :: a -> IO (MVar a)
 514 newMVar value =
 515     newEmptyMVar        >>= \ mvar ->
 516     putMVar mvar value  >>
 517     return mvar
 518
 519 -- |Return the contents of the 'MVar'.  If the 'MVar' is currently
 520 -- empty, 'takeMVar' will wait until it is full.  After a 'takeMVar',
 521 -- the 'MVar' is left empty.
 522 --
 523 -- There are two further important properties of 'takeMVar':
 524 --
 525 --   * 'takeMVar' is single-wakeup.  That is, if there are multiple
 526 --     threads blocked in 'takeMVar', and the 'MVar' becomes full,
 527 --     only one thread will be woken up.  The runtime guarantees that
 528 --     the woken thread completes its 'takeMVar' operation.
 529 --
 530 --   * When multiple threads are blocked on an 'MVar', they are
 531 --     woken up in FIFO order.  This is useful for providing
 532 --     fairness properties of abstractions built using 'MVar's.
 533 --
 534 takeMVar :: MVar a -> IO a
 535 takeMVar (MVar mvar#) = IO $ \ s# -> takeMVar# mvar# s#
 536
 537 -- |Put a value into an 'MVar'.  If the 'MVar' is currently full,
 538 -- 'putMVar' will wait until it becomes empty.
 539 --
 540 -- There are two further important properties of 'putMVar':
 541 --
 542 --   * 'putMVar' is single-wakeup.  That is, if there are multiple
 543 --     threads blocked in 'putMVar', and the 'MVar' becomes empty,
 544 --     only one thread will be woken up.  The runtime guarantees that
 545 --     the woken thread completes its 'putMVar' operation.
 546 --
 547 --   * When multiple threads are blocked on an 'MVar', they are
 548 --     woken up in FIFO order.  This is useful for providing
 549 --     fairness properties of abstractions built using 'MVar's.
 550 --
 551 putMVar  :: MVar a -> a -> IO ()
 552 putMVar (MVar mvar#) x = IO $ \ s# ->
 553     case putMVar# mvar# x s# of
 554         s2# -> (# s2#, () #)
 555
 556 -- |A non-blocking version of 'takeMVar'.  The 'tryTakeMVar' function
 557 -- returns immediately, with 'Nothing' if the 'MVar' was empty, or
 558 -- @'Just' a@ if the 'MVar' was full with contents @a@.  After 'tryTakeMVar',
 559 -- the 'MVar' is left empty.
 560 tryTakeMVar :: MVar a -> IO (Maybe a)
 561 tryTakeMVar (MVar m) = IO $ \ s ->
 562     case tryTakeMVar# m s of
 563         (# s, 0#, _ #) -> (# s, Nothing #)      -- MVar is empty
 564         (# s, _,  a #) -> (# s, Just a  #)      -- MVar is full
 565
 566 -- |A non-blocking version of 'putMVar'.  The 'tryPutMVar' function
 567 -- attempts to put the value @a@ into the 'MVar', returning 'True' if
 568 -- it was successful, or 'False' otherwise.
 569 tryPutMVar  :: MVar a -> a -> IO Bool
 570 tryPutMVar (MVar mvar#) x = IO $ \ s# ->
 571     case tryPutMVar# mvar# x s# of
 572         (# s, 0# #) -> (# s, False #)
 573         (# s, _  #) -> (# s, True #)
 574
 575 -- |Check whether a given 'MVar' is empty.
 576 --
 577 -- Notice that the boolean value returned  is just a snapshot of
 578 -- the state of the MVar. By the time you get to react on its result,
 579 -- the MVar may have been filled (or emptied) - so be extremely
 580 -- careful when using this operation.   Use 'tryTakeMVar' instead if possible.
 581 isEmptyMVar :: MVar a -> IO Bool
 582 isEmptyMVar (MVar mv#) = IO $ \ s# ->
 583     case isEmptyMVar# mv# s# of
 584         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 585
 586 -- |Add a finalizer to an 'MVar' (GHC only).  See "Foreign.ForeignPtr" and
 587 -- "System.Mem.Weak" for more about finalizers.
 588 addMVarFinalizer :: MVar a -> IO () -> IO ()
 589 addMVarFinalizer (MVar m) finalizer =
 590   IO $ \s -> case mkWeak# m () finalizer s of { (# s1, w #) -> (# s1, () #) }
 591
 592 withMVar :: MVar a -> (a -> IO b) -> IO b
 593 withMVar m io =
 594   block $ do
 595     a <- takeMVar m
 596     b <- catchException (unblock (io a))
 597             (\e -> do putMVar m a; throw e)
 598     putMVar m a
 599     return b
 600 \end{code}
 601
 602
 603 %************************************************************************
 604 %*                                                                      *
 605 \subsection{Thread waiting}
 606 %*                                                                      *
 607 %************************************************************************
 608
 609 \begin{code}
 610 #ifdef mingw32_HOST_OS
 611
 612 -- Note: threadWaitRead and threadWaitWrite aren't really functional
 613 -- on Win32, but left in there because lib code (still) uses them (the manner
 614 -- in which they're used doesn't cause problems on a Win32 platform though.)
 615
 616 asyncRead :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 617 asyncRead  (I# fd) (I# isSock) (I# len) (Ptr buf) =
 618   IO $ \s -> case asyncRead# fd isSock len buf s of
 619                (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 620
 621 asyncWrite :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 622 asyncWrite  (I# fd) (I# isSock) (I# len) (Ptr buf) =
 623   IO $ \s -> case asyncWrite# fd isSock len buf s of
 624                (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 625
 626 asyncDoProc :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
 627 asyncDoProc (FunPtr proc) (Ptr param) =
 628     -- the 'length' value is ignored; simplifies implementation of
 629     -- the async*# primops to have them all return the same result.
 630   IO $ \s -> case asyncDoProc# proc param s  of
 631                (# s, len#, err# #) -> (# s, I# err# #)
 632
 633 -- to aid the use of these primops by the IO Handle implementation,
 634 -- provide the following convenience funs:
 635
 636 -- this better be a pinned byte array!
 637 asyncReadBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 638 asyncReadBA fd isSock len off bufB =
 639   asyncRead fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 640
 641 asyncWriteBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 642 asyncWriteBA fd isSock len off bufB =
 643   asyncWrite fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 644
 645 #endif
 646
 647 -- -----------------------------------------------------------------------------
 648 -- Thread IO API
 649
 650 -- | Block the current thread until data is available to read on the
 651 -- given file descriptor (GHC only).
 652 threadWaitRead :: Fd -> IO ()
 653 threadWaitRead fd
 654 #ifndef mingw32_HOST_OS
 655   | threaded  = waitForReadEvent fd
 656 #endif
 657   | otherwise = IO $ \s ->
 658         case fromIntegral fd of { I# fd# ->
 659         case waitRead# fd# s of { s -> (# s, () #)
 660         }}
 661
 662 -- | Block the current thread until data can be written to the
 663 -- given file descriptor (GHC only).
 664 threadWaitWrite :: Fd -> IO ()
 665 threadWaitWrite fd
 666 #ifndef mingw32_HOST_OS
 667   | threaded  = waitForWriteEvent fd
 668 #endif
 669   | otherwise = IO $ \s ->
 670         case fromIntegral fd of { I# fd# ->
 671         case waitWrite# fd# s of { s -> (# s, () #)
 672         }}
 673
 674 -- | Suspends the current thread for a given number of microseconds
 675 -- (GHC only).
 676 --
 677 -- There is no guarantee that the thread will be rescheduled promptly
 678 -- when the delay has expired, but the thread will never continue to
 679 -- run /earlier/ than specified.
 680 --
 681 threadDelay :: Int -> IO ()
 682 threadDelay time
 683   | threaded  = waitForDelayEvent time
 684   | otherwise = IO $ \s ->
 685         case fromIntegral time of { I# time# ->
 686         case delay# time# s of { s -> (# s, () #)
 687         }}
 688
 689
 690 -- | Set the value of returned TVar to True after a given number of
 691 -- microseconds. The caveats associated with threadDelay also apply.
 692 --
 693 registerDelay :: Int -> IO (TVar Bool)
 694 registerDelay usecs
 695   | threaded = waitForDelayEventSTM usecs
 696   | otherwise = error "registerDelay: requires -threaded"
 697
 698 foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
 699
 700 waitForDelayEvent :: Int -> IO ()
 701 waitForDelayEvent usecs = do
 702   m <- newEmptyMVar
 703   target <- calculateTarget usecs
 704   atomicModifyIORef pendingDelays (\xs -> (Delay target m : xs, ()))
 705   prodServiceThread
 706   takeMVar m
 707
 708 -- Delays for use in STM
 709 waitForDelayEventSTM :: Int -> IO (TVar Bool)
 710 waitForDelayEventSTM usecs = do
 711    t <- atomically $ newTVar False
 712    target <- calculateTarget usecs
 713    atomicModifyIORef pendingDelays (\xs -> (DelaySTM target t : xs, ()))
 714    prodServiceThread
 715    return t
 716
 717 calculateTarget :: Int -> IO USecs
 718 calculateTarget usecs = do
 719     now <- getUSecOfDay
 720     return $ now + (fromIntegral usecs)
 721
 722
 723 -- ----------------------------------------------------------------------------
 724 -- Threaded RTS implementation of threadWaitRead, threadWaitWrite, threadDelay
 725
 726 -- In the threaded RTS, we employ a single IO Manager thread to wait
 727 -- for all outstanding IO requests (threadWaitRead,threadWaitWrite)
 728 -- and delays (threadDelay).
 729 --
 730 -- We can do this because in the threaded RTS the IO Manager can make
 731 -- a non-blocking call to select(), so we don't have to do select() in
 732 -- the scheduler as we have to in the non-threaded RTS.  We get performance
 733 -- benefits from doing it this way, because we only have to restart the select()
 734 -- when a new request arrives, rather than doing one select() each time
 735 -- around the scheduler loop.  Furthermore, the scheduler can be simplified
 736 -- by not having to check for completed IO requests.
 737
 738 -- Issues, possible problems:
 739 --
 740 --      - we might want bound threads to just do the blocking
 741 --        operation rather than communicating with the IO manager
 742 --        thread.  This would prevent simgle-threaded programs which do
 743 --        IO from requiring multiple OS threads.  However, it would also
 744 --        prevent bound threads waiting on IO from being killed or sent
 745 --        exceptions.
 746 --
 747 --      - Apprently exec() doesn't work on Linux in a multithreaded program.
 748 --        I couldn't repeat this.
 749 --
 750 --      - How do we handle signal delivery in the multithreaded RTS?
 751 --
 752 --      - forkProcess will kill the IO manager thread.  Let's just
 753 --        hope we don't need to do any blocking IO between fork & exec.
 754
 755 #ifndef mingw32_HOST_OS
 756 data IOReq
 757   = Read   {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 758   | Write  {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 759 #endif
 760
 761 data DelayReq
 762   = Delay    {-# UNPACK #-} !USecs {-# UNPACK #-} !(MVar ())
 763   | DelaySTM {-# UNPACK #-} !USecs {-# UNPACK #-} !(TVar Bool)
 764
 765 #ifndef mingw32_HOST_OS
 766 pendingEvents :: IORef [IOReq]
 767 #endif
 768 pendingDelays :: IORef [DelayReq]
 769         -- could use a strict list or array here
 770 {-# NOINLINE pendingEvents #-}
 771 {-# NOINLINE pendingDelays #-}
 772 (pendingEvents,pendingDelays) = unsafePerformIO $ do
 773   startIOManagerThread
 774   reqs <- newIORef []
 775   dels <- newIORef []
 776   return (reqs, dels)
 777         -- the first time we schedule an IO request, the service thread
 778         -- will be created (cool, huh?)
 779
 780 ensureIOManagerIsRunning :: IO ()
 781 ensureIOManagerIsRunning
 782   | threaded  = seq pendingEvents $ return ()
 783   | otherwise = return ()
 784
 785 insertDelay :: DelayReq -> [DelayReq] -> [DelayReq]
 786 insertDelay d [] = [d]
 787 insertDelay d1 ds@(d2 : rest)
 788   | delayTime d1 <= delayTime d2 = d1 : ds
 789   | otherwise                    = d2 : insertDelay d1 rest
 790
 791 delayTime :: DelayReq -> USecs
 792 delayTime (Delay t _) = t
 793 delayTime (DelaySTM t _) = t
 794
 795 type USecs = Word64
 796
 797 -- XXX: move into GHC.IOBase from Data.IORef?
 798 atomicModifyIORef :: IORef a -> (a -> (a,b)) -> IO b
 799 atomicModifyIORef (IORef (STRef r#)) f = IO $ \s -> atomicModifyMutVar# r# f s
 800
 801 foreign import ccall unsafe "getUSecOfDay"
 802   getUSecOfDay :: IO USecs
 803
 804 prodding :: IORef Bool
 805 {-# NOINLINE prodding #-}
 806 prodding = unsafePerformIO (newIORef False)
 807
 808 prodServiceThread :: IO ()
 809 prodServiceThread = do
 810   was_set <- atomicModifyIORef prodding (\a -> (True,a))
 811   if (not (was_set)) then wakeupIOManager else return ()
 812
 813 #ifdef mingw32_HOST_OS
 814 -- ----------------------------------------------------------------------------
 815 -- Windows IO manager thread
 816
 817 startIOManagerThread :: IO ()
 818 startIOManagerThread = do
 819   wakeup <- c_getIOManagerEvent
 820   forkIO $ service_loop wakeup []
 821   return ()
 822
 823 service_loop :: HANDLE          -- read end of pipe
 824              -> [DelayReq]      -- current delay requests
 825              -> IO ()
 826
 827 service_loop wakeup old_delays = do
 828   -- pick up new delay requests
 829   new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
 830   let  delays = foldr insertDelay old_delays new_delays
 831
 832   now <- getUSecOfDay
 833   (delays', timeout) <- getDelay now delays
 834
 835   r <- c_WaitForSingleObject wakeup timeout
 836   case r of
 837     0xffffffff -> do c_maperrno; throwErrno "service_loop"
 838     0 -> do
 839         r <- c_readIOManagerEvent
 840         exit <-
 841               case r of
 842                 _ | r == io_MANAGER_WAKEUP -> return False
 843                 _ | r == io_MANAGER_DIE    -> return True
 844                 0 -> return False -- spurious wakeup
 845                 r -> do start_console_handler (r `shiftR` 1); return False
 846         if exit
 847           then return ()
 848           else service_cont wakeup delays'
 849
 850     _other -> service_cont wakeup delays' -- probably timeout
 851
 852 service_cont wakeup delays = do
 853   atomicModifyIORef prodding (\_ -> (False,False))
 854   service_loop wakeup delays
 855
 856 -- must agree with rts/win32/ThrIOManager.c
 857 io_MANAGER_WAKEUP = 0xffffffff :: Word32
 858 io_MANAGER_DIE    = 0xfffffffe :: Word32
 859
 860 data ConsoleEvent
 861  = ControlC
 862  | Break
 863  | Close
 864     -- these are sent to Services only.
 865  | Logoff
 866  | Shutdown
 867  deriving (Eq, Ord, Enum, Show, Read, Typeable)
 868
 869 start_console_handler :: Word32 -> IO ()
 870 start_console_handler r =
 871   case toWin32ConsoleEvent r of
 872      Just x  -> withMVar win32ConsoleHandler $ \handler -> do
 873                     forkIO (handler x)
 874                     return ()
 875      Nothing -> return ()
 876
 877 toWin32ConsoleEvent ev =
 878    case ev of
 879        0 {- CTRL_C_EVENT-}        -> Just ControlC
 880        1 {- CTRL_BREAK_EVENT-}    -> Just Break
 881        2 {- CTRL_CLOSE_EVENT-}    -> Just Close
 882        5 {- CTRL_LOGOFF_EVENT-}   -> Just Logoff
 883        6 {- CTRL_SHUTDOWN_EVENT-} -> Just Shutdown
 884        _ -> Nothing
 885
 886 win32ConsoleHandler :: MVar (ConsoleEvent -> IO ())
 887 win32ConsoleHandler = unsafePerformIO (newMVar (error "win32ConsoleHandler"))
 888
 889 stick :: IORef HANDLE
 890 {-# NOINLINE stick #-}
 891 stick = unsafePerformIO (newIORef nullPtr)
 892
 893 wakeupIOManager = do
 894   hdl <- readIORef stick
 895   c_sendIOManagerEvent io_MANAGER_WAKEUP
 896
 897 -- Walk the queue of pending delays, waking up any that have passed
 898 -- and return the smallest delay to wait for.  The queue of pending
 899 -- delays is kept ordered.
 900 getDelay :: USecs -> [DelayReq] -> IO ([DelayReq], DWORD)
 901 getDelay now [] = return ([], iNFINITE)
 902 getDelay now all@(d : rest)
 903   = case d of
 904      Delay time m | now >= time -> do
 905         putMVar m ()
 906         getDelay now rest
 907      DelaySTM time t | now >= time -> do
 908         atomically $ writeTVar t True
 909         getDelay now rest
 910      _otherwise ->
 911         -- delay is in millisecs for WaitForSingleObject
 912         let micro_seconds = delayTime d - now
 913             milli_seconds = (micro_seconds + 999) `div` 1000
 914         in return (all, fromIntegral milli_seconds)
 915
 916 -- ToDo: this just duplicates part of System.Win32.Types, which isn't
 917 -- available yet.  We should move some Win32 functionality down here,
 918 -- maybe as part of the grand reorganisation of the base package...
 919 type HANDLE       = Ptr ()
 920 type DWORD        = Word32
 921
 922 iNFINITE = 0xFFFFFFFF :: DWORD -- urgh
 923
 924 foreign import ccall unsafe "getIOManagerEvent" -- in the RTS (ThrIOManager.c)
 925   c_getIOManagerEvent :: IO HANDLE
 926
 927 foreign import ccall unsafe "readIOManagerEvent" -- in the RTS (ThrIOManager.c)
 928   c_readIOManagerEvent :: IO Word32
 929
 930 foreign import ccall unsafe "sendIOManagerEvent" -- in the RTS (ThrIOManager.c)
 931   c_sendIOManagerEvent :: Word32 -> IO ()
 932
 933 foreign import ccall unsafe "maperrno"             -- in Win32Utils.c
 934    c_maperrno :: IO ()
 935
 936 foreign import stdcall "WaitForSingleObject"
 937    c_WaitForSingleObject :: HANDLE -> DWORD -> IO DWORD
 938
 939 #else
 940 -- ----------------------------------------------------------------------------
 941 -- Unix IO manager thread, using select()
 942
 943 startIOManagerThread :: IO ()
 944 startIOManagerThread = do
 945         allocaArray 2 $ \fds -> do
 946         throwErrnoIfMinus1 "startIOManagerThread" (c_pipe fds)
 947         rd_end <- peekElemOff fds 0
 948         wr_end <- peekElemOff fds 1
 949         writeIORef stick (fromIntegral wr_end)
 950         c_setIOManagerPipe wr_end
 951         forkIO $ do
 952             allocaBytes sizeofFdSet   $ \readfds -> do
 953             allocaBytes sizeofFdSet   $ \writefds -> do
 954             allocaBytes sizeofTimeVal $ \timeval -> do
 955             service_loop (fromIntegral rd_end) readfds writefds timeval [] []
 956         return ()
 957
 958 service_loop
 959    :: Fd                -- listen to this for wakeup calls
 960    -> Ptr CFdSet
 961    -> Ptr CFdSet
 962    -> Ptr CTimeVal
 963    -> [IOReq]
 964    -> [DelayReq]
 965    -> IO ()
 966 service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do
 967
 968   -- pick up new IO requests
 969   new_reqs <- atomicModifyIORef pendingEvents (\a -> ([],a))
 970   let reqs = new_reqs ++ old_reqs
 971
 972   -- pick up new delay requests
 973   new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
 974   let  delays = foldr insertDelay old_delays new_delays
 975
 976   -- build the FDSets for select()
 977   fdZero readfds
 978   fdZero writefds
 979   fdSet wakeup readfds
 980   maxfd <- buildFdSets 0 readfds writefds reqs
 981
 982   -- perform the select()
 983   let do_select delays = do
 984           -- check the current time and wake up any thread in
 985           -- threadDelay whose timeout has expired.  Also find the
 986           -- timeout value for the select() call.
 987           now <- getUSecOfDay
 988           (delays', timeout) <- getDelay now ptimeval delays
 989
 990           res <- c_select (fromIntegral ((max wakeup maxfd)+1)) readfds writefds
 991                         nullPtr timeout
 992           if (res == -1)
 993              then do
 994                 err <- getErrno
 995                 case err of
 996                   _ | err == eINTR ->  do_select delays'
 997                         -- EINTR: just redo the select()
 998                   _ | err == eBADF ->  return (True, delays)
 999                         -- EBADF: one of the file descriptors is closed or bad,
1000                         -- we don't know which one, so wake everyone up.
1001                   _ | otherwise    ->  throwErrno "select"
1002                         -- otherwise (ENOMEM or EINVAL) something has gone
1003                         -- wrong; report the error.
1004              else
1005                 return (False,delays')
1006
1007   (wakeup_all,delays') <- do_select delays
1008
1009   exit <-
1010     if wakeup_all then return False
1011       else do
1012         b <- fdIsSet wakeup readfds
1013         if b == 0
1014           then return False
1015           else alloca $ \p -> do
1016                  c_read (fromIntegral wakeup) p 1; return ()
1017                  s <- peek p
1018                  case s of
1019                   _ | s == io_MANAGER_WAKEUP -> return False
1020                   _ | s == io_MANAGER_DIE    -> return True
1021                   _ -> withMVar signalHandlerLock $ \_ -> do
1022                           handler_tbl <- peek handlers
1023                           sp <- peekElemOff handler_tbl (fromIntegral s)
1024                           io <- deRefStablePtr sp
1025                           forkIO io
1026                           return False
1027
1028   if exit then return () else do
1029
1030   atomicModifyIORef prodding (\_ -> (False,False))
1031
1032   reqs' <- if wakeup_all then do wakeupAll reqs; return []
1033                          else completeRequests reqs readfds writefds []
1034
1035   service_loop wakeup readfds writefds ptimeval reqs' delays'
1036
1037 io_MANAGER_WAKEUP = 0xff :: CChar
1038 io_MANAGER_DIE    = 0xfe :: CChar
1039
1040 stick :: IORef Fd
1041 {-# NOINLINE stick #-}
1042 stick = unsafePerformIO (newIORef 0)
1043
1044 wakeupIOManager :: IO ()
1045 wakeupIOManager = do
1046   fd <- readIORef stick
1047   with io_MANAGER_WAKEUP $ \pbuf -> do
1048     c_write (fromIntegral fd) pbuf 1; return ()
1049
1050 -- Lock used to protect concurrent access to signal_handlers.  Symptom of
1051 -- this race condition is #1922, although that bug was on Windows a similar
1052 -- bug also exists on Unix.
1053 signalHandlerLock :: MVar ()
1054 signalHandlerLock = unsafePerformIO (newMVar ())
1055
1056 foreign import ccall "&signal_handlers" handlers :: Ptr (Ptr (StablePtr (IO ())))
1057
1058 foreign import ccall "setIOManagerPipe"
1059   c_setIOManagerPipe :: CInt -> IO ()
1060
1061 -- -----------------------------------------------------------------------------
1062 -- IO requests
1063
1064 buildFdSets maxfd readfds writefds [] = return maxfd
1065 buildFdSets maxfd readfds writefds (Read fd m : reqs)
1066   | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
1067   | otherwise        =  do
1068         fdSet fd readfds
1069         buildFdSets (max maxfd fd) readfds writefds reqs
1070 buildFdSets maxfd readfds writefds (Write fd m : reqs)
1071   | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
1072   | otherwise        =  do
1073         fdSet fd writefds
1074         buildFdSets (max maxfd fd) readfds writefds reqs
1075
1076 completeRequests [] _ _ reqs' = return reqs'
1077 completeRequests (Read fd m : reqs) readfds writefds reqs' = do
1078   b <- fdIsSet fd readfds
1079   if b /= 0
1080     then do putMVar m (); completeRequests reqs readfds writefds reqs'
1081     else completeRequests reqs readfds writefds (Read fd m : reqs')
1082 completeRequests (Write fd m : reqs) readfds writefds reqs' = do
1083   b <- fdIsSet fd writefds
1084   if b /= 0
1085     then do putMVar m (); completeRequests reqs readfds writefds reqs'
1086     else completeRequests reqs readfds writefds (Write fd m : reqs')
1087
1088 wakeupAll [] = return ()
1089 wakeupAll (Read  fd m : reqs) = do putMVar m (); wakeupAll reqs
1090 wakeupAll (Write fd m : reqs) = do putMVar m (); wakeupAll reqs
1091
1092 waitForReadEvent :: Fd -> IO ()
1093 waitForReadEvent fd = do
1094   m <- newEmptyMVar
1095   atomicModifyIORef pendingEvents (\xs -> (Read fd m : xs, ()))
1096   prodServiceThread
1097   takeMVar m
1098
1099 waitForWriteEvent :: Fd -> IO ()
1100 waitForWriteEvent fd = do
1101   m <- newEmptyMVar
1102   atomicModifyIORef pendingEvents (\xs -> (Write fd m : xs, ()))
1103   prodServiceThread
1104   takeMVar m
1105
1106 -- -----------------------------------------------------------------------------
1107 -- Delays
1108
1109 -- Walk the queue of pending delays, waking up any that have passed
1110 -- and return the smallest delay to wait for.  The queue of pending
1111 -- delays is kept ordered.
1112 getDelay :: USecs -> Ptr CTimeVal -> [DelayReq] -> IO ([DelayReq], Ptr CTimeVal)
1113 getDelay now ptimeval [] = return ([],nullPtr)
1114 getDelay now ptimeval all@(d : rest)
1115   = case d of
1116      Delay time m | now >= time -> do
1117         putMVar m ()
1118         getDelay now ptimeval rest
1119      DelaySTM time t | now >= time -> do
1120         atomically $ writeTVar t True
1121         getDelay now ptimeval rest
1122      _otherwise -> do
1123         setTimevalTicks ptimeval (delayTime d - now)
1124         return (all,ptimeval)
1125
1126 newtype CTimeVal = CTimeVal ()
1127
1128 foreign import ccall unsafe "sizeofTimeVal"
1129   sizeofTimeVal :: Int
1130
1131 foreign import ccall unsafe "setTimevalTicks"
1132   setTimevalTicks :: Ptr CTimeVal -> USecs -> IO ()
1133
1134 {-
1135   On Win32 we're going to have a single Pipe, and a
1136   waitForSingleObject with the delay time.  For signals, we send a
1137   byte down the pipe just like on Unix.
1138 -}
1139
1140 -- ----------------------------------------------------------------------------
1141 -- select() interface
1142
1143 -- ToDo: move to System.Posix.Internals?
1144
1145 newtype CFdSet = CFdSet ()
1146
1147 foreign import ccall safe "select"
1148   c_select :: CInt -> Ptr CFdSet -> Ptr CFdSet -> Ptr CFdSet -> Ptr CTimeVal
1149            -> IO CInt
1150
1151 foreign import ccall unsafe "hsFD_SETSIZE"
1152   c_fD_SETSIZE :: CInt
1153
1154 fD_SETSIZE :: Fd
1155 fD_SETSIZE = fromIntegral c_fD_SETSIZE
1156
1157 foreign import ccall unsafe "hsFD_CLR"
1158   c_fdClr :: CInt -> Ptr CFdSet -> IO ()
1159
1160 fdClr :: Fd -> Ptr CFdSet -> IO ()
1161 fdClr (Fd fd) fdset = c_fdClr fd fdset
1162
1163 foreign import ccall unsafe "hsFD_ISSET"
1164   c_fdIsSet :: CInt -> Ptr CFdSet -> IO CInt
1165
1166 fdIsSet :: Fd -> Ptr CFdSet -> IO CInt
1167 fdIsSet (Fd fd) fdset = c_fdIsSet fd fdset
1168
1169 foreign import ccall unsafe "hsFD_SET"
1170   c_fdSet :: CInt -> Ptr CFdSet -> IO ()
1171
1172 fdSet :: Fd -> Ptr CFdSet -> IO ()
1173 fdSet (Fd fd) fdset = c_fdSet fd fdset
1174
1175 foreign import ccall unsafe "hsFD_ZERO"
1176   fdZero :: Ptr CFdSet -> IO ()
1177
1178 foreign import ccall unsafe "sizeof_fd_set"
1179   sizeofFdSet :: Int
1180
1181 #endif
1182
1183 \end{code}