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