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