Control/Concurrent.hs

   1 {-# OPTIONS_GHC -fno-warn-unused-imports #-}
   2 -----------------------------------------------------------------------------
   3 -- |
   4 -- Module      :  Control.Concurrent
   5 -- Copyright   :  (c) The University of Glasgow 2001
   6 -- License     :  BSD-style (see the file libraries/base/LICENSE)
   7 --
   8 -- Maintainer  :  libraries@haskell.org
   9 -- Stability   :  experimental
  10 -- Portability :  non-portable (concurrency)
  11 --
  12 -- A common interface to a collection of useful concurrency
  13 -- abstractions.
  14 --
  15 -----------------------------------------------------------------------------
  16
  17 module Control.Concurrent (
  18         -- * Concurrent Haskell
  19
  20         -- $conc_intro
  21
  22         -- * Basic concurrency operations
  23
  24         ThreadId,
  25 #ifdef __GLASGOW_HASKELL__
  26         myThreadId,
  27 #endif
  28
  29         forkIO,
  30 #ifdef __GLASGOW_HASKELL__
  31         killThread,
  32         throwTo,
  33 #endif
  34
  35         -- * Scheduling
  36
  37         -- $conc_scheduling
  38         yield,                  -- :: IO ()
  39
  40         -- ** Blocking
  41
  42         -- $blocking
  43
  44 #ifdef __GLASGOW_HASKELL__
  45         -- ** Waiting
  46         threadDelay,            -- :: Int -> IO ()
  47         threadWaitRead,         -- :: Int -> IO ()
  48         threadWaitWrite,        -- :: Int -> IO ()
  49 #endif
  50
  51         -- * Communication abstractions
  52
  53         module Control.Concurrent.MVar,
  54         module Control.Concurrent.Chan,
  55         module Control.Concurrent.QSem,
  56         module Control.Concurrent.QSemN,
  57         module Control.Concurrent.SampleVar,
  58
  59         -- * Merging of streams
  60 #ifndef __HUGS__
  61         mergeIO,                -- :: [a]   -> [a] -> IO [a]
  62         nmergeIO,               -- :: [[a]] -> IO [a]
  63 #endif
  64         -- $merge
  65
  66 #ifdef __GLASGOW_HASKELL__
  67         -- * Bound Threads
  68         -- $boundthreads
  69         rtsSupportsBoundThreads,
  70         forkOS,
  71         isCurrentThreadBound,
  72         runInBoundThread,
  73         runInUnboundThread
  74 #endif
  75
  76         -- * GHC's implementation of concurrency
  77
  78         -- |This section describes features specific to GHC's
  79         -- implementation of Concurrent Haskell.
  80
  81         -- ** Haskell threads and Operating System threads
  82
  83         -- $osthreads
  84
  85         -- ** Terminating the program
  86
  87         -- $termination
  88
  89         -- ** Pre-emption
  90
  91         -- $preemption
  92     ) where
  93
  94 import Prelude
  95
  96 import Control.Exception.Base as Exception
  97
  98 #ifdef __GLASGOW_HASKELL__
  99 import GHC.Exception
 100 import GHC.Conc         ( ThreadId(..), myThreadId, killThread, yield,
 101                           threadDelay, forkIO, childHandler )
 102 import qualified GHC.Conc
 103 import GHC.IO           ( IO(..), unsafeInterleaveIO )
 104 import GHC.IORef        ( newIORef, readIORef, writeIORef )
 105 import GHC.Base
 106
 107 import System.Posix.Types ( Fd )
 108 import Foreign.StablePtr
 109 import Foreign.C.Types  ( CInt )
 110 import Control.Monad    ( when )
 111
 112 #ifdef mingw32_HOST_OS
 113 import Foreign.C
 114 import System.IO
 115 #endif
 116 #endif
 117
 118 #ifdef __HUGS__
 119 import Hugs.ConcBase
 120 #endif
 121
 122 import Control.Concurrent.MVar
 123 import Control.Concurrent.Chan
 124 import Control.Concurrent.QSem
 125 import Control.Concurrent.QSemN
 126 import Control.Concurrent.SampleVar
 127
 128 #ifdef __HUGS__
 129 type ThreadId = ()
 130 #endif
 131
 132 {- $conc_intro
 133
 134 The concurrency extension for Haskell is described in the paper
 135 /Concurrent Haskell/
 136 <http://www.haskell.org/ghc/docs/papers/concurrent-haskell.ps.gz>.
 137
 138 Concurrency is \"lightweight\", which means that both thread creation
 139 and context switching overheads are extremely low.  Scheduling of
 140 Haskell threads is done internally in the Haskell runtime system, and
 141 doesn't make use of any operating system-supplied thread packages.
 142
 143 However, if you want to interact with a foreign library that expects your
 144 program to use the operating system-supplied thread package, you can do so
 145 by using 'forkOS' instead of 'forkIO'.
 146
 147 Haskell threads can communicate via 'MVar's, a kind of synchronised
 148 mutable variable (see "Control.Concurrent.MVar").  Several common
 149 concurrency abstractions can be built from 'MVar's, and these are
 150 provided by the "Control.Concurrent" library.
 151 In GHC, threads may also communicate via exceptions.
 152 -}
 153
 154 {- $conc_scheduling
 155
 156     Scheduling may be either pre-emptive or co-operative,
 157     depending on the implementation of Concurrent Haskell (see below
 158     for information related to specific compilers).  In a co-operative
 159     system, context switches only occur when you use one of the
 160     primitives defined in this module.  This means that programs such
 161     as:
 162
 163
 164 >   main = forkIO (write 'a') >> write 'b'
 165 >     where write c = putChar c >> write c
 166
 167     will print either @aaaaaaaaaaaaaa...@ or @bbbbbbbbbbbb...@,
 168     instead of some random interleaving of @a@s and @b@s.  In
 169     practice, cooperative multitasking is sufficient for writing
 170     simple graphical user interfaces.
 171 -}
 172
 173 {- $blocking
 174 Different Haskell implementations have different characteristics with
 175 regard to which operations block /all/ threads.
 176
 177 Using GHC without the @-threaded@ option, all foreign calls will block
 178 all other Haskell threads in the system, although I\/O operations will
 179 not.  With the @-threaded@ option, only foreign calls with the @unsafe@
 180 attribute will block all other threads.
 181
 182 Using Hugs, all I\/O operations and foreign calls will block all other
 183 Haskell threads.
 184 -}
 185
 186 #ifndef __HUGS__
 187 max_buff_size :: Int
 188 max_buff_size = 1
 189
 190 mergeIO :: [a] -> [a] -> IO [a]
 191 nmergeIO :: [[a]] -> IO [a]
 192
 193 -- $merge
 194 -- The 'mergeIO' and 'nmergeIO' functions fork one thread for each
 195 -- input list that concurrently evaluates that list; the results are
 196 -- merged into a single output list.
 197 --
 198 -- Note: Hugs does not provide these functions, since they require
 199 -- preemptive multitasking.
 200
 201 mergeIO ls rs
 202  = newEmptyMVar                >>= \ tail_node ->
 203    newMVar tail_node           >>= \ tail_list ->
 204    newQSem max_buff_size       >>= \ e ->
 205    newMVar 2                   >>= \ branches_running ->
 206    let
 207     buff = (tail_list,e)
 208    in
 209     forkIO (suckIO branches_running buff ls) >>
 210     forkIO (suckIO branches_running buff rs) >>
 211     takeMVar tail_node  >>= \ val ->
 212     signalQSem e        >>
 213     return val
 214
 215 type Buffer a
 216  = (MVar (MVar [a]), QSem)
 217
 218 suckIO :: MVar Int -> Buffer a -> [a] -> IO ()
 219
 220 suckIO branches_running buff@(tail_list,e) vs
 221  = case vs of
 222         [] -> takeMVar branches_running >>= \ val ->
 223               if val == 1 then
 224                  takeMVar tail_list     >>= \ node ->
 225                  putMVar node []        >>
 226                  putMVar tail_list node
 227               else
 228                  putMVar branches_running (val-1)
 229         (x:xs) ->
 230                 waitQSem e                       >>
 231                 takeMVar tail_list               >>= \ node ->
 232                 newEmptyMVar                     >>= \ next_node ->
 233                 unsafeInterleaveIO (
 234                         takeMVar next_node  >>= \ y ->
 235                         signalQSem e        >>
 236                         return y)                >>= \ next_node_val ->
 237                 putMVar node (x:next_node_val)   >>
 238                 putMVar tail_list next_node      >>
 239                 suckIO branches_running buff xs
 240
 241 nmergeIO lss
 242  = let
 243     len = length lss
 244    in
 245     newEmptyMVar          >>= \ tail_node ->
 246     newMVar tail_node     >>= \ tail_list ->
 247     newQSem max_buff_size >>= \ e ->
 248     newMVar len           >>= \ branches_running ->
 249     let
 250      buff = (tail_list,e)
 251     in
 252     mapIO (\ x -> forkIO (suckIO branches_running buff x)) lss >>
 253     takeMVar tail_node  >>= \ val ->
 254     signalQSem e        >>
 255     return val
 256   where
 257     mapIO f xs = sequence (map f xs)
 258 #endif /* __HUGS__ */
 259
 260 #ifdef __GLASGOW_HASKELL__
 261 -- ---------------------------------------------------------------------------
 262 -- Bound Threads
 263
 264 {- $boundthreads
 265    #boundthreads#
 266
 267 Support for multiple operating system threads and bound threads as described
 268 below is currently only available in the GHC runtime system if you use the
 269 /-threaded/ option when linking.
 270
 271 Other Haskell systems do not currently support multiple operating system threads.
 272
 273 A bound thread is a haskell thread that is /bound/ to an operating system
 274 thread. While the bound thread is still scheduled by the Haskell run-time
 275 system, the operating system thread takes care of all the foreign calls made
 276 by the bound thread.
 277
 278 To a foreign library, the bound thread will look exactly like an ordinary
 279 operating system thread created using OS functions like @pthread_create@
 280 or @CreateThread@.
 281
 282 Bound threads can be created using the 'forkOS' function below. All foreign
 283 exported functions are run in a bound thread (bound to the OS thread that
 284 called the function). Also, the @main@ action of every Haskell program is
 285 run in a bound thread.
 286
 287 Why do we need this? Because if a foreign library is called from a thread
 288 created using 'forkIO', it won't have access to any /thread-local state/ -
 289 state variables that have specific values for each OS thread
 290 (see POSIX's @pthread_key_create@ or Win32's @TlsAlloc@). Therefore, some
 291 libraries (OpenGL, for example) will not work from a thread created using
 292 'forkIO'. They work fine in threads created using 'forkOS' or when called
 293 from @main@ or from a @foreign export@.
 294
 295 In terms of performance, 'forkOS' (aka bound) threads are much more
 296 expensive than 'forkIO' (aka unbound) threads, because a 'forkOS'
 297 thread is tied to a particular OS thread, whereas a 'forkIO' thread
 298 can be run by any OS thread.  Context-switching between a 'forkOS'
 299 thread and a 'forkIO' thread is many times more expensive than between
 300 two 'forkIO' threads.
 301
 302 Note in particular that the main program thread (the thread running
 303 @Main.main@) is always a bound thread, so for good concurrency
 304 performance you should ensure that the main thread is not doing
 305 repeated communication with other threads in the system.  Typically
 306 this means forking subthreads to do the work using 'forkIO', and
 307 waiting for the results in the main thread.
 308
 309 -}
 310
 311 -- | 'True' if bound threads are supported.
 312 -- If @rtsSupportsBoundThreads@ is 'False', 'isCurrentThreadBound'
 313 -- will always return 'False' and both 'forkOS' and 'runInBoundThread' will
 314 -- fail.
 315 foreign import ccall rtsSupportsBoundThreads :: Bool
 316
 317
 318 {- |
 319 Like 'forkIO', this sparks off a new thread to run the 'IO'
 320 computation passed as the first argument, and returns the 'ThreadId'
 321 of the newly created thread.
 322
 323 However, 'forkOS' creates a /bound/ thread, which is necessary if you
 324 need to call foreign (non-Haskell) libraries that make use of
 325 thread-local state, such as OpenGL (see "Control.Concurrent#boundthreads").
 326
 327 Using 'forkOS' instead of 'forkIO' makes no difference at all to the
 328 scheduling behaviour of the Haskell runtime system.  It is a common
 329 misconception that you need to use 'forkOS' instead of 'forkIO' to
 330 avoid blocking all the Haskell threads when making a foreign call;
 331 this isn't the case.  To allow foreign calls to be made without
 332 blocking all the Haskell threads (with GHC), it is only necessary to
 333 use the @-threaded@ option when linking your program, and to make sure
 334 the foreign import is not marked @unsafe@.
 335 -}
 336
 337 forkOS :: IO () -> IO ThreadId
 338
 339 foreign export ccall forkOS_entry
 340     :: StablePtr (IO ()) -> IO ()
 341
 342 foreign import ccall "forkOS_entry" forkOS_entry_reimported
 343     :: StablePtr (IO ()) -> IO ()
 344
 345 forkOS_entry :: StablePtr (IO ()) -> IO ()
 346 forkOS_entry stableAction = do
 347         action <- deRefStablePtr stableAction
 348         action
 349
 350 foreign import ccall forkOS_createThread
 351     :: StablePtr (IO ()) -> IO CInt
 352
 353 failNonThreaded :: IO a
 354 failNonThreaded = fail $ "RTS doesn't support multiple OS threads "
 355                        ++"(use ghc -threaded when linking)"
 356
 357 forkOS action0
 358     | rtsSupportsBoundThreads = do
 359         mv <- newEmptyMVar
 360         b <- Exception.blocked
 361         let
 362             -- async exceptions are blocked in the child if they are blocked
 363             -- in the parent, as for forkIO (see #1048). forkOS_createThread
 364             -- creates a thread with exceptions blocked by default.
 365             action1 | b = action0
 366                     | otherwise = unblock action0
 367
 368             action_plus = Exception.catch action1 childHandler
 369
 370         entry <- newStablePtr (myThreadId >>= putMVar mv >> action_plus)
 371         err <- forkOS_createThread entry
 372         when (err /= 0) $ fail "Cannot create OS thread."
 373         tid <- takeMVar mv
 374         freeStablePtr entry
 375         return tid
 376     | otherwise = failNonThreaded
 377
 378 -- | Returns 'True' if the calling thread is /bound/, that is, if it is
 379 -- safe to use foreign libraries that rely on thread-local state from the
 380 -- calling thread.
 381 isCurrentThreadBound :: IO Bool
 382 isCurrentThreadBound = IO $ \ s# ->
 383     case isCurrentThreadBound# s# of
 384         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 385
 386
 387 {- |
 388 Run the 'IO' computation passed as the first argument. If the calling thread
 389 is not /bound/, a bound thread is created temporarily. @runInBoundThread@
 390 doesn't finish until the 'IO' computation finishes.
 391
 392 You can wrap a series of foreign function calls that rely on thread-local state
 393 with @runInBoundThread@ so that you can use them without knowing whether the
 394 current thread is /bound/.
 395 -}
 396 runInBoundThread :: IO a -> IO a
 397
 398 runInBoundThread action
 399     | rtsSupportsBoundThreads = do
 400         bound <- isCurrentThreadBound
 401         if bound
 402             then action
 403             else do
 404                 ref <- newIORef undefined
 405                 let action_plus = Exception.try action >>= writeIORef ref
 406                 resultOrException <-
 407                     bracket (newStablePtr action_plus)
 408                             freeStablePtr
 409                             (\cEntry -> forkOS_entry_reimported cEntry >> readIORef ref)
 410                 case resultOrException of
 411                     Left exception -> Exception.throw (exception :: SomeException)
 412                     Right result -> return result
 413     | otherwise = failNonThreaded
 414
 415 {- |
 416 Run the 'IO' computation passed as the first argument. If the calling thread
 417 is /bound/, an unbound thread is created temporarily using 'forkIO'.
 418 @runInBoundThread@ doesn't finish until the 'IO' computation finishes.
 419
 420 Use this function /only/ in the rare case that you have actually observed a
 421 performance loss due to the use of bound threads. A program that
 422 doesn't need it's main thread to be bound and makes /heavy/ use of concurrency
 423 (e.g. a web server), might want to wrap it's @main@ action in
 424 @runInUnboundThread@.
 425 -}
 426 runInUnboundThread :: IO a -> IO a
 427
 428 runInUnboundThread action = do
 429     bound <- isCurrentThreadBound
 430     if bound
 431         then do
 432             mv <- newEmptyMVar
 433             forkIO (Exception.try action >>= putMVar mv)
 434             takeMVar mv >>= \ei -> case ei of
 435                 Left exception -> Exception.throw (exception :: SomeException)
 436                 Right result -> return result
 437         else action
 438
 439 #endif /* __GLASGOW_HASKELL__ */
 440
 441 #ifdef __GLASGOW_HASKELL__
 442 -- ---------------------------------------------------------------------------
 443 -- threadWaitRead/threadWaitWrite
 444
 445 -- | Block the current thread until data is available to read on the
 446 -- given file descriptor (GHC only).
 447 threadWaitRead :: Fd -> IO ()
 448 threadWaitRead fd
 449 #ifdef mingw32_HOST_OS
 450   -- we have no IO manager implementing threadWaitRead on Windows.
 451   -- fdReady does the right thing, but we have to call it in a
 452   -- separate thread, otherwise threadWaitRead won't be interruptible,
 453   -- and this only works with -threaded.
 454   | threaded  = withThread (waitFd fd 0)
 455   | otherwise = case fd of
 456                   0 -> do hWaitForInput stdin (-1); return ()
 457                         -- hWaitForInput does work properly, but we can only
 458                         -- do this for stdin since we know its FD.
 459                   _ -> error "threadWaitRead requires -threaded on Windows, or use System.IO.hWaitForInput"
 460 #else
 461   = GHC.Conc.threadWaitRead fd
 462 #endif
 463
 464 -- | Block the current thread until data can be written to the
 465 -- given file descriptor (GHC only).
 466 threadWaitWrite :: Fd -> IO ()
 467 threadWaitWrite fd
 468 #ifdef mingw32_HOST_OS
 469   | threaded  = withThread (waitFd fd 1)
 470   | otherwise = error "threadWaitWrite requires -threaded on Windows"
 471 #else
 472   = GHC.Conc.threadWaitWrite fd
 473 #endif
 474
 475 #ifdef mingw32_HOST_OS
 476 foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
 477
 478 withThread :: IO a -> IO a
 479 withThread io = do
 480   m <- newEmptyMVar
 481   forkIO $ try io >>= putMVar m
 482   x <- takeMVar m
 483   case x of
 484     Right a -> return a
 485     Left e  -> throwIO (e :: IOException)
 486
 487 waitFd :: Fd -> CInt -> IO ()
 488 waitFd fd write = do
 489    throwErrnoIfMinus1 "fdReady" $
 490         fdReady (fromIntegral fd) write (fromIntegral iNFINITE) 0
 491    return ()
 492
 493 iNFINITE :: CInt
 494 iNFINITE = 0xFFFFFFFF -- urgh
 495
 496 foreign import ccall safe "fdReady"
 497   fdReady :: CInt -> CInt -> CInt -> CInt -> IO CInt
 498 #endif
 499
 500 -- ---------------------------------------------------------------------------
 501 -- More docs
 502
 503 {- $osthreads
 504
 505       #osthreads# In GHC, threads created by 'forkIO' are lightweight threads, and
 506       are managed entirely by the GHC runtime.  Typically Haskell
 507       threads are an order of magnitude or two more efficient (in
 508       terms of both time and space) than operating system threads.
 509
 510       The downside of having lightweight threads is that only one can
 511       run at a time, so if one thread blocks in a foreign call, for
 512       example, the other threads cannot continue.  The GHC runtime
 513       works around this by making use of full OS threads where
 514       necessary.  When the program is built with the @-threaded@
 515       option (to link against the multithreaded version of the
 516       runtime), a thread making a @safe@ foreign call will not block
 517       the other threads in the system; another OS thread will take
 518       over running Haskell threads until the original call returns.
 519       The runtime maintains a pool of these /worker/ threads so that
 520       multiple Haskell threads can be involved in external calls
 521       simultaneously.
 522
 523       The "System.IO" library manages multiplexing in its own way.  On
 524       Windows systems it uses @safe@ foreign calls to ensure that
 525       threads doing I\/O operations don't block the whole runtime,
 526       whereas on Unix systems all the currently blocked I\/O requests
 527       are managed by a single thread (the /IO manager thread/) using
 528       @select@.
 529
 530       The runtime will run a Haskell thread using any of the available
 531       worker OS threads.  If you need control over which particular OS
 532       thread is used to run a given Haskell thread, perhaps because
 533       you need to call a foreign library that uses OS-thread-local
 534       state, then you need bound threads (see "Control.Concurrent#boundthreads").
 535
 536       If you don't use the @-threaded@ option, then the runtime does
 537       not make use of multiple OS threads.  Foreign calls will block
 538       all other running Haskell threads until the call returns.  The
 539       "System.IO" library still does multiplexing, so there can be multiple
 540       threads doing I\/O, and this is handled internally by the runtime using
 541       @select@.
 542 -}
 543
 544 {- $termination
 545
 546       In a standalone GHC program, only the main thread is
 547       required to terminate in order for the process to terminate.
 548       Thus all other forked threads will simply terminate at the same
 549       time as the main thread (the terminology for this kind of
 550       behaviour is \"daemonic threads\").
 551
 552       If you want the program to wait for child threads to
 553       finish before exiting, you need to program this yourself.  A
 554       simple mechanism is to have each child thread write to an
 555       'MVar' when it completes, and have the main
 556       thread wait on all the 'MVar's before
 557       exiting:
 558
 559 >   myForkIO :: IO () -> IO (MVar ())
 560 >   myForkIO io = do
 561 >     mvar <- newEmptyMVar
 562 >     forkIO (io `finally` putMVar mvar ())
 563 >     return mvar
 564
 565       Note that we use 'finally' from the
 566       "Control.Exception" module to make sure that the
 567       'MVar' is written to even if the thread dies or
 568       is killed for some reason.
 569
 570       A better method is to keep a global list of all child
 571       threads which we should wait for at the end of the program:
 572
 573 >    children :: MVar [MVar ()]
 574 >    children = unsafePerformIO (newMVar [])
 575 >
 576 >    waitForChildren :: IO ()
 577 >    waitForChildren = do
 578 >      cs <- takeMVar children
 579 >      case cs of
 580 >        []   -> return ()
 581 >        m:ms -> do
 582 >           putMVar children ms
 583 >           takeMVar m
 584 >           waitForChildren
 585 >
 586 >    forkChild :: IO () -> IO ThreadId
 587 >    forkChild io = do
 588 >        mvar <- newEmptyMVar
 589 >        childs <- takeMVar children
 590 >        putMVar children (mvar:childs)
 591 >        forkIO (io `finally` putMVar mvar ())
 592 >
 593 >     main =
 594 >       later waitForChildren $
 595 >       ...
 596
 597       The main thread principle also applies to calls to Haskell from
 598       outside, using @foreign export@.  When the @foreign export@ed
 599       function is invoked, it starts a new main thread, and it returns
 600       when this main thread terminates.  If the call causes new
 601       threads to be forked, they may remain in the system after the
 602       @foreign export@ed function has returned.
 603 -}
 604
 605 {- $preemption
 606
 607       GHC implements pre-emptive multitasking: the execution of
 608       threads are interleaved in a random fashion.  More specifically,
 609       a thread may be pre-empted whenever it allocates some memory,
 610       which unfortunately means that tight loops which do no
 611       allocation tend to lock out other threads (this only seems to
 612       happen with pathological benchmark-style code, however).
 613
 614       The rescheduling timer runs on a 20ms granularity by
 615       default, but this may be altered using the
 616       @-i\<n\>@ RTS option.  After a rescheduling
 617       \"tick\" the running thread is pre-empted as soon as
 618       possible.
 619
 620       One final note: the
 621       @aaaa@ @bbbb@ example may not
 622       work too well on GHC (see Scheduling, above), due
 623       to the locking on a 'System.IO.Handle'.  Only one thread
 624       may hold the lock on a 'System.IO.Handle' at any one
 625       time, so if a reschedule happens while a thread is holding the
 626       lock, the other thread won't be able to run.  The upshot is that
 627       the switch from @aaaa@ to
 628       @bbbbb@ happens infrequently.  It can be
 629       improved by lowering the reschedule tick period.  We also have a
 630       patch that causes a reschedule whenever a thread waiting on a
 631       lock is woken up, but haven't found it to be useful for anything
 632       other than this example :-)
 633 -}
 634 #endif /* __GLASGOW_HASKELL__ */