Safe Haskell	None
Language	Haskell2010

RIO.ProcessPool.Pool

Description

Launch- and Dispatch messages to processes.

A pool has an Input for Multiplexed messages, and dispatches incoming messges to concurrent processes using user defined MessageBoxes.

The pool starts and stops the processes and creates the message boxes.

The user supplied PoolWorkerCallback usually runs a loop that receives messages from the MessageBox created by the pool for that worker.

When a worker process dies, e.g. because the PoolWorkerCallback returns, the pool process will also cancel the process (just to make sure...) and cleanup the internal Broker.

Synopsis

data Pool poolBox k w = MkPool {
- poolInput :: !(Input (MessageBox poolBox) (Multiplexed k (Maybe w)))
- poolAsync :: !(Async BrokerResult)
}
spawnPool :: forall k w poolBox workerBox m. (IsMessageBoxArg poolBox, IsMessageBoxArg workerBox, Ord k, Display k, HasLogFunc m) => poolBox -> workerBox -> PoolWorkerCallback workerBox w k m -> RIO m (Either SomeException (Pool poolBox k w))
newtype PoolWorkerCallback workerBox w k m = MkPoolWorkerCallback {
- runPoolWorkerCallback :: k -> MessageBox workerBox w -> RIO m ()
}
removePoolWorkerMessage :: k -> Multiplexed k (Maybe w)

Documentation

data Pool poolBox k w Source #

A record containing the message box Input of the Broker and the Async value required to cancel the pools broker process.

Constructors

MkPool
Fields poolInput :: !(Input (MessageBox poolBox) (Multiplexed k (Maybe w))) Message sent to this input are dispatched to workers. If the message is an `Initialize` message, a new `async` process will be started. If the message value is `Nothing`, the processes is killed. poolAsync :: !(Async BrokerResult) The async of the internal `Broker`.

spawnPool :: forall k w poolBox workerBox m. (IsMessageBoxArg poolBox, IsMessageBoxArg workerBox, Ord k, Display k, HasLogFunc m) => poolBox -> workerBox -> PoolWorkerCallback workerBox w k m -> RIO m (Either SomeException (Pool poolBox k w)) Source #

Start a Pool.

Start a process that receives messages sent to the poolInput and dispatches them to the Input of pool member processes. If necessary the pool worker processes are started.

Each pool worker process is started using async and executes the PoolWorkerCallback.

When the callback returns, the process will exit.

Internally the pool uses the async function to wrap the callback.

When a Multiplixed Dispatch message is received with a Nothing then the worker is cancelled and the worker is removed from the map.

Such a message is automatically sent after the PoolWorkerCallback has returned, even when an exception was thrown. See finally.

newtype PoolWorkerCallback workerBox w k m Source #

The function that processes a MessageBox of a worker for a specific key.

Constructors

MkPoolWorkerCallback
Fields runPoolWorkerCallback :: k -> MessageBox workerBox w -> RIO m ()

removePoolWorkerMessage :: k -> Multiplexed k (Maybe w) Source #

This message will cancel the worker with the given key. If the PoolWorkerCallback wants to do cleanup it should use finally or onException.