BoundedChan.hs

-- |Implements bounded channels. In these channels, you can give a
-- rough maximum number of elements, and you will be guaranteed that no more
-- than that number of elements will be pending within the channel.
--
-- This boundedness is ideal when you will be (or may be) writing to a channel
-- faster than you are able to read from it.
--
-- This module supports all the functions of Control.Concurrent.Chan except
-- unGetChan and dupChan, which are not supported for bounded channels.
module Control.Concurrent.BoundedChan(
         BoundedChan
       , newBoundedChan
       , writeChan
       , readChan
       , isEmptyChan
       , getChanContents
       , writeList2Chan
       )
  where

import Control.Concurrent.MVar
import Control.Monad(replicateM)
import Data.Array
import System.IO.Unsafe(unsafeInterleaveIO)

-- |BoundedChan is an abstract data type representing an unbounded channel.
data BoundedChan a = BC {
       _size     :: Int
     , _contents :: Array Int (MVar a)
     , _writePos :: MVar Int
     , _readPos  :: MVar Int
     }
-- LOCK ORDERING: A -> B -> C

-- |Create a new bounded chan with size n. 
newBoundedChan :: Int -> IO (BoundedChan a)
newBoundedChan x = do
  entls   <- replicateM x newEmptyMVar
  wpos    <- newMVar 0
  rpos    <- newMVar 0
  let entries = listArray (0, x - 1) entls
  return (BC x entries wpos rpos)

-- |Write an element to the channel. If the channel is full, this routine will
-- block until it is able to write. If you have multiple writers, be careful
-- here, because the unlocking is not guaranteed to avoid starvation.
writeChan :: BoundedChan a -> a -> IO ()
writeChan (BC size contents wposMV _) x = do
  wpos <- takeMVar wposMV
  putMVar wposMV $! (wpos + 1) `mod` size
  putMVar (contents ! wpos) x

-- |Read an element from the channel. If the channel is empty, this routine 
-- will block until it is able to read.
readChan :: BoundedChan a -> IO a
readChan (BC size contents _ rposMV) = do
  rpos <- takeMVar rposMV
  putMVar rposMV $! (rpos + 1) `mod` size
  takeMVar (contents ! rpos)

isEmptyChan :: BoundedChan a -> IO Bool
isEmptyChan (BC _ contents _ rposMV) = do
  rpos <- takeMVar rposMV
  res <- isEmptyMVar (contents ! rpos)
  putMVar rposMV rpos
  return res

getChanContents :: BoundedChan a -> IO [a]
getChanContents ch = unsafeInterleaveIO $ do
                       x  <- readChan ch
                       xs <- getChanContents ch
                       return (x:xs)

-- |Write a list of elements to the channel. Note that this may block as it
-- writes the list into the channel.
writeList2Chan :: BoundedChan a -> [a] -> IO ()
writeList2Chan ch ls = mapM_ (writeChan ch) ls