BroadcastChan

A closable, fair, single-wakeup channel that avoids the 0 reader space leak that Control.Concurrent.Chan from base suffers from.

The Chan type from Control.Concurrent.Chan consists of both a read and write end combined into a single value. This means there is always at least 1 read end for a Chan, which keeps any values written to it alive. This is a problem for applications/libraries that want to have a channel that can have zero listeners.

Suppose we have an library that produces events and we want to let users register to receive events. If we use a channel and write all events to it, we would like to drop and garbage collect any events that take place when there are 0 listeners. The always present read end of Chan from base makes this impossible. We end up with a Chan that forever accumulates more and more events that will never get removed, resulting in a memory leak.

BroadcastChan splits channels into separate read and write ends. Any message written to a a channel with no existing read end is immediately dropped so it can be garbage collected. Once a read end is created, all messages written to the channel will be accessible to that read end.

Once all read ends for a channel have disappeared and been garbage collected, the channel will return to dropping messages as soon as they are written.

Why should I use BroadcastChan over Control.Concurrent.Chan?

• BroadcastChan is closable,
• BroadcastChan has no 0 reader space leak,
• BroadcastChan has comparable or better performance.

Why should I use BroadcastChan over various (closable) STM channels?

• BroadcastChan is single-wakeup,
• BroadcastChan is fair,
• BroadcastChan performs better under contention.

Datatypes

data BroadcastChan (dir :: Direction) a

The abstract type representing the read or write end of a BroadcastChan.

Instances

Eq (BroadcastChan dir a)

Defined in BroadcastChan.Internal

Methods

(==) :: BroadcastChan dir a -> BroadcastChan dir a -> Bool

(/=) :: BroadcastChan dir a -> BroadcastChan dir a -> Bool

data Direction

Used with DataKinds as phantom type indicating whether a BroadcastChan value is a read or write end.

Constructors

In

Indicates a write BroadcastChan

Out

Indicates a read BroadcastChan

type In = 'In

Alias for the In type from the Direction kind, allows users to write the BroadcastChan In a type without enabling DataKinds.

type Out = 'Out

Alias for the Out type from the Direction kind, allows users to write the BroadcastChan Out a type without enabling DataKinds.

Construction

newBroadcastChan :: MonadIO m => m (BroadcastChan In a)

Creates a new BroadcastChan write end.

newBChanListener :: MonadIO m => BroadcastChan dir a -> m (BroadcastChan Out a)

Create a new read end for a BroadcastChan.

BroadcastChan In:

Will receive all messages written to the channel after this read end is created.

BroadcastChan Out:

Will receive all currently unread messages and all future messages.

Basic Operations

readBChan :: MonadIO m => BroadcastChan Out a -> m (Maybe a)

Read the next value from the read end of a BroadcastChan. Returns Nothing if the BroadcastChan is closed and empty. See BroadcastChan.Throw.readBChan for an exception throwing variant.

writeBChan :: MonadIO m => BroadcastChan In a -> a -> m Bool

Write a value to write end of a BroadcastChan. Any messages written while there are no live read ends are dropped on the floor and can be immediately garbage collected, thus avoiding space leaks.

The return value indicates whether the write succeeded, i.e., True if the message was written, False is the channel is closed. See BroadcastChan.Throw.writeBChan for an exception throwing variant.

closeBChan :: MonadIO m => BroadcastChan In a -> m Bool

Close a BroadcastChan, disallowing further writes. Returns True if the BroadcastChan was closed. Returns False if the BroadcastChan was already closed.

isClosedBChan :: MonadIO m => BroadcastChan dir a -> m Bool

Check whether a BroadcastChan is closed. True meaning that future read/write operations on the channel will always fail.

BroadcastChan In:

True indicates the channel is closed and writes will always fail.

Beware of TOC-TOU races: It is possible for a BroadcastChan to be closed by another thread. If multiple threads use the same channel a closeBChan from another thread can result in the channel being closed right after isClosedBChan returns.

BroadcastChan Out:

True indicates the channel is both closed and empty, meaning reads will always fail.

getBChanContents :: BroadcastChan dir a -> IO [a]

