ghc/lib/concurrent/Channel.hs

   1 {-
   2 %
   3 % (c) The GRASP/AQUA Project, Glasgow University, 1995
   4 %
   5 \section[Channel]{Unbounded Channels}
   6
   7 Standard, unbounded channel abstraction.
   8
   9 \begin{code}
  10 -}
  11 module Channel
  12        (
  13          {- abstract type defined -}
  14         Chan,
  15
  16          {- creator -}
  17         newChan,         -- :: IO (Chan a)
  18
  19          {- operators -}
  20         putChan,         -- :: Chan a -> a -> IO ()
  21         getChan,         -- :: Chan a -> IO a
  22         dupChan,         -- :: Chan a -> IO (Chan a)
  23         unGetChan,       -- :: Chan a -> a -> IO ()
  24
  25          {- stream interface -}
  26         getChanContents, -- :: Chan a -> IO [a]
  27         putList2Chan     -- :: Chan a -> [a] -> IO ()
  28
  29        ) where
  30
  31 import GHCbase
  32 {-
  33 \end{code}
  34
  35 A channel is represented by two @MVar@s keeping track of the two ends
  36 of the channel contents,i.e.,  the read- and write ends. Empty @MVar@s
  37 are used to handle consumers trying to read from an empty channel.
  38
  39 \begin{code}
  40 -}
  41
  42 data Chan a
  43  = Chan (MVar (Stream a))
  44         (MVar (Stream a))
  45
  46 type Stream a = MVar (ChItem a)
  47
  48 data ChItem a = ChItem a (Stream a)
  49
  50
  51 {-
  52 \end{code}
  53
  54 See the Concurrent Haskell paper for a diagram explaining the
  55 how the different channel operations proceed.
  56
  57 @newChan@ sets up the read and write end of a channel by initialising
  58 these two @MVar@s with an empty @MVar@.
  59
  60 \begin{code}
  61 -}
  62
  63 newChan :: IO (Chan a)
  64 newChan
  65  = newEmptyMVar      >>= \ hole ->
  66    newMVar hole      >>= \ read ->
  67    newMVar hole      >>= \ write ->
  68    return (Chan read write)
  69
  70 {-
  71 \end{code}
  72
  73 To put an element on a channel, a new hole at the write end is created.
  74 What was previously the empty @MVar@ at the back of the channel is then
  75 filled in with a new stream element holding the entered value and the
  76 new hole.
  77
  78 \begin{code}
  79 -}
  80
  81 putChan :: Chan a -> a -> IO ()
  82 putChan (Chan read write) val
  83  = newEmptyMVar             >>= \ new_hole ->
  84    takeMVar write           >>= \ old_hole ->
  85    putMVar write new_hole   >>
  86    putMVar old_hole (ChItem val new_hole) >>
  87    return ()
  88
  89
  90 getChan :: Chan a -> IO a
  91 getChan (Chan read write)
  92  = takeMVar read          >>= \ rend ->
  93    takeMVar rend          >>= \ (ChItem val new_rend) ->
  94    putMVar read new_rend  >>
  95    return val
  96
  97
  98 dupChan :: Chan a -> IO (Chan a)
  99 dupChan (Chan read write)
 100  = newEmptyMVar           >>= \ new_read ->
 101    readMVar write         >>= \ hole ->
 102    putMVar new_read hole  >>
 103    return (Chan new_read write)
 104
 105 unGetChan :: Chan a -> a -> IO ()
 106 unGetChan (Chan read write) val
 107  = newEmptyMVar                       >>= \ new_rend ->
 108    takeMVar read                      >>= \ rend ->
 109    putMVar new_rend (ChItem val rend) >>
 110    putMVar read new_rend              >>
 111    return ()
 112
 113 {-
 114 \end{code}
 115
 116 Operators for interfacing with functional streams.
 117
 118 \begin{code}
 119 -}
 120
 121 getChanContents :: Chan a -> IO [a]
 122 getChanContents ch
 123 {- WAS:
 124   = unsafeInterleavePrimIO (
 125       getChan ch                                   `thenPrimIO` \ ~(Right x) ->
 126       unsafeInterleavePrimIO (getChanContents ch)  `thenPrimIO` \ ~(Right xs) ->
 127       returnPrimIO  (Right (x:xs)))
 128 -}
 129   = my_2_IO $ unsafeInterleavePrimIO (
 130         getChan_prim ch                                  >>= \ ~(Right x) ->
 131         unsafeInterleavePrimIO (getChanContents_prim ch) >>= \ ~(Right xs) ->
 132         returnPrimIO  (Right (x:xs)))
 133
 134 my_2_IO :: PrimIO (Either IOError a) -> IO a -- simple; primIOToIO does too much!
 135 my_2_IO m = IO m
 136
 137 getChan_prim         :: Chan a -> PrimIO (Either IOError  a)
 138 getChanContents_prim :: Chan a -> PrimIO (Either IOError [a])
 139
 140 getChan_prim ch = ST $ \ s ->
 141     case (getChan ch) of { IO (ST get) ->
 142     get s }
 143
 144 getChanContents_prim ch = ST $ \ s ->
 145     case (getChanContents ch) of { IO (ST get) ->
 146     get s }
 147
 148 -------------
 149 putList2Chan :: Chan a -> [a] -> IO ()
 150 putList2Chan ch ls = sequence (map (putChan ch) ls)