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