-- Hoogle documentation, generated by Haddock -- See Hoogle, http://www.haskell.org/hoogle/ -- | Cloud Haskell: Erlang-style concurrency in 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. -- You will probably also want to install a Cloud Haskell backend such as -- distributed-process-simplelocalnet. @package distributed-process @version 0.4.2 -- | Spine and element strict list module Control.Distributed.Process.Internal.StrictList -- | Strict list data StrictList a Cons :: !a -> !(StrictList a) -> StrictList a Nil :: StrictList a Snoc :: !(StrictList a) -> !a -> StrictList a Append :: !(StrictList a) -> !(StrictList a) -> StrictList a append :: StrictList a -> StrictList a -> StrictList a foldr :: (a -> b -> b) -> b -> StrictList a -> b module Control.Distributed.Process.Internal.StrictContainerAccessors mapMaybe :: Ord key => key -> Accessor (Map key elem) (Maybe elem) mapDefault :: Ord key => elem -> key -> Accessor (Map key elem) elem -- | Clone of Control.Concurrent.STM.TQueue with support for mkWeakTQueue -- -- Not all functionality from the original module is available: -- unGetTQueue, peekTQueue and tryPeekTQueue are missing. In order to -- implement these we'd need to be able to touch# the write end of the -- queue inside unGetTQueue, but that means we need a version of touch# -- that works within the STM monad. module Control.Distributed.Process.Internal.WeakTQueue -- | TQueue is an abstract type representing an unbounded FIFO -- channel. data TQueue a -- | Build and returns a new instance of TQueue newTQueue :: STM (TQueue a) -- | IO version of newTQueue. This is useful for creating -- top-level TQueues using unsafePerformIO, because using -- atomically inside unsafePerformIO isn't possible. newTQueueIO :: IO (TQueue a) -- | Read the next value from the TQueue. readTQueue :: TQueue a -> STM a -- | A version of readTQueue which does not retry. Instead it -- returns Nothing if no value is available. tryReadTQueue :: TQueue a -> STM (Maybe a) -- | Write a value to a TQueue. writeTQueue :: TQueue a -> a -> STM () -- | Returns True if the supplied TQueue is empty. isEmptyTQueue :: TQueue a -> STM Bool mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a)) instance Typeable1 TQueue instance Eq (TQueue a) -- | Like Control.Concurrent.MVar.Strict but reduce to HNF, not NF module Control.Distributed.Process.Internal.StrictMVar newtype StrictMVar a StrictMVar :: (MVar a) -> StrictMVar a newEmptyMVar :: IO (StrictMVar a) newMVar :: a -> IO (StrictMVar a) takeMVar :: StrictMVar a -> IO a putMVar :: StrictMVar a -> a -> IO () withMVar :: StrictMVar a -> (a -> IO b) -> IO b modifyMVar_ :: StrictMVar a -> (a -> IO a) -> IO () modifyMVar :: StrictMVar a -> (a -> IO (a, b)) -> IO b mkWeakMVar :: StrictMVar a -> IO () -> IO (Weak (StrictMVar a)) -- | Concurrent queue for single reader, single writer module Control.Distributed.Process.Internal.CQueue data CQueue a data BlockSpec NonBlocking :: BlockSpec Blocking :: BlockSpec Timeout :: Int -> BlockSpec data MatchOn m a MatchMsg :: (m -> Maybe a) -> MatchOn m a MatchChan :: (STM a) -> MatchOn m a newCQueue :: IO (CQueue a) -- | Enqueue an element -- -- Enqueue is strict. enqueue :: CQueue a -> a -> IO () -- | Dequeue an element -- -- The timeout (if any) is applied only to waiting for incoming messages, -- not to checking messages that have already arrived dequeue :: CQueue m -> BlockSpec -> [MatchOn m a] -> IO (Maybe a) -- | Weak reference to a CQueue mkWeakCQueue :: CQueue a -> IO () -> IO (Weak (CQueue a)) module Control.Distributed.Process.Serializable -- | Objects that can be sent across the network class (Binary a, Typeable a) => Serializable a -- | Encode type representation as a bytestring encodeFingerprint :: Fingerprint -> ByteString -- | Decode a bytestring into a fingerprint. Throws an IO exception on -- failure decodeFingerprint :: ByteString -> Fingerprint -- | The fingerprint of the typeRep of the argument fingerprint :: Typeable a => a -> Fingerprint -- | Size of a fingerprint sizeOfFingerprint :: Int data Fingerprint :: * -- | Show fingerprint (for debugging purposes) showFingerprint :: Fingerprint -> ShowS -- | Reification of Serializable (see -- Control.Distributed.Process.Closure) data SerializableDict a SerializableDict :: SerializableDict a instance Typeable1 SerializableDict instance (Binary a, Typeable a) => Serializable a -- | Types used throughout the Cloud Haskell framework -- -- We collect all types used internally in a single module because many -- of these data types are mutually recursive and cannot be split across -- modules. module Control.Distributed.Process.Internal.Types -- | Node identifier newtype NodeId NodeId :: EndPointAddress -> NodeId nodeAddress :: NodeId -> EndPointAddress -- | A local process ID consists of a seed which distinguishes processes -- from different instances of the same local node and a counter data LocalProcessId LocalProcessId :: {-# UNPACK #-} !Int32 -> {-# UNPACK #-} !Int32 -> LocalProcessId lpidUnique :: LocalProcessId -> {-# UNPACK #-} !Int32 lpidCounter :: LocalProcessId -> {-# UNPACK #-} !Int32 -- | Process identifier data ProcessId ProcessId :: !NodeId -> {-# UNPACK #-} !LocalProcessId -> ProcessId -- | The ID of the node the process is running on processNodeId :: ProcessId -> !NodeId -- | Node-local identifier for the process processLocalId :: ProcessId -> {-# UNPACK #-} !LocalProcessId -- | Union of all kinds of identifiers data Identifier NodeIdentifier :: !NodeId -> Identifier ProcessIdentifier :: !ProcessId -> Identifier SendPortIdentifier :: !SendPortId -> Identifier nodeOf :: Identifier -> NodeId firstNonReservedProcessId :: Int32 nullProcessId :: NodeId -> ProcessId -- | Local nodes data LocalNode LocalNode :: !NodeId -> !EndPoint -> !(StrictMVar LocalNodeState) -> !(Chan NCMsg) -> !Tracer -> !RemoteTable -> LocalNode -- | NodeId of the node localNodeId :: LocalNode -> !NodeId -- | The network endpoint associated with this node localEndPoint :: LocalNode -> !EndPoint -- | Local node state localState :: LocalNode -> !(StrictMVar LocalNodeState) -- | Channel for the node controller localCtrlChan :: LocalNode -> !(Chan NCMsg) -- | Current active system debug/trace log localTracer :: LocalNode -> !Tracer -- | Runtime lookup table for supporting closures TODO: this should be part -- of the CH state, not the local endpoint state remoteTable :: LocalNode -> !RemoteTable -- | Required for system tracing in the node controller data Tracer LogFileTracer :: !ThreadId -> !(TQueue String) -> !Handle -> Tracer EventLogTracer :: !(String -> IO ()) -> Tracer LocalNodeTracer :: !LocalNode -> Tracer InactiveTracer :: Tracer -- | Local node state data LocalNodeState LocalNodeState :: !(Map LocalProcessId LocalProcess) -> !Int32 -> !Int32 -> !(Map (Identifier, Identifier) (Connection, ImplicitReconnect)) -> LocalNodeState -- | Processes running on this node _localProcesses :: LocalNodeState -> !(Map LocalProcessId LocalProcess) -- | Counter to assign PIDs _localPidCounter :: LocalNodeState -> !Int32 -- | The unique value used to create PIDs (so that processes on -- restarted nodes have new PIDs) _localPidUnique :: LocalNodeState -> !Int32 -- | Outgoing connections _localConnections :: LocalNodeState -> !(Map (Identifier, Identifier) (Connection, ImplicitReconnect)) -- | Processes running on our local node data LocalProcess LocalProcess :: !(CQueue Message) -> !(Weak (CQueue Message)) -> !ProcessId -> !(StrictMVar LocalProcessState) -> !ThreadId -> !LocalNode -> LocalProcess processQueue :: LocalProcess -> !(CQueue Message) processWeakQ :: LocalProcess -> !(Weak (CQueue Message)) processId :: LocalProcess -> !ProcessId processState :: LocalProcess -> !(StrictMVar LocalProcessState) processThread :: LocalProcess -> !ThreadId processNode :: LocalProcess -> !LocalNode -- | Local process state data LocalProcessState LocalProcessState :: !Int32 -> !Int32 -> !Int32 -> !(Map LocalSendPortId TypedChannel) -> LocalProcessState _monitorCounter :: LocalProcessState -> !Int32 _spawnCounter :: LocalProcessState -> !Int32 _channelCounter :: LocalProcessState -> !Int32 _typedChannels :: LocalProcessState -> !(Map LocalSendPortId TypedChannel) -- | The Cloud Haskell Process type newtype Process a Process :: ReaderT LocalProcess IO a -> Process a unProcess :: Process a -> ReaderT LocalProcess IO a -- | Deconstructor for Process (not exported to the public API) runLocalProcess :: LocalProcess -> Process a -> IO a data ImplicitReconnect WithImplicitReconnect :: ImplicitReconnect NoImplicitReconnect :: ImplicitReconnect type LocalSendPortId = Int32 -- | A send port is identified by a SendPortId. -- -- You cannot send directly to a SendPortId; instead, use -- newChan to create a SendPort. data SendPortId SendPortId :: {-# UNPACK #-} !ProcessId -> {-# UNPACK #-} !LocalSendPortId -> SendPortId -- | The ID of the process that will receive messages sent on this port sendPortProcessId :: SendPortId -> {-# UNPACK #-} !ProcessId -- | Process-local ID of the channel sendPortLocalId :: SendPortId -> {-# UNPACK #-} !LocalSendPortId data TypedChannel TypedChannel :: (Weak (TQueue a)) -> TypedChannel -- | The send send of a typed channel (serializable) newtype SendPort a SendPort :: SendPortId -> SendPort a -- | The (unique) ID of this send port sendPortId :: SendPort a -> SendPortId -- | 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. newtype ReceivePort a ReceivePort :: STM a -> ReceivePort a receiveSTM :: ReceivePort a -> STM a -- | Messages consist of their typeRep fingerprint and their encoding data Message Message :: !Fingerprint -> !ByteString -> Message messageFingerprint :: Message -> !Fingerprint messageEncoding :: Message -> !ByteString -- | Turn any serialiable term into a message createMessage :: Serializable a => a -> Message -- | Serialize a message messageToPayload :: Message -> [ByteString] -- | Deserialize a message payloadToMessage :: [ByteString] -> Message -- | MonitorRef is opaque for regular Cloud Haskell processes data MonitorRef MonitorRef :: !Identifier -> !Int32 -> MonitorRef -- | ID of the entity to be monitored monitorRefIdent :: MonitorRef -> !Identifier -- | Unique to distinguish multiple monitor requests by the same process monitorRefCounter :: MonitorRef -> !Int32 -- | Message sent by process monitors data ProcessMonitorNotification ProcessMonitorNotification :: !MonitorRef -> !ProcessId -> !DiedReason -> ProcessMonitorNotification -- | Message sent by node monitors data NodeMonitorNotification NodeMonitorNotification :: !MonitorRef -> !NodeId -> !DiedReason -> NodeMonitorNotification -- | Message sent by channel (port) monitors data PortMonitorNotification PortMonitorNotification :: !MonitorRef -> !SendPortId -> !DiedReason -> PortMonitorNotification -- | Internal exception thrown indirectly by exit data ProcessExitException ProcessExitException :: !ProcessId -> !Message -> ProcessExitException -- | Exceptions thrown when a linked process dies data ProcessLinkException ProcessLinkException :: !ProcessId -> !DiedReason -> ProcessLinkException -- | Exception thrown when a linked node dies data NodeLinkException NodeLinkException :: !NodeId -> !DiedReason -> NodeLinkException -- | Exception thrown when a linked channel (port) dies data PortLinkException PortLinkException :: !SendPortId -> !DiedReason -> PortLinkException -- | 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 ProcessRegistrationException ProcessRegistrationException :: !String -> ProcessRegistrationException -- | Why did a process die? data DiedReason -- | Normal termination DiedNormal :: DiedReason -- | The process exited with an exception (provided as String -- because Exception does not implement Binary) DiedException :: !String -> DiedReason -- | We got disconnected from the process node DiedDisconnect :: DiedReason -- | The process node died DiedNodeDown :: DiedReason -- | Invalid (processnodechannel) identifier DiedUnknownId :: DiedReason -- | (Asynchronous) reply from unmonitor newtype DidUnmonitor DidUnmonitor :: MonitorRef -> DidUnmonitor -- | (Asynchronous) reply from unlink newtype DidUnlinkProcess DidUnlinkProcess :: ProcessId -> DidUnlinkProcess -- | (Asynchronous) reply from unlinkNode newtype DidUnlinkNode DidUnlinkNode :: NodeId -> DidUnlinkNode -- | (Asynchronous) reply from unlinkPort newtype DidUnlinkPort DidUnlinkPort :: SendPortId -> DidUnlinkPort -- | SpawnRef are used to return pids of spawned processes newtype SpawnRef SpawnRef :: Int32 -> SpawnRef -- | (Asynchronius) reply from spawn data DidSpawn DidSpawn :: SpawnRef -> ProcessId -> DidSpawn -- | (Asynchronous) reply from whereis data WhereIsReply WhereIsReply :: String -> (Maybe ProcessId) -> WhereIsReply -- | (Asynchronous) reply from register and unregister data RegisterReply RegisterReply :: String -> Bool -> RegisterReply -- | Provide information about a running process data ProcessInfo ProcessInfo :: NodeId -> [String] -> Maybe Int -> [(ProcessId, MonitorRef)] -> [ProcessId] -> ProcessInfo infoNode :: ProcessInfo -> NodeId infoRegisteredNames :: ProcessInfo -> [String] infoMessageQueueLength :: ProcessInfo -> Maybe Int infoMonitors :: ProcessInfo -> [(ProcessId, MonitorRef)] infoLinks :: ProcessInfo -> [ProcessId] data ProcessInfoNone ProcessInfoNone :: DiedReason -> ProcessInfoNone -- | Messages to the node controller data NCMsg NCMsg :: !Identifier -> !ProcessSignal -> NCMsg ctrlMsgSender :: NCMsg -> !Identifier ctrlMsgSignal :: NCMsg -> !ProcessSignal -- | Signals to the node controller (see NCMsg) data ProcessSignal Link :: !Identifier -> ProcessSignal Unlink :: !Identifier -> ProcessSignal Monitor :: !MonitorRef -> ProcessSignal Unmonitor :: !MonitorRef -> ProcessSignal Died :: Identifier -> !DiedReason -> ProcessSignal Spawn :: !(Closure (Process ())) -> !SpawnRef -> ProcessSignal WhereIs :: !String -> ProcessSignal Register :: !String -> !NodeId -> !(Maybe ProcessId) -> !Bool -> ProcessSignal NamedSend :: !String -> !Message -> ProcessSignal Kill :: !ProcessId -> !String -> ProcessSignal Exit :: !ProcessId -> !Message -> ProcessSignal GetInfo :: !ProcessId -> ProcessSignal localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess) localPidCounter :: Accessor LocalNodeState Int32 localPidUnique :: Accessor LocalNodeState Int32 localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (Connection, ImplicitReconnect)) localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess) localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (Connection, ImplicitReconnect)) monitorCounter :: Accessor LocalProcessState Int32 spawnCounter :: Accessor LocalProcessState Int32 channelCounter :: Accessor LocalProcessState LocalSendPortId typedChannels :: Accessor LocalProcessState (Map LocalSendPortId TypedChannel) typedChannelWithId :: LocalSendPortId -> Accessor LocalProcessState (Maybe TypedChannel) forever' :: Monad m => m a -> m b instance Typeable NodeId instance Typeable LocalProcessId instance Typeable ProcessId instance Typeable1 SendPort instance Typeable1 ReceivePort instance Typeable ProcessRegistrationException instance Typeable ProcessExitException instance Typeable PortLinkException instance Typeable NodeLinkException instance Typeable ProcessLinkException instance Typeable PortMonitorNotification instance Typeable NodeMonitorNotification instance Typeable ProcessMonitorNotification instance Typeable DidUnmonitor instance Typeable DidUnlinkProcess instance Typeable DidUnlinkNode instance Typeable DidUnlinkPort instance Typeable SpawnRef instance Typeable DidSpawn instance Typeable WhereIsReply instance Typeable RegisterReply instance Typeable ProcessInfo instance Typeable ProcessInfoNone instance Typeable1 Process instance Eq NodeId instance Ord NodeId instance Binary NodeId instance Eq LocalProcessId instance Ord LocalProcessId instance Show LocalProcessId instance Eq ProcessId instance Ord ProcessId instance Eq ImplicitReconnect instance Show ImplicitReconnect instance Eq SendPortId instance Ord SendPortId instance Eq Identifier instance Ord Identifier instance Binary (SendPort a) instance Show (SendPort a) instance Eq (SendPort a) instance Ord (SendPort a) instance Functor ReceivePort instance Applicative ReceivePort instance Alternative ReceivePort instance Monad ReceivePort instance Eq MonitorRef instance Ord MonitorRef instance Show MonitorRef instance Show ProcessRegistrationException instance Show DiedReason instance Eq DiedReason instance Show PortLinkException instance Show NodeLinkException instance Show ProcessLinkException instance Show PortMonitorNotification instance Show NodeMonitorNotification instance Show ProcessMonitorNotification instance Binary DidUnmonitor instance Binary DidUnlinkProcess instance Binary DidUnlinkNode instance Binary DidUnlinkPort instance Show SpawnRef instance Binary SpawnRef instance Eq SpawnRef instance Show DidSpawn instance Show WhereIsReply instance Show RegisterReply instance Show ProcessInfo instance Eq ProcessInfo instance Show ProcessInfoNone instance Show ProcessSignal instance Functor Process instance Monad Process instance MonadIO Process instance MonadReader LocalProcess Process instance Applicative Process instance Show NCMsg instance Binary ProcessInfoNone instance Binary ProcessInfo instance Binary RegisterReply instance Binary WhereIsReply instance Binary Identifier instance Binary SendPortId instance Binary DidSpawn instance Binary DiedReason instance Binary ProcessSignal instance Binary MonitorRef instance Binary NCMsg instance Binary PortMonitorNotification instance Binary NodeMonitorNotification instance Binary ProcessMonitorNotification instance Binary ProcessId instance Binary LocalProcessId instance Exception ProcessRegistrationException instance Exception PortLinkException instance Exception NodeLinkException instance Exception ProcessLinkException instance Show ProcessExitException instance Exception ProcessExitException instance Show Message instance Show SendPortId instance Show Identifier instance Show ProcessId instance Show NodeId -- | Simple (internal) system logging/tracing support. module Control.Distributed.Process.Internal.Trace -- | Required for system tracing in the node controller data Tracer data TraceArg TraceStr :: String -> TraceArg Trace :: a -> TraceArg trace :: Tracer -> String -> IO () traceFormat :: Tracer -> String -> [TraceArg] -> IO () startTracing :: LocalNode -> IO LocalNode stopTracer :: Tracer -> IO () module Control.Distributed.Process.Internal.Messaging sendPayload :: LocalNode -> Identifier -> Identifier -> ImplicitReconnect -> [ByteString] -> IO () sendBinary :: Binary a => LocalNode -> Identifier -> Identifier -> ImplicitReconnect -> a -> IO () sendMessage :: Serializable a => LocalNode -> Identifier -> Identifier -> ImplicitReconnect -> a -> IO () disconnect :: LocalNode -> Identifier -> Identifier -> IO () closeImplicitReconnections :: LocalNode -> Identifier -> IO () -- | a impliesDeathOf b is true if the death of a -- (for instance, a node) implies the death of b (for instance, -- a process on that node) impliesDeathOf :: Identifier -> Identifier -> Bool -- | Cloud Haskell primitives -- -- We define these in a separate module so that we don't have to rely on -- the closure combinators module Control.Distributed.Process.Internal.Primitives -- | Send a message send :: Serializable a => ProcessId -> a -> Process () -- | Wait for a message of a specific type expect :: Serializable a => Process a -- | Create a new typed channel newChan :: Serializable a => Process (SendPort a, ReceivePort a) -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process () -- | Wait for a message on a typed channel receiveChan :: Serializable a => ReceivePort a -> Process 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. mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a) -- | Like mergePortsBiased, but with a round-robin scheduler (rather -- than left-biased) mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a) -- | Opaque type used in receiveWait and receiveTimeout data Match b -- | Test the matches in order against each message in the queue receiveWait :: [Match b] -> Process 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). receiveTimeout :: Int -> [Match b] -> Process (Maybe b) -- | Match against any message of the right type match :: Serializable a => (a -> Process b) -> Match b -- | Match against any message of the right type that satisfies a predicate matchIf :: Serializable a => (a -> Bool) -> (a -> Process b) -> Match b -- | Remove any message from the queue matchUnknown :: Process b -> Match b -- | Represents a received message and provides two basic operations on it. data AbstractMessage AbstractMessage :: (ProcessId -> Process ()) -> (forall a b. Serializable a => (a -> Process b) -> Process (Maybe b)) -> AbstractMessage -- | forward the message to ProcessId forward :: AbstractMessage -> ProcessId -> Process () -- | Handle the message. If the type of the message matches the type of the -- first argument to the supplied expression, then the expression will be -- evaluated against it. If this runtime type checking fails, then -- Nothing will be returned to indicate the fact. If the check -- succeeds and evaluation proceeds however, the resulting value with be -- wrapped with Just. maybeHandleMessage :: AbstractMessage -> forall a b. Serializable a => (a -> Process b) -> Process (Maybe b) -- | Match against an arbitrary message. matchAny removes the first -- available message from the process mailbox, and via the -- AbstractMessage type, supports forwarding or handling -- the message if it is of the correct type. If not of the -- right type, then the AbstractMessage -- maybeHandleMessage function will not evaluate the supplied -- expression, but the message will still have been removed from -- the process mailbox! matchAny :: (AbstractMessage -> Process b) -> Match b -- | Match against an arbitrary message. matchAnyIf will only -- remove the message from the process mailbox, if the supplied -- condition matches. The success (or failure) of runtime type checks in -- maybeHandleMessage does not count here, i.e., if the -- condition evaluates to True then the message will be removed -- from the process mailbox and decoded, but that does not -- guarantee that an expression passed to maybeHandleMessage -- will pass the runtime type checks and therefore be evaluated. If the -- types do not match up, then maybeHandleMessage returns -- Nothing. matchAnyIf :: Serializable a => (a -> Bool) -> (AbstractMessage -> Process b) -> Match b matchChan :: ReceivePort a -> (a -> Process b) -> Match b -- | Terminate immediately (throws a ProcessTerminationException) terminate :: Process a -- | Thrown by terminate data ProcessTerminationException ProcessTerminationException :: ProcessTerminationException -- | Die immediately - throws a ProcessExitException with the given -- reason. die :: Serializable a => a -> Process b -- | Forceful request to kill a process. Where exit provides an -- exception that can be caught and handled, kill throws an -- unexposed exception type which cannot be handled explicitly (by type). kill :: ProcessId -> String -> Process () -- | Graceful request to exit a process. Throws ProcessExitException -- with the supplied reason encoded as a message. Any exit -- signal raised in this manner can be handled using the -- catchExit family of functions. exit :: Serializable a => ProcessId -> a -> Process () -- | Catches ProcessExitException. The handler will not be applied -- unless its type matches the encoded data stored in the exception (see -- the reason argument given to the exit primitive). If the -- handler cannot be applied, the exception will be re-thrown. -- -- To handle ProcessExitException without regard for -- reason, see catch. To handle multiple reasons of -- differing types, see catchesExit. catchExit :: (Show a, Serializable a) => Process b -> (ProcessId -> a -> Process b) -> Process b -- | Lift catches (almost). -- -- As ProcessExitException stores the exit reason as a -- typed, encoded message, a handler must accept an input of the expected -- type. In order to handle a list of potentially different handlers (and -- therefore input types), a handler passed to catchesExit must -- accept AbstractMessage and return Maybe (i.e., -- Just p if it handled the exit reason, otherwise -- Nothing). -- -- See maybeHandleMessage and AsbtractMessage for more -- details. catchesExit :: Process b -> [(ProcessId -> AbstractMessage -> (Process (Maybe b)))] -> Process b -- | Internal exception thrown indirectly by exit data ProcessExitException -- | Our own process ID getSelfPid :: Process ProcessId -- | Get the node ID of our local node getSelfNode :: Process NodeId -- | Provide information about a running process data ProcessInfo ProcessInfo :: NodeId -> [String] -> Maybe Int -> [(ProcessId, MonitorRef)] -> [ProcessId] -> ProcessInfo infoNode :: ProcessInfo -> NodeId infoRegisteredNames :: ProcessInfo -> [String] infoMessageQueueLength :: ProcessInfo -> Maybe Int infoMonitors :: ProcessInfo -> [(ProcessId, MonitorRef)] infoLinks :: ProcessInfo -> [ProcessId] -- | Get information about the specified process getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo) -- | 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. link :: 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. unlink :: ProcessId -> Process () -- | 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. monitor :: ProcessId -> Process MonitorRef -- | Remove a monitor -- -- This has the same synchronous/asynchronous nature as unlink. unmonitor :: MonitorRef -> Process () -- | Establishes temporary monitoring of another process. -- -- withMonitor pid code sets up monitoring of pid for -- the duration of code. Note: although monitoring is no longer -- active when withMonitor returns, there might still be -- unreceived monitor messages in the queue. withMonitor :: ProcessId -> Process a -> Process a -- | 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. say :: String -> Process () -- | 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. register :: String -> ProcessId -> Process () -- | Like register, but will replace an existing registration. The -- name must already be registered. reregister :: String -> ProcessId -> Process () -- | 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. unregister :: String -> Process () -- | Query the local process registry whereis :: String -> Process (Maybe ProcessId) -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process () -- | 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 registerRemoteAsync :: NodeId -> String -> ProcessId -> Process () reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process () -- | Remove a process from a remote registry (asynchronous). -- -- Reply wil come in the form of a RegisterReply message -- -- See comments in whereisRemoteAsync unregisterRemoteAsync :: 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). whereisRemoteAsync :: NodeId -> String -> Process () -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () -- | Resolve a closure unClosure :: Typeable a => Closure a -> Process a -- | Resolve a static value unStatic :: Typeable a => Static a -> Process a -- | Lift catch catch :: Exception e => Process a -> (e -> Process a) -> Process a -- | You need this when using catches data Handler a Handler :: (e -> Process a) -> Handler a -- | Lift catches catches :: Process a -> [Handler a] -> Process a -- | Lift try try :: Exception e => Process a -> Process (Either e a) -- | Lift mask mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b -- | Lift onException onException :: Process a -> Process b -> Process a -- | Lift bracket bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c -- | Lift bracket_ bracket_ :: Process a -> Process b -> Process c -> Process c -- | Lift finally finally :: Process a -> Process b -> Process a -- | Like expect but with a timeout expectTimeout :: Serializable a => Int -> Process (Maybe a) -- | Like receiveChan but with a timeout. If the timeout is 0, do a -- non-blocking check for a message. receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a) -- | Asynchronous version of spawn -- -- (spawn is defined in terms of spawnAsync and -- expect) spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef -- | Link to a node (asynchronous) linkNode :: NodeId -> Process () -- | Link to a channel (asynchronous) linkPort :: SendPort a -> Process () -- | Remove a node link -- -- This has the same synchronous/asynchronous nature as unlink. unlinkNode :: NodeId -> Process () -- | Remove a channel (send port) link -- -- This has the same synchronous/asynchronous nature as unlink. unlinkPort :: SendPort a -> Process () -- | Monitor a node (asynchronous) monitorNode :: NodeId -> Process MonitorRef -- | Monitor a typed channel (asynchronous) monitorPort :: Serializable a => SendPort a -> Process MonitorRef -- | 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. reconnect :: ProcessId -> Process () -- | Reconnect to a sendport. See reconnect for more information. reconnectPort :: SendPort a -> Process () -- | Send a message to the internal (system) trace facility. If tracing is -- enabled, this will create a custom trace event. Note that several -- Cloud Haskell sub-systems also generate trace events for -- informational/debugging purposes, thus traces generated this way will -- not be the only output seen. -- -- Just as with the Debug.Trace module, this is a -- debugging/tracing facility for use in development, and should not be -- used in a production setting - which is why the default behaviour is -- to trace to the GHC eventlog. For a general purpose logging facility, -- you should consider say. -- -- Trace events can be written to the GHC event log, a text file, or to -- the standard system logger process (see say). The default -- behaviour for writing to the eventlog requires specific intervention -- to work, without which traces are silently dropped/ignored and no -- output will be generated. The GHC eventlog documentation provides -- information about enabling, viewing and working with event traces: -- http://hackage.haskell.org/trac/ghc/wiki/EventLog. -- -- When a new local node is started, the contents of the environment -- variable DISTRIBUTED_PROCESS_TRACE_FILE are checked for a -- valid file path. If this exists and the file can be opened for -- writing, all trace output will be directed thence. If the environment -- variable is empty, the path invalid, or the file unavailable for -- writing - e.g., because another node has already started tracing to it -- - then the DISTRIBUTED_PROCESS_TRACE_CONSOLE environment -- variable is checked for any non-empty value. If this is set, -- then all trace output will be directed to the system logger process. -- If neither evironment variable provides a valid trace configuration, -- all internal traces are written to Debug.Trace.traceEventIO, -- which writes to the GHC eventlog. -- -- Users of the simplelocalnet Cloud Haskell backend should also -- note that because the trace file option only supports trace output -- from a single node (so as to avoid interleaving), a file trace -- configured for the master node will prevent slaves from tracing to the -- file and they will fall back to using the console/say or -- eventlog instead. trace :: String -> Process () instance Typeable ProcessTerminationException instance Show ProcessTerminationException instance Functor Handler instance Exception ProcessTerminationException module Control.Distributed.Process.Internal.Closure.BuiltIn remoteTable :: RemoteTable -> RemoteTable -- | Static decoder, given a static serialization dictionary. -- -- See module documentation of Control.Distributed.Process.Closure -- for an example. staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a) -- | Serialization dictionary for '()' sdictUnit :: Static (SerializableDict ()) -- | Serialization dictionary for ProcessId sdictProcessId :: Static (SerializableDict ProcessId) -- | Serialization dictionary for SendPort sdictSendPort :: Typeable a => Static (SerializableDict a) -> Static (SerializableDict (SendPort a)) sndStatic :: Static ((a, b) -> b) -- | CP a b is a process with input of type a and output -- of type b type CP a b = Closure (a -> Process b) -- | CP version of id idCP :: Typeable a => CP a a -- | CP version of (***) splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d) => CP a c -> CP b d -> CP (a, b) (c, d) -- | CP version of return returnCP :: Serializable a => Static (SerializableDict a) -> a -> Closure (Process a) -- | (Not quite the) CP version of (>>=) bindCP :: (Typeable a, Typeable b) => Closure (Process a) -> CP a b -> Closure (Process b) -- | CP version of (>>) seqCP :: (Typeable a, Typeable b) => Closure (Process a) -> Closure (Process b) -> Closure (Process b) -- | CP version of link cpLink :: ProcessId -> Closure (Process ()) -- | CP version of unlink cpUnlink :: ProcessId -> Closure (Process ()) -- | CP version of send cpSend :: Typeable a => Static (SerializableDict a) -> ProcessId -> CP a () -- | CP version of expect cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a) -- | CP version of newChan cpNewChan :: Typeable a => Static (SerializableDict a) -> Closure (Process (SendPort a, ReceivePort a)) -- | CP version of delay cpDelay :: ProcessId -> Closure (Process ()) -> Closure (Process ()) -- | Local nodes module Control.Distributed.Process.Node -- | Local nodes data LocalNode -- | Initialize a new local node. newLocalNode :: Transport -> RemoteTable -> IO LocalNode -- | Force-close a local node -- -- TODO: for now we just close the associated endpoint closeLocalNode :: LocalNode -> IO () -- | Spawn a new process on a local node forkProcess :: LocalNode -> Process () -> IO ProcessId -- | Run a process on a local node and wait for it to finish runProcess :: LocalNode -> Process () -> IO () initRemoteTable :: RemoteTable -- | NodeId of the node localNodeId :: LocalNode -> NodeId instance Typeable ProcessKillException instance Functor NC instance Monad NC instance MonadIO NC instance MonadState NCState NC instance MonadReader LocalNode NC instance Show ProcessKillException instance Exception ProcessKillException -- | -- -- 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. -- -- For a detailed description of the package and other reference -- materials, please see the distributed-process wiki page on github: -- https://github.com/haskell-distributed/distributed-process/wiki. module Control.Distributed.Process -- | Process identifier data ProcessId -- | Node identifier data NodeId -- | The Cloud Haskell Process type data Process a -- | A send port is identified by a SendPortId. -- -- You cannot send directly to a SendPortId; instead, use -- newChan to create a SendPort. data SendPortId -- | The ID of the node the process is running on processNodeId :: ProcessId -> NodeId -- | The ID of the process that will receive messages sent on this port sendPortProcessId :: SendPortId -> ProcessId -- | Lift a computation from the IO monad. liftIO :: MonadIO m => forall a. IO a -> m a -- | Send a message send :: Serializable a => ProcessId -> a -> Process () -- | Wait for a message of a specific type expect :: Serializable a => Process a -- | Like expect but with a timeout expectTimeout :: Serializable a => Int -> Process (Maybe a) -- | 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 ReceivePort a -- | The send send of a typed channel (serializable) data SendPort a -- | The (unique) ID of this send port sendPortId :: SendPort a -> SendPortId -- | Create a new typed channel newChan :: Serializable a => Process (SendPort a, ReceivePort a) -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process () -- | Wait for a message on a typed channel receiveChan :: Serializable a => ReceivePort a -> Process a -- | Like receiveChan but with a timeout. If the timeout is 0, do a -- non-blocking check for a message. receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe 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. mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a) -- | Like mergePortsBiased, but with a round-robin scheduler (rather -- than left-biased) mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a) -- | Opaque type used in receiveWait and receiveTimeout data Match b -- | Test the matches in order against each message in the queue receiveWait :: [Match b] -> Process 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). receiveTimeout :: Int -> [Match b] -> Process (Maybe b) -- | Match against any message of the right type match :: Serializable a => (a -> Process b) -> Match b -- | Match against any message of the right type that satisfies a predicate matchIf :: Serializable a => (a -> Bool) -> (a -> Process b) -> Match b -- | Remove any message from the queue matchUnknown :: Process b -> Match b -- | Represents a received message and provides two basic operations on it. data AbstractMessage AbstractMessage :: (ProcessId -> Process ()) -> (forall a b. Serializable a => (a -> Process b) -> Process (Maybe b)) -> AbstractMessage -- | forward the message to ProcessId forward :: AbstractMessage -> ProcessId -> Process () -- | Handle the message. If the type of the message matches the type of the -- first argument to the supplied expression, then the expression will be -- evaluated against it. If this runtime type checking fails, then -- Nothing will be returned to indicate the fact. If the check -- succeeds and evaluation proceeds however, the resulting value with be -- wrapped with Just. maybeHandleMessage :: AbstractMessage -> forall a b. Serializable a => (a -> Process b) -> Process (Maybe b) -- | Match against an arbitrary message. matchAny removes the first -- available message from the process mailbox, and via the -- AbstractMessage type, supports forwarding or handling -- the message if it is of the correct type. If not of the -- right type, then the AbstractMessage -- maybeHandleMessage function will not evaluate the supplied -- expression, but the message will still have been removed from -- the process mailbox! matchAny :: (AbstractMessage -> Process b) -> Match b -- | Match against an arbitrary message. matchAnyIf will only -- remove the message from the process mailbox, if the supplied -- condition matches. The success (or failure) of runtime type checks in -- maybeHandleMessage does not count here, i.e., if the -- condition evaluates to True then the message will be removed -- from the process mailbox and decoded, but that does not -- guarantee that an expression passed to maybeHandleMessage -- will pass the runtime type checks and therefore be evaluated. If the -- types do not match up, then maybeHandleMessage returns -- Nothing. matchAnyIf :: Serializable a => (a -> Bool) -> (AbstractMessage -> Process b) -> Match b matchChan :: ReceivePort a -> (a -> Process b) -> Match b -- | Spawn a process -- -- For more information about Closure, see -- Control.Distributed.Process.Closure. -- -- See also call. spawn :: NodeId -> Closure (Process ()) -> Process ProcessId -- | 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. call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a -- | Terminate immediately (throws a ProcessTerminationException) terminate :: Process a -- | Die immediately - throws a ProcessExitException with the given -- reason. die :: Serializable a => a -> Process b -- | Forceful request to kill a process. Where exit provides an -- exception that can be caught and handled, kill throws an -- unexposed exception type which cannot be handled explicitly (by type). kill :: ProcessId -> String -> Process () -- | Graceful request to exit a process. Throws ProcessExitException -- with the supplied reason encoded as a message. Any exit -- signal raised in this manner can be handled using the -- catchExit family of functions. exit :: Serializable a => ProcessId -> a -> Process () -- | Catches ProcessExitException. The handler will not be applied -- unless its type matches the encoded data stored in the exception (see -- the reason argument given to the exit primitive). If the -- handler cannot be applied, the exception will be re-thrown. -- -- To handle ProcessExitException without regard for -- reason, see catch. To handle multiple reasons of -- differing types, see catchesExit. catchExit :: (Show a, Serializable a) => Process b -> (ProcessId -> a -> Process b) -> Process b -- | Lift catches (almost). -- -- As ProcessExitException stores the exit reason as a -- typed, encoded message, a handler must accept an input of the expected -- type. In order to handle a list of potentially different handlers (and -- therefore input types), a handler passed to catchesExit must -- accept AbstractMessage and return Maybe (i.e., -- Just p if it handled the exit reason, otherwise -- Nothing). -- -- See maybeHandleMessage and AsbtractMessage for more -- details. catchesExit :: Process b -> [(ProcessId -> AbstractMessage -> (Process (Maybe b)))] -> Process b -- | Thrown by terminate data ProcessTerminationException ProcessTerminationException :: ProcessTerminationException -- | 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 ProcessRegistrationException ProcessRegistrationException :: !String -> ProcessRegistrationException -- | SpawnRef are used to return pids of spawned processes data SpawnRef -- | Our own process ID getSelfPid :: Process ProcessId -- | Get the node ID of our local node getSelfNode :: Process NodeId -- | Provide information about a running process data ProcessInfo ProcessInfo :: NodeId -> [String] -> Maybe Int -> [(ProcessId, MonitorRef)] -> [ProcessId] -> ProcessInfo infoNode :: ProcessInfo -> NodeId infoRegisteredNames :: ProcessInfo -> [String] infoMessageQueueLength :: ProcessInfo -> Maybe Int infoMonitors :: ProcessInfo -> [(ProcessId, MonitorRef)] infoLinks :: ProcessInfo -> [ProcessId] -- | Get information about the specified process getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo) -- | 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. link :: ProcessId -> Process () -- | Link to a node (asynchronous) linkNode :: NodeId -> Process () -- | Link to a channel (asynchronous) linkPort :: SendPort a -> 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. unlink :: ProcessId -> Process () -- | Remove a node link -- -- This has the same synchronous/asynchronous nature as unlink. unlinkNode :: NodeId -> Process () -- | Remove a channel (send port) link -- -- This has the same synchronous/asynchronous nature as unlink. unlinkPort :: SendPort a -> Process () -- | 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. monitor :: ProcessId -> Process MonitorRef -- | Monitor a node (asynchronous) monitorNode :: NodeId -> Process MonitorRef -- | Monitor a typed channel (asynchronous) monitorPort :: Serializable a => SendPort a -> Process MonitorRef -- | Remove a monitor -- -- This has the same synchronous/asynchronous nature as unlink. unmonitor :: MonitorRef -> Process () -- | Establishes temporary monitoring of another process. -- -- withMonitor pid code sets up monitoring of pid for -- the duration of code. Note: although monitoring is no longer -- active when withMonitor returns, there might still be -- unreceived monitor messages in the queue. withMonitor :: ProcessId -> Process a -> Process a -- | MonitorRef is opaque for regular Cloud Haskell processes data MonitorRef -- | Exceptions thrown when a linked process dies data ProcessLinkException ProcessLinkException :: !ProcessId -> !DiedReason -> ProcessLinkException -- | Exception thrown when a linked node dies data NodeLinkException NodeLinkException :: !NodeId -> !DiedReason -> NodeLinkException -- | Exception thrown when a linked channel (port) dies data PortLinkException PortLinkException :: !SendPortId -> !DiedReason -> PortLinkException -- | Message sent by process monitors data ProcessMonitorNotification ProcessMonitorNotification :: !MonitorRef -> !ProcessId -> !DiedReason -> ProcessMonitorNotification -- | Message sent by node monitors data NodeMonitorNotification NodeMonitorNotification :: !MonitorRef -> !NodeId -> !DiedReason -> NodeMonitorNotification -- | Message sent by channel (port) monitors data PortMonitorNotification PortMonitorNotification :: !MonitorRef -> !SendPortId -> !DiedReason -> PortMonitorNotification -- | Why did a process die? data DiedReason -- | Normal termination DiedNormal :: DiedReason -- | The process exited with an exception (provided as String -- because Exception does not implement Binary) DiedException :: !String -> DiedReason -- | We got disconnected from the process node DiedDisconnect :: DiedReason -- | The process node died DiedNodeDown :: DiedReason -- | Invalid (processnodechannel) identifier DiedUnknownId :: DiedReason -- | A closure is a static value and an encoded environment data Closure a :: * -> * closure :: Static (ByteString -> a) -> ByteString -> Closure a -- | A static value. Static is opaque; see staticLabel and -- staticApply. data Static a :: * -> * -- | Resolve a static value unStatic :: Typeable a => Static a -> Process a -- | Resolve a closure unClosure :: Typeable a => Closure a -> Process a -- | Runtime dictionary for unstatic lookups data RemoteTable :: * -- | 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. say :: String -> Process () -- | 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. register :: String -> ProcessId -> Process () -- | Like register, but will replace an existing registration. The -- name must already be registered. reregister :: String -> ProcessId -> Process () -- | 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. unregister :: String -> Process () -- | Query the local process registry whereis :: String -> Process (Maybe ProcessId) -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process () -- | 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 registerRemoteAsync :: NodeId -> String -> ProcessId -> Process () reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process () -- | Remove a process from a remote registry (asynchronous). -- -- Reply wil come in the form of a RegisterReply message -- -- See comments in whereisRemoteAsync unregisterRemoteAsync :: 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). whereisRemoteAsync :: NodeId -> String -> Process () -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () -- | (Asynchronous) reply from whereis data WhereIsReply WhereIsReply :: String -> (Maybe ProcessId) -> WhereIsReply -- | (Asynchronous) reply from register and unregister data RegisterReply RegisterReply :: String -> Bool -> RegisterReply -- | Lift catch catch :: Exception e => Process a -> (e -> Process a) -> Process a -- | You need this when using catches data Handler a Handler :: (e -> Process a) -> Handler a -- | Lift catches catches :: Process a -> [Handler a] -> Process a -- | Lift try try :: Exception e => Process a -> Process (Either e a) -- | Lift mask mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b -- | Lift onException onException :: Process a -> Process b -> Process a -- | Lift bracket bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c -- | Lift bracket_ bracket_ :: Process a -> Process b -> Process c -> Process c -- | Lift finally finally :: Process a -> Process b -> Process a -- | Asynchronous version of spawn -- -- (spawn is defined in terms of spawnAsync and -- expect) spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef -- | Spawn a child process, have the child link to the parent and the -- parent monitor the child spawnSupervised :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef) -- | 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) spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId -- | Like spawnLink, but monitor the spawned process spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef) -- | Spawn a new process, supplying it with a new ReceivePort and -- return the corresponding SendPort. spawnChannel :: Typeable a => Static (SerializableDict a) -> NodeId -> Closure (ReceivePort a -> Process ()) -> Process (SendPort a) -- | (Asynchronius) reply from spawn data DidSpawn DidSpawn :: SpawnRef -> ProcessId -> DidSpawn -- | Spawn a process on the local node spawnLocal :: Process () -> Process ProcessId -- | Create a new typed channel, spawn a process on the local node, passing -- it the receive port, and return the send port spawnChannelLocal :: Serializable a => (ReceivePort a -> Process ()) -> Process (SendPort a) -- | 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. reconnect :: ProcessId -> Process () -- | Reconnect to a sendport. See reconnect for more information. reconnectPort :: SendPort a -> Process () -- | Template Haskell support module Control.Distributed.Process.Internal.Closure.TH -- | Create the closure, decoder, and metadata definitions for the given -- list of functions remotable :: [Name] -> Q [Dec] -- | Like remotable, but parameterized by the declaration of a -- function instead of the function name. So where for remotable -- you'd do -- -- 
--   f :: T1 -> T2
--   f = ...
--   
--   remotable ['f]
--
-- -- with remotableDecl you would instead do -- -- 
--   remotableDecl [
--      [d| f :: T1 -> T2 ;
--          f = ...
--        |]
--    ]
--
-- -- remotableDecl creates the function specified as well as the -- various dictionaries and static versions that remotable also -- creates. remotableDecl is sometimes necessary when you want to -- refer to, say, $(mkClosure 'f) within the definition of -- f itself. -- -- NOTE: remotableDecl creates __remoteTableDecl instead -- of __remoteTable so that you can use both remotable -- and remotableDecl within the same module. remotableDecl :: [Q [Dec]] -> Q [Dec] -- | Construct a static value. -- -- If f : forall a1 .. an. T then $(mkStatic 'f) :: forall -- a1 .. an. Static T. Be sure to pass f to -- remotable. mkStatic :: Name -> Q Exp -- | If f : T1 -> T2 is a monomorphic function then -- $(functionSDict 'f) :: Static (SerializableDict T1). -- -- Be sure to pass f to remotable. functionSDict :: Name -> Q Exp -- | If f : T1 -> Process T2 is a monomorphic function then -- $(functionTDict 'f) :: Static (SerializableDict T2). -- -- Be sure to pass f to remotable. functionTDict :: Name -> Q Exp mkClosure :: Name -> Q Exp -- | Make a Closure from a static function. This is useful for -- making a closure for a top-level Process () function, because -- using mkClosure would require adding a dummy () -- argument. mkStaticClosure :: Name -> Q Exp -- | Towards Haskell in the Cloud (Epstein et al., Haskell Symposium -- 2011) proposes a new type construct called static that -- characterizes values that are known statically. Cloud Haskell uses the -- Static implementation from Control.Distributed.Static. -- That module comes with its own extensive documentation, which you -- should read if you want to know the details. Here we explain the -- Template Haskell support only. -- -- -- -- Given a top-level (possibly polymorphic, but unqualified) definition -- -- 
--   f :: forall a1 .. an. T
--   f = ...
--
-- -- you can use a Template Haskell splice to create a static version of -- f: -- -- 
--   $(mkStatic 'f) :: forall a1 .. an. (Typeable a1, .., Typeable an) => Static T
--
-- -- Every module that you write that contains calls to mkStatic -- needs to have a call to remotable: -- -- 
--   remotable [ 'f, 'g, ... ]
--
-- -- where you must pass every function (or other value) that you pass as -- an argument to mkStatic. The call to remotable will -- create a definition -- -- 
--   __remoteTable :: RemoteTable -> RemoteTable
--
-- -- which can be used to construct the RemoteTable used to -- initialize Cloud Haskell. You should have (at most) one call to -- remotable per module, and compose all created functions when -- initializing Cloud Haskell: -- -- 
--   let rtable :: RemoteTable
--       rtable = M1.__remoteTable
--              . M2.__remoteTable
--              . ...
--              . Mn.__remoteTable
--              $ initRemoteTable
--
-- -- NOTE: If you get a type error from ghc along these lines -- -- 
--   The exact Name `a_a30k' is not in scope
--        Probable cause: you used a unique name (NameU) in Template Haskell but did not bind it
--
-- -- then you need to enable the ScopedTypeVariables language -- extension. -- -- -- -- Some Cloud Haskell primitives require static serialization -- dictionaries (**): -- -- 
--   call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a
--
-- -- Given some serializable type T you can define -- -- 
--   sdictT :: SerializableDict T
--   sdictT = SerializableDict
--
-- -- and then have -- -- 
--   $(mkStatic 'sdictT) :: Static (SerializableDict T)
--
-- -- However, since these dictionaries are so frequently required Cloud -- Haskell provides special support for them. When you call -- remotable on a monomorphic function f :: T1 -> -- T2 -- -- 
--   remotable ['f]
--
-- -- then a serialization dictionary is automatically created for you, -- which you can access with -- -- 
--   $(functionSDict 'f) :: Static (SerializableDict T1)
--
-- -- In addition, if f :: T1 -> Process T2, then a second -- dictionary is created -- -- 
--   $(functionTDict 'f) :: Static (SerializableDict T2)
--
-- -- -- -- Suppose you have a process -- -- 
--   isPrime :: Integer -> Process Bool
--
-- -- Then -- -- 
--   $(mkClosure 'isPrime) :: Integer -> Closure (Process Bool)
--
-- -- which you can then call, for example, to have a remote node -- check if a number is prime. -- -- In general, if you have a monomorphic function -- -- 
--   f :: T1 -> T2
--
-- -- then -- -- 
--   $(mkClosure 'f) :: T1 -> Closure T2
--
-- -- provided that T1 is serializable (*) (remember to pass -- f to remotable). -- -- (You can also create closures manually--see the documentation of -- Control.Distributed.Static for examples.) -- -- -- -- Here is a small self-contained example that uses closures and -- serialization dictionaries. It makes use of the -- Control.Distributed.Process.SimpleLocalnet Cloud Haskell backend. -- -- 
--   {-# LANGUAGE TemplateHaskell #-}
--   import System.Environment (getArgs)
--   import Control.Distributed.Process
--   import Control.Distributed.Process.Closure
--   import Control.Distributed.Process.Backend.SimpleLocalnet
--   import Control.Distributed.Process.Node (initRemoteTable)
--   
--   isPrime :: Integer -> Process Bool
--   isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]
--     where
--       sieve :: [Integer] -> [Integer]
--       sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]
--   
--   remotable ['isPrime]
--   
--   master :: [NodeId] -> Process ()
--   master [] = liftIO $ putStrLn "no slaves"
--   master (slave:_) = do
--     isPrime79 <- call $(functionTDict 'isPrime) slave ($(mkClosure 'isPrime) (79 :: Integer))
--     liftIO $ print isPrime79
--   
--   main :: IO ()
--   main = do
--     args <- getArgs
--     case args of
--       ["master", host, port] -> do
--         backend <- initializeBackend host port rtable
--         startMaster backend master
--       ["slave", host, port] -> do
--         backend <- initializeBackend host port rtable
--         startSlave backend
--     where
--       rtable :: RemoteTable
--       rtable = __remoteTable initRemoteTable
--
-- -- -- -- (*) If T1 is not serializable you will get a type error in -- the generated code. Unfortunately, the Template Haskell infrastructure -- cannot check a priori if T1 is serializable or not due to a -- bug in the Template Haskell libraries -- (http://hackage.haskell.org/trac/ghc/ticket/7066) -- -- (**) Even though call is passed an explicit serialization -- dictionary, we still need the Serializable constraint because -- Static is not the true static. If it was, we could -- unstatic the dictionary and pattern match on it to bring the -- Typeable instance into scope, but unless proper -- static support is added to ghc we need both the type class -- argument and the explicit dictionary. module Control.Distributed.Process.Closure -- | Reification of Serializable (see -- Control.Distributed.Process.Closure) data SerializableDict a SerializableDict :: SerializableDict a -- | Static decoder, given a static serialization dictionary. -- -- See module documentation of Control.Distributed.Process.Closure -- for an example. staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a) -- | Serialization dictionary for '()' sdictUnit :: Static (SerializableDict ()) -- | Serialization dictionary for ProcessId sdictProcessId :: Static (SerializableDict ProcessId) -- | Serialization dictionary for SendPort sdictSendPort :: Typeable a => Static (SerializableDict a) -> Static (SerializableDict (SendPort a)) -- | CP a b is a process with input of type a and output -- of type b type CP a b = Closure (a -> Process b) -- | CP version of id idCP :: Typeable a => CP a a -- | CP version of (***) splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d) => CP a c -> CP b d -> CP (a, b) (c, d) -- | CP version of return returnCP :: Serializable a => Static (SerializableDict a) -> a -> Closure (Process a) -- | (Not quite the) CP version of (>>=) bindCP :: (Typeable a, Typeable b) => Closure (Process a) -> CP a b -> Closure (Process b) -- | CP version of (>>) seqCP :: (Typeable a, Typeable b) => Closure (Process a) -> Closure (Process b) -> Closure (Process b) -- | CP version of link cpLink :: ProcessId -> Closure (Process ()) -- | CP version of unlink cpUnlink :: ProcessId -> Closure (Process ()) -- | CP version of send cpSend :: Typeable a => Static (SerializableDict a) -> ProcessId -> CP a () -- | CP version of expect cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a) -- | CP version of newChan cpNewChan :: Typeable a => Static (SerializableDict a) -> Closure (Process (SendPort a, ReceivePort a)) -- | Create the closure, decoder, and metadata definitions for the given -- list of functions remotable :: [Name] -> Q [Dec] -- | Like remotable, but parameterized by the declaration of a -- function instead of the function name. So where for remotable -- you'd do -- -- 
--   f :: T1 -> T2
--   f = ...
--   
--   remotable ['f]
--
-- -- with remotableDecl you would instead do -- -- 
--   remotableDecl [
--      [d| f :: T1 -> T2 ;
--          f = ...
--        |]
--    ]
--
-- -- remotableDecl creates the function specified as well as the -- various dictionaries and static versions that remotable also -- creates. remotableDecl is sometimes necessary when you want to -- refer to, say, $(mkClosure 'f) within the definition of -- f itself. -- -- NOTE: remotableDecl creates __remoteTableDecl instead -- of __remoteTable so that you can use both remotable -- and remotableDecl within the same module. remotableDecl :: [Q [Dec]] -> Q [Dec] -- | Construct a static value. -- -- If f : forall a1 .. an. T then $(mkStatic 'f) :: forall -- a1 .. an. Static T. Be sure to pass f to -- remotable. mkStatic :: Name -> Q Exp mkClosure :: Name -> Q Exp -- | Make a Closure from a static function. This is useful for -- making a closure for a top-level Process () function, because -- using mkClosure would require adding a dummy () -- argument. mkStaticClosure :: Name -> Q Exp -- | If f : T1 -> T2 is a monomorphic function then -- $(functionSDict 'f) :: Static (SerializableDict T1). -- -- Be sure to pass f to remotable. functionSDict :: Name -> Q Exp -- | If f : T1 -> Process T2 is a monomorphic function then -- $(functionTDict 'f) :: Static (SerializableDict T2). -- -- Be sure to pass f to remotable. functionTDict :: Name -> Q Exp