Return a lazy list representing the messages written to the channel.

Uses unsafeInterleaveIO to defer the IO operations.

BroadcastChan In:

The list contains every message written to the channel after this IO action completes.

BroadcastChan Out:

The list contains every currently unread message and all future messages. It's safe to keep using the original channel in any thread.

Unlike getChanContents from Control.Concurrent, the list resulting from this function is not affected by reads on the input channel. Every message that is unread or written after the IO action completes will end up in the result list.

Parallel processing

data Action

Action to take when an exception occurs while processing an element.

Constructors

Drop

Drop the current element and continue processing.

Retry

Retry by appending the current element to the queue of remaining elements.

Terminate

Stop all processing and reraise the exception.

Instances

Eq Action

Defined in BroadcastChan.Extra

Methods

(==) :: Action -> Action -> Bool

(/=) :: Action -> Action -> Bool

Show Action

Defined in BroadcastChan.Extra

Methods

showsPrec :: Int -> Action -> ShowS

show :: Action -> String

showList :: [Action] -> ShowS

data Handler m a

Exception handler for parallel processing.

Constructors

Simple Action

Always take the specified Action.

Handle (a -> SomeException -> m Action)

Allow inspection of the element, exception, and execution of monadic actions before deciding the Action to take.

parMapM_

Arguments

:: (Foldable f, MonadUnliftIO m)

=> Handler m a

Exception handler

-> Int

Number of parallel threads to use

-> (a -> m ())

Function to run in parallel

-> f a

The Foldable to process in parallel

-> m ()

Map a monadic function over a Foldable, processing elements in parallel.

This function does NOT guarantee that elements are processed in a deterministic order!

parFoldMap

Arguments

:: (Foldable f, MonadUnliftIO m)

=> Handler m a

Exception handler

-> Int

Number of parallel threads to use

-> (a -> m b)

Function to run in parallel

-> (r -> b -> r)

Function to fold results with

-> r

Zero element for the fold

-> f a

The Foldable to process

-> m r

Like parMapM_, but folds the individual results into single result value.

This function does NOT guarantee that elements are processed in a deterministic order!

parFoldMapM

Arguments

:: forall a b f m r. (Foldable f, MonadUnliftIO m)

=> Handler m a

Exception handler

-> Int

Number of parallel threads to use

-> (a -> m b)

Function to run in parallel

-> (r -> b -> m r)

Monadic function to fold results with

-> r

Zero element for the fold

-> f a

The Foldable to process

-> m r

Like parFoldMap, but uses a monadic fold function.

This function does NOT guarantee that elements are processed in a deterministic order!

Foldl combinators

Combinators for use with Tekmo's foldl package.

foldBChan :: (MonadIO m, MonadIO n) => (x -> a -> x) -> x -> (x -> b) -> BroadcastChan d a -> n (m b)

Strict fold of the BroadcastChan​'s messages. Can be used with Control.Foldl from Tekmo's foldl package:

Control.Foldl.purely foldBChan :: (MonadIO m, MonadIO n) => Fold a b -> BroadcastChan d a -> n (m b)

The result of this function is a nested monadic value to give more fine-grained control/separation between the start of listening for messages and the start of processing. The inner action folds the actual messages and completes when the channel is closed and exhausted. The outer action controls from when on messages are received. Specifically:

BroadcastChan In:

Will process all messages sent after the outer action completes.

BroadcastChan Out:

Will process all messages that are unread when the outer action completes, as well as all future messages.

After the outer action completes the fold is unaffected by other (concurrent) reads performed on the original channel. So it's safe to reuse the channel.

foldBChanM :: (MonadIO m, MonadIO n) => (x -> a -> m x) -> m x -> (x -> m b) -> BroadcastChan d a -> n (m b)

Strict, monadic fold of the BroadcastChan​'s messages. Can be used with Control.Foldl from Tekmo's foldl package:

Control.Foldl.impurely foldBChanM :: (MonadIO m, MonadIO n) => FoldM m a b -> BroadcastChan d a -> n (m b)

Has the same behaviour and guarantees as foldBChan.