@@ -56,6 +56,7 @@ module Control.Distributed.Process
5656 , expect
5757 , send
5858 , getSelfPid
59+ , monitor
5960 , monitorProcess
6061 , linkProcess
6162 -- * Monitoring
@@ -127,56 +128,58 @@ import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
127128import qualified Data.Accessor.Container as DAC (mapMaybe)
128129import System.Random (randomIO)
129130
131+-- | A local process ID consists of a seed which distinguishes processes from
132+-- different instances of the same local node and a counter
130133data LocalProcessId = LocalProcessId
131− { pidUnique :: Int32
132− , pidCounter :: Int32
134+ { lpidUnique :: Int32
135+ , lpidCounter :: Int32
133136 }
134137 deriving (Eq, Ord, Typeable, Show)
135138
136−instance Binary LocalProcessId where
137− put lpid = put (pidUnique lpid) >> put (pidCounter lpid)
138− get = liftM2 LocalProcessId get get
139−
139+-- | A process ID combines a local process with with an endpoint address
140+-- (in other words, we identify nodes and endpoints)
140141data ProcessId = ProcessId
141142 { processAddress :: NT.EndPointAddress
142143 , processLocalId :: LocalProcessId
143144 }
144145 deriving (Eq, Ord, Typeable, Show)
145146
146−instance Binary ProcessId where
147− put pid = put (processAddress pid) >> put (processLocalId pid)
148− get = liftM2 ProcessId get get
149−
147+-- | Messages consist of their typeRep fingerprint and their encoding
150148data Message = Message
151149 { messageFingerprint :: Fingerprint
152150 , messageEncoding :: BSL.ByteString
153151 }
154152
153+-- | Local nodes
155154data LocalNode = LocalNode
156155 { localEndPoint :: NT.EndPoint
157156 , localState :: MVar LocalNodeState
158157 }
159158
159+-- | Local node state
160160data LocalNodeState = LocalNodeState
161161 { _localConnections :: Map ProcessId NT.Connection
162− , _localProcesses :: Map LocalProcessId ProcessState
162+ , _localProcesses :: Map LocalProcessId LocalProcess
163163 , _localPidCounter :: Int32
164164 , _localPidUnique :: Int32
165165 }
166166
167−data ProcessState = ProcessState
167+-- | Processes running on our local node
168+data LocalProcess = LocalProcess
168169 { processQueue :: CQueue Message
169170 , processNode :: LocalNode
170171 , processId :: ProcessId
171172 }
172173
173−newtype Process a = Process { unProcess :: ReaderT ProcessState IO a }
174− deriving (Functor, Monad, MonadIO, MonadReader ProcessState)
174+-- The Cloud Haskell 'Process' type
175+newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
176+ deriving (Functor, Monad, MonadIO, MonadReader LocalProcess)
175177
176178--------------------------------------------------------------------------------
177179-- Basic Cloud Haskell API --
178180--------------------------------------------------------------------------------
179181
182+-- | Wait for a message of a specific type
180183expect :: forall a. Serializable a => Process a
181184expect = do
182185 queue <- processQueue <$> ask
@@ -184,6 +187,7 @@ expect = do
184187 msg <- liftIO $ dequeueMatching queue ((== fp) . messageFingerprint)
185188 return (decode . messageEncoding $ msg)
186189
190+-- | Send a message
187191send :: Serializable a => ProcessId -> a -> Process ()
188192send pid msg = do
189193 -- This requires a lookup on every send. If we want to avoid that we need to
@@ -195,12 +199,19 @@ send pid msg = do
195199 : BSL.toChunks (encode msg)
196200 )
197201
202+-- | Our own process ID
198203getSelfPid :: Process ProcessId
199204getSelfPid = processId <$> ask
200205
206+-- | Variation on 'monitorProcess' where the monitor is the current process
207+monitor :: ProcessId -> MonitorAction -> Process ()
208+monitor _them = undefined
209+
210+-- | Have one process monitor another
201211monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()
202212monitorProcess = undefined
203213
214+-- | Link failure in two processes
204215linkProcess :: ProcessId -> Process ()
205216linkProcess = undefined
206217
@@ -325,13 +336,13 @@ forkProcess :: LocalNode -> Process () -> IO ProcessId
325336forkProcess node proc = do
326337 queue <- newCQueue
327338 state <- modifyMVar (localState node) $ \st -> do
328− let lpid = LocalProcessId { pidCounter = st ^. localPidCounter
329− , pidUnique = st ^. localPidUnique
339+ let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
340+ , lpidUnique = st ^. localPidUnique
330341 }
331342 let pid = ProcessId { processAddress = NT.address (localEndPoint node)
332343 , processLocalId = lpid
333344 }
334− let state = ProcessState { processQueue = queue
345+ let state = LocalProcess { processQueue = queue
335346 , processNode = node
336347 , processId = pid
337348 }
@@ -444,16 +455,28 @@ payloadToMessage payload = Message fp msg
444455 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
445456
446457lpidToPayload :: LocalProcessId -> [BSS.ByteString]
447−lpidToPayload lpid = [ NTI.encodeInt32 (pidCounter lpid)
448− , NTI.encodeInt32 (pidUnique lpid)
458+lpidToPayload lpid = [ NTI.encodeInt32 (lpidCounter lpid)
459+ , NTI.encodeInt32 (lpidUnique lpid)
449460 ]
450461
451462payloadToLpid :: [BSS.ByteString] -> LocalProcessId
452463payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
453− in LocalProcessId { pidCounter = NTI.decodeInt32 bs1
454− , pidUnique = NTI.decodeInt32 bs2
464+ in LocalProcessId { lpidCounter = NTI.decodeInt32 bs1
465+ , lpidUnique = NTI.decodeInt32 bs2
455466 }
456467
468+--------------------------------------------------------------------------------
469+-- Binary instances --
470+--------------------------------------------------------------------------------
471+
472+instance Binary LocalProcessId where
473+ put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)
474+ get = liftM2 LocalProcessId get get
475+
476+instance Binary ProcessId where
477+ put pid = put (processAddress pid) >> put (processLocalId pid)
478+ get = liftM2 ProcessId get get
479+
457480--------------------------------------------------------------------------------
458481-- Accessors --
459482--------------------------------------------------------------------------------
@@ -461,7 +484,7 @@ payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
461484localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)
462485localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })
463486
464−localProcesses :: Accessor LocalNodeState (Map LocalProcessId ProcessState)
487+localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)
465488localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
466489
467490localPidCounter :: Accessor LocalNodeState Int32
@@ -473,5 +496,5 @@ localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq
473496localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)
474497localConnectionTo pid = localConnections >>> DAC.mapMaybe pid
475498
476−localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe ProcessState)
499+localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
477500localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid