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         -- * GHC's implementation of concurrency
  66
  67         -- |This section describes features specific to GHC's
  68         -- implementation of Concurrent Haskell.
  69
  70         -- ** Terminating the program
  71
  72         -- $termination
  73
  74         -- ** Pre-emption
  75
  76         -- $preemption
  77
  78     ) where
  79
  80 import Prelude
  81
  82 import Control.Exception as Exception
  83
  84 #ifdef __GLASGOW_HASKELL__
  85 import GHC.Conc
  86 import GHC.TopHandler   ( reportStackOverflow, reportError )
  87 import GHC.IOBase       ( IO(..) )
  88 import GHC.IOBase       ( unsafeInterleaveIO )
  89 import GHC.Base
  90 #endif
  91
  92 #ifdef __HUGS__
  93 import Hugs.ConcBase
  94 #endif
  95
  96 import Control.Concurrent.MVar
  97 import Control.Concurrent.Chan
  98 import Control.Concurrent.QSem
  99 import Control.Concurrent.QSemN
 100 import Control.Concurrent.SampleVar
 101
 102 #ifdef __HUGS__
 103 type ThreadId = ()
 104 #endif
 105
 106 {- $conc_intro
 107
 108 The concurrency extension for Haskell is described in the paper
 109 /Concurrent Haskell/
 110 <http://www.haskell.org/ghc/docs/papers/concurrent-haskell.ps.gz>.
 111
 112 Concurrency is \"lightweight\", which means that both thread creation
 113 and context switching overheads are extremely low.  Scheduling of
 114 Haskell threads is done internally in the Haskell runtime system, and
 115 doesn't make use of any operating system-supplied thread packages.
 116
 117 Haskell threads can communicate via 'MVar's, a kind of synchronised
 118 mutable variable (see "Control.Concurrent.MVar").  Several common
 119 concurrency abstractions can be built from 'MVar's, and these are
 120 provided by the "Control.Concurrent" library.
 121 In GHC, threads may also communicate via exceptions.
 122 -}
 123
 124 {- $conc_scheduling
 125
 126     Scheduling may be either pre-emptive or co-operative,
 127     depending on the implementation of Concurrent Haskell (see below
 128     for imformation related to specific compilers).  In a co-operative
 129     system, context switches only occur when you use one of the
 130     primitives defined in this module.  This means that programs such
 131     as:
 132
 133
 134 >   main = forkIO (write 'a') >> write 'b'
 135 >     where write c = putChar c >> write c
 136
 137     will print either @aaaaaaaaaaaaaa...@ or @bbbbbbbbbbbb...@,
 138     instead of some random interleaving of @a@s and @b@s.  In
 139     practice, cooperative multitasking is sufficient for writing
 140     simple graphical user interfaces.
 141 -}
 142
 143 {- $blocking
 144 Calling a foreign C procedure (such as @getchar@) that blocks waiting
 145 for input will block /all/ threads, unless the @threadsafe@ attribute
 146 is used on the foreign call (and your compiler \/ operating system
 147 supports it).  GHC's I\/O system uses non-blocking I\/O internally to
 148 implement thread-friendly I\/O, so calling standard Haskell I\/O
 149 functions blocks only the thread making the call.
 150 -}
 151
 152 -- Thread Ids, specifically the instances of Eq and Ord for these things.
 153 -- The ThreadId type itself is defined in std/PrelConc.lhs.
 154
 155 -- Rather than define a new primitve, we use a little helper function
 156 -- cmp_thread in the RTS.
 157
 158 #ifdef __GLASGOW_HASKELL__
 159 id2TSO :: ThreadId -> ThreadId#
 160 id2TSO (ThreadId t) = t
 161
 162 foreign import ccall unsafe "cmp_thread" cmp_thread :: ThreadId# -> ThreadId# -> Int
 163 -- Returns -1, 0, 1
 164
 165 cmpThread :: ThreadId -> ThreadId -> Ordering
 166 cmpThread t1 t2 =
 167    case cmp_thread (id2TSO t1) (id2TSO t2) of
 168       -1 -> LT
 169       0  -> EQ
 170       _  -> GT -- must be 1
 171
 172 instance Eq ThreadId where
 173    t1 == t2 =
 174       case t1 `cmpThread` t2 of
 175          EQ -> True
 176          _  -> False
 177
 178 instance Ord ThreadId where
 179    compare = cmpThread
 180
 181 foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> Int
 182
 183 instance Show ThreadId where
 184    showsPrec d t =
 185         showString "ThreadId " .
 186         showsPrec d (getThreadId (id2TSO t))
 187
 188 {- |
 189 This sparks off a new thread to run the 'IO' computation passed as the
 190 first argument, and returns the 'ThreadId' of the newly created
 191 thread.
 192 -}
 193 forkIO :: IO () -> IO ThreadId
 194 forkIO action = IO $ \ s ->
 195    case (fork# action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
 196  where
 197   action_plus = Exception.catch action childHandler
 198
 199 childHandler :: Exception -> IO ()
 200 childHandler err = Exception.catch (real_handler err) childHandler
 201
 202 real_handler :: Exception -> IO ()
 203 real_handler ex =
 204   case ex of
 205         -- ignore thread GC and killThread exceptions:
 206         BlockedOnDeadMVar            -> return ()
 207         AsyncException ThreadKilled  -> return ()
 208
 209         -- report all others:
 210         AsyncException StackOverflow -> reportStackOverflow False
 211         ErrorCall s -> reportError False s
 212         other       -> reportError False (showsPrec 0 other "\n")
 213
 214 #endif /* __GLASGOW_HASKELL__ */
 215
 216 #ifndef __HUGS__
 217 max_buff_size :: Int
 218 max_buff_size = 1
 219
 220 mergeIO :: [a] -> [a] -> IO [a]
 221 nmergeIO :: [[a]] -> IO [a]
 222
 223 -- $merge
 224 -- The 'mergeIO' and 'nmergeIO' functions fork one thread for each
 225 -- input list that concurrently evaluates that list; the results are
 226 -- merged into a single output list.
 227 --
 228 -- Note: Hugs does not provide these functions, since they require
 229 -- preemptive multitasking.
 230
 231 mergeIO ls rs
 232  = newEmptyMVar                >>= \ tail_node ->
 233    newMVar tail_node           >>= \ tail_list ->
 234    newQSem max_buff_size       >>= \ e ->
 235    newMVar 2                   >>= \ branches_running ->
 236    let
 237     buff = (tail_list,e)
 238    in
 239     forkIO (suckIO branches_running buff ls) >>
 240     forkIO (suckIO branches_running buff rs) >>
 241     takeMVar tail_node  >>= \ val ->
 242     signalQSem e        >>
 243     return val
 244
 245 type Buffer a
 246  = (MVar (MVar [a]), QSem)
 247
 248 suckIO :: MVar Int -> Buffer a -> [a] -> IO ()
 249
 250 suckIO branches_running buff@(tail_list,e) vs
 251  = case vs of
 252         [] -> takeMVar branches_running >>= \ val ->
 253               if val == 1 then
 254                  takeMVar tail_list     >>= \ node ->
 255                  putMVar node []        >>
 256                  putMVar tail_list node
 257               else
 258                  putMVar branches_running (val-1)
 259         (x:xs) ->
 260                 waitQSem e                       >>
 261                 takeMVar tail_list               >>= \ node ->
 262                 newEmptyMVar                     >>= \ next_node ->
 263                 unsafeInterleaveIO (
 264                         takeMVar next_node  >>= \ y ->
 265                         signalQSem e        >>
 266                         return y)                >>= \ next_node_val ->
 267                 putMVar node (x:next_node_val)   >>
 268                 putMVar tail_list next_node      >>
 269                 suckIO branches_running buff xs
 270
 271 nmergeIO lss
 272  = let
 273     len = length lss
 274    in
 275     newEmptyMVar          >>= \ tail_node ->
 276     newMVar tail_node     >>= \ tail_list ->
 277     newQSem max_buff_size >>= \ e ->
 278     newMVar len           >>= \ branches_running ->
 279     let
 280      buff = (tail_list,e)
 281     in
 282     mapIO (\ x -> forkIO (suckIO branches_running buff x)) lss >>
 283     takeMVar tail_node  >>= \ val ->
 284     signalQSem e        >>
 285     return val
 286   where
 287     mapIO f xs = sequence (map f xs)
 288 #endif /* __HUGS__ */
 289
 290 -- ---------------------------------------------------------------------------
 291 -- More docs
 292
 293 {- $termination
 294
 295       In a standalone GHC program, only the main thread is
 296       required to terminate in order for the process to terminate.
 297       Thus all other forked threads will simply terminate at the same
 298       time as the main thread (the terminology for this kind of
 299       behaviour is \"daemonic threads\").
 300
 301       If you want the program to wait for child threads to
 302       finish before exiting, you need to program this yourself.  A
 303       simple mechanism is to have each child thread write to an
 304       'MVar' when it completes, and have the main
 305       thread wait on all the 'MVar's before
 306       exiting:
 307
 308 >   myForkIO :: IO () -> IO (MVar ())
 309 >   myForkIO io = do
 310 >     mvar \<- newEmptyMVar
 311 >     forkIO (io \`finally\` putMVar mvar ())
 312 >     return mvar
 313
 314       Note that we use 'finally' from the
 315       "Control.Exception" module to make sure that the
 316       'MVar' is written to even if the thread dies or
 317       is killed for some reason.
 318
 319       A better method is to keep a global list of all child
 320       threads which we should wait for at the end of the program:
 321
 322 >     children :: MVar [MVar ()]
 323 >     children = unsafePerformIO (newMVar [])
 324 >
 325 >     waitForChildren :: IO ()
 326 >     waitForChildren = do
 327 >       (mvar:mvars) \<- takeMVar children
 328 >       putMVar children mvars
 329 >       takeMVar mvar
 330 >       waitForChildren
 331 >
 332 >     forkChild :: IO () -> IO ()
 333 >     forkChild io = do
 334 >        mvar \<- newEmptyMVar
 335 >        forkIO (p \`finally\` putMVar mvar ())
 336 >        childs \<- takeMVar children
 337 >        putMVar children (mvar:childs)
 338 >
 339 >     later = flip finally
 340 >
 341 >     main =
 342 >       later waitForChildren $
 343 >       ...
 344
 345       The main thread principle also applies to calls to Haskell from
 346       outside, using @foreign export@.  When the @foreign export@ed
 347       function is invoked, it starts a new main thread, and it returns
 348       when this main thread terminates.  If the call causes new
 349       threads to be forked, they may remain in the system after the
 350       @foreign export@ed function has returned.
 351 -}
 352
 353 {- $preemption
 354
 355       GHC implements pre-emptive multitasking: the execution of
 356       threads are interleaved in a random fashion.  More specifically,
 357       a thread may be pre-empted whenever it allocates some memory,
 358       which unfortunately means that tight loops which do no
 359       allocation tend to lock out other threads (this only seems to
 360       happen with pathalogical benchmark-style code, however).
 361
 362       The rescheduling timer runs on a 20ms granularity by
 363       default, but this may be altered using the
 364       @-i\<n\>@ RTS option.  After a rescheduling
 365       \"tick\" the running thread is pre-empted as soon as
 366       possible.
 367
 368       One final note: the
 369       @aaaa@ @bbbb@ example may not
 370       work too well on GHC (see Scheduling, above), due
 371       to the locking on a 'System.IO.Handle'.  Only one thread
 372       may hold the lock on a 'System.IO.Handle' at any one
 373       time, so if a reschedule happens while a thread is holding the
 374       lock, the other thread won't be able to run.  The upshot is that
 375       the switch from @aaaa@ to
 376       @bbbbb@ happens infrequently.  It can be
 377       improved by lowering the reschedule tick period.  We also have a
 378       patch that causes a reschedule whenever a thread waiting on a
 379       lock is woken up, but haven't found it to be useful for anything
 380       other than this example :-)
 381 -}