Control.Distributed.Process

Cloud Haskell

This is an implementation of Cloud Haskell, as described in Towards Haskell in the Cloud by Jeff Epstein, Andrew Black, and Simon Peyton Jones (http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/), although some of the details are different. The precise message passing semantics are based on A unified semantics for future Erlang by Hans Svensson, Lars-Åke Fredlund and Clara Benac Earle.

Basic types

data ProcessId

Process identifier

Instances

Eq ProcessId

Ord ProcessId

Show ProcessId

Typeable ProcessId

Binary ProcessId

data NodeId

Node identifier

Instances

Eq NodeId

Ord NodeId

Show NodeId

Typeable NodeId

Binary NodeId

data Process a

The Cloud Haskell Process type

Instances

Monad Process

Functor Process

Typeable1 Process

Applicative Process

MonadIO Process

MonadReader LocalProcess Process

data SendPortId

A send port is identified by a SendPortId.

You cannot send directly to a SendPortId; instead, use newChan to create a SendPort.

Instances

Eq SendPortId

Ord SendPortId

Show SendPortId

Binary SendPortId

processNodeId :: ProcessId -> NodeId

The ID of the node the process is running on

sendPortProcessId :: SendPortId -> ProcessId

The ID of the process that will receive messages sent on this port

liftIO :: MonadIO m => forall a. IO a -> m a

Lift a computation from the IO monad.

Basic messaging

send :: Serializable a => ProcessId -> a -> Process ()

Send a message

expect :: forall a. Serializable a => Process a

Wait for a message of a specific type

Channels

data ReceivePort a

The receive end of a typed channel (not serializable)

Instances

Typeable1 ReceivePort

data SendPort a

The send send of a typed channel (serializable)

Instances

Typeable1 SendPort

Eq (SendPort a)

Eq (SendPort a) => Ord (SendPort a)

Show (SendPort a)

Binary (SendPort a)

sendPortId :: SendPort a -> SendPortId

The (unique) ID of this send port

newChan :: Serializable a => Process (SendPort a, ReceivePort a)

Create a new typed channel

sendChan :: Serializable a => SendPort a -> a -> Process ()

Send a message on a typed channel

receiveChan :: Serializable a => ReceivePort a -> Process a

Wait for a message on a typed channel

mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

Merge a list of typed channels.

The result port is left-biased: if there are messages available on more than one port, the first available message is returned.

mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

Like mergePortsBiased, but with a round-robin scheduler (rather than left-biased)

Advanced messaging

data Match b

Opaque type used in receiveWait and receiveTimeout

receiveWait :: [Match b] -> Process b

Test the matches in order against each message in the queue

receiveTimeout :: Int -> [Match b] -> Process (Maybe b)

Like receiveWait but with a timeout.

If the timeout is zero do a non-blocking check for matching messages. A non-zero timeout is applied only when waiting for incoming messages (that is, after we have checked the messages that are already in the mailbox).

match :: forall a b. Serializable a => (a -> Process b) -> Match b

Match against any message of the right type

matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b

Match against any message of the right type that satisfies a predicate

matchUnknown :: Process b -> Match b

Remove any message from the queue

data AbstractMessage

Constructors

AbstractMessage

Fields

forward :: ProcessId -> Process ()

matchAny :: forall b. (AbstractMessage -> Process b) -> Match b

Match against an arbitrary message

Process management

spawn :: NodeId -> Closure (Process ()) -> Process ProcessId

Spawn a process

For more information about Closure, see Control.Distributed.Process.Closure.

See also call.

call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a

Run a process remotely and wait for it to reply

We monitor the remote process: if it dies before it can send a reply, we die too.

For more information about Static, SerializableDict, and Closure, see Control.Distributed.Process.Closure.

See also spawn.

terminate :: Process a

Terminate (throws a ProcessTerminationException)

data ProcessTerminationException

Thrown by terminate

Constructors

ProcessTerminationException

Instances

Show ProcessTerminationException

Typeable ProcessTerminationException

Exception ProcessTerminationException

data SpawnRef

SpawnRef are used to return pids of spawned processes

Instances

Eq SpawnRef

Show SpawnRef

Typeable SpawnRef

Binary SpawnRef

getSelfPid :: Process ProcessId

Our own process ID

getSelfNode :: Process NodeId

Get the node ID of our local node

Monitoring and linking

link :: ProcessId -> Process ()

Link to a remote process (asynchronous)

Note that link provides unidirectional linking (see spawnSupervised). Linking makes no distinction between normal and abnormal termination of the remote process.

linkNode :: NodeId -> Process ()

Link to a node (asynchronous)

linkPort :: SendPort a -> Process ()

Link to a channel (asynchronous)

unlink :: ProcessId -> Process ()

Remove a link

This is synchronous in the sense that once it returns you are guaranteed that no exception will be raised if the remote process dies. However, it is asynchronous in the sense that we do not wait for a response from the remote node.

unlinkNode :: NodeId -> Process ()

Remove a node link

This has the same synchronous/asynchronous nature as unlink.

unlinkPort :: SendPort a -> Process ()

Remove a channel (send port) link

This has the same synchronous/asynchronous nature as unlink.

monitor :: ProcessId -> Process MonitorRef

Monitor another process (asynchronous)

monitorNode :: NodeId -> Process MonitorRef

Monitor a node (asynchronous)

monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef

Monitor a typed channel (asynchronous)

unmonitor :: MonitorRef -> Process ()

Remove a monitor

This has the same synchronous/asynchronous nature as unlink.

data ProcessLinkException

Exceptions thrown when a linked process dies

Constructors

ProcessLinkException !ProcessId !DiedReason

Instances

Show ProcessLinkException

Typeable ProcessLinkException

Exception ProcessLinkException

data NodeLinkException

Exception thrown when a linked node dies

Constructors

NodeLinkException !NodeId !DiedReason

Instances

Show NodeLinkException

Typeable NodeLinkException

Exception NodeLinkException

data PortLinkException

Exception thrown when a linked channel (port) dies

Constructors

PortLinkException !SendPortId !DiedReason

Instances

Show PortLinkException

Typeable PortLinkException

Exception PortLinkException

data MonitorRef

MonitorRef is opaque for regular Cloud Haskell processes

Instances

Eq MonitorRef

Ord MonitorRef

Show MonitorRef

Binary MonitorRef

data ProcessMonitorNotification

Message sent by process monitors

Constructors

ProcessMonitorNotification !MonitorRef !ProcessId !DiedReason

Instances

Show ProcessMonitorNotification

Typeable ProcessMonitorNotification

Binary ProcessMonitorNotification

data NodeMonitorNotification

Message sent by node monitors

Constructors

NodeMonitorNotification !MonitorRef !NodeId !DiedReason

Instances

Show NodeMonitorNotification

Typeable NodeMonitorNotification

Binary NodeMonitorNotification

data PortMonitorNotification

Message sent by channel (port) monitors

Constructors

PortMonitorNotification !MonitorRef !SendPortId !DiedReason

Instances

Show PortMonitorNotification

Typeable PortMonitorNotification

Binary PortMonitorNotification

data DiedReason

Why did a process die?

Constructors

DiedNormal

Normal termination

DiedException !String

The process exited with an exception (provided as String because Exception does not implement Binary)

DiedDisconnect

We got disconnected from the process node

DiedNodeDown

The process node died

DiedUnknownId

Invalid (processnodechannel) identifier

Instances

Eq DiedReason

Show DiedReason

Binary DiedReason

Closures

data Closure a

A closure is a static value and an encoded environment

Instances

Typeable1 Closure

Show (Closure a)

Typeable a => Binary (Closure a)

closure

Arguments

:: Static (ByteString -> a)

Decoder

-> ByteString

Encoded closure environment

-> Closure a

data Static a

A static value. Static is opaque; see staticLabel and staticApply.

Instances

Typeable1 Static

Show (Static a)

Typeable a => Binary (Static a)

unStatic :: Typeable a => Static a -> Process a

Resolve a static value

unClosure :: Typeable a => Closure a -> Process a

Resolve a closure

data RemoteTable

Runtime dictionary for unstatic lookups

Logging

say :: String -> Process ()

Log a string

say message sends a message (time, pid of the current process, message) to the process registered as logger. By default, this process simply sends the string to stderr. Individual Cloud Haskell backends might replace this with a different logger process, however.

Registry

register :: String -> ProcessId -> Process ()

Register a process with the local registry (asynchronous).

The process to be registered does not have to be local itself.

unregister :: String -> Process ()

Remove a process from the local registry (asynchronous).

whereis :: String -> Process (Maybe ProcessId)

Query the local process registry

nsend :: Serializable a => String -> a -> Process ()

Named send to a process in the local registry (asynchronous)

registerRemote :: NodeId -> String -> ProcessId -> Process ()

Register a process with a remote registry (asynchronous).

The process to be registered does not have to live on the same remote node.

unregisterRemote :: NodeId -> String -> Process ()

Remove a process from a remote registry (asynchronous).

whereisRemoteAsync :: NodeId -> String -> Process ()

Query a remote process registry (asynchronous)

Reply will come in the form of a WhereIsReply message.

There is currently no synchronous version of whereisRemoteAsync: if you implement one yourself, be sure to take into account that the remote node might die or get disconnect before it can respond (i.e. you should use monitorNode and take appropriate action when you receive a NodeMonitorNotification).

nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

Named send to a process in a remote registry (asynchronous)

data WhereIsReply

(Asynchronous) reply from whereis

Constructors

WhereIsReply String (Maybe ProcessId)

Instances

Show WhereIsReply

Typeable WhereIsReply

Binary WhereIsReply

Exception handling

catch :: Exception e => Process a -> (e -> Process a) -> Process a

Lift catch

mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b

Lift mask

onException :: Process a -> Process b -> Process a

Lift onException

bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c

Lift bracket

bracket_ :: Process a -> Process b -> Process c -> Process c

Lift bracket_

finally :: Process a -> Process b -> Process a

Lift finally

Auxiliary API

expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)

Like expect but with a timeout

spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef

Asynchronous version of spawn

(spawn is defined in terms of spawnAsync and expect)

spawnSupervised :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)

Spawn a child process, have the child link to the parent and the parent monitor the child

spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId

Spawn a process and link to it

Note that this is just the sequential composition of spawn and link. (The Unified semantics that underlies Cloud Haskell does not even support a synchronous link operation)

spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)

Like spawnLink, but monitor the spawned process

spawnChannel :: forall a. Typeable a => Static (SerializableDict a) -> NodeId -> Closure (ReceivePort a -> Process ()) -> Process (SendPort a)

Spawn a new process, supplying it with a new ReceivePort and return the corresponding SendPort.

data DidSpawn

(Asynchronius) reply from spawn

Constructors

DidSpawn SpawnRef ProcessId

Instances

Show DidSpawn

Typeable DidSpawn

Binary DidSpawn

Local versions of spawn

spawnLocal :: Process () -> Process ProcessId

Spawn a process on the local node

spawnChannelLocal :: Serializable a => (ReceivePort a -> Process ()) -> Process (SendPort a)

Create a new typed channel, spawn a process on the local node, passing it the receive port, and return the send port

Reconnecting

reconnect :: ProcessId -> Process ()

Cloud Haskell provides the illusion of connection-less, reliable, ordered message passing. However, when network connections get disrupted this illusion cannot always be maintained. Once a network connection breaks (even temporarily) no further communication on that connection will be possible. For example, if process A sends a message to process B, and A is then notified (by monitor notification) that it got disconnected from B, A will not be able to send any further messages to B, unless A explicitly indicates that it is acceptable to attempt to reconnect to B using the Cloud Haskell reconnect primitive.

Importantly, when A calls reconnect it acknowledges that some messages to B might have been lost. For instance, if A sends messages m1 and m2 to B, then receives a monitor notification that its connection to B has been lost, calls reconnect and then sends m3, it is possible that B will receive m1 and m3 but not m2.

Note that reconnect does not mean reconnect now but rather /it is okay to attempt to reconnect on the next send/. In particular, if no further communication attempts are made to B then A can use reconnect to clean up its connection to B.

reconnectPort :: SendPort a -> Process ()

Reconnect to a sendport. See reconnect for more information.