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         myThreadId,
  25
  26         forkIO,
  27         killThread,
  28         throwTo,
  29
  30         -- * Scheduling
  31
  32         -- $conc_scheduling
  33         yield,                  -- :: IO ()
  34
  35         -- ** Blocking
  36
  37         -- $blocking
  38
  39 #ifdef __GLASGOW_HASKELL__
  40         -- ** Waiting
  41         threadDelay,            -- :: Int -> IO ()
  42         threadWaitRead,         -- :: Int -> IO ()
  43         threadWaitWrite,        -- :: Int -> IO ()
  44 #endif
  45
  46         -- * Communication abstractions
  47
  48         module Control.Concurrent.MVar,
  49         module Control.Concurrent.Chan,
  50         module Control.Concurrent.QSem,
  51         module Control.Concurrent.QSemN,
  52         module Control.Concurrent.SampleVar,
  53
  54         -- * Merging of streams
  55         mergeIO,                -- :: [a]   -> [a] -> IO [a]
  56         nmergeIO,               -- :: [[a]] -> IO [a]
  57         -- $merge
  58
  59         -- * GHC's implementation of concurrency
  60
  61         -- |This section describes features specific to GHC's
  62         -- implementation of Concurrent Haskell.
  63
  64         -- ** Terminating the program
  65
  66         -- $termination
  67
  68         -- ** Pre-emption
  69
  70         -- $preemption
  71
  72     ) where
  73
  74 import Prelude
  75
  76 import Control.Exception as Exception
  77
  78 #ifdef __GLASGOW_HASKELL__
  79 import GHC.Conc
  80 import GHC.TopHandler   ( reportStackOverflow, reportError )
  81 import GHC.IOBase       ( IO(..) )
  82 import GHC.IOBase       ( unsafeInterleaveIO )
  83 import GHC.Base
  84 import GHC.Ptr
  85 #endif
  86
  87 #ifdef __HUGS__
  88 import IOExts ( unsafeInterleaveIO )
  89 import ConcBase
  90 #endif
  91
  92 import Control.Concurrent.MVar
  93 import Control.Concurrent.Chan
  94 import Control.Concurrent.QSem
  95 import Control.Concurrent.QSemN
  96 import Control.Concurrent.SampleVar
  97
  98 {- $conc_intro
  99
 100 The concurrency extension for Haskell is described in the paper
 101 /Concurrent Haskell/
 102 <http://www.haskell.org/ghc/docs/papers/concurrent-haskell.ps.gz>.
 103
 104 Concurrency is \"lightweight\", which means that both thread creation
 105 and context switching overheads are extremely low.  Scheduling of
 106 Haskell threads is done internally in the Haskell runtime system, and
 107 doesn't make use of any operating system-supplied thread packages.
 108
 109 Haskell threads can communicate via 'MVar's, a kind of synchronised
 110 mutable variable (see "Control.Concurrent.MVar").  Several common
 111 concurrency abstractions can be built from 'MVar's, and these are
 112 provided by the "Concurrent" library.  Threads may also communicate
 113 via exceptions.
 114 -}
 115
 116 {- $conc_scheduling
 117
 118     Scheduling may be either pre-emptive or co-operative,
 119     depending on the implementation of Concurrent Haskell (see below
 120     for imformation related to specific compilers).  In a co-operative
 121     system, context switches only occur when you use one of the
 122     primitives defined in this module.  This means that programs such
 123     as:
 124
 125
 126 >   main = forkIO (write 'a') >> write 'b'
 127 >     where write c = putChar c >> write c
 128
 129     will print either @aaaaaaaaaaaaaa...@ or @bbbbbbbbbbbb...@,
 130     instead of some random interleaving of @a@s and @b@s.  In
 131     practice, cooperative multitasking is sufficient for writing
 132     simple graphical user interfaces.
 133 -}
 134
 135 {- $blocking
 136 Calling a foreign C procedure (such as @getchar@) that blocks waiting
 137 for input will block /all/ threads, unless the @threadsafe@ attribute
 138 is used on the foreign call (and your compiler \/ operating system
 139 supports it).  GHC's I\/O system uses non-blocking I\/O internally to
 140 implement thread-friendly I\/O, so calling standard Haskell I\/O
 141 functions blocks only the thread making the call.
 142 -}
 143
 144 -- Thread Ids, specifically the instances of Eq and Ord for these things.
 145 -- The ThreadId type itself is defined in std/PrelConc.lhs.
 146
 147 -- Rather than define a new primitve, we use a little helper function
 148 -- cmp_thread in the RTS.
 149
 150 #ifdef __GLASGOW_HASKELL__
 151 type StgTSO = Ptr ()
 152
 153 id2TSO :: ThreadId -> StgTSO
 154 id2TSO (ThreadId t) = unsafeCoerce# t
 155
 156 foreign import ccall unsafe "cmp_thread" cmp_thread :: StgTSO -> StgTSO -> Int
 157 -- Returns -1, 0, 1
 158
 159 cmpThread :: ThreadId -> ThreadId -> Ordering
 160 cmpThread t1 t2 =
 161    case cmp_thread (id2TSO t1) (id2TSO t2) of
 162       -1 -> LT
 163       0  -> EQ
 164       _  -> GT -- must be 1
 165
 166 instance Eq ThreadId where
 167    t1 == t2 =
 168       case t1 `cmpThread` t2 of
 169          EQ -> True
 170          _  -> False
 171
 172 instance Ord ThreadId where
 173    compare = cmpThread
 174
 175 foreign import ccall unsafe "rts_getThreadId" getThreadId :: StgTSO -> Int
 176
 177 instance Show ThreadId where
 178    showsPrec d t =
 179         showString "ThreadId " .
 180         showsPrec d (getThreadId (id2TSO t))
 181
 182 {- |
 183 This sparks off a new thread to run the 'IO' computation passed as the
 184 first argument, and returns the 'ThreadId' of the newly created
 185 thread.
 186 -}
 187 forkIO :: IO () -> IO ThreadId
 188 forkIO action = IO $ \ s ->
 189    case (fork# action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
 190  where
 191   action_plus = Exception.catch action childHandler
 192
 193 childHandler :: Exception -> IO ()
 194 childHandler err = Exception.catch (real_handler err) childHandler
 195
 196 real_handler :: Exception -> IO ()
 197 real_handler ex =
 198   case ex of
 199         -- ignore thread GC and killThread exceptions:
 200         BlockedOnDeadMVar            -> return ()
 201         AsyncException ThreadKilled  -> return ()
 202
 203         -- report all others:
 204         AsyncException StackOverflow -> reportStackOverflow False
 205         ErrorCall s -> reportError False s
 206         other       -> reportError False (showsPrec 0 other "\n")
 207
 208 #endif /* __GLASGOW_HASKELL__ */
 209
 210
 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
 283 -- ---------------------------------------------------------------------------
 284 -- More docs
 285
 286 {- $termination
 287
 288       In a standalone GHC program, only the main thread is
 289       required to terminate in order for the process to terminate.
 290       Thus all other forked threads will simply terminate at the same
 291       time as the main thread (the terminology for this kind of
 292       behaviour is \"daemonic threads\").
 293
 294       If you want the program to wait for child threads to
 295       finish before exiting, you need to program this yourself.  A
 296       simple mechanism is to have each child thread write to an
 297       'MVar' when it completes, and have the main
 298       thread wait on all the 'MVar's before
 299       exiting:
 300
 301 >   myForkIO :: IO () -> IO (MVar ())
 302 >   myForkIO io = do
 303 >     mvar \<- newEmptyMVar
 304 >     forkIO (io \`finally\` putMVar mvar ())
 305 >     return mvar
 306
 307       Note that we use 'finally' from the
 308       "Exception" module to make sure that the
 309       'MVar' is written to even if the thread dies or
 310       is killed for some reason.
 311
 312       A better method is to keep a global list of all child
 313       threads which we should wait for at the end of the program:
 314
 315 >     children :: MVar [MVar ()]
 316 >     children = unsafePerformIO (newMVar [])
 317 >
 318 >     waitForChildren :: IO ()
 319 >     waitForChildren = do
 320 >       (mvar:mvars) \<- takeMVar children
 321 >       putMVar children mvars
 322 >       takeMVar mvar
 323 >       waitForChildren
 324 >
 325 >     forkChild :: IO () -> IO ()
 326 >     forkChild io = do
 327 >        mvar \<- newEmptyMVar
 328 >        forkIO (p \`finally\` putMVar mvar ())
 329 >        childs \<- takeMVar children
 330 >        putMVar children (mvar:childs)
 331 >
 332 >     later = flip finally
 333 >
 334 >     main =
 335 >       later waitForChildren $
 336 >       ...
 337
 338       The main thread principle also applies to calls to Haskell from
 339       outside, using @foreign export@.  When the @foreign export@ed
 340       function is invoked, it starts a new main thread, and it returns
 341       when this main thread terminates.  If the call causes new
 342       threads to be forked, they may remain in the system after the
 343       @foreign export@ed function has returned.
 344 -}
 345
 346 {- $preemption
 347
 348       GHC implements pre-emptive multitasking: the execution of
 349       threads are interleaved in a random fashion.  More specifically,
 350       a thread may be pre-empted whenever it allocates some memory,
 351       which unfortunately means that tight loops which do no
 352       allocation tend to lock out other threads (this only seems to
 353       happen with pathalogical benchmark-style code, however).
 354
 355       The rescheduling timer runs on a 20ms granularity by
 356       default, but this may be altered using the
 357       @-i<n>@ RTS option.  After a rescheduling
 358       \"tick\" the running thread is pre-empted as soon as
 359       possible.
 360
 361       One final note: the
 362       @aaaa@ @bbbb@ example may not
 363       work too well on GHC (see Scheduling, above), due
 364       to the locking on a 'Handle'.  Only one thread
 365       may hold the lock on a 'Handle' at any one
 366       time, so if a reschedule happens while a thread is holding the
 367       lock, the other thread won't be able to run.  The upshot is that
 368       the switch from @aaaa@ to
 369       @bbbbb@ happens infrequently.  It can be
 370       improved by lowering the reschedule tick period.  We also have a
 371       patch that causes a reschedule whenever a thread waiting on a
 372       lock is woken up, but haven't found it to be useful for anything
 373       other than this example :-)
 374 -}