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 (d :: Direction) a

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

Instances

Eq (BroadcastChan d a)

Methods

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

(/=) :: BroadcastChan d a -> BroadcastChan d 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 In a -> m (BroadcastChan Out a)

Create a new read end for a BroadcastChan. Will receive all messages written to the channel after this read end is created.

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 In a -> m Bool

Check whether a BroadcastChan is closed. True means it's closed, False means it's writable. However:

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

Utility functions

foldBChan :: MonadIO m => (x -> a -> x) -> x -> (x -> b) -> BroadcastChan In a -> m (m b)

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

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

foldBChanM :: MonadIO m => (x -> a -> m x) -> m x -> (x -> m b) -> BroadcastChan In a -> m (m b)

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

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

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

Return a lazy list representing everything written to the supplied BroadcastChan after this IO action returns. Similar to getChanContents.

Uses unsafeInterleaveIO to defer the IO operations.

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

Methods

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

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

Show Action

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

:: (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!