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 -- Thread Ids, specifically the instances of Eq and Ord for these things.
 173 -- The ThreadId type itself is defined in std/PrelConc.lhs.
 174
 175 -- Rather than define a new primitve, we use a little helper function
 176 -- cmp_thread in the RTS.
 177
 178 #ifdef __GLASGOW_HASKELL__
 179 id2TSO :: ThreadId -> ThreadId#
 180 id2TSO (ThreadId t) = t
 181
 182 foreign import ccall unsafe "cmp_thread" cmp_thread :: ThreadId# -> ThreadId# -> CInt
 183 -- Returns -1, 0, 1
 184
 185 cmpThread :: ThreadId -> ThreadId -> Ordering
 186 cmpThread t1 t2 =
 187    case cmp_thread (id2TSO t1) (id2TSO t2) of
 188       -1 -> LT
 189       0  -> EQ
 190       _  -> GT -- must be 1
 191
 192 instance Eq ThreadId where
 193    t1 == t2 =
 194       case t1 `cmpThread` t2 of
 195          EQ -> True
 196          _  -> False
 197
 198 instance Ord ThreadId where
 199    compare = cmpThread
 200
 201 foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> Int
 202
 203 instance Show ThreadId where
 204    showsPrec d t =
 205         showString "ThreadId " .
 206         showsPrec d (getThreadId (id2TSO t))
 207
 208 #endif /* __GLASGOW_HASKELL__ */
 209
 210 #ifndef __HUGS__
 211 max_buff_size :: Int
 212 max_buff_size = 1
 213
 214 mergeIO :: [a] -> [a] -> IO [a]
 215 nmergeIO :: [[a]] -> IO [a]
 216
 217 -- $merge
 218 -- The 'mergeIO' and 'nmergeIO' functions fork one thread for each
 219 -- input list that concurrently evaluates that list; the results are
 220 -- merged into a single output list.
 221 --
 222 -- Note: Hugs does not provide these functions, since they require
 223 -- preemptive multitasking.
 224
 225 mergeIO ls rs
 226  = newEmptyMVar                >>= \ tail_node ->
 227    newMVar tail_node           >>= \ tail_list ->
 228    newQSem max_buff_size       >>= \ e ->
 229    newMVar 2                   >>= \ branches_running ->
 230    let
 231     buff = (tail_list,e)
 232    in
 233     forkIO (suckIO branches_running buff ls) >>
 234     forkIO (suckIO branches_running buff rs) >>
 235     takeMVar tail_node  >>= \ val ->
 236     signalQSem e        >>
 237     return val
 238
 239 type Buffer a
 240  = (MVar (MVar [a]), QSem)
 241
 242 suckIO :: MVar Int -> Buffer a -> [a] -> IO ()
 243
 244 suckIO branches_running buff@(tail_list,e) vs
 245  = case vs of
 246         [] -> takeMVar branches_running >>= \ val ->
 247               if val == 1 then
 248                  takeMVar tail_list     >>= \ node ->
 249                  putMVar node []        >>
 250                  putMVar tail_list node
 251               else
 252                  putMVar branches_running (val-1)
 253         (x:xs) ->
 254                 waitQSem e                       >>
 255                 takeMVar tail_list               >>= \ node ->
 256                 newEmptyMVar                     >>= \ next_node ->
 257                 unsafeInterleaveIO (
 258                         takeMVar next_node  >>= \ y ->
 259                         signalQSem e        >>
 260                         return y)                >>= \ next_node_val ->
 261                 putMVar node (x:next_node_val)   >>
 262                 putMVar tail_list next_node      >>
 263                 suckIO branches_running buff xs
 264
 265 nmergeIO lss
 266  = let
 267     len = length lss
 268    in
 269     newEmptyMVar          >>= \ tail_node ->
 270     newMVar tail_node     >>= \ tail_list ->
 271     newQSem max_buff_size >>= \ e ->
 272     newMVar len           >>= \ branches_running ->
 273     let
 274      buff = (tail_list,e)
 275     in
 276     mapIO (\ x -> forkIO (suckIO branches_running buff x)) lss >>
 277     takeMVar tail_node  >>= \ val ->
 278     signalQSem e        >>
 279     return val
 280   where
 281     mapIO f xs = sequence (map f xs)
 282 #endif /* __HUGS__ */
 283
 284 #ifdef __GLASGOW_HASKELL__
 285 -- ---------------------------------------------------------------------------
 286 -- Bound Threads
 287
 288 {- $boundthreads
 289
 290 Support for multiple operating system threads and bound threads as described
 291 below is currently only available in the GHC runtime system if you use the
 292 /-threaded/ option when linking.
 293
 294 Other Haskell systems do not currently support multiple operating system threads.
 295
 296 A bound thread is a haskell thread that is /bound/ to an operating system
 297 thread. While the bound thread is still scheduled by the Haskell run-time
 298 system, the operating system thread takes care of all the foreign calls made
 299 by the bound thread.
 300
 301 To a foreign library, the bound thread will look exactly like an ordinary
 302 operating system thread created using OS functions like @pthread_create@
 303 or @CreateThread@.
 304
 305 Bound threads can be created using the 'forkOS' function below. All foreign
 306 exported functions are run in a bound thread (bound to the OS thread that
 307 called the function). Also, the @main@ action of every Haskell program is
 308 run in a bound thread.
 309
 310 Why do we need this? Because if a foreign library is called from a thread
 311 created using 'forkIO', it won't have access to any /thread-local state/ -
 312 state variables that have specific values for each OS thread
 313 (see POSIX's @pthread_key_create@ or Win32's @TlsAlloc@). Therefore, some
 314 libraries (OpenGL, for example) will not work from a thread created using
 315 'forkIO'. They work fine in threads created using 'forkOS' or when called
 316 from @main@ or from a @foreign export@.
 317 -}
 318
 319 -- | 'True' if bound threads are supported.
 320 -- If @rtsSupportsBoundThreads@ is 'False', 'isCurrentThreadBound'
 321 -- will always return 'False' and both 'forkOS' and 'runInBoundThread' will
 322 -- fail.
 323 foreign import ccall rtsSupportsBoundThreads :: Bool
 324
 325
 326 {- |
 327 Like 'forkIO', this sparks off a new thread to run the 'IO' computation passed as the
 328 first argument, and returns the 'ThreadId' of the newly created
 329 thread.
 330
 331 However, @forkOS@ uses operating system-supplied multithreading support to create
 332 a new operating system thread. The new thread is /bound/, which means that
 333 all foreign calls made by the 'IO' computation are guaranteed to be executed
 334 in this new operating system thread; also, the operating system thread is not
 335 used for any other foreign calls.
 336
 337 This means that you can use all kinds of foreign libraries from this thread
 338 (even those that rely on thread-local state), without the limitations of 'forkIO'.
 339 -}
 340 forkOS :: IO () -> IO ThreadId
 341
 342 foreign export ccall forkOS_entry
 343     :: StablePtr (IO ()) -> IO ()
 344
 345 foreign import ccall "forkOS_entry" forkOS_entry_reimported
 346     :: StablePtr (IO ()) -> IO ()
 347
 348 forkOS_entry stableAction = do
 349         action <- deRefStablePtr stableAction
 350         action
 351
 352 foreign import ccall forkOS_createThread
 353     :: StablePtr (IO ()) -> IO CInt
 354
 355 failNonThreaded = fail $ "RTS doesn't support multiple OS threads "
 356                        ++"(use ghc -threaded when linking)"
 357
 358 forkOS action
 359     | rtsSupportsBoundThreads = do
 360         mv <- newEmptyMVar
 361         let action_plus = Exception.catch action childHandler
 362         entry <- newStablePtr (myThreadId >>= putMVar mv >> action_plus)
 363         err <- forkOS_createThread entry
 364         when (err /= 0) $ fail "Cannot create OS thread."
 365         tid <- takeMVar mv
 366         freeStablePtr entry
 367         return tid
 368     | otherwise = failNonThreaded
 369
 370 -- | Returns 'True' if the calling thread is /bound/, that is, if it is
 371 -- safe to use foreign libraries that rely on thread-local state from the
 372 -- calling thread.
 373 isCurrentThreadBound :: IO Bool
 374 isCurrentThreadBound = IO $ \ s# ->
 375     case isCurrentThreadBound# s# of
 376         (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
 377
 378
 379 {- |
 380 Run the 'IO' computation passed as the first argument. If the calling thread
 381 is not /bound/, a bound thread is created temporarily. @runInBoundThread@
 382 doesn't finish until the 'IO' computation finishes.
 383
 384 You can wrap a series of foreign function calls that rely on thread-local state
 385 with @runInBoundThread@ so that you can use them without knowing whether the
 386 current thread is /bound/.
 387 -}
 388 runInBoundThread :: IO a -> IO a
 389
 390 runInBoundThread action
 391     | rtsSupportsBoundThreads = do
 392         bound <- isCurrentThreadBound
 393         if bound
 394             then action
 395             else do
 396                 ref <- newIORef undefined
 397                 let action_plus = Exception.try action >>= writeIORef ref
 398                 resultOrException <-
 399                     bracket (newStablePtr action_plus)
 400                             freeStablePtr
 401                             (\cEntry -> forkOS_entry_reimported cEntry >> readIORef ref)
 402                 case resultOrException of
 403                     Left exception -> Exception.throw exception
 404                     Right result -> return result
 405     | otherwise = failNonThreaded
 406
 407 {- |
 408 Run the 'IO' computation passed as the first argument. If the calling thread
 409 is /bound/, an unbound thread is created temporarily using 'forkIO'.
 410 @runInBoundThread@ doesn't finish until the 'IO' computation finishes.
 411
 412 Use this function /only/ in the rare case that you have actually observed a
 413 performance loss due to the use of bound threads. A program that
 414 doesn't need it's main thread to be bound and makes /heavy/ use of concurrency
 415 (e.g. a web server), might want to wrap it's @main@ action in
 416 @runInUnboundThread@.
 417 -}
 418 runInUnboundThread :: IO a -> IO a
 419
 420 runInUnboundThread action = do
 421     bound <- isCurrentThreadBound
 422     if bound
 423         then do
 424             mv <- newEmptyMVar
 425             forkIO (Exception.try action >>= putMVar mv)
 426             takeMVar mv >>= \either -> case either of
 427                 Left exception -> Exception.throw exception
 428                 Right result -> return result
 429         else action
 430
 431 #endif /* __GLASGOW_HASKELL__ */
 432
 433 -- ---------------------------------------------------------------------------
 434 -- More docs
 435
 436 {- $termination
 437
 438       In a standalone GHC program, only the main thread is
 439       required to terminate in order for the process to terminate.
 440       Thus all other forked threads will simply terminate at the same
 441       time as the main thread (the terminology for this kind of
 442       behaviour is \"daemonic threads\").
 443
 444       If you want the program to wait for child threads to
 445       finish before exiting, you need to program this yourself.  A
 446       simple mechanism is to have each child thread write to an
 447       'MVar' when it completes, and have the main
 448       thread wait on all the 'MVar's before
 449       exiting:
 450
 451 >   myForkIO :: IO () -> IO (MVar ())
 452 >   myForkIO io = do
 453 >     mvar <- newEmptyMVar
 454 >     forkIO (io `finally` putMVar mvar ())
 455 >     return mvar
 456
 457       Note that we use 'finally' from the
 458       "Control.Exception" module to make sure that the
 459       'MVar' is written to even if the thread dies or
 460       is killed for some reason.
 461
 462       A better method is to keep a global list of all child
 463       threads which we should wait for at the end of the program:
 464
 465 >    children :: MVar [MVar ()]
 466 >    children = unsafePerformIO (newMVar [])
 467 >
 468 >    waitForChildren :: IO ()
 469 >    waitForChildren = do
 470 >      cs <- takeMVar children
 471 >      case cs of
 472 >        []   -> return ()
 473 >        m:ms -> do
 474 >           putMVar children ms
 475 >           takeMVar m
 476 >           waitForChildren
 477 >
 478 >    forkChild :: IO () -> IO ()
 479 >    forkChild io = do
 480 >        mvar <- newEmptyMVar
 481 >        childs <- takeMVar children
 482 >        putMVar children (mvar:childs)
 483 >        forkIO (io `finally` putMVar mvar ())
 484 >
 485 >     main =
 486 >       later waitForChildren $
 487 >       ...
 488
 489       The main thread principle also applies to calls to Haskell from
 490       outside, using @foreign export@.  When the @foreign export@ed
 491       function is invoked, it starts a new main thread, and it returns
 492       when this main thread terminates.  If the call causes new
 493       threads to be forked, they may remain in the system after the
 494       @foreign export@ed function has returned.
 495 -}
 496
 497 {- $preemption
 498
 499       GHC implements pre-emptive multitasking: the execution of
 500       threads are interleaved in a random fashion.  More specifically,
 501       a thread may be pre-empted whenever it allocates some memory,
 502       which unfortunately means that tight loops which do no
 503       allocation tend to lock out other threads (this only seems to
 504       happen with pathological benchmark-style code, however).
 505
 506       The rescheduling timer runs on a 20ms granularity by
 507       default, but this may be altered using the
 508       @-i\<n\>@ RTS option.  After a rescheduling
 509       \"tick\" the running thread is pre-empted as soon as
 510       possible.
 511
 512       One final note: the
 513       @aaaa@ @bbbb@ example may not
 514       work too well on GHC (see Scheduling, above), due
 515       to the locking on a 'System.IO.Handle'.  Only one thread
 516       may hold the lock on a 'System.IO.Handle' at any one
 517       time, so if a reschedule happens while a thread is holding the
 518       lock, the other thread won't be able to run.  The upshot is that
 519       the switch from @aaaa@ to
 520       @bbbbb@ happens infrequently.  It can be
 521       improved by lowering the reschedule tick period.  We also have a
 522       patch that causes a reschedule whenever a thread waiting on a
 523       lock is woken up, but haven't found it to be useful for anything
 524       other than this example :-)
 525 -}