distributed-process-0.4.1: Cloud Haskell: Erlang-style concurrency in Haskell

Safe HaskellNone

Control.Distributed.Process

Contents

Description

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.

Synopsis

Basic types

data NodeId Source

Node identifier

data SendPortId Source

A send port is identified by a SendPortId.

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

processNodeId :: ProcessId -> NodeIdSource

The ID of the node the process is running on

sendPortProcessId :: SendPortId -> ProcessIdSource

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 ()Source

Send a message

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

Wait for a message of a specific type

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

Like expect but with a timeout

Channels

data ReceivePort a Source

The receive end of a typed channel (not serializable)

Note that ReceivePort implements Functor, Applicative, Alternative and Monad. This is especially useful when merging receive ports.

data SendPort a Source

The send send of a typed channel (serializable)

sendPortId :: SendPort a -> SendPortIdSource

The (unique) ID of this send port

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

Create a new typed channel

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

Send a message on a typed channel

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

Wait for a message on a typed channel

receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)Source

Like receiveChan but with a timeout. If the timeout is 0, do a non-blocking check for a message.

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

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)Source

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

Advanced messaging

data Match b Source

Opaque type used in receiveWait and receiveTimeout

receiveWait :: [Match b] -> Process bSource

Test the matches in order against each message in the queue

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

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 bSource

Match against any message of the right type

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

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

matchUnknown :: Process b -> Match bSource

Remove any message from the queue

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

Match against an arbitrary message

Process management

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

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 aSource

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 aSource

Terminate (throws a ProcessTerminationException)

data ProcessRegistrationException Source

Exception thrown when a process attempts to register a process under an already-registered name or to unregister a name that hasn't been registered

data SpawnRef Source

SpawnRef are used to return pids of spawned processes

getSelfPid :: Process ProcessIdSource

Our own process ID

getSelfNode :: Process NodeIdSource

Get the node ID of our local node

Monitoring and linking

link :: ProcessId -> Process ()Source

Link to a remote process (asynchronous)

When process A links to process B (that is, process A calls link pidB) then an asynchronous exception will be thrown to process A when process B terminates (normally or abnormally), or when process A gets disconnected from process B. Although it is technically possible to catch these exceptions, chances are if you find yourself trying to do so you should probably be using monitor rather than link. In particular, code such as

 link pidB   -- Link to process B
 expect      -- Wait for a message from process B
 unlink pidB -- Unlink again

doesn't quite do what one might expect: if process B sends a message to process A, and subsequently terminates, then process A might or might not be terminated too, depending on whether the exception is thrown before or after the unlink (i.e., this code has a race condition).

Linking is all-or-nothing: A is either linked to B, or it's not. A second call to link has no effect.

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

linkNode :: NodeId -> Process ()Source

Link to a node (asynchronous)

linkPort :: SendPort a -> Process ()Source

Link to a channel (asynchronous)

unlink :: ProcessId -> Process ()Source

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 ()Source

Remove a node link

This has the same synchronous/asynchronous nature as unlink.

unlinkPort :: SendPort a -> Process ()Source

Remove a channel (send port) link

This has the same synchronous/asynchronous nature as unlink.

monitor :: ProcessId -> Process MonitorRefSource

Monitor another process (asynchronous)

When process A monitors process B (that is, process A calls monitor pidB) then process A will receive a ProcessMonitorNotification when process B terminates (normally or abnormally), or when process A gets disconnected from process B. You receive this message like any other (using expect); the notification includes a reason (DiedNormal, DiedException, DiedDisconnect, etc.).

Every call to monitor returns a new monitor reference MonitorRef; if multiple monitors are set up, multiple notifications will be delivered and monitors can be disabled individually using unmonitor.

monitorNode :: NodeId -> Process MonitorRefSource

Monitor a node (asynchronous)

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

Monitor a typed channel (asynchronous)

unmonitor :: MonitorRef -> Process ()Source

Remove a monitor

This has the same synchronous/asynchronous nature as unlink.

data PortLinkException Source

Exception thrown when a linked channel (port) dies

data MonitorRef Source

MonitorRef is opaque for regular Cloud Haskell processes

data DiedReason Source

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

Closures

data Closure a

A closure is a static value and an encoded environment

Instances

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

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

Resolve a static value

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

Resolve a closure

data RemoteTable

Runtime dictionary for unstatic lookups

Logging

say :: String -> Process ()Source

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 ()Source

Register a process with the local registry (asynchronous). This version will wait until a response is gotten from the management process. The name must not already be registered. The process need not be on this node. A bad registration will result in a ProcessRegistrationException

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

reregister :: String -> ProcessId -> Process ()Source

Like register, but will replace an existing registration. The name must already be registered.

unregister :: String -> Process ()Source

Remove a process from the local registry (asynchronous). This version will wait until a response is gotten from the management process. The name must already be registered.

whereis :: String -> Process (Maybe ProcessId)Source

Query the local process registry

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

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

registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()Source

Register a process with a remote registry (asynchronous).

The process to be registered does not have to live on the same remote node. Reply wil come in the form of a RegisterReply message

See comments in whereisRemoteAsync

unregisterRemoteAsync :: NodeId -> String -> Process ()Source

Remove a process from a remote registry (asynchronous).

Reply wil come in the form of a RegisterReply message

See comments in whereisRemoteAsync

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

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 ()Source

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

data WhereIsReply Source

(Asynchronous) reply from whereis

Exception handling

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

Lift catch

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

Lift mask

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

Lift bracket

Auxiliary API

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

Asynchronous version of spawn

(spawn is defined in terms of spawnAsync and expect)

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

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

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

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)Source

Like spawnLink, but monitor the spawned process

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

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

data DidSpawn Source

(Asynchronius) reply from spawn

Local versions of spawn

spawnLocal :: Process () -> Process ProcessIdSource

Spawn a process on the local node

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

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 ()Source

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 ()Source

Reconnect to a sendport. See reconnect for more information.