Copyright	No rights reserved
License	UNLICENSE
Maintainer	xenog@protonmail.com
Stability	experimental
Portability	POSIX
Safe Haskell	None
Language	Haskell2010

Control.Concurrent.NQE.Process

Description

This is the core of the NQE library. It is composed of code to deal with processes and mailboxes. Processes represent concurrent threads that receive messages via a mailbox, also referred to as a channel. NQE is inspired by Erlang/OTP and it stands for “Not Quite Erlang”. A process is analogous to an actor in Scala, or an object in the original (Alan Kay) sense of the word. To implement synchronous communication NQE makes use of STM actions embedded in asynchronous messages.

Synopsis

type Listen a = a -> STM ()
data Mailbox msg = OutChan mbox => Mailbox !(mbox msg) !Unique
data Inbox msg = (OutChan mbox, InChan mbox) => Inbox !(mbox msg) !Unique
data Process msg = Process {
- getProcessAsync :: Async ()
- getProcessMailbox :: Mailbox msg
}
class InChan mbox where
- mailboxEmptySTM :: mbox msg -> STM Bool
- receiveSTM :: mbox msg -> STM msg
- requeueSTM :: msg -> mbox msg -> STM ()
class OutChan mbox where
- mailboxFullSTM :: mbox msg -> STM Bool
- sendSTM :: msg -> mbox msg -> STM ()
inboxToMailbox :: Inbox msg -> Mailbox msg
wrapChannel :: (MonadIO m, InChan mbox, OutChan mbox) => mbox msg -> m (Inbox msg)
newInbox :: MonadIO m => m (Inbox msg)
newBoundedInbox :: MonadIO m => Natural -> m (Inbox msg)
send :: (MonadIO m, OutChan mbox) => msg -> mbox msg -> m ()
receive :: (InChan mbox, MonadIO m) => mbox msg -> m msg
query :: (MonadIO m, OutChan mbox) => (Listen response -> request) -> mbox request -> m response
queryU :: (MonadUnliftIO m, OutChan mbox) => Int -> (Listen response -> request) -> mbox request -> m (Maybe response)
queryS :: (MonadUnliftIO m, OutChan mbox) => Int -> (Listen response -> request) -> mbox request -> m (Maybe response)
receiveMatch :: (MonadIO m, InChan mbox) => mbox msg -> (msg -> Maybe a) -> m a
receiveMatchU :: (MonadUnliftIO m, InChan mbox) => Int -> mbox msg -> (msg -> Maybe a) -> m (Maybe a)
receiveMatchS :: (MonadUnliftIO m, InChan mbox) => Int -> mbox msg -> (msg -> Maybe a) -> m (Maybe a)
receiveMatchSTM :: InChan mbox => mbox msg -> (msg -> Maybe a) -> STM a
mailboxEmpty :: (MonadIO m, InChan mbox) => mbox msg -> m Bool
requeueListSTM :: InChan mbox => [msg] -> mbox msg -> STM ()
withProcess :: MonadUnliftIO m => (Inbox msg -> m ()) -> (Process msg -> m a) -> m a
process :: MonadUnliftIO m => (Inbox msg -> m ()) -> m (Process msg)
newMailbox :: MonadUnliftIO m => m (Inbox msg, Mailbox msg)

Documentation

type Listen a = a -> STM () #

STM function that receives an event and does something with it.

data Mailbox msg #

Channel that only allows messages to be sent to it.

Constructors

OutChan mbox => Mailbox !(mbox msg) !Unique

Instances

OutChan Mailbox #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: Mailbox msg -> STM Bool # sendSTM :: msg -> Mailbox msg -> STM () #
Eq (Mailbox msg) #
Instance details Defined in Control.Concurrent.NQE.Process Methods (==) :: Mailbox msg -> Mailbox msg -> Bool # (/=) :: Mailbox msg -> Mailbox msg -> Bool #
Hashable (Mailbox msg) #
Instance details Defined in Control.Concurrent.NQE.Process Methods hashWithSalt :: Int -> Mailbox msg -> Int # hash :: Mailbox msg -> Int #

data Inbox msg #

Channel that allows to send or receive messages.

Constructors

(OutChan mbox, InChan mbox) => Inbox !(mbox msg) !Unique

Instances

OutChan Inbox #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: Inbox msg -> STM Bool # sendSTM :: msg -> Inbox msg -> STM () #
InChan Inbox #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxEmptySTM :: Inbox msg -> STM Bool # receiveSTM :: Inbox msg -> STM msg # requeueSTM :: msg -> Inbox msg -> STM () #
Eq (Inbox msg) #
Instance details Defined in Control.Concurrent.NQE.Process Methods (==) :: Inbox msg -> Inbox msg -> Bool # (/=) :: Inbox msg -> Inbox msg -> Bool #

data Process msg #

Async handle and Mailbox for a process.

Constructors

Process
Fields getProcessAsync :: Async () getProcessMailbox :: Mailbox msg

Instances

OutChan Process #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: Process msg -> STM Bool # sendSTM :: msg -> Process msg -> STM () #
Eq (Process msg) #
Instance details Defined in Control.Concurrent.NQE.Process Methods (==) :: Process msg -> Process msg -> Bool # (/=) :: Process msg -> Process msg -> Bool #
Hashable (Process msg) #
Instance details Defined in Control.Concurrent.NQE.Process Methods hashWithSalt :: Int -> Process msg -> Int # hash :: Process msg -> Int #

class InChan mbox where #

Class for implementation of an Inbox.

Methods

mailboxEmptySTM :: mbox msg -> STM Bool #

Are there messages queued?

receiveSTM :: mbox msg -> STM msg #

Receive a message.

requeueSTM :: msg -> mbox msg -> STM () #

Put a message in the mailbox such that it is received next.

Instances

InChan TQueue #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxEmptySTM :: TQueue msg -> STM Bool # receiveSTM :: TQueue msg -> STM msg # requeueSTM :: msg -> TQueue msg -> STM () #
InChan TBQueue #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxEmptySTM :: TBQueue msg -> STM Bool # receiveSTM :: TBQueue msg -> STM msg # requeueSTM :: msg -> TBQueue msg -> STM () #
InChan Inbox #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxEmptySTM :: Inbox msg -> STM Bool # receiveSTM :: Inbox msg -> STM msg # requeueSTM :: msg -> Inbox msg -> STM () #

class OutChan mbox where #

Class for implementation of a Mailbox.

Methods

mailboxFullSTM :: mbox msg -> STM Bool #

Is this bounded channel full? Always False for unbounded channels.

sendSTM :: msg -> mbox msg -> STM () #

Send a message to this channel.

Instances

OutChan TQueue #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: TQueue msg -> STM Bool # sendSTM :: msg -> TQueue msg -> STM () #
OutChan TBQueue #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: TBQueue msg -> STM Bool # sendSTM :: msg -> TBQueue msg -> STM () #
OutChan Process #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: Process msg -> STM Bool # sendSTM :: msg -> Process msg -> STM () #
OutChan Inbox #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: Inbox msg -> STM Bool # sendSTM :: msg -> Inbox msg -> STM () #
OutChan Mailbox #
Instance details Defined in Control.Concurrent.NQE.Process Methods mailboxFullSTM :: Mailbox msg -> STM Bool # sendSTM :: msg -> Mailbox msg -> STM () #

inboxToMailbox :: Inbox msg -> Mailbox msg #

Get a send-only Mailbox for an Inbox.

wrapChannel :: (MonadIO m, InChan mbox, OutChan mbox) => mbox msg -> m (Inbox msg) #

Wrap a channel in an Inbox

newInbox :: MonadIO m => m (Inbox msg) #

Create an unbounded Inbox.

newBoundedInbox :: MonadIO m => Natural -> m (Inbox msg) #

Inbox with upper bound on number of allowed queued messages.

send :: (MonadIO m, OutChan mbox) => msg -> mbox msg -> m () #

Send a message to a channel.

receive :: (InChan mbox, MonadIO m) => mbox msg -> m msg #

Receive a message from a channel.

query :: (MonadIO m, OutChan mbox) => (Listen response -> request) -> mbox request -> m response #

Send request to channel and wait for a response. The request STM action will be created by this function.

queryU :: (MonadUnliftIO m, OutChan mbox) => Int -> (Listen response -> request) -> mbox request -> m (Maybe response) #

Do a query but timeout after u microseconds. Return Nothing if timeout reached.

queryS :: (MonadUnliftIO m, OutChan mbox) => Int -> (Listen response -> request) -> mbox request -> m (Maybe response) #

Do a query but timeout after s seconds. Return Nothing if timeout reached.

receiveMatch :: (MonadIO m, InChan mbox) => mbox msg -> (msg -> Maybe a) -> m a #

Test all messages in a channel against the supplied function and return the first matching message. Will block until a match is found. Messages that do not match remain in the channel.

receiveMatchU :: (MonadUnliftIO m, InChan mbox) => Int -> mbox msg -> (msg -> Maybe a) -> m (Maybe a) #

Like receiveMatch but with a timeout set at u microseconds. Returns Nothing if timeout is reached.

receiveMatchS :: (MonadUnliftIO m, InChan mbox) => Int -> mbox msg -> (msg -> Maybe a) -> m (Maybe a) #

Like receiveMatch but with a timeout set at s seconds. Returns Nothing if timeout is reached.

receiveMatchSTM :: InChan mbox => mbox msg -> (msg -> Maybe a) -> STM a #

Match a message in the channel as an atomic STM action.

mailboxEmpty :: (MonadIO m, InChan mbox) => mbox msg -> m Bool #

Check if the channel is empty.

requeueListSTM :: InChan mbox => [msg] -> mbox msg -> STM () #

Put a list of messages at the start of a channel, so that the last element of the list is the next message to be received.

withProcess :: MonadUnliftIO m => (Inbox msg -> m ()) -> (Process msg -> m a) -> m a #

Run a process in the background and pass it to a function. Stop the background process once the function returns. Background process exceptions are re-thrown in the current thread.

process :: MonadUnliftIO m => (Inbox msg -> m ()) -> m (Process msg) #

Run a process in the background and return the Process handle. Background process exceptions are re-thrown in the current thread.

newMailbox :: MonadUnliftIO m => m (Inbox msg, Mailbox msg) #

Create an unbounded inbox and corresponding mailbox.