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