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
 256 Support for multiple operating system threads and bound threads as described
 257 below is currently only available in the GHC runtime system if you use the
 258 /-threaded/ option when linking.
 259
 260 Other Haskell systems do not currently support multiple operating system threads.
 261
 262 A bound thread is a haskell thread that is /bound/ to an operating system
 263 thread. While the bound thread is still scheduled by the Haskell run-time
 264 system, the operating system thread takes care of all the foreign calls made
 265 by the bound thread.
 266
 267 To a foreign library, the bound thread will look exactly like an ordinary
 268 operating system thread created using OS functions like @pthread_create@
 269 or @CreateThread@.
 270
 271 Bound threads can be created using the 'forkOS' function below. All foreign
 272 exported functions are run in a bound thread (bound to the OS thread that
 273 called the function). Also, the @main@ action of every Haskell program is
 274 run in a bound thread.
 275
 276 Why do we need this? Because if a foreign library is called from a thread
 277 created using 'forkIO', it won't have access to any /thread-local state/ -
 278 state variables that have specific values for each OS thread
 279 (see POSIX's @pthread_key_create@ or Win32's @TlsAlloc@). Therefore, some
 280 libraries (OpenGL, for example) will not work from a thread created using
 281 'forkIO'. They work fine in threads created using 'forkOS' or when called
 282 from @main@ or from a @foreign export@.
 283 -}
 284
 285 -- | 'True' if bound threads are supported.
 286 -- If @rtsSupportsBoundThreads@ is 'False', 'isCurrentThreadBound'
 287 -- will always return 'False' and both 'forkOS' and 'runInBoundThread' will
 288 -- fail.
 289 foreign import ccall rtsSupportsBoundThreads :: Bool
 290
 291
 292 {- |
 293 Like 'forkIO', this sparks off a new thread to run the 'IO' computation passed as the
 294 first argument, and returns the 'ThreadId' of the newly created
 295 thread.
 296
 297 However, @forkOS@ uses operating system-supplied multithreading support to create
 298 a new operating system thread. The new thread is /bound/, which means that
 299 all foreign calls made by the 'IO' computation are guaranteed to be executed
 300 in this new operating system thread; also, the operating system thread is not
 301 used for any other foreign calls.
 302
 303 This means that you can use all kinds of foreign libraries from this thread
 304 (even those that rely on thread-local state), without the limitations of 'forkIO'.
 305
 306 Just to clarify, 'forkOS' is /only/ necessary if you need to associate
 307 a Haskell thread with a particular OS thread.  It is not necessary if
 308 you only need to make non-blocking foreign calls (see "Control.Concurrent#osthreads").
 309
 310 -}
 311 forkOS :: IO () -> IO ThreadId
 312
 313 foreign export ccall forkOS_entry
 314     :: StablePtr (IO ()) -> IO ()
 315
 316 foreign import ccall "forkOS_entry" forkOS_entry_reimported
 317     :: StablePtr (IO ()) -> IO ()
 318
 319 forkOS_entry stableAction = do
 320         action <- deRefStablePtr stableAction
 321         action
 322
 323 foreign import ccall forkOS_createThread
 324     :: StablePtr (IO ()) -> IO CInt
 325
 326 failNonThreaded = fail $ "RTS doesn't support multiple OS threads "
 327                        ++"(use ghc -threaded when linking)"
 328
 329 forkOS action
 330     | rtsSupportsBoundThreads = do
 331         mv <- newEmptyMVar
 332         let action_plus = Exception.catch action childHandler
 333         entry <- newStablePtr (myThreadId >>= putMVar mv >> action_plus)
 334         err <- forkOS_createThread entry
 335         when (err /= 0) $ fail "Cannot create OS thread."
 336         tid <- takeMVar mv
 337         freeStablePtr entry
 338         return tid
 339     | otherwise = failNonThreaded
 340
 341 -- | Returns 'True' if the calling thread is /bound/, that is, if it is
 342 -- safe to use foreign libraries that rely on thread-local state from the
 343 -- calling thread.
 344 isCurrentThreadBound :: IO Bool
 345 isCurrentThreadBound = IO $ \ s# ->
 346     case isCurrentThreadBound# s# of
 347         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 348
 349
 350 {- |
 351 Run the 'IO' computation passed as the first argument. If the calling thread
 352 is not /bound/, a bound thread is created temporarily. @runInBoundThread@
 353 doesn't finish until the 'IO' computation finishes.
 354
 355 You can wrap a series of foreign function calls that rely on thread-local state
 356 with @runInBoundThread@ so that you can use them without knowing whether the
 357 current thread is /bound/.
 358 -}
 359 runInBoundThread :: IO a -> IO a
 360
 361 runInBoundThread action
 362     | rtsSupportsBoundThreads = do
 363         bound <- isCurrentThreadBound
 364         if bound
 365             then action
 366             else do
 367                 ref <- newIORef undefined
 368                 let action_plus = Exception.try action >>= writeIORef ref
 369                 resultOrException <-
 370                     bracket (newStablePtr action_plus)
 371                             freeStablePtr
 372                             (\cEntry -> forkOS_entry_reimported cEntry >> readIORef ref)
 373                 case resultOrException of
 374                     Left exception -> Exception.throw exception
 375                     Right result -> return result
 376     | otherwise = failNonThreaded
 377
 378 {- |
 379 Run the 'IO' computation passed as the first argument. If the calling thread
 380 is /bound/, an unbound thread is created temporarily using 'forkIO'.
 381 @runInBoundThread@ doesn't finish until the 'IO' computation finishes.
 382
 383 Use this function /only/ in the rare case that you have actually observed a
 384 performance loss due to the use of bound threads. A program that
 385 doesn't need it's main thread to be bound and makes /heavy/ use of concurrency
 386 (e.g. a web server), might want to wrap it's @main@ action in
 387 @runInUnboundThread@.
 388 -}
 389 runInUnboundThread :: IO a -> IO a
 390
 391 runInUnboundThread action = do
 392     bound <- isCurrentThreadBound
 393     if bound
 394         then do
 395             mv <- newEmptyMVar
 396             forkIO (Exception.try action >>= putMVar mv)
 397             takeMVar mv >>= \either -> case either of
 398                 Left exception -> Exception.throw exception
 399                 Right result -> return result
 400         else action
 401
 402 #endif /* __GLASGOW_HASKELL__ */
 403
 404 -- ---------------------------------------------------------------------------
 405 -- More docs
 406
 407 {- $osthreads
 408
 409       #osthreads# In GHC, threads created by 'forkIO' are lightweight threads, and
 410       are managed entirely by the GHC runtime.  Typically Haskell
 411       threads are an order of magnitude or two more efficient (in
 412       terms of both time and space) than operating system threads.
 413
 414       The downside of having lightweight threads is that only one can
 415       run at a time, so if one thread blocks in a foreign call, for
 416       example, the other threads cannot continue.  The GHC runtime
 417       works around this by making use of full OS threads where
 418       necessary.  When the program is built with the @-threaded@
 419       option (to link against the multithreaded version of the
 420       runtime), a thread making a @safe@ foreign call will not block
 421       the other threads in the system; another OS thread will take
 422       over running Haskell threads until the original call returns.
 423       The runtime maintains a pool of these /worker/ threads so that
 424       multiple Haskell threads can be involved in external calls
 425       simultaneously.
 426
 427       The "System.IO" library manages multiplexing in its own way.  On
 428       Windows systems it uses @safe@ foreign calls to ensure that
 429       threads doing I\/O operations don't block the whole runtime,
 430       whereas on Unix systems all the currently blocked I\/O reqwests
 431       are managed by a single thread (the /IO manager thread/) using
 432       @select@.
 433
 434       The runtime will run a Haskell thread using any of the available
 435       worker OS threads.  If you need control over which particular OS
 436       thread is used to run a given Haskell thread, perhaps because
 437       you need to call a foreign library that uses OS-thread-local
 438       state, then you need "bound threads" (see above).
 439
 440       If you don't use the @-threaded@ option, then the runtime does
 441       not make use of multiple OS threads.  Foreign calls will block
 442       all other running Haskell threads until the call returns.  The
 443       "System.IO" library still does multiplexing, so there can be multiple
 444       threads doing I\/O, and this is handled internally by the runtime using
 445       @select@.
 446 -}
 447
 448 {- $termination
 449
 450       In a standalone GHC program, only the main thread is
 451       required to terminate in order for the process to terminate.
 452       Thus all other forked threads will simply terminate at the same
 453       time as the main thread (the terminology for this kind of
 454       behaviour is \"daemonic threads\").
 455
 456       If you want the program to wait for child threads to
 457       finish before exiting, you need to program this yourself.  A
 458       simple mechanism is to have each child thread write to an
 459       'MVar' when it completes, and have the main
 460       thread wait on all the 'MVar's before
 461       exiting:
 462
 463 >   myForkIO :: IO () -> IO (MVar ())
 464 >   myForkIO io = do
 465 >     mvar <- newEmptyMVar
 466 >     forkIO (io `finally` putMVar mvar ())
 467 >     return mvar
 468
 469       Note that we use 'finally' from the
 470       "Control.Exception" module to make sure that the
 471       'MVar' is written to even if the thread dies or
 472       is killed for some reason.
 473
 474       A better method is to keep a global list of all child
 475       threads which we should wait for at the end of the program:
 476
 477 >    children :: MVar [MVar ()]
 478 >    children = unsafePerformIO (newMVar [])
 479 >
 480 >    waitForChildren :: IO ()
 481 >    waitForChildren = do
 482 >      cs <- takeMVar children
 483 >      case cs of
 484 >        []   -> return ()
 485 >        m:ms -> do
 486 >           putMVar children ms
 487 >           takeMVar m
 488 >           waitForChildren
 489 >
 490 >    forkChild :: IO () -> IO ()
 491 >    forkChild io = do
 492 >        mvar <- newEmptyMVar
 493 >        childs <- takeMVar children
 494 >        putMVar children (mvar:childs)
 495 >        forkIO (io `finally` putMVar mvar ())
 496 >
 497 >     main =
 498 >       later waitForChildren $
 499 >       ...
 500
 501       The main thread principle also applies to calls to Haskell from
 502       outside, using @foreign export@.  When the @foreign export@ed
 503       function is invoked, it starts a new main thread, and it returns
 504       when this main thread terminates.  If the call causes new
 505       threads to be forked, they may remain in the system after the
 506       @foreign export@ed function has returned.
 507 -}
 508
 509 {- $preemption
 510
 511       GHC implements pre-emptive multitasking: the execution of
 512       threads are interleaved in a random fashion.  More specifically,
 513       a thread may be pre-empted whenever it allocates some memory,
 514       which unfortunately means that tight loops which do no
 515       allocation tend to lock out other threads (this only seems to
 516       happen with pathological benchmark-style code, however).
 517
 518       The rescheduling timer runs on a 20ms granularity by
 519       default, but this may be altered using the
 520       @-i\<n\>@ RTS option.  After a rescheduling
 521       \"tick\" the running thread is pre-empted as soon as
 522       possible.
 523
 524       One final note: the
 525       @aaaa@ @bbbb@ example may not
 526       work too well on GHC (see Scheduling, above), due
 527       to the locking on a 'System.IO.Handle'.  Only one thread
 528       may hold the lock on a 'System.IO.Handle' at any one
 529       time, so if a reschedule happens while a thread is holding the
 530       lock, the other thread won't be able to run.  The upshot is that
 531       the switch from @aaaa@ to
 532       @bbbbb@ happens infrequently.  It can be
 533       improved by lowering the reschedule tick period.  We also have a
 534       patch that causes a reschedule whenever a thread waiting on a
 535       lock is woken up, but haven't found it to be useful for anything
 536       other than this example :-)
 537 -}