GHC/Conc.lhs

   1 \begin{code}
   2 {-# OPTIONS_GHC -fno-implicit-prelude #-}
   3 -----------------------------------------------------------------------------
   4 -- |
   5 -- Module      :  GHC.Conc
   6 -- Copyright   :  (c) The University of Glasgow, 1994-2002
   7 -- License     :  see libraries/base/LICENSE
   8 --
   9 -- Maintainer  :  cvs-ghc@haskell.org
  10 -- Stability   :  internal
  11 -- Portability :  non-portable (GHC extensions)
  12 --
  13 -- Basic concurrency stuff.
  14 --
  15 -----------------------------------------------------------------------------
  16
  17 -- No: #hide, because bits of this module are exposed by the stm package.
  18 -- However, we don't want this module to be the home location for the
  19 -- bits it exports, we'd rather have Control.Concurrent and the other
  20 -- higher level modules be the home.  Hence:
  21
  22 -- #not-home
  23 module GHC.Conc
  24         ( ThreadId(..)
  25
  26         -- Forking and suchlike
  27         , forkIO        -- :: IO a -> IO ThreadId
  28         , childHandler  -- :: Exception -> IO ()
  29         , myThreadId    -- :: IO ThreadId
  30         , killThread    -- :: ThreadId -> IO ()
  31         , throwTo       -- :: ThreadId -> Exception -> IO ()
  32         , par           -- :: a -> b -> b
  33         , pseq          -- :: a -> b -> b
  34         , yield         -- :: IO ()
  35         , labelThread   -- :: ThreadId -> String -> IO ()
  36
  37         -- Waiting
  38         , threadDelay           -- :: Int -> IO ()
  39         , registerDelay         -- :: Int -> IO (TVar Bool)
  40         , threadWaitRead        -- :: Int -> IO ()
  41         , threadWaitWrite       -- :: Int -> IO ()
  42
  43         -- MVars
  44         , MVar          -- abstract
  45         , newMVar       -- :: a -> IO (MVar a)
  46         , newEmptyMVar  -- :: IO (MVar a)
  47         , takeMVar      -- :: MVar a -> IO a
  48         , putMVar       -- :: MVar a -> a -> IO ()
  49         , tryTakeMVar   -- :: MVar a -> IO (Maybe a)
  50         , tryPutMVar    -- :: MVar a -> a -> IO Bool
  51         , isEmptyMVar   -- :: MVar a -> IO Bool
  52         , addMVarFinalizer -- :: MVar a -> IO () -> IO ()
  53
  54         -- TVars
  55         , STM           -- abstract
  56         , atomically    -- :: STM a -> IO a
  57         , retry         -- :: STM a
  58         , orElse        -- :: STM a -> STM a -> STM a
  59         , catchSTM      -- :: STM a -> (Exception -> STM a) -> STM a
  60         , TVar          -- abstract
  61         , newTVar       -- :: a -> STM (TVar a)
  62         , newTVarIO     -- :: a -> STM (TVar a)
  63         , readTVar      -- :: TVar a -> STM a
  64         , writeTVar     -- :: a -> TVar a -> STM ()
  65         , unsafeIOToSTM -- :: IO a -> STM a
  66
  67 #ifdef mingw32_HOST_OS
  68         , asyncRead     -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
  69         , asyncWrite    -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
  70         , asyncDoProc   -- :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
  71
  72         , asyncReadBA   -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
  73         , asyncWriteBA  -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
  74 #endif
  75
  76 #ifndef mingw32_HOST_OS
  77         , ensureIOManagerIsRunning
  78 #endif
  79         ) where
  80
  81 import System.Posix.Types
  82 import System.Posix.Internals
  83 import Foreign
  84 import Foreign.C
  85
  86 #ifndef __HADDOCK__
  87 import {-# SOURCE #-} GHC.TopHandler ( reportError, reportStackOverflow )
  88 #endif
  89
  90 import Data.Maybe
  91
  92 import GHC.Base
  93 import GHC.IOBase
  94 import GHC.Num          ( Num(..) )
  95 import GHC.Real         ( fromIntegral, quot )
  96 import GHC.Base         ( Int(..) )
  97 import GHC.Exception    ( catchException, Exception(..), AsyncException(..) )
  98 import GHC.Pack         ( packCString# )
  99 import GHC.Ptr          ( Ptr(..), plusPtr, FunPtr(..) )
 100 import GHC.STRef
 101 import Data.Typeable
 102
 103 infixr 0 `par`, `pseq`
 104 \end{code}
 105
 106 %************************************************************************
 107 %*                                                                      *
 108 \subsection{@ThreadId@, @par@, and @fork@}
 109 %*                                                                      *
 110 %************************************************************************
 111
 112 \begin{code}
 113 data ThreadId = ThreadId ThreadId# deriving( Typeable )
 114 -- ToDo: data ThreadId = ThreadId (Weak ThreadId#)
 115 -- But since ThreadId# is unlifted, the Weak type must use open
 116 -- type variables.
 117 {- ^
 118 A 'ThreadId' is an abstract type representing a handle to a thread.
 119 'ThreadId' is an instance of 'Eq', 'Ord' and 'Show', where
 120 the 'Ord' instance implements an arbitrary total ordering over
 121 'ThreadId's. The 'Show' instance lets you convert an arbitrary-valued
 122 'ThreadId' to string form; showing a 'ThreadId' value is occasionally
 123 useful when debugging or diagnosing the behaviour of a concurrent
 124 program.
 125
 126 /Note/: in GHC, if you have a 'ThreadId', you essentially have
 127 a pointer to the thread itself.  This means the thread itself can\'t be
 128 garbage collected until you drop the 'ThreadId'.
 129 This misfeature will hopefully be corrected at a later date.
 130
 131 /Note/: Hugs does not provide any operations on other threads;
 132 it defines 'ThreadId' as a synonym for ().
 133 -}
 134
 135 {- |
 136 This sparks off a new thread to run the 'IO' computation passed as the
 137 first argument, and returns the 'ThreadId' of the newly created
 138 thread.
 139
 140 The new thread will be a lightweight thread; if you want to use a foreign
 141 library that uses thread-local storage, use 'forkOS' instead.
 142 -}
 143 forkIO :: IO () -> IO ThreadId
 144 forkIO action = IO $ \ s ->
 145    case (fork# action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
 146  where
 147   action_plus = catchException action childHandler
 148
 149 childHandler :: Exception -> IO ()
 150 childHandler err = catchException (real_handler err) childHandler
 151
 152 real_handler :: Exception -> IO ()
 153 real_handler ex =
 154   case ex of
 155         -- ignore thread GC and killThread exceptions:
 156         BlockedOnDeadMVar            -> return ()
 157         BlockedIndefinitely          -> return ()
 158         AsyncException ThreadKilled  -> return ()
 159
 160         -- report all others:
 161         AsyncException StackOverflow -> reportStackOverflow
 162         other       -> reportError other
 163
 164 {- | 'killThread' terminates the given thread (GHC only).
 165 Any work already done by the thread isn\'t
 166 lost: the computation is suspended until required by another thread.
 167 The memory used by the thread will be garbage collected if it isn\'t
 168 referenced from anywhere.  The 'killThread' function is defined in
 169 terms of 'throwTo':
 170
 171 > killThread tid = throwTo tid (AsyncException ThreadKilled)
 172
 173 -}
 174 killThread :: ThreadId -> IO ()
 175 killThread tid = throwTo tid (AsyncException ThreadKilled)
 176
 177 {- | 'throwTo' raises an arbitrary exception in the target thread (GHC only).
 178
 179 'throwTo' does not return until the exception has been raised in the
 180 target thread.  The calling thread can thus be certain that the target
 181 thread has received the exception.  This is a useful property to know
 182 when dealing with race conditions: eg. if there are two threads that
 183 can kill each other, it is guaranteed that only one of the threads
 184 will get to kill the other.
 185
 186 If the target thread is currently making a foreign call, then the
 187 exception will not be raised (and hence 'throwTo' will not return)
 188 until the call has completed.  This is the case regardless of whether
 189 the call is inside a 'block' or not.
 190  -}
 191 throwTo :: ThreadId -> Exception -> IO ()
 192 throwTo (ThreadId id) ex = IO $ \ s ->
 193    case (killThread# id ex s) of s1 -> (# s1, () #)
 194
 195 -- | Returns the 'ThreadId' of the calling thread (GHC only).
 196 myThreadId :: IO ThreadId
 197 myThreadId = IO $ \s ->
 198    case (myThreadId# s) of (# s1, id #) -> (# s1, ThreadId id #)
 199
 200
 201 -- |The 'yield' action allows (forces, in a co-operative multitasking
 202 -- implementation) a context-switch to any other currently runnable
 203 -- threads (if any), and is occasionally useful when implementing
 204 -- concurrency abstractions.
 205 yield :: IO ()
 206 yield = IO $ \s ->
 207    case (yield# s) of s1 -> (# s1, () #)
 208
 209 {- | 'labelThread' stores a string as identifier for this thread if
 210 you built a RTS with debugging support. This identifier will be used in
 211 the debugging output to make distinction of different threads easier
 212 (otherwise you only have the thread state object\'s address in the heap).
 213
 214 Other applications like the graphical Concurrent Haskell Debugger
 215 (<http://www.informatik.uni-kiel.de/~fhu/chd/>) may choose to overload
 216 'labelThread' for their purposes as well.
 217 -}
 218
 219 labelThread :: ThreadId -> String -> IO ()
 220 labelThread (ThreadId t) str = IO $ \ s ->
 221    let ps  = packCString# str
 222        adr = byteArrayContents# ps in
 223      case (labelThread# t adr s) of s1 -> (# s1, () #)
 224
 225 --      Nota Bene: 'pseq' used to be 'seq'
 226 --                 but 'seq' is now defined in PrelGHC
 227 --
 228 -- "pseq" is defined a bit weirdly (see below)
 229 --
 230 -- The reason for the strange "lazy" call is that
 231 -- it fools the compiler into thinking that pseq  and par are non-strict in
 232 -- their second argument (even if it inlines pseq at the call site).
 233 -- If it thinks pseq is strict in "y", then it often evaluates
 234 -- "y" before "x", which is totally wrong.
 235
 236 {-# INLINE pseq  #-}
 237 pseq :: a -> b -> b
 238 pseq  x y = x `seq` lazy y
 239
 240 {-# INLINE par  #-}
 241 par :: a -> b -> b
 242 par  x y = case (par# x) of { _ -> lazy y }
 243 \end{code}
 244
 245
 246 %************************************************************************
 247 %*                                                                      *
 248 \subsection[stm]{Transactional heap operations}
 249 %*                                                                      *
 250 %************************************************************************
 251
 252 TVars are shared memory locations which support atomic memory
 253 transactions.
 254
 255 \begin{code}
 256 newtype STM a = STM (State# RealWorld -> (# State# RealWorld, a #)) deriving( Typeable )
 257
 258 unSTM :: STM a -> (State# RealWorld -> (# State# RealWorld, a #))
 259 unSTM (STM a) = a
 260
 261 instance  Functor STM where
 262    fmap f x = x >>= (return . f)
 263
 264 instance  Monad STM  where
 265     {-# INLINE return #-}
 266     {-# INLINE (>>)   #-}
 267     {-# INLINE (>>=)  #-}
 268     m >> k      = thenSTM m k
 269     return x    = returnSTM x
 270     m >>= k     = bindSTM m k
 271
 272 bindSTM :: STM a -> (a -> STM b) -> STM b
 273 bindSTM (STM m) k = STM ( \s ->
 274   case m s of
 275     (# new_s, a #) -> unSTM (k a) new_s
 276   )
 277
 278 thenSTM :: STM a -> STM b -> STM b
 279 thenSTM (STM m) k = STM ( \s ->
 280   case m s of
 281     (# new_s, a #) -> unSTM k new_s
 282   )
 283
 284 returnSTM :: a -> STM a
 285 returnSTM x = STM (\s -> (# s, x #))
 286
 287 -- | Unsafely performs IO in the STM monad.
 288 unsafeIOToSTM :: IO a -> STM a
 289 unsafeIOToSTM (IO m) = STM m
 290
 291 -- |Perform a series of STM actions atomically.
 292 atomically :: STM a -> IO a
 293 atomically (STM m) = IO (\s -> (atomically# m) s )
 294
 295 -- |Retry execution of the current memory transaction because it has seen
 296 -- values in TVars which mean that it should not continue (e.g. the TVars
 297 -- represent a shared buffer that is now empty).  The implementation may
 298 -- block the thread until one of the TVars that it has read from has been
 299 -- udpated.
 300 retry :: STM a
 301 retry = STM $ \s# -> retry# s#
 302
 303 -- |Compose two alternative STM actions.  If the first action completes without
 304 -- retrying then it forms the result of the orElse.  Otherwise, if the first
 305 -- action retries, then the second action is tried in its place.  If both actions
 306 -- retry then the orElse as a whole retries.
 307 orElse :: STM a -> STM a -> STM a
 308 orElse (STM m) e = STM $ \s -> catchRetry# m (unSTM e) s
 309
 310 -- |Exception handling within STM actions.
 311 catchSTM :: STM a -> (Exception -> STM a) -> STM a
 312 catchSTM (STM m) k = STM $ \s -> catchSTM# m (\ex -> unSTM (k ex)) s
 313
 314 data TVar a = TVar (TVar# RealWorld a) deriving( Typeable )
 315
 316 instance Eq (TVar a) where
 317         (TVar tvar1#) == (TVar tvar2#) = sameTVar# tvar1# tvar2#
 318
 319 -- |Create a new TVar holding a value supplied
 320 newTVar :: a -> STM (TVar a)
 321 newTVar val = STM $ \s1# ->
 322     case newTVar# val s1# of
 323          (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
 324
 325 -- |@IO@ version of 'newTVar'.  This is useful for creating top-level
 326 -- 'TVar's using 'System.IO.Unsafe.unsafePerformIO', because using
 327 -- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't
 328 -- possible.
 329 newTVarIO :: a -> IO (TVar a)
 330 newTVarIO val = IO $ \s1# ->
 331     case newTVar# val s1# of
 332          (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
 333
 334 -- |Return the current value stored in a TVar
 335 readTVar :: TVar a -> STM a
 336 readTVar (TVar tvar#) = STM $ \s# -> readTVar# tvar# s#
 337
 338 -- |Write the supplied value into a TVar
 339 writeTVar :: TVar a -> a -> STM ()
 340 writeTVar (TVar tvar#) val = STM $ \s1# ->
 341     case writeTVar# tvar# val s1# of
 342          s2# -> (# s2#, () #)
 343
 344 \end{code}
 345
 346 %************************************************************************
 347 %*                                                                      *
 348 \subsection[mvars]{M-Structures}
 349 %*                                                                      *
 350 %************************************************************************
 351
 352 M-Vars are rendezvous points for concurrent threads.  They begin
 353 empty, and any attempt to read an empty M-Var blocks.  When an M-Var
 354 is written, a single blocked thread may be freed.  Reading an M-Var
 355 toggles its state from full back to empty.  Therefore, any value
 356 written to an M-Var may only be read once.  Multiple reads and writes
 357 are allowed, but there must be at least one read between any two
 358 writes.
 359
 360 \begin{code}
 361 --Defined in IOBase to avoid cycle: data MVar a = MVar (SynchVar# RealWorld a)
 362
 363 -- |Create an 'MVar' which is initially empty.
 364 newEmptyMVar  :: IO (MVar a)
 365 newEmptyMVar = IO $ \ s# ->
 366     case newMVar# s# of
 367          (# s2#, svar# #) -> (# s2#, MVar svar# #)
 368
 369 -- |Create an 'MVar' which contains the supplied value.
 370 newMVar :: a -> IO (MVar a)
 371 newMVar value =
 372     newEmptyMVar        >>= \ mvar ->
 373     putMVar mvar value  >>
 374     return mvar
 375
 376 -- |Return the contents of the 'MVar'.  If the 'MVar' is currently
 377 -- empty, 'takeMVar' will wait until it is full.  After a 'takeMVar',
 378 -- the 'MVar' is left empty.
 379 --
 380 -- There are two further important properties of 'takeMVar':
 381 --
 382 --   * 'takeMVar' is single-wakeup.  That is, if there are multiple
 383 --     threads blocked in 'takeMVar', and the 'MVar' becomes full,
 384 --     only one thread will be woken up.  The runtime guarantees that
 385 --     the woken thread completes its 'takeMVar' operation.
 386 --
 387 --   * When multiple threads are blocked on an 'MVar', they are
 388 --     woken up in FIFO order.  This is useful for providing
 389 --     fairness properties of abstractions built using 'MVar's.
 390 --
 391 takeMVar :: MVar a -> IO a
 392 takeMVar (MVar mvar#) = IO $ \ s# -> takeMVar# mvar# s#
 393
 394 -- |Put a value into an 'MVar'.  If the 'MVar' is currently full,
 395 -- 'putMVar' will wait until it becomes empty.
 396 --
 397 -- There are two further important properties of 'putMVar':
 398 --
 399 --   * 'putMVar' is single-wakeup.  That is, if there are multiple
 400 --     threads blocked in 'putMVar', and the 'MVar' becomes empty,
 401 --     only one thread will be woken up.  The runtime guarantees that
 402 --     the woken thread completes its 'putMVar' operation.
 403 --
 404 --   * When multiple threads are blocked on an 'MVar', they are
 405 --     woken up in FIFO order.  This is useful for providing
 406 --     fairness properties of abstractions built using 'MVar's.
 407 --
 408 putMVar  :: MVar a -> a -> IO ()
 409 putMVar (MVar mvar#) x = IO $ \ s# ->
 410     case putMVar# mvar# x s# of
 411         s2# -> (# s2#, () #)
 412
 413 -- |A non-blocking version of 'takeMVar'.  The 'tryTakeMVar' function
 414 -- returns immediately, with 'Nothing' if the 'MVar' was empty, or
 415 -- @'Just' a@ if the 'MVar' was full with contents @a@.  After 'tryTakeMVar',
 416 -- the 'MVar' is left empty.
 417 tryTakeMVar :: MVar a -> IO (Maybe a)
 418 tryTakeMVar (MVar m) = IO $ \ s ->
 419     case tryTakeMVar# m s of
 420         (# s, 0#, _ #) -> (# s, Nothing #)      -- MVar is empty
 421         (# s, _,  a #) -> (# s, Just a  #)      -- MVar is full
 422
 423 -- |A non-blocking version of 'putMVar'.  The 'tryPutMVar' function
 424 -- attempts to put the value @a@ into the 'MVar', returning 'True' if
 425 -- it was successful, or 'False' otherwise.
 426 tryPutMVar  :: MVar a -> a -> IO Bool
 427 tryPutMVar (MVar mvar#) x = IO $ \ s# ->
 428     case tryPutMVar# mvar# x s# of
 429         (# s, 0# #) -> (# s, False #)
 430         (# s, _  #) -> (# s, True #)
 431
 432 -- |Check whether a given 'MVar' is empty.
 433 --
 434 -- Notice that the boolean value returned  is just a snapshot of
 435 -- the state of the MVar. By the time you get to react on its result,
 436 -- the MVar may have been filled (or emptied) - so be extremely
 437 -- careful when using this operation.   Use 'tryTakeMVar' instead if possible.
 438 isEmptyMVar :: MVar a -> IO Bool
 439 isEmptyMVar (MVar mv#) = IO $ \ s# ->
 440     case isEmptyMVar# mv# s# of
 441         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 442
 443 -- |Add a finalizer to an 'MVar' (GHC only).  See "Foreign.ForeignPtr" and
 444 -- "System.Mem.Weak" for more about finalizers.
 445 addMVarFinalizer :: MVar a -> IO () -> IO ()
 446 addMVarFinalizer (MVar m) finalizer =
 447   IO $ \s -> case mkWeak# m () finalizer s of { (# s1, w #) -> (# s1, () #) }
 448 \end{code}
 449
 450
 451 %************************************************************************
 452 %*                                                                      *
 453 \subsection{Thread waiting}
 454 %*                                                                      *
 455 %************************************************************************
 456
 457 \begin{code}
 458 #ifdef mingw32_HOST_OS
 459
 460 -- Note: threadDelay, threadWaitRead and threadWaitWrite aren't really functional
 461 -- on Win32, but left in there because lib code (still) uses them (the manner
 462 -- in which they're used doesn't cause problems on a Win32 platform though.)
 463
 464 asyncRead :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 465 asyncRead  (I# fd) (I# isSock) (I# len) (Ptr buf) =
 466   IO $ \s -> case asyncRead# fd isSock len buf s of
 467                (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 468
 469 asyncWrite :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 470 asyncWrite  (I# fd) (I# isSock) (I# len) (Ptr buf) =
 471   IO $ \s -> case asyncWrite# fd isSock len buf s of
 472                (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 473
 474 asyncDoProc :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
 475 asyncDoProc (FunPtr proc) (Ptr param) =
 476     -- the 'length' value is ignored; simplifies implementation of
 477     -- the async*# primops to have them all return the same result.
 478   IO $ \s -> case asyncDoProc# proc param s  of
 479                (# s, len#, err# #) -> (# s, I# err# #)
 480
 481 -- to aid the use of these primops by the IO Handle implementation,
 482 -- provide the following convenience funs:
 483
 484 -- this better be a pinned byte array!
 485 asyncReadBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 486 asyncReadBA fd isSock len off bufB =
 487   asyncRead fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 488
 489 asyncWriteBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 490 asyncWriteBA fd isSock len off bufB =
 491   asyncWrite fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 492
 493 #endif
 494
 495 -- -----------------------------------------------------------------------------
 496 -- Thread IO API
 497
 498 -- | Block the current thread until data is available to read on the
 499 -- given file descriptor (GHC only).
 500 threadWaitRead :: Fd -> IO ()
 501 threadWaitRead fd
 502 #ifndef mingw32_HOST_OS
 503   | threaded  = waitForReadEvent fd
 504 #endif
 505   | otherwise = IO $ \s ->
 506         case fromIntegral fd of { I# fd# ->
 507         case waitRead# fd# s of { s -> (# s, () #)
 508         }}
 509
 510 -- | Block the current thread until data can be written to the
 511 -- given file descriptor (GHC only).
 512 threadWaitWrite :: Fd -> IO ()
 513 threadWaitWrite fd
 514 #ifndef mingw32_HOST_OS
 515   | threaded  = waitForWriteEvent fd
 516 #endif
 517   | otherwise = IO $ \s ->
 518         case fromIntegral fd of { I# fd# ->
 519         case waitWrite# fd# s of { s -> (# s, () #)
 520         }}
 521
 522 -- | Suspends the current thread for a given number of microseconds
 523 -- (GHC only).
 524 --
 525 -- Note that the resolution used by the Haskell runtime system's
 526 -- internal timer is 1\/50 second, and 'threadDelay' will round its
 527 -- argument up to the nearest multiple of this resolution.
 528 --
 529 -- There is no guarantee that the thread will be rescheduled promptly
 530 -- when the delay has expired, but the thread will never continue to
 531 -- run /earlier/ than specified.
 532 --
 533 threadDelay :: Int -> IO ()
 534 threadDelay time
 535 #ifndef mingw32_HOST_OS
 536   | threaded  = waitForDelayEvent time
 537 #else
 538   | threaded  = c_Sleep (fromIntegral (time `quot` 1000))
 539 #endif
 540   | otherwise = IO $ \s ->
 541         case fromIntegral time of { I# time# ->
 542         case delay# time# s of { s -> (# s, () #)
 543         }}
 544
 545 registerDelay usecs
 546 #ifndef mingw32_HOST_OS
 547   | threaded = waitForDelayEventSTM usecs
 548   | otherwise = error "registerDelay: requires -threaded"
 549 #else
 550   = error "registerDelay: not currently supported on Windows"
 551 #endif
 552
 553 -- On Windows, we just make a safe call to 'Sleep' to implement threadDelay.
 554 #ifdef mingw32_HOST_OS
 555 foreign import stdcall safe "Sleep" c_Sleep :: CInt -> IO ()
 556 #endif
 557
 558 foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
 559
 560 -- ----------------------------------------------------------------------------
 561 -- Threaded RTS implementation of threadWaitRead, threadWaitWrite, threadDelay
 562
 563 -- In the threaded RTS, we employ a single IO Manager thread to wait
 564 -- for all outstanding IO requests (threadWaitRead,threadWaitWrite)
 565 -- and delays (threadDelay).
 566 --
 567 -- We can do this because in the threaded RTS the IO Manager can make
 568 -- a non-blocking call to select(), so we don't have to do select() in
 569 -- the scheduler as we have to in the non-threaded RTS.  We get performance
 570 -- benefits from doing it this way, because we only have to restart the select()
 571 -- when a new request arrives, rather than doing one select() each time
 572 -- around the scheduler loop.  Furthermore, the scheduler can be simplified
 573 -- by not having to check for completed IO requests.
 574
 575 -- Issues, possible problems:
 576 --
 577 --      - we might want bound threads to just do the blocking
 578 --        operation rather than communicating with the IO manager
 579 --        thread.  This would prevent simgle-threaded programs which do
 580 --        IO from requiring multiple OS threads.  However, it would also
 581 --        prevent bound threads waiting on IO from being killed or sent
 582 --        exceptions.
 583 --
 584 --      - Apprently exec() doesn't work on Linux in a multithreaded program.
 585 --        I couldn't repeat this.
 586 --
 587 --      - How do we handle signal delivery in the multithreaded RTS?
 588 --
 589 --      - forkProcess will kill the IO manager thread.  Let's just
 590 --        hope we don't need to do any blocking IO between fork & exec.
 591
 592 #ifndef mingw32_HOST_OS
 593
 594 data IOReq
 595   = Read   {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 596   | Write  {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 597
 598 data DelayReq
 599   = Delay    {-# UNPACK #-} !Int {-# UNPACK #-} !(MVar ())
 600   | DelaySTM {-# UNPACK #-} !Int {-# UNPACK #-} !(TVar Bool)
 601
 602 pendingEvents :: IORef [IOReq]
 603 pendingDelays :: IORef [DelayReq]
 604         -- could use a strict list or array here
 605 {-# NOINLINE pendingEvents #-}
 606 {-# NOINLINE pendingDelays #-}
 607 (pendingEvents,pendingDelays) = unsafePerformIO $ do
 608   startIOManagerThread
 609   reqs <- newIORef []
 610   dels <- newIORef []
 611   return (reqs, dels)
 612         -- the first time we schedule an IO request, the service thread
 613         -- will be created (cool, huh?)
 614
 615 ensureIOManagerIsRunning :: IO ()
 616 ensureIOManagerIsRunning
 617   | threaded  = seq pendingEvents $ return ()
 618   | otherwise = return ()
 619
 620 startIOManagerThread :: IO ()
 621 startIOManagerThread = do
 622         allocaArray 2 $ \fds -> do
 623         throwErrnoIfMinus1 "startIOManagerThread" (c_pipe fds)
 624         rd_end <- peekElemOff fds 0
 625         wr_end <- peekElemOff fds 1
 626         writeIORef stick (fromIntegral wr_end)
 627         c_setIOManagerPipe wr_end
 628         forkIO $ do
 629             allocaBytes sizeofFdSet   $ \readfds -> do
 630             allocaBytes sizeofFdSet   $ \writefds -> do
 631             allocaBytes sizeofTimeVal $ \timeval -> do
 632             service_loop (fromIntegral rd_end) readfds writefds timeval [] []
 633         return ()
 634
 635 service_loop
 636    :: Fd                -- listen to this for wakeup calls
 637    -> Ptr CFdSet
 638    -> Ptr CFdSet
 639    -> Ptr CTimeVal
 640    -> [IOReq]
 641    -> [DelayReq]
 642    -> IO ()
 643 service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do
 644
 645   -- pick up new IO requests
 646   new_reqs <- atomicModifyIORef pendingEvents (\a -> ([],a))
 647   let reqs = new_reqs ++ old_reqs
 648
 649   -- pick up new delay requests
 650   new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
 651   let  delays = foldr insertDelay old_delays new_delays
 652
 653   -- build the FDSets for select()
 654   fdZero readfds
 655   fdZero writefds
 656   fdSet wakeup readfds
 657   maxfd <- buildFdSets 0 readfds writefds reqs
 658
 659   -- perform the select()
 660   let do_select delays = do
 661           -- check the current time and wake up any thread in
 662           -- threadDelay whose timeout has expired.  Also find the
 663           -- timeout value for the select() call.
 664           now <- getTicksOfDay
 665           (delays', timeout) <- getDelay now ptimeval delays
 666
 667           res <- c_select ((max wakeup maxfd)+1) readfds writefds
 668                         nullPtr timeout
 669           if (res == -1)
 670              then do
 671                 err <- getErrno
 672                 if err == eINTR
 673                         then do_select delays'
 674                         else return (res,delays')
 675              else
 676                 return (res,delays')
 677
 678   (res,delays') <- do_select delays
 679   -- ToDo: check result
 680
 681   b <- fdIsSet wakeup readfds
 682   if b == 0
 683     then return ()
 684     else alloca $ \p -> do
 685             c_read (fromIntegral wakeup) p 1; return ()
 686             s <- peek p
 687             if (s == 0xff)
 688               then return ()
 689               else do handler_tbl <- peek handlers
 690                       sp <- peekElemOff handler_tbl (fromIntegral s)
 691                       forkIO (do io <- deRefStablePtr sp; io)
 692                       return ()
 693
 694   takeMVar prodding
 695   putMVar prodding False
 696
 697   reqs' <- completeRequests reqs readfds writefds []
 698   service_loop wakeup readfds writefds ptimeval reqs' delays'
 699
 700 stick :: IORef Fd
 701 {-# NOINLINE stick #-}
 702 stick = unsafePerformIO (newIORef 0)
 703
 704 prodding :: MVar Bool
 705 {-# NOINLINE prodding #-}
 706 prodding = unsafePerformIO (newMVar False)
 707
 708 prodServiceThread :: IO ()
 709 prodServiceThread = do
 710   b <- takeMVar prodding
 711   if (not b)
 712     then do fd <- readIORef stick
 713             with 0xff $ \pbuf -> do c_write (fromIntegral fd) pbuf 1; return ()
 714     else return ()
 715   putMVar prodding True
 716
 717 foreign import ccall "&signal_handlers" handlers :: Ptr (Ptr (StablePtr (IO ())))
 718
 719 foreign import ccall "setIOManagerPipe"
 720   c_setIOManagerPipe :: CInt -> IO ()
 721
 722 -- -----------------------------------------------------------------------------
 723 -- IO requests
 724
 725 buildFdSets maxfd readfds writefds [] = return maxfd
 726 buildFdSets maxfd readfds writefds (Read fd m : reqs)
 727   | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
 728   | otherwise        =  do
 729         fdSet fd readfds
 730         buildFdSets (max maxfd fd) readfds writefds reqs
 731 buildFdSets maxfd readfds writefds (Write fd m : reqs)
 732   | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
 733   | otherwise        =  do
 734         fdSet fd writefds
 735         buildFdSets (max maxfd fd) readfds writefds reqs
 736
 737 completeRequests [] _ _ reqs' = return reqs'
 738 completeRequests (Read fd m : reqs) readfds writefds reqs' = do
 739   b <- fdIsSet fd readfds
 740   if b /= 0
 741     then do putMVar m (); completeRequests reqs readfds writefds reqs'
 742     else completeRequests reqs readfds writefds (Read fd m : reqs')
 743 completeRequests (Write fd m : reqs) readfds writefds reqs' = do
 744   b <- fdIsSet fd writefds
 745   if b /= 0
 746     then do putMVar m (); completeRequests reqs readfds writefds reqs'
 747     else completeRequests reqs readfds writefds (Write fd m : reqs')
 748
 749 waitForReadEvent :: Fd -> IO ()
 750 waitForReadEvent fd = do
 751   m <- newEmptyMVar
 752   atomicModifyIORef pendingEvents (\xs -> (Read fd m : xs, ()))
 753   prodServiceThread
 754   takeMVar m
 755
 756 waitForWriteEvent :: Fd -> IO ()
 757 waitForWriteEvent fd = do
 758   m <- newEmptyMVar
 759   atomicModifyIORef pendingEvents (\xs -> (Write fd m : xs, ()))
 760   prodServiceThread
 761   takeMVar m
 762
 763 -- XXX: move into GHC.IOBase from Data.IORef?
 764 atomicModifyIORef :: IORef a -> (a -> (a,b)) -> IO b
 765 atomicModifyIORef (IORef (STRef r#)) f = IO $ \s -> atomicModifyMutVar# r# f s
 766
 767 -- -----------------------------------------------------------------------------
 768 -- Delays
 769
 770 waitForDelayEvent :: Int -> IO ()
 771 waitForDelayEvent usecs = do
 772   m <- newEmptyMVar
 773   now <- getTicksOfDay
 774   let target = now + usecs `quot` tick_usecs
 775   atomicModifyIORef pendingDelays (\xs -> (Delay target m : xs, ()))
 776   prodServiceThread
 777   takeMVar m
 778
 779 -- Delays for use in STM
 780 waitForDelayEventSTM :: Int -> IO (TVar Bool)
 781 waitForDelayEventSTM usecs = do
 782    t <- atomically $ newTVar False
 783    now <- getTicksOfDay
 784    let target = now + usecs `quot` tick_usecs
 785    atomicModifyIORef pendingDelays (\xs -> (DelaySTM target t : xs, ()))
 786    prodServiceThread
 787    return t
 788
 789 -- Walk the queue of pending delays, waking up any that have passed
 790 -- and return the smallest delay to wait for.  The queue of pending
 791 -- delays is kept ordered.
 792 getDelay :: Ticks -> Ptr CTimeVal -> [DelayReq] -> IO ([DelayReq], Ptr CTimeVal)
 793 getDelay now ptimeval [] = return ([],nullPtr)
 794 getDelay now ptimeval all@(d : rest)
 795   = case d of
 796      Delay time m | now >= time -> do
 797         putMVar m ()
 798         getDelay now ptimeval rest
 799      DelaySTM time t | now >= time -> do
 800         atomically $ writeTVar t True
 801         getDelay now ptimeval rest
 802      _otherwise -> do
 803         setTimevalTicks ptimeval (delayTime d - now)
 804         return (all,ptimeval)
 805
 806 insertDelay :: DelayReq -> [DelayReq] -> [DelayReq]
 807 insertDelay d [] = [d]
 808 insertDelay d1 ds@(d2 : rest)
 809   | delayTime d1 <= delayTime d2 = d1 : ds
 810   | otherwise                    = d2 : insertDelay d1 rest
 811
 812 delayTime (Delay t _) = t
 813 delayTime (DelaySTM t _) = t
 814
 815 type Ticks = Int
 816 tick_freq  = 50 :: Ticks  -- accuracy of threadDelay (ticks per sec)
 817 tick_usecs = 1000000 `quot` tick_freq :: Int
 818
 819 newtype CTimeVal = CTimeVal ()
 820
 821 foreign import ccall unsafe "sizeofTimeVal"
 822   sizeofTimeVal :: Int
 823
 824 foreign import ccall unsafe "getTicksOfDay"
 825   getTicksOfDay :: IO Ticks
 826
 827 foreign import ccall unsafe "setTimevalTicks"
 828   setTimevalTicks :: Ptr CTimeVal -> Ticks -> IO ()
 829
 830 -- ----------------------------------------------------------------------------
 831 -- select() interface
 832
 833 -- ToDo: move to System.Posix.Internals?
 834
 835 newtype CFdSet = CFdSet ()
 836
 837 foreign import ccall safe "select"
 838   c_select :: Fd -> Ptr CFdSet -> Ptr CFdSet -> Ptr CFdSet -> Ptr CTimeVal
 839            -> IO CInt
 840
 841 foreign import ccall unsafe "hsFD_SETSIZE"
 842   fD_SETSIZE :: Fd
 843
 844 foreign import ccall unsafe "hsFD_CLR"
 845   fdClr :: Fd -> Ptr CFdSet -> IO ()
 846
 847 foreign import ccall unsafe "hsFD_ISSET"
 848   fdIsSet :: Fd -> Ptr CFdSet -> IO CInt
 849
 850 foreign import ccall unsafe "hsFD_SET"
 851   fdSet :: Fd -> Ptr CFdSet -> IO ()
 852
 853 foreign import ccall unsafe "hsFD_ZERO"
 854   fdZero :: Ptr CFdSet -> IO ()
 855
 856 foreign import ccall unsafe "sizeof_fd_set"
 857   sizeofFdSet :: Int
 858
 859 #endif
 860 \end{code}