GHC/Conc.lhs

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