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