Control/Concurrent.hs

   1 -----------------------------------------------------------------------------
   2 -- |
   3 -- Module      :  Control.Concurrent
   4 -- Copyright   :  (c) The University of Glasgow 2001
   5 -- License     :  BSD-style (see the file libraries/base/LICENSE)
   6 --
   7 -- Maintainer  :  libraries@haskell.org
   8 -- Stability   :  experimental
   9 -- Portability :  non-portable (concurrency)
  10 --
  11 -- A common interface to a collection of useful concurrency
  12 -- abstractions.
  13 --
  14 -----------------------------------------------------------------------------
  15
  16 module Control.Concurrent (
  17         -- * Concurrent Haskell
  18
  19         -- $conc_intro
  20
  21         -- * Basic concurrency operations
  22
  23         ThreadId,
  24 #ifdef __GLASGOW_HASKELL__
  25         myThreadId,
  26 #endif
  27
  28         forkIO,
  29 #ifdef __GLASGOW_HASKELL__
  30         killThread,
  31         throwTo,
  32 #endif
  33
  34         -- * Scheduling
  35
  36         -- $conc_scheduling
  37         yield,                  -- :: IO ()
  38
  39         -- ** Blocking
  40
  41         -- $blocking
  42
  43 #ifdef __GLASGOW_HASKELL__
  44         -- ** Waiting
  45         threadDelay,            -- :: Int -> IO ()
  46         threadWaitRead,         -- :: Int -> IO ()
  47         threadWaitWrite,        -- :: Int -> IO ()
  48 #endif
  49
  50         -- * Communication abstractions
  51
  52         module Control.Concurrent.MVar,
  53         module Control.Concurrent.Chan,
  54         module Control.Concurrent.QSem,
  55         module Control.Concurrent.QSemN,
  56         module Control.Concurrent.SampleVar,
  57
  58         -- * Merging of streams
  59 #ifndef __HUGS__
  60         mergeIO,                -- :: [a]   -> [a] -> IO [a]
  61         nmergeIO,               -- :: [[a]] -> IO [a]
  62 #endif
  63         -- $merge
  64
  65 #ifdef __GLASGOW_HASKELL__
  66         -- * Bound Threads
  67         -- $boundthreads
  68         rtsSupportsBoundThreads,
  69         forkOS,
  70         isCurrentThreadBound,
  71         runInBoundThread,
  72         runInUnboundThread
  73 #endif
  74
  75         -- * GHC's implementation of concurrency
  76
  77         -- |This section describes features specific to GHC's
  78         -- implementation of Concurrent Haskell.
  79
  80         -- ** Terminating the program
  81
  82         -- $termination
  83
  84         -- ** Pre-emption
  85
  86         -- $preemption
  87     ) where
  88
  89 import Prelude
  90
  91 import Control.Exception as Exception
  92
  93 #ifdef __GLASGOW_HASKELL__
  94 import GHC.Conc         ( ThreadId(..), myThreadId, killThread, yield,
  95                           threadDelay, threadWaitRead, threadWaitWrite,
  96                           forkIO, childHandler )
  97 import GHC.TopHandler   ( reportStackOverflow, reportError )
  98 import GHC.IOBase       ( IO(..) )
  99 import GHC.IOBase       ( unsafeInterleaveIO )
 100 import GHC.IOBase       ( newIORef, readIORef, writeIORef )
 101 import GHC.Base
 102
 103 import Foreign.StablePtr
 104 import Foreign.C.Types  ( CInt )
 105 import Control.Monad    ( when )
 106 #endif
 107
 108 #ifdef __HUGS__
 109 import Hugs.ConcBase
 110 #endif
 111
 112 import Control.Concurrent.MVar
 113 import Control.Concurrent.Chan
 114 import Control.Concurrent.QSem
 115 import Control.Concurrent.QSemN
 116 import Control.Concurrent.SampleVar
 117
 118 #ifdef __HUGS__
 119 type ThreadId = ()
 120 #endif
 121
 122 {- $conc_intro
 123
 124 The concurrency extension for Haskell is described in the paper
 125 /Concurrent Haskell/
 126 <http://www.haskell.org/ghc/docs/papers/concurrent-haskell.ps.gz>.
 127
 128 Concurrency is \"lightweight\", which means that both thread creation
 129 and context switching overheads are extremely low.  Scheduling of
 130 Haskell threads is done internally in the Haskell runtime system, and
 131 doesn't make use of any operating system-supplied thread packages.
 132
 133 However, if you want to interact with a foreign library that expects your
 134 program to use the operating system-supplied thread package, you can do so
 135 by using 'forkOS' instead of 'forkIO'.
 136
 137 Haskell threads can communicate via 'MVar's, a kind of synchronised
 138 mutable variable (see "Control.Concurrent.MVar").  Several common
 139 concurrency abstractions can be built from 'MVar's, and these are
 140 provided by the "Control.Concurrent" library.
 141 In GHC, threads may also communicate via exceptions.
 142 -}
 143
 144 {- $conc_scheduling
 145
 146     Scheduling may be either pre-emptive or co-operative,
 147     depending on the implementation of Concurrent Haskell (see below
 148     for information related to specific compilers).  In a co-operative
 149     system, context switches only occur when you use one of the
 150     primitives defined in this module.  This means that programs such
 151     as:
 152
 153
 154 >   main = forkIO (write 'a') >> write 'b'
 155 >     where write c = putChar c >> write c
 156
 157     will print either @aaaaaaaaaaaaaa...@ or @bbbbbbbbbbbb...@,
 158     instead of some random interleaving of @a@s and @b@s.  In
 159     practice, cooperative multitasking is sufficient for writing
 160     simple graphical user interfaces.
 161 -}
 162
 163 {- $blocking
 164 Calling a foreign C procedure (such as @getchar@) that blocks waiting
 165 for input will block /all/ threads, unless the @threadsafe@ attribute
 166 is used on the foreign call (and your compiler \/ operating system
 167 supports it).  GHC's I\/O system uses non-blocking I\/O internally to
 168 implement thread-friendly I\/O, so calling standard Haskell I\/O
 169 functions blocks only the thread making the call.
 170 -}
 171
 172 #ifndef __HUGS__
 173 max_buff_size :: Int
 174 max_buff_size = 1
 175
 176 mergeIO :: [a] -> [a] -> IO [a]
 177 nmergeIO :: [[a]] -> IO [a]
 178
 179 -- $merge
 180 -- The 'mergeIO' and 'nmergeIO' functions fork one thread for each
 181 -- input list that concurrently evaluates that list; the results are
 182 -- merged into a single output list.
 183 --
 184 -- Note: Hugs does not provide these functions, since they require
 185 -- preemptive multitasking.
 186
 187 mergeIO ls rs
 188  = newEmptyMVar                >>= \ tail_node ->
 189    newMVar tail_node           >>= \ tail_list ->
 190    newQSem max_buff_size       >>= \ e ->
 191    newMVar 2                   >>= \ branches_running ->
 192    let
 193     buff = (tail_list,e)
 194    in
 195     forkIO (suckIO branches_running buff ls) >>
 196     forkIO (suckIO branches_running buff rs) >>
 197     takeMVar tail_node  >>= \ val ->
 198     signalQSem e        >>
 199     return val
 200
 201 type Buffer a
 202  = (MVar (MVar [a]), QSem)
 203
 204 suckIO :: MVar Int -> Buffer a -> [a] -> IO ()
 205
 206 suckIO branches_running buff@(tail_list,e) vs
 207  = case vs of
 208         [] -> takeMVar branches_running >>= \ val ->
 209               if val == 1 then
 210                  takeMVar tail_list     >>= \ node ->
 211                  putMVar node []        >>
 212                  putMVar tail_list node
 213               else
 214                  putMVar branches_running (val-1)
 215         (x:xs) ->
 216                 waitQSem e                       >>
 217                 takeMVar tail_list               >>= \ node ->
 218                 newEmptyMVar                     >>= \ next_node ->
 219                 unsafeInterleaveIO (
 220                         takeMVar next_node  >>= \ y ->
 221                         signalQSem e        >>
 222                         return y)                >>= \ next_node_val ->
 223                 putMVar node (x:next_node_val)   >>
 224                 putMVar tail_list next_node      >>
 225                 suckIO branches_running buff xs
 226
 227 nmergeIO lss
 228  = let
 229     len = length lss
 230    in
 231     newEmptyMVar          >>= \ tail_node ->
 232     newMVar tail_node     >>= \ tail_list ->
 233     newQSem max_buff_size >>= \ e ->
 234     newMVar len           >>= \ branches_running ->
 235     let
 236      buff = (tail_list,e)
 237     in
 238     mapIO (\ x -> forkIO (suckIO branches_running buff x)) lss >>
 239     takeMVar tail_node  >>= \ val ->
 240     signalQSem e        >>
 241     return val
 242   where
 243     mapIO f xs = sequence (map f xs)
 244 #endif /* __HUGS__ */
 245
 246 #ifdef __GLASGOW_HASKELL__
 247 -- ---------------------------------------------------------------------------
 248 -- Bound Threads
 249
 250 {- $boundthreads
 251
 252 Support for multiple operating system threads and bound threads as described
 253 below is currently only available in the GHC runtime system if you use the
 254 /-threaded/ option when linking.
 255
 256 Other Haskell systems do not currently support multiple operating system threads.
 257
 258 A bound thread is a haskell thread that is /bound/ to an operating system
 259 thread. While the bound thread is still scheduled by the Haskell run-time
 260 system, the operating system thread takes care of all the foreign calls made
 261 by the bound thread.
 262
 263 To a foreign library, the bound thread will look exactly like an ordinary
 264 operating system thread created using OS functions like @pthread_create@
 265 or @CreateThread@.
 266
 267 Bound threads can be created using the 'forkOS' function below. All foreign
 268 exported functions are run in a bound thread (bound to the OS thread that
 269 called the function). Also, the @main@ action of every Haskell program is
 270 run in a bound thread.
 271
 272 Why do we need this? Because if a foreign library is called from a thread
 273 created using 'forkIO', it won't have access to any /thread-local state/ -
 274 state variables that have specific values for each OS thread
 275 (see POSIX's @pthread_key_create@ or Win32's @TlsAlloc@). Therefore, some
 276 libraries (OpenGL, for example) will not work from a thread created using
 277 'forkIO'. They work fine in threads created using 'forkOS' or when called
 278 from @main@ or from a @foreign export@.
 279 -}
 280
 281 -- | 'True' if bound threads are supported.
 282 -- If @rtsSupportsBoundThreads@ is 'False', 'isCurrentThreadBound'
 283 -- will always return 'False' and both 'forkOS' and 'runInBoundThread' will
 284 -- fail.
 285 foreign import ccall rtsSupportsBoundThreads :: Bool
 286
 287
 288 {- |
 289 Like 'forkIO', this sparks off a new thread to run the 'IO' computation passed as the
 290 first argument, and returns the 'ThreadId' of the newly created
 291 thread.
 292
 293 However, @forkOS@ uses operating system-supplied multithreading support to create
 294 a new operating system thread. The new thread is /bound/, which means that
 295 all foreign calls made by the 'IO' computation are guaranteed to be executed
 296 in this new operating system thread; also, the operating system thread is not
 297 used for any other foreign calls.
 298
 299 This means that you can use all kinds of foreign libraries from this thread
 300 (even those that rely on thread-local state), without the limitations of 'forkIO'.
 301 -}
 302 forkOS :: IO () -> IO ThreadId
 303
 304 foreign export ccall forkOS_entry
 305     :: StablePtr (IO ()) -> IO ()
 306
 307 foreign import ccall "forkOS_entry" forkOS_entry_reimported
 308     :: StablePtr (IO ()) -> IO ()
 309
 310 forkOS_entry stableAction = do
 311         action <- deRefStablePtr stableAction
 312         action
 313
 314 foreign import ccall forkOS_createThread
 315     :: StablePtr (IO ()) -> IO CInt
 316
 317 failNonThreaded = fail $ "RTS doesn't support multiple OS threads "
 318                        ++"(use ghc -threaded when linking)"
 319
 320 forkOS action
 321     | rtsSupportsBoundThreads = do
 322         mv <- newEmptyMVar
 323         let action_plus = Exception.catch action childHandler
 324         entry <- newStablePtr (myThreadId >>= putMVar mv >> action_plus)
 325         err <- forkOS_createThread entry
 326         when (err /= 0) $ fail "Cannot create OS thread."
 327         tid <- takeMVar mv
 328         freeStablePtr entry
 329         return tid
 330     | otherwise = failNonThreaded
 331
 332 -- | Returns 'True' if the calling thread is /bound/, that is, if it is
 333 -- safe to use foreign libraries that rely on thread-local state from the
 334 -- calling thread.
 335 isCurrentThreadBound :: IO Bool
 336 isCurrentThreadBound = IO $ \ s# ->
 337     case isCurrentThreadBound# s# of
 338         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 339
 340
 341 {- |
 342 Run the 'IO' computation passed as the first argument. If the calling thread
 343 is not /bound/, a bound thread is created temporarily. @runInBoundThread@
 344 doesn't finish until the 'IO' computation finishes.
 345
 346 You can wrap a series of foreign function calls that rely on thread-local state
 347 with @runInBoundThread@ so that you can use them without knowing whether the
 348 current thread is /bound/.
 349 -}
 350 runInBoundThread :: IO a -> IO a
 351
 352 runInBoundThread action
 353     | rtsSupportsBoundThreads = do
 354         bound <- isCurrentThreadBound
 355         if bound
 356             then action
 357             else do
 358                 ref <- newIORef undefined
 359                 let action_plus = Exception.try action >>= writeIORef ref
 360                 resultOrException <-
 361                     bracket (newStablePtr action_plus)
 362                             freeStablePtr
 363                             (\cEntry -> forkOS_entry_reimported cEntry >> readIORef ref)
 364                 case resultOrException of
 365                     Left exception -> Exception.throw exception
 366                     Right result -> return result
 367     | otherwise = failNonThreaded
 368
 369 {- |
 370 Run the 'IO' computation passed as the first argument. If the calling thread
 371 is /bound/, an unbound thread is created temporarily using 'forkIO'.
 372 @runInBoundThread@ doesn't finish until the 'IO' computation finishes.
 373
 374 Use this function /only/ in the rare case that you have actually observed a
 375 performance loss due to the use of bound threads. A program that
 376 doesn't need it's main thread to be bound and makes /heavy/ use of concurrency
 377 (e.g. a web server), might want to wrap it's @main@ action in
 378 @runInUnboundThread@.
 379 -}
 380 runInUnboundThread :: IO a -> IO a
 381
 382 runInUnboundThread action = do
 383     bound <- isCurrentThreadBound
 384     if bound
 385         then do
 386             mv <- newEmptyMVar
 387             forkIO (Exception.try action >>= putMVar mv)
 388             takeMVar mv >>= \either -> case either of
 389                 Left exception -> Exception.throw exception
 390                 Right result -> return result
 391         else action
 392
 393 #endif /* __GLASGOW_HASKELL__ */
 394
 395 -- ---------------------------------------------------------------------------
 396 -- More docs
 397
 398 {- $termination
 399
 400       In a standalone GHC program, only the main thread is
 401       required to terminate in order for the process to terminate.
 402       Thus all other forked threads will simply terminate at the same
 403       time as the main thread (the terminology for this kind of
 404       behaviour is \"daemonic threads\").
 405
 406       If you want the program to wait for child threads to
 407       finish before exiting, you need to program this yourself.  A
 408       simple mechanism is to have each child thread write to an
 409       'MVar' when it completes, and have the main
 410       thread wait on all the 'MVar's before
 411       exiting:
 412
 413 >   myForkIO :: IO () -> IO (MVar ())
 414 >   myForkIO io = do
 415 >     mvar <- newEmptyMVar
 416 >     forkIO (io `finally` putMVar mvar ())
 417 >     return mvar
 418
 419       Note that we use 'finally' from the
 420       "Control.Exception" module to make sure that the
 421       'MVar' is written to even if the thread dies or
 422       is killed for some reason.
 423
 424       A better method is to keep a global list of all child
 425       threads which we should wait for at the end of the program:
 426
 427 >    children :: MVar [MVar ()]
 428 >    children = unsafePerformIO (newMVar [])
 429 >
 430 >    waitForChildren :: IO ()
 431 >    waitForChildren = do
 432 >      cs <- takeMVar children
 433 >      case cs of
 434 >        []   -> return ()
 435 >        m:ms -> do
 436 >           putMVar children ms
 437 >           takeMVar m
 438 >           waitForChildren
 439 >
 440 >    forkChild :: IO () -> IO ()
 441 >    forkChild io = do
 442 >        mvar <- newEmptyMVar
 443 >        childs <- takeMVar children
 444 >        putMVar children (mvar:childs)
 445 >        forkIO (io `finally` putMVar mvar ())
 446 >
 447 >     main =
 448 >       later waitForChildren $
 449 >       ...
 450
 451       The main thread principle also applies to calls to Haskell from
 452       outside, using @foreign export@.  When the @foreign export@ed
 453       function is invoked, it starts a new main thread, and it returns
 454       when this main thread terminates.  If the call causes new
 455       threads to be forked, they may remain in the system after the
 456       @foreign export@ed function has returned.
 457 -}
 458
 459 {- $preemption
 460
 461       GHC implements pre-emptive multitasking: the execution of
 462       threads are interleaved in a random fashion.  More specifically,
 463       a thread may be pre-empted whenever it allocates some memory,
 464       which unfortunately means that tight loops which do no
 465       allocation tend to lock out other threads (this only seems to
 466       happen with pathological benchmark-style code, however).
 467
 468       The rescheduling timer runs on a 20ms granularity by
 469       default, but this may be altered using the
 470       @-i\<n\>@ RTS option.  After a rescheduling
 471       \"tick\" the running thread is pre-empted as soon as
 472       possible.
 473
 474       One final note: the
 475       @aaaa@ @bbbb@ example may not
 476       work too well on GHC (see Scheduling, above), due
 477       to the locking on a 'System.IO.Handle'.  Only one thread
 478       may hold the lock on a 'System.IO.Handle' at any one
 479       time, so if a reschedule happens while a thread is holding the
 480       lock, the other thread won't be able to run.  The upshot is that
 481       the switch from @aaaa@ to
 482       @bbbbb@ happens infrequently.  It can be
 483       improved by lowering the reschedule tick period.  We also have a
 484       patch that causes a reschedule whenever a thread waiting on a
 485       lock is woken up, but haven't found it to be useful for anything
 486       other than this example :-)
 487 -}