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 -- If several threads are competing to take the same 'MVar', one is chosen
 381 -- to continue at random when the 'MVar' becomes full.
 382 takeMVar :: MVar a -> IO a
 383 takeMVar (MVar mvar#) = IO $ \ s# -> takeMVar# mvar# s#
 384
 385 -- |Put a value into an 'MVar'.  If the 'MVar' is currently full,
 386 -- 'putMVar' will wait until it becomes empty.
 387 --
 388 -- If several threads are competing to fill the same 'MVar', one is
 389 -- chosen to continue at random when the 'MVar' becomes empty.
 390 putMVar  :: MVar a -> a -> IO ()
 391 putMVar (MVar mvar#) x = IO $ \ s# ->
 392     case putMVar# mvar# x s# of
 393         s2# -> (# s2#, () #)
 394
 395 -- |A non-blocking version of 'takeMVar'.  The 'tryTakeMVar' function
 396 -- returns immediately, with 'Nothing' if the 'MVar' was empty, or
 397 -- @'Just' a@ if the 'MVar' was full with contents @a@.  After 'tryTakeMVar',
 398 -- the 'MVar' is left empty.
 399 tryTakeMVar :: MVar a -> IO (Maybe a)
 400 tryTakeMVar (MVar m) = IO $ \ s ->
 401     case tryTakeMVar# m s of
 402         (# s, 0#, _ #) -> (# s, Nothing #)      -- MVar is empty
 403         (# s, _,  a #) -> (# s, Just a  #)      -- MVar is full
 404
 405 -- |A non-blocking version of 'putMVar'.  The 'tryPutMVar' function
 406 -- attempts to put the value @a@ into the 'MVar', returning 'True' if
 407 -- it was successful, or 'False' otherwise.
 408 tryPutMVar  :: MVar a -> a -> IO Bool
 409 tryPutMVar (MVar mvar#) x = IO $ \ s# ->
 410     case tryPutMVar# mvar# x s# of
 411         (# s, 0# #) -> (# s, False #)
 412         (# s, _  #) -> (# s, True #)
 413
 414 -- |Check whether a given 'MVar' is empty.
 415 --
 416 -- Notice that the boolean value returned  is just a snapshot of
 417 -- the state of the MVar. By the time you get to react on its result,
 418 -- the MVar may have been filled (or emptied) - so be extremely
 419 -- careful when using this operation.   Use 'tryTakeMVar' instead if possible.
 420 isEmptyMVar :: MVar a -> IO Bool
 421 isEmptyMVar (MVar mv#) = IO $ \ s# ->
 422     case isEmptyMVar# mv# s# of
 423         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 424
 425 -- |Add a finalizer to an 'MVar' (GHC only).  See "Foreign.ForeignPtr" and
 426 -- "System.Mem.Weak" for more about finalizers.
 427 addMVarFinalizer :: MVar a -> IO () -> IO ()
 428 addMVarFinalizer (MVar m) finalizer =
 429   IO $ \s -> case mkWeak# m () finalizer s of { (# s1, w #) -> (# s1, () #) }
 430 \end{code}
 431
 432
 433 %************************************************************************
 434 %*                                                                      *
 435 \subsection{Thread waiting}
 436 %*                                                                      *
 437 %************************************************************************
 438
 439 \begin{code}
 440 #ifdef mingw32_HOST_OS
 441
 442 -- Note: threadDelay, threadWaitRead and threadWaitWrite aren't really functional
 443 -- on Win32, but left in there because lib code (still) uses them (the manner
 444 -- in which they're used doesn't cause problems on a Win32 platform though.)
 445
 446 asyncRead :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 447 asyncRead  (I# fd) (I# isSock) (I# len) (Ptr buf) =
 448   IO $ \s -> case asyncRead# fd isSock len buf s of
 449                (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 450
 451 asyncWrite :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 452 asyncWrite  (I# fd) (I# isSock) (I# len) (Ptr buf) =
 453   IO $ \s -> case asyncWrite# fd isSock len buf s of
 454                (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 455
 456 asyncDoProc :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
 457 asyncDoProc (FunPtr proc) (Ptr param) =
 458     -- the 'length' value is ignored; simplifies implementation of
 459     -- the async*# primops to have them all return the same result.
 460   IO $ \s -> case asyncDoProc# proc param s  of
 461                (# s, len#, err# #) -> (# s, I# err# #)
 462
 463 -- to aid the use of these primops by the IO Handle implementation,
 464 -- provide the following convenience funs:
 465
 466 -- this better be a pinned byte array!
 467 asyncReadBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 468 asyncReadBA fd isSock len off bufB =
 469   asyncRead fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 470
 471 asyncWriteBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
 472 asyncWriteBA fd isSock len off bufB =
 473   asyncWrite fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 474
 475 #endif
 476
 477 -- -----------------------------------------------------------------------------
 478 -- Thread IO API
 479
 480 -- | Block the current thread until data is available to read on the
 481 -- given file descriptor (GHC only).
 482 threadWaitRead :: Fd -> IO ()
 483 threadWaitRead fd
 484 #ifndef mingw32_HOST_OS
 485   | threaded  = waitForReadEvent fd
 486 #endif
 487   | otherwise = IO $ \s ->
 488         case fromIntegral fd of { I# fd# ->
 489         case waitRead# fd# s of { s -> (# s, () #)
 490         }}
 491
 492 -- | Block the current thread until data can be written to the
 493 -- given file descriptor (GHC only).
 494 threadWaitWrite :: Fd -> IO ()
 495 threadWaitWrite fd
 496 #ifndef mingw32_HOST_OS
 497   | threaded  = waitForWriteEvent fd
 498 #endif
 499   | otherwise = IO $ \s ->
 500         case fromIntegral fd of { I# fd# ->
 501         case waitWrite# fd# s of { s -> (# s, () #)
 502         }}
 503
 504 -- | Suspends the current thread for a given number of microseconds
 505 -- (GHC only).
 506 --
 507 -- Note that the resolution used by the Haskell runtime system's
 508 -- internal timer is 1\/50 second, and 'threadDelay' will round its
 509 -- argument up to the nearest multiple of this resolution.
 510 --
 511 -- There is no guarantee that the thread will be rescheduled promptly
 512 -- when the delay has expired, but the thread will never continue to
 513 -- run /earlier/ than specified.
 514 --
 515 threadDelay :: Int -> IO ()
 516 threadDelay time
 517 #ifndef mingw32_HOST_OS
 518   | threaded  = waitForDelayEvent time
 519 #else
 520   | threaded  = c_Sleep (fromIntegral (time `quot` 1000))
 521 #endif
 522   | otherwise = IO $ \s ->
 523         case fromIntegral time of { I# time# ->
 524         case delay# time# s of { s -> (# s, () #)
 525         }}
 526
 527 registerDelay usecs
 528 #ifndef mingw32_HOST_OS
 529   | threaded = waitForDelayEventSTM usecs
 530   | otherwise = error "registerDelay: requires -threaded"
 531 #else
 532   = error "registerDelay: not currently supported on Windows"
 533 #endif
 534
 535 -- On Windows, we just make a safe call to 'Sleep' to implement threadDelay.
 536 #ifdef mingw32_HOST_OS
 537 foreign import stdcall safe "Sleep" c_Sleep :: CInt -> IO ()
 538 #endif
 539
 540 foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
 541
 542 -- ----------------------------------------------------------------------------
 543 -- Threaded RTS implementation of threadWaitRead, threadWaitWrite, threadDelay
 544
 545 -- In the threaded RTS, we employ a single IO Manager thread to wait
 546 -- for all outstanding IO requests (threadWaitRead,threadWaitWrite)
 547 -- and delays (threadDelay).
 548 --
 549 -- We can do this because in the threaded RTS the IO Manager can make
 550 -- a non-blocking call to select(), so we don't have to do select() in
 551 -- the scheduler as we have to in the non-threaded RTS.  We get performance
 552 -- benefits from doing it this way, because we only have to restart the select()
 553 -- when a new request arrives, rather than doing one select() each time
 554 -- around the scheduler loop.  Furthermore, the scheduler can be simplified
 555 -- by not having to check for completed IO requests.
 556
 557 -- Issues, possible problems:
 558 --
 559 --      - we might want bound threads to just do the blocking
 560 --        operation rather than communicating with the IO manager
 561 --        thread.  This would prevent simgle-threaded programs which do
 562 --        IO from requiring multiple OS threads.  However, it would also
 563 --        prevent bound threads waiting on IO from being killed or sent
 564 --        exceptions.
 565 --
 566 --      - Apprently exec() doesn't work on Linux in a multithreaded program.
 567 --        I couldn't repeat this.
 568 --
 569 --      - How do we handle signal delivery in the multithreaded RTS?
 570 --
 571 --      - forkProcess will kill the IO manager thread.  Let's just
 572 --        hope we don't need to do any blocking IO between fork & exec.
 573
 574 #ifndef mingw32_HOST_OS
 575
 576 data IOReq
 577   = Read   {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 578   | Write  {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
 579
 580 data DelayReq
 581   = Delay    {-# UNPACK #-} !Int {-# UNPACK #-} !(MVar ())
 582   | DelaySTM {-# UNPACK #-} !Int {-# UNPACK #-} !(TVar Bool)
 583
 584 pendingEvents :: IORef [IOReq]
 585 pendingDelays :: IORef [DelayReq]
 586         -- could use a strict list or array here
 587 {-# NOINLINE pendingEvents #-}
 588 {-# NOINLINE pendingDelays #-}
 589 (pendingEvents,pendingDelays) = unsafePerformIO $ do
 590   startIOManagerThread
 591   reqs <- newIORef []
 592   dels <- newIORef []
 593   return (reqs, dels)
 594         -- the first time we schedule an IO request, the service thread
 595         -- will be created (cool, huh?)
 596
 597 ensureIOManagerIsRunning :: IO ()
 598 ensureIOManagerIsRunning
 599   | threaded  = seq pendingEvents $ return ()
 600   | otherwise = return ()
 601
 602 startIOManagerThread :: IO ()
 603 startIOManagerThread = do
 604         allocaArray 2 $ \fds -> do
 605         throwErrnoIfMinus1 "startIOManagerThread" (c_pipe fds)
 606         rd_end <- peekElemOff fds 0
 607         wr_end <- peekElemOff fds 1
 608         writeIORef stick (fromIntegral wr_end)
 609         c_setIOManagerPipe wr_end
 610         forkIO $ do
 611             allocaBytes sizeofFdSet   $ \readfds -> do
 612             allocaBytes sizeofFdSet   $ \writefds -> do
 613             allocaBytes sizeofTimeVal $ \timeval -> do
 614             service_loop (fromIntegral rd_end) readfds writefds timeval [] []
 615         return ()
 616
 617 service_loop
 618    :: Fd                -- listen to this for wakeup calls
 619    -> Ptr CFdSet
 620    -> Ptr CFdSet
 621    -> Ptr CTimeVal
 622    -> [IOReq]
 623    -> [DelayReq]
 624    -> IO ()
 625 service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do
 626
 627   -- pick up new IO requests
 628   new_reqs <- atomicModifyIORef pendingEvents (\a -> ([],a))
 629   let reqs = new_reqs ++ old_reqs
 630
 631   -- pick up new delay requests
 632   new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
 633   let  delays = foldr insertDelay old_delays new_delays
 634
 635   -- build the FDSets for select()
 636   fdZero readfds
 637   fdZero writefds
 638   fdSet wakeup readfds
 639   maxfd <- buildFdSets 0 readfds writefds reqs
 640
 641   -- perform the select()
 642   let do_select delays = do
 643           -- check the current time and wake up any thread in
 644           -- threadDelay whose timeout has expired.  Also find the
 645           -- timeout value for the select() call.
 646           now <- getTicksOfDay
 647           (delays', timeout) <- getDelay now ptimeval delays
 648
 649           res <- c_select ((max wakeup maxfd)+1) readfds writefds
 650                         nullPtr timeout
 651           if (res == -1)
 652              then do
 653                 err <- getErrno
 654                 if err == eINTR
 655                         then do_select delays'
 656                         else return (res,delays')
 657              else
 658                 return (res,delays')
 659
 660   (res,delays') <- do_select delays
 661   -- ToDo: check result
 662
 663   b <- fdIsSet wakeup readfds
 664   if b == 0
 665     then return ()
 666     else alloca $ \p -> do
 667             c_read (fromIntegral wakeup) p 1; return ()
 668             s <- peek p
 669             if (s == 0xff)
 670               then return ()
 671               else do handler_tbl <- peek handlers
 672                       sp <- peekElemOff handler_tbl (fromIntegral s)
 673                       forkIO (do io <- deRefStablePtr sp; io)
 674                       return ()
 675
 676   takeMVar prodding
 677   putMVar prodding False
 678
 679   reqs' <- completeRequests reqs readfds writefds []
 680   service_loop wakeup readfds writefds ptimeval reqs' delays'
 681
 682 stick :: IORef Fd
 683 {-# NOINLINE stick #-}
 684 stick = unsafePerformIO (newIORef 0)
 685
 686 prodding :: MVar Bool
 687 {-# NOINLINE prodding #-}
 688 prodding = unsafePerformIO (newMVar False)
 689
 690 prodServiceThread :: IO ()
 691 prodServiceThread = do
 692   b <- takeMVar prodding
 693   if (not b)
 694     then do fd <- readIORef stick
 695             with 0xff $ \pbuf -> do c_write (fromIntegral fd) pbuf 1; return ()
 696     else return ()
 697   putMVar prodding True
 698
 699 foreign import ccall "&signal_handlers" handlers :: Ptr (Ptr (StablePtr (IO ())))
 700
 701 foreign import ccall "setIOManagerPipe"
 702   c_setIOManagerPipe :: CInt -> IO ()
 703
 704 -- -----------------------------------------------------------------------------
 705 -- IO requests
 706
 707 buildFdSets maxfd readfds writefds [] = return maxfd
 708 buildFdSets maxfd readfds writefds (Read fd m : reqs)
 709   | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
 710   | otherwise        =  do
 711         fdSet fd readfds
 712         buildFdSets (max maxfd fd) readfds writefds reqs
 713 buildFdSets maxfd readfds writefds (Write fd m : reqs)
 714   | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
 715   | otherwise        =  do
 716         fdSet fd writefds
 717         buildFdSets (max maxfd fd) readfds writefds reqs
 718
 719 completeRequests [] _ _ reqs' = return reqs'
 720 completeRequests (Read fd m : reqs) readfds writefds reqs' = do
 721   b <- fdIsSet fd readfds
 722   if b /= 0
 723     then do putMVar m (); completeRequests reqs readfds writefds reqs'
 724     else completeRequests reqs readfds writefds (Read fd m : reqs')
 725 completeRequests (Write fd m : reqs) readfds writefds reqs' = do
 726   b <- fdIsSet fd writefds
 727   if b /= 0
 728     then do putMVar m (); completeRequests reqs readfds writefds reqs'
 729     else completeRequests reqs readfds writefds (Write fd m : reqs')
 730
 731 waitForReadEvent :: Fd -> IO ()
 732 waitForReadEvent fd = do
 733   m <- newEmptyMVar
 734   atomicModifyIORef pendingEvents (\xs -> (Read fd m : xs, ()))
 735   prodServiceThread
 736   takeMVar m
 737
 738 waitForWriteEvent :: Fd -> IO ()
 739 waitForWriteEvent fd = do
 740   m <- newEmptyMVar
 741   atomicModifyIORef pendingEvents (\xs -> (Write fd m : xs, ()))
 742   prodServiceThread
 743   takeMVar m
 744
 745 -- XXX: move into GHC.IOBase from Data.IORef?
 746 atomicModifyIORef :: IORef a -> (a -> (a,b)) -> IO b
 747 atomicModifyIORef (IORef (STRef r#)) f = IO $ \s -> atomicModifyMutVar# r# f s
 748
 749 -- -----------------------------------------------------------------------------
 750 -- Delays
 751
 752 waitForDelayEvent :: Int -> IO ()
 753 waitForDelayEvent usecs = do
 754   m <- newEmptyMVar
 755   now <- getTicksOfDay
 756   let target = now + usecs `quot` tick_usecs
 757   atomicModifyIORef pendingDelays (\xs -> (Delay target m : xs, ()))
 758   prodServiceThread
 759   takeMVar m
 760
 761 -- Delays for use in STM
 762 waitForDelayEventSTM :: Int -> IO (TVar Bool)
 763 waitForDelayEventSTM usecs = do
 764    t <- atomically $ newTVar False
 765    now <- getTicksOfDay
 766    let target = now + usecs `quot` tick_usecs
 767    atomicModifyIORef pendingDelays (\xs -> (DelaySTM target t : xs, ()))
 768    prodServiceThread
 769    return t
 770
 771 -- Walk the queue of pending delays, waking up any that have passed
 772 -- and return the smallest delay to wait for.  The queue of pending
 773 -- delays is kept ordered.
 774 getDelay :: Ticks -> Ptr CTimeVal -> [DelayReq] -> IO ([DelayReq], Ptr CTimeVal)
 775 getDelay now ptimeval [] = return ([],nullPtr)
 776 getDelay now ptimeval all@(d : rest)
 777   = case d of
 778      Delay time m | now >= time -> do
 779         putMVar m ()
 780         getDelay now ptimeval rest
 781      DelaySTM time t | now >= time -> do
 782         atomically $ writeTVar t True
 783         getDelay now ptimeval rest
 784      _otherwise -> do
 785         setTimevalTicks ptimeval (delayTime d - now)
 786         return (all,ptimeval)
 787
 788 insertDelay :: DelayReq -> [DelayReq] -> [DelayReq]
 789 insertDelay d [] = [d]
 790 insertDelay d1 ds@(d2 : rest)
 791   | delayTime d1 <= delayTime d2 = d1 : ds
 792   | otherwise                    = d2 : insertDelay d1 rest
 793
 794 delayTime (Delay t _) = t
 795 delayTime (DelaySTM t _) = t
 796
 797 type Ticks = Int
 798 tick_freq  = 50 :: Ticks  -- accuracy of threadDelay (ticks per sec)
 799 tick_usecs = 1000000 `quot` tick_freq :: Int
 800
 801 newtype CTimeVal = CTimeVal ()
 802
 803 foreign import ccall unsafe "sizeofTimeVal"
 804   sizeofTimeVal :: Int
 805
 806 foreign import ccall unsafe "getTicksOfDay"
 807   getTicksOfDay :: IO Ticks
 808
 809 foreign import ccall unsafe "setTimevalTicks"
 810   setTimevalTicks :: Ptr CTimeVal -> Ticks -> IO ()
 811
 812 -- ----------------------------------------------------------------------------
 813 -- select() interface
 814
 815 -- ToDo: move to System.Posix.Internals?
 816
 817 newtype CFdSet = CFdSet ()
 818
 819 foreign import ccall safe "select"
 820   c_select :: Fd -> Ptr CFdSet -> Ptr CFdSet -> Ptr CFdSet -> Ptr CTimeVal
 821            -> IO CInt
 822
 823 foreign import ccall unsafe "hsFD_SETSIZE"
 824   fD_SETSIZE :: Fd
 825
 826 foreign import ccall unsafe "hsFD_CLR"
 827   fdClr :: Fd -> Ptr CFdSet -> IO ()
 828
 829 foreign import ccall unsafe "hsFD_ISSET"
 830   fdIsSet :: Fd -> Ptr CFdSet -> IO CInt
 831
 832 foreign import ccall unsafe "hsFD_SET"
 833   fdSet :: Fd -> Ptr CFdSet -> IO ()
 834
 835 foreign import ccall unsafe "hsFD_ZERO"
 836   fdZero :: Ptr CFdSet -> IO ()
 837
 838 foreign import ccall unsafe "sizeof_fd_set"
 839   sizeofFdSet :: Int
 840
 841 #endif
 842 \end{code}