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/core/LICENSE)
   6 --
   7 -- Maintainer  :  libraries@haskell.org
   8 -- Stability   :  experimental
   9 -- Portability :  non-portable
  10 --
  11 -- $Id: Concurrent.hs,v 1.4 2002/04/10 15:57:16 simonmar Exp $
  12 --
  13 -- A common interface to a collection of useful concurrency
  14 -- abstractions.
  15 --
  16 -----------------------------------------------------------------------------
  17
  18 module Control.Concurrent (
  19         module Control.Concurrent.Chan,
  20         module Control.Concurrent.CVar,
  21         module Control.Concurrent.MVar,
  22         module Control.Concurrent.QSem,
  23         module Control.Concurrent.QSemN,
  24         module Control.Concurrent.SampleVar,
  25
  26         forkIO,                 -- :: IO () -> IO ()
  27         yield,                  -- :: IO ()
  28
  29 #ifdef __GLASGOW_HASKELL__
  30         ThreadId,
  31
  32         -- Forking and suchlike
  33         myThreadId,             -- :: IO ThreadId
  34         killThread,             -- :: ThreadId -> IO ()
  35         throwTo,                -- :: ThreadId -> Exception -> IO ()
  36
  37         threadDelay,            -- :: Int -> IO ()
  38         threadWaitRead,         -- :: Int -> IO ()
  39         threadWaitWrite,        -- :: Int -> IO ()
  40 #endif
  41
  42          -- merging of streams
  43         mergeIO,                -- :: [a]   -> [a] -> IO [a]
  44         nmergeIO                -- :: [[a]] -> IO [a]
  45     ) where
  46
  47 import Prelude
  48
  49 import Control.Exception as Exception
  50
  51 #ifdef __GLASGOW_HASKELL__
  52 import GHC.Conc
  53 import GHC.TopHandler   ( reportStackOverflow, reportError )
  54 import GHC.IOBase       ( IO(..) )
  55 import GHC.IOBase       ( unsafeInterleaveIO )
  56 import GHC.Base
  57 #endif
  58
  59 #ifdef __HUGS__
  60 import IOExts ( unsafeInterleaveIO )
  61 import ConcBase
  62 #endif
  63
  64 import Control.Concurrent.MVar
  65 import Control.Concurrent.CVar
  66 import Control.Concurrent.Chan
  67 import Control.Concurrent.QSem
  68 import Control.Concurrent.QSemN
  69 import Control.Concurrent.SampleVar
  70
  71 -- Thread Ids, specifically the instances of Eq and Ord for these things.
  72 -- The ThreadId type itself is defined in std/PrelConc.lhs.
  73
  74 -- Rather than define a new primitve, we use a little helper function
  75 -- cmp_thread in the RTS.
  76
  77 #ifdef __GLASGOW_HASKELL__
  78 foreign import ccall unsafe "cmp_thread" cmp_thread :: Addr# -> Addr# -> Int
  79 -- Returns -1, 0, 1
  80
  81 cmpThread :: ThreadId -> ThreadId -> Ordering
  82 cmpThread (ThreadId t1) (ThreadId t2) =
  83    case cmp_thread (unsafeCoerce# t1) (unsafeCoerce# t2) of
  84       -1 -> LT
  85       0  -> EQ
  86       _  -> GT -- must be 1
  87
  88 instance Eq ThreadId where
  89    t1 == t2 =
  90       case t1 `cmpThread` t2 of
  91          EQ -> True
  92          _  -> False
  93
  94 instance Ord ThreadId where
  95    compare = cmpThread
  96
  97 forkIO :: IO () -> IO ThreadId
  98 forkIO action = IO $ \ s ->
  99    case (fork# action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
 100  where
 101   action_plus = Exception.catch action childHandler
 102
 103 childHandler :: Exception -> IO ()
 104 childHandler err = Exception.catch (real_handler err) childHandler
 105
 106 real_handler :: Exception -> IO ()
 107 real_handler ex =
 108   case ex of
 109         -- ignore thread GC and killThread exceptions:
 110         BlockedOnDeadMVar            -> return ()
 111         AsyncException ThreadKilled  -> return ()
 112
 113         -- report all others:
 114         AsyncException StackOverflow -> reportStackOverflow False
 115         ErrorCall s -> reportError False s
 116         other       -> reportError False (showsPrec 0 other "\n")
 117
 118 #endif /* __GLASGOW_HASKELL__ */
 119
 120
 121 max_buff_size :: Int
 122 max_buff_size = 1
 123
 124 mergeIO :: [a] -> [a] -> IO [a]
 125 nmergeIO :: [[a]] -> IO [a]
 126
 127 mergeIO ls rs
 128  = newEmptyMVar                >>= \ tail_node ->
 129    newMVar tail_node           >>= \ tail_list ->
 130    newQSem max_buff_size       >>= \ e ->
 131    newMVar 2                   >>= \ branches_running ->
 132    let
 133     buff = (tail_list,e)
 134    in
 135     forkIO (suckIO branches_running buff ls) >>
 136     forkIO (suckIO branches_running buff rs) >>
 137     takeMVar tail_node  >>= \ val ->
 138     signalQSem e        >>
 139     return val
 140
 141 type Buffer a
 142  = (MVar (MVar [a]), QSem)
 143
 144 suckIO :: MVar Int -> Buffer a -> [a] -> IO ()
 145
 146 suckIO branches_running buff@(tail_list,e) vs
 147  = case vs of
 148         [] -> takeMVar branches_running >>= \ val ->
 149               if val == 1 then
 150                  takeMVar tail_list     >>= \ node ->
 151                  putMVar node []        >>
 152                  putMVar tail_list node
 153               else
 154                  putMVar branches_running (val-1)
 155         (x:xs) ->
 156                 waitQSem e                       >>
 157                 takeMVar tail_list               >>= \ node ->
 158                 newEmptyMVar                     >>= \ next_node ->
 159                 unsafeInterleaveIO (
 160                         takeMVar next_node  >>= \ y ->
 161                         signalQSem e        >>
 162                         return y)                >>= \ next_node_val ->
 163                 putMVar node (x:next_node_val)   >>
 164                 putMVar tail_list next_node      >>
 165                 suckIO branches_running buff xs
 166
 167 nmergeIO lss
 168  = let
 169     len = length lss
 170    in
 171     newEmptyMVar          >>= \ tail_node ->
 172     newMVar tail_node     >>= \ tail_list ->
 173     newQSem max_buff_size >>= \ e ->
 174     newMVar len           >>= \ branches_running ->
 175     let
 176      buff = (tail_list,e)
 177     in
 178     mapIO (\ x -> forkIO (suckIO branches_running buff x)) lss >>
 179     takeMVar tail_node  >>= \ val ->
 180     signalQSem e        >>
 181     return val
 182   where
 183     mapIO f xs = sequence (map f xs)