ghc/lib/concurrent/Channel.lhs

   1 %
   2 % (c) The GRASP/AQUA Project, Glasgow University, 1995-97
   3 %
   4 \section[Channel]{Unbounded Channels}
   5
   6 Standard, unbounded channel abstraction.
   7
   8 \begin{code}
   9 module Channel
  10        (
  11          {- abstract type defined -}
  12         Chan,
  13
  14          {- creator -}
  15         newChan,         -- :: IO (Chan a)
  16
  17          {- operators -}
  18         writeChan,       -- :: Chan a -> a -> IO ()
  19         readChan,        -- :: Chan a -> IO a
  20         dupChan,         -- :: Chan a -> IO (Chan a)
  21         unGetChan,       -- :: Chan a -> a -> IO ()
  22
  23         isEmptyChan,     -- :: Chan a -> IO Bool
  24
  25          {- stream interface -}
  26         getChanContents, -- :: Chan a -> IO [a]
  27         writeList2Chan   -- :: Chan a -> [a] -> IO ()
  28
  29        ) where
  30
  31 import Prelude
  32 import PrelConc
  33 import PrelST
  34 import PrelIOBase ( unsafeInterleaveIO )
  35 \end{code}
  36
  37 A channel is represented by two @MVar@s keeping track of the two ends
  38 of the channel contents,i.e.,  the read- and write ends. Empty @MVar@s
  39 are used to handle consumers trying to read from an empty channel.
  40
  41 \begin{code}
  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 \end{code}
  50
  51 See the Concurrent Haskell paper for a diagram explaining the
  52 how the different channel operations proceed.
  53
  54 @newChan@ sets up the read and write end of a channel by initialising
  55 these two @MVar@s with an empty @MVar@.
  56
  57 \begin{code}
  58 newChan :: IO (Chan a)
  59 newChan = do
  60    hole  <- newEmptyMVar
  61    read  <- newMVar hole
  62    write <- newMVar hole
  63    return (Chan read write)
  64 \end{code}
  65
  66 To put an element on a channel, a new hole at the write end is created.
  67 What was previously the empty @MVar@ at the back of the channel is then
  68 filled in with a new stream element holding the entered value and the
  69 new hole.
  70
  71 \begin{code}
  72 writeChan :: Chan a -> a -> IO ()
  73 writeChan (Chan read write) val = do
  74    new_hole <- newEmptyMVar
  75    old_hole <- takeMVar write
  76    putMVar write new_hole
  77    putMVar old_hole (ChItem val new_hole)
  78
  79 readChan :: Chan a -> IO a
  80 readChan (Chan read write) = do
  81   read_end                  <- takeMVar read
  82   (ChItem val new_read_end) <- takeMVar read_end
  83   putMVar read new_read_end
  84   return val
  85
  86
  87 dupChan :: Chan a -> IO (Chan a)
  88 dupChan (Chan read write) = do
  89    new_read <- newEmptyMVar
  90    hole     <- readMVar write
  91    putMVar new_read hole
  92    return (Chan new_read write)
  93
  94 unGetChan :: Chan a -> a -> IO ()
  95 unGetChan (Chan read write) val = do
  96    new_read_end <- newEmptyMVar
  97    read_end     <- takeMVar read
  98    putMVar new_read_end (ChItem val read_end)
  99    putMVar read new_read_end
 100
 101 isEmptyChan :: Chan a -> IO Bool
 102 isEmptyChan (Chan read write) = do
 103    r <- takeMVar read
 104    w <- readMVar write
 105    let eq = r == w
 106    eq `seq` putMVar read r
 107    return eq
 108
 109 \end{code}
 110
 111 Operators for interfacing with functional streams.
 112
 113 \begin{code}
 114 getChanContents :: Chan a -> IO [a]
 115 getChanContents ch
 116   = unsafeInterleaveIO (do
 117         x  <- readChan ch
 118         xs <- getChanContents ch
 119         return (x:xs)
 120     )
 121
 122 -------------
 123 writeList2Chan :: Chan a -> [a] -> IO ()
 124 writeList2Chan ch ls = sequence_ (map (writeChan ch) ls)
 125
 126 \end{code}