GHC/Conc.lhs

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