Hi Mark, your question made me take a look at the TChan implementation, which I always had wanted to do (but never had the time). To test my understanding, I sketched a TChan variation that should solve the problem. (A test with one sender and two receivers showed expected output but I did not measure memory usage.) The sender replaces older messages with a marker. This should make the content available to the garbage collector (if it is not referenced by a receiver who has read it). On reading a marker, a receiver skips directly to the next valid message. On the downside, the sender keeps a reference to the last n messages, so they will not be garbage collected even if every receiver has read them. Thomas {-# LANGUAGE CPP, DeriveDataTypeable #-} module Control.Concurrent.STM.TBBroadcast( #ifdef __GLASGOW_HASKELL__ TSender, TReceiver, newSender, newSenderIO, writeBC, newReceiver, readBC #endif ) where #ifdef __GLASGOW_HASKELL__ import GHC.Conc import Data.Typeable (Typeable) data TSender a = Sender {-# UNPACK #-} !(TVar Int) {-# UNPACK #-} !(TVar (TVarList a)) {-# UNPACK #-} !(TVar (TVarList a)) deriving (Eq, Typeable) type TVarList a = TVar (TList a) data TList a = TNil | TCons a {-# UNPACK #-} !(TVarList a) | Outdated {-# UNPACK #-} !(TVar (TVarList a)) newSender :: Int -> STM (TSender a) newSender n | n <= 0 = error "windows size must be >=0" | otherwise = do hole <- newTVar TNil first <- newTVar hole end <- newTVar hole count <- newTVar n return (Sender count first end) newSenderIO :: Int -> IO (TSender a) newSenderIO n | n <= 0 = error "windows size must be >=0" | otherwise = do hole <- newTVarIO TNil first <- newTVarIO hole end <- newTVarIO hole count <- newTVarIO n return (Sender count first end) writeBC :: TSender a -> a -> STM () writeBC (Sender count first end) a = do listend <- readTVar end new_listend <- newTVar TNil writeTVar listend (TCons a new_listend) writeTVar end new_listend n <- readTVar count case n of 0 -> do listhead <- readTVar first head <- readTVar listhead case head of TCons _ tl -> writeTVar first tl writeTVar listhead (Outdated first) _ -> writeTVar count $! (n-1) data TReceiver a = Receiver {-# UNPACK #-} !(TVar (TVarList a)) newReceiver :: TSender a -> STM (TReceiver a) newReceiver (Sender _ _ end) = do hole <- readTVar end first <-newTVar hole return (Receiver first) readBC :: TReceiver a -> STM a readBC (Receiver first) = do listhead <- readTVar first head <- readTVar listhead case head of TNil -> retry TCons a tl -> do writeTVar first tl return a Outdated next -> do next' <- readTVar next writeTVar first next' readBC (Receiver first) #endif Am 28.01.2016 um 20:30 schrieb Mark Fine:
We're currently using a TMChan to broadcast from a single producer thread to many consumer threads. This works well! However, we're seeing issues with a fast producer and/or a slow consumer, with the channel growing unbounded. Fortunately, our producer-consumer communication is time-sensitive and tolerant of loss: we're ok with the producer always writing at the expense of dropping communication to a slow consumer.
A TMBChan provides a bounded channel (but no means to dupe/broadcast) where a writer will block once the channel fills up. In our use case, we'd like to continue writing to the channel but dropping off the end of the channel. Clojure's core-async module has some related concepts, in particular the notion of a sliding buffer https://clojure.github.io/core.async/#clojure.core.async/sliding-buffer that drops the oldest elements once full. Has anyone encountered something similar in working with channels and/or have any solutions? Thanks!
Mark
_______________________________________________ Haskell-Cafe mailing list Haskell-Cafe@haskell.org http://mail.haskell.org/cgi-bin/mailman/listinfo/haskell-cafe