@@ -6,22 +6,27 @@ module Control.Distributed.Process.Internal
66 NodeId(..)
77 , LocalProcessId(..)
88 , ProcessId(..)
9+ , Identifier
910 -- * Local nodes and processes
1011 , LocalNode(..)
1112 , LocalNodeState(..)
1213 , LocalProcess(..)
1314 , LocalProcessState(..)
15+ , Process(..)
16+ -- * Closures
17+ , Static(..)
18+ , Closure(..)
1419 -- * Messages
1520 , Message(..)
16− -- * Monitoring and linking
21+ -- * Node controller user-visible data types
1722 , MonitorRef(..)
1823 , MonitorNotification(..)
1924 , LinkException(..)
2025 , DiedReason(..)
2126 , DidUnmonitor(..)
2227 , DidUnlink(..)
23− -- * Node controller data types
24− , Identifier
28+ , SpawnRef(..)
29+ -- * Node controller internal data types
2530 , NCMsg(..)
2631 , ProcessSignal(..)
2732 -- * Serialization/deserialization
@@ -34,9 +39,10 @@ module Control.Distributed.Process.Internal
3439 , localProcesses
3540 , localPidCounter
3641 , localPidUnique
37− , localMonitorCounter
3842 , localProcessWithId
3943 , connections
44+ , monitorCounter
45+ , spawnCounter
4046 , connectionTo
4147 ) where
4248
@@ -51,6 +57,7 @@ import Data.Int (Int32)
5157import Data.Typeable (Typeable)
5258import Data.Dynamic (Dynamic)
5359import Data.Binary (Binary, encode, put, get, putWord8, getWord8)
60+import Data.ByteString.Lazy (ByteString)
5461import Data.Accessor (Accessor, accessor)
5562import qualified Data.Accessor.Container as DAC (mapMaybe)
5663import qualified Data.ByteString.Lazy as BSL (ByteString, toChunks, fromChunks, splitAt)
@@ -58,6 +65,8 @@ import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
5865import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
5966import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
6067import Control.Applicative ((<$>), (<*>))
68+import Control.Monad.Reader (MonadReader(..), ReaderT)
69+import Control.Monad.IO.Class (MonadIO)
6170import Control.Distributed.Process.Serializable ( Fingerprint
6271 , Serializable
6372 , encodeFingerprint
@@ -67,6 +76,21 @@ import Control.Distributed.Process.Serializable ( Fingerprint
6776 )
6877import Control.Distributed.Process.Internal.CQueue (CQueue)
6978
79+--------------------------------------------------------------------------------
80+-- Global state --
81+--------------------------------------------------------------------------------
82+
83+-- | Used to fake 'static' (see paper)
84+type RemoteTable = Map String Dynamic
85+
86+-- | Initial (empty) remote-call meta data
87+initRemoteTable :: RemoteTable
88+initRemoteTable = Map.empty
89+
90+--------------------------------------------------------------------------------
91+-- Node and process identifiers --
92+--------------------------------------------------------------------------------
93+
7094-- | Node identifier
7195newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }
7296 deriving (Eq, Ord, Binary)
@@ -95,12 +119,12 @@ data ProcessId = ProcessId
95119instance Show ProcessId where
96120 show pid = show (processNodeId pid) ++ ":" ++ show (processLocalId pid)
97121
98−-- | Used to fake 'static' (see paper)
99−type RemoteTable = Map String Dynamic
122+-- | Node or process identifier
123+type Identifier = Either ProcessId NodeId
100124
101−-- | Initial (empty) remote-call meta data
102−initRemoteTable :: RemoteTable
103−initRemoteTable = Map.empty
125+--------------------------------------------------------------------------------
126+-- Local nodes and processes --
127+--------------------------------------------------------------------------------
104128
105129-- | Local nodes
106130data LocalNode = LocalNode
@@ -117,7 +141,6 @@ data LocalNodeState = LocalNodeState
117141 { _localProcesses :: Map LocalProcessId LocalProcess
118142 , _localPidCounter :: Int32
119143 , _localPidUnique :: Int32
120− , _localMonitorCounter :: Int32
121144 }
122145
123146-- | Processes running on our local node
@@ -132,17 +155,46 @@ data LocalProcess = LocalProcess
132155-- | Local process state
133156data LocalProcessState = LocalProcessState
134157 { _connections :: Map ProcessId NT.Connection
158+ , _monitorCounter :: Int32
159+ , _spawnCounter :: Int32
135160 }
136161
162+-- | The Cloud Haskell 'Process' type
163+newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
164+ deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable)
165+
166+--------------------------------------------------------------------------------
167+-- Closures --
168+--------------------------------------------------------------------------------
169+
170+-- | A static value is one that is bound at top-level.
171+-- We represent it simply by a string
172+newtype Static a = Static String
173+ deriving (Typeable, Show)
174+
175+-- | A closure is a static value and an encoded environment
176+data Closure a = Closure (Static (ByteString -> a)) ByteString
177+ deriving (Typeable, Show)
178+
179+--------------------------------------------------------------------------------
180+-- Messages --
181+--------------------------------------------------------------------------------
182+
137183-- | Messages consist of their typeRep fingerprint and their encoding
138184data Message = Message
139185 { messageFingerprint :: Fingerprint
140186 , messageEncoding :: BSL.ByteString
141187 }
142188
189+--------------------------------------------------------------------------------
190+-- Node controller user-visible data types --
191+--------------------------------------------------------------------------------
192+
143193-- | MonitorRef is opaque for regular Cloud Haskell processes
144194data MonitorRef = MonitorRef
145− { monitorRefPid :: ProcessId
195+ { -- | PID of the process to be monitored (for routing purposes)
196+ monitorRefPid :: ProcessId
197+ -- | Unique to distinguish multiple monitor requests by the same process
146198 , monitorRefCounter :: Int32
147199 }
148200 deriving (Eq, Ord, Show)
@@ -157,14 +209,6 @@ data LinkException = LinkException ProcessId DiedReason
157209
158210instance Exception LinkException
159211
160−-- | (Asynchronous) reply from unmonitor
161−newtype DidUnmonitor = DidUnmonitor MonitorRef
162− deriving (Typeable, Binary)
163−
164−-- | (Asynchronous) reply from unlink
165−newtype DidUnlink = DidUnlink ProcessId
166− deriving (Typeable, Binary)
167−
168212-- | Why did a process die?
169213data DiedReason =
170214 -- | Normal termination
@@ -180,8 +224,21 @@ data DiedReason =
180224 | DiedNoProc
181225 deriving (Show, Eq)
182226
183−-- | Node or process identifier
184−type Identifier = Either ProcessId NodeId
227+-- | (Asynchronous) reply from unmonitor
228+newtype DidUnmonitor = DidUnmonitor MonitorRef
229+ deriving (Typeable, Binary)
230+
231+-- | (Asynchronous) reply from unlink
232+newtype DidUnlink = DidUnlink ProcessId
233+ deriving (Typeable, Binary)
234+
235+-- | 'SpawnRef' are used to return pids of spawned processes
236+newtype SpawnRef = SpawnRef Int32
237+ deriving (Show, Binary, Typeable)
238+
239+--------------------------------------------------------------------------------
240+-- Node controller internal data types --
241+--------------------------------------------------------------------------------
185242
186243-- | Messages to the node controller
187244data NCMsg = NCMsg
@@ -197,6 +254,7 @@ data ProcessSignal =
197254 | Monitor ProcessId MonitorRef
198255 | Unmonitor MonitorRef
199256 | Died Identifier DiedReason
257+ | Spawn (Closure (Process ())) SpawnRef
200258 deriving Show
201259
202260--------------------------------------------------------------------------------
@@ -241,6 +299,10 @@ payloadToId bss = let (bs1, bss') = BSS.splitAt 4 . BSS.concat $ bss
241299 }
242300 _ -> fail "payloadToId"
243301
302+--------------------------------------------------------------------------------
303+-- Binary instances --
304+--------------------------------------------------------------------------------
305+
244306instance Binary LocalProcessId where
245307 put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)
246308 get = LocalProcessId <$> get <*> get
@@ -267,6 +329,7 @@ instance Binary ProcessSignal where
267329 put (Monitor pid ref) = putWord8 2 >> put pid >> put ref
268330 put (Unmonitor ref) = putWord8 3 >> put ref
269331 put (Died who reason) = putWord8 4 >> put who >> put reason
332+ put (Spawn proc ref) = putWord8 5 >> put proc >> put ref
270333 get = do
271334 header <- getWord8
272335 case header of
@@ -275,6 +338,7 @@ instance Binary ProcessSignal where
275338 2 -> Monitor <$> get <*> get
276339 3 -> Unmonitor <$> get
277340 4 -> Died <$> get <*> get
341+ 5 -> Spawn <$> get <*> get
278342 _ -> fail "ProcessSignal.get: invalid"
279343
280344instance Binary DiedReason where
@@ -293,6 +357,10 @@ instance Binary DiedReason where
293357 4 -> return DiedNoProc
294358 _ -> fail "DiedReason.get: invalid"
295359
360+instance Binary (Closure a) where
361+ put (Closure (Static label) env) = put label >> put env
362+ get = Closure <$> (Static <$> get) <*> get
363+
296364--------------------------------------------------------------------------------
297365-- Accessors --
298366--------------------------------------------------------------------------------
@@ -306,14 +374,19 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =
306374localPidUnique :: Accessor LocalNodeState Int32
307375localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
308376
309−localMonitorCounter :: Accessor LocalNodeState Int32
310−localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })
311−
312377localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
313378localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
314379
315380connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)
316381connections = accessor _connections (\conns st -> st { _connections = conns })
317382
383+monitorCounter :: Accessor LocalProcessState Int32
384+monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt })
385+
386+spawnCounter :: Accessor LocalProcessState Int32
387+spawnCounter = accessor _spawnCounter (\cnt st -> st { _spawnCounter = cnt })
388+
318389connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
319390connectionTo pid = connections >>> DAC.mapMaybe pid
391+
392+