[Haskell-cafe] Help with "shootout"

[Chris Kuklewicz]

[ Deeply nested replies are starting to look similar to runListT $
runStateT $ runWriter .... ]

matth at mindspring.com wrote:
> On Tue, Jan 03, 2006 at 12:07:43AM +0000, Joel Reymont wrote:
> 
>>On Jan 2, 2006, at 9:20 PM, Chris Kuklewicz wrote:
>>
>>
>>> This makes me ponder one of the things that Joel was trying to do:
>>>efficiently pass data to a logging thread.  It may be that a custom
>>>channel would be helpful for that as well.
>>
>>I have not taken the time to analyze the Chameneos code but need to  
>>point out that my problem was not with efficiently passing data to  
>>the logging thread. The issue was with data accumulating in the  
>>channel and the logger thread not reading it out fast enough.
>>
>>The TChan implementation is a single-linked list implemented on top  
>>of TVar's. That would seem pretty efficient to me.
> 
> 
> It's simple and efficient but does nothing to prevent the channel from
> growing out of control.  A slightly modified (custom) channel based on
> TChan, but enforcing a maximum size (blocking on insert if the channel
> is too full), probably would have solved the problem.
> 
> I assume that Erlang either does that or increases the priority of
> threads with large event queues, or both.
> 
> Thanks,
> Matt Harden

Given that actually controlling priorities is not an option, adding
blocking like that makes sense.  One can make a ring buffer instead of a
singly linked list very easily.  In fact, I have that code lying around
(now attached).  It has not been speed optimized, but I did like being
able to express:

> type Node a = [TMVar a]
>
> make :: (Integral k) => k -> STM (Node a)
> make k = liftM cycle $ sequence $ genericReplicate k newEmptyTMVar

It has the usual operations, but you need to pass a fixed size to
new/newEmpty and you also have an isFull test.  It has no operations to
resize the ring buffer created by "make".
-------------- next part --------------
module ProdCons (PC,new,newEmpty, put,ProdCons.take,ProdCons.read,
                   tryPut,tryTake,tryRead, isEmpty,isFull) where

{-  Fixed bounded-buffer size solution of producer/consumer problem.

    Acts like a FIFO TMVar, blocking when capacity is reached.  So a
    capacity of 1 behaves like a TMVar.

    For arbitrary capacity just use a TChan. -}

import Control.Concurrent.STM
import Control.Concurrent
import Control.Monad.Fix
import Control.Monad
import Data.List(cycle,genericReplicate)

type Node a = [TMVar a]
newtype PC a = PC (TVar (Node a),TVar (Node a))

newEmpty :: (Integral k) => k -> IO (PC a)
newEmpty k | k <=0 = error "Need capacity > 0"
           | otherwise = do 
  node <- atomically $ make k
  atomically $ do
    tv1 <- newTVar node
    tv2 <- newTVar node
    return (PC (tv1,tv2))

new :: (Integral k) => k -> a -> IO (PC a)
new k v | k <=0 = error "Need capacity > 0"
      v | otherwise = do 
  pc <- newEmpty k 
  atomically $ put pc v 
  return pc

put ::PC a -> a -> STM ()
put (PC (tvar,_)) value = do 
  (tmvar:next) <- readTVar tvar
  putTMVar tmvar value
  writeTVar tvar next

take :: PC a -> STM a
take (PC (_,tvar)) = do
  (tmvar:next) <- readTVar tvar
  value <- takeTMVar tmvar
  writeTVar tvar next
  return value

read :: PC a -> STM a
read (PC (_,tvar)) = do 
  (tmvar:_) <- readTVar tvar
  readTMVar tmvar

tryTake :: PC a -> STM (Maybe a)
tryTake pc = (ProdCons.take pc >>= return.Just) `orElse` (return Nothing)

tryRead :: PC a -> STM (Maybe a)
tryRead pc = (ProdCons.read pc >>= return.Just) `orElse` (return Nothing)

tryPut :: PC a -> a -> STM Bool
tryPut pc v = (put pc v >> return True) `orElse` (return False)

isEmpty :: PC a -> STM Bool
isEmpty (PC (_,tvar)) = do
   (tmvar:_) <- readTVar tvar
   isEmptyTMVar tmvar

isFull (PC (tvar,_)) = do
   (tmvar:_) <- readTVar tvar
   empty <- isEmptyTMVar tmvar
   return (not empty)

-- -- -- Internal -- -- --

make :: (Integral k) => k -> STM (Node a)
make k = liftM cycle $ sequence $ genericReplicate k newEmptyTMVar