@@ -1,50 +1,245 @@
1−module Control.Distributed.Process.Internal.NodeController
2− ( runNodeController
3− , NCMsg(..)
1+-- | Local nodes
2+module Control.Distributed.Process.Node
3+ ( newLocalNode
4+ , closeLocalNode
5+ , forkProcess
6+ , runProcess
7+ , initRemoteTable
8+ , localNodeId
49 ) where
510
11+import Prelude hiding (catch)
12+import System.IO (fixIO)
13+import qualified Data.ByteString.Lazy as BSL (fromChunks)
14+import Data.Binary (decode)
615import Data.Map (Map)
7−import qualified Data.Map as Map (empty, toList)
16+import qualified Data.Map as Map (empty, lookup, insert, delete, toList)
17+import qualified Data.List as List (delete)
818import Data.Set (Set)
9−import qualified Data.Set as Set (empty, insert, delete)
19+import qualified Data.Set as Set (empty, insert, delete, member)
1020import Data.Foldable (forM_)
1121import Data.Maybe (isJust)
12−import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
13−import qualified Data.Accessor.Container as DAC (mapDefault)
22+import Data.Typeable (Typeable)
23+import Data.Dynamic (fromDynamic)
1424import Control.Category ((>>>))
15−import Control.Monad (when, forever)
25+import Control.Applicative ((<$>))
26+import Control.Monad (void, when, forever)
1627import Control.Monad.IO.Class (MonadIO, liftIO)
1728import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
1829import qualified Control.Monad.Trans.Class as Trans (lift)
19−import Control.Concurrent.MVar (withMVar)
20−import Control.Concurrent.Chan (readChan)
21−import Control.Exception (Exception, throwTo)
22−import Control.Distributed.Process.Serializable (Serializable)
23−import Control.Distributed.Process.Internal.CQueue (enqueue)
24−import Control.Distributed.Process.Internal.Types
25− ( NodeId(..)
30+import Control.Exception (throwIO, SomeException, Exception, throwTo)
31+import qualified Control.Exception as Exception (catch)
32+import Control.Concurrent (forkIO)
33+import Control.Concurrent.MVar
34+ ( newMVar
35+ , withMVar
36+ , modifyMVar
37+ , modifyMVar_
38+ , newEmptyMVar
39+ , putMVar
40+ , takeMVar
41+ )
42+import Control.Concurrent.Chan (newChan, writeChan, readChan)
43+import Control.Distributed.Process.Internal.CQueue (enqueue, newCQueue)
44+import qualified Network.Transport as NT
45+ ( Transport
46+ , EndPoint
47+ , newEndPoint
48+ , receive
49+ , Event(..)
50+ , EventErrorCode(..)
51+ , TransportError(..)
52+ , address
53+ , closeEndPoint
54+ , ConnectionId
55+ )
56+import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
57+import qualified Data.Accessor.Container as DAC (mapDefault)
58+import System.Random (randomIO)
59+import Control.Distributed.Process.Internal.Types
60+ ( RemoteTable
61+ , NodeId(..)
2662 , LocalProcessId(..)
2763 , ProcessId(..)
2864 , LocalNode(..)
65+ , LocalNodeState(..)
2966 , LocalProcess(..)
67+ , LocalProcessState(..)
68+ , Process(..)
69+ , runLocalProcess
70+ , DiedReason(..)
71+ , NCMsg(..)
72+ , ProcessSignal(..)
73+ , payloadToMessage
74+ , payloadToId
75+ , localPidCounter
76+ , localPidUnique
77+ , localProcessWithId
78+ , initRemoteTable
3079 , MonitorRef(..)
3180 , MonitorNotification(..)
3281 , LinkException(..)
3382 , DidUnmonitor(..)
3483 , DidUnlink(..)
35− , DiedReason(..)
36− , NCMsg(..)
37− , ProcessSignal(..)
84+ , SpawnRef
85+ , DidSpawn(..)
86+ , Closure(..)
87+ , Static(..)
3888 , createMessage
39− , localProcessWithId
4089 , MessageT
4190 , runMessageT
4291 )
92+import Control.Distributed.Process.Serializable (Serializable)
4393import Control.Distributed.Process.Internal.MessageT
4494 ( sendBinary
95+ , sendMessage
4596 , getLocalNode
4697 )
4798
99+--------------------------------------------------------------------------------
100+-- Initialization --
101+--------------------------------------------------------------------------------
102+
103+-- | Initialize a new local node.
104+--
105+-- Note that proper Cloud Haskell initialization and configuration is still
106+-- to do.
107+newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
108+newLocalNode transport rtable = do
109+ mEndPoint <- NT.newEndPoint transport
110+ case mEndPoint of
111+ Left ex -> throwIO ex
112+ Right endPoint -> do
113+ unq <- randomIO
114+ state <- newMVar LocalNodeState
115+ { _localProcesses = Map.empty
116+ , _localPidCounter = 0
117+ , _localPidUnique = unq
118+ }
119+ ctrlChan <- newChan
120+ let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
121+ , localEndPoint = endPoint
122+ , localState = state
123+ , localCtrlChan = ctrlChan
124+ , remoteTable = rtable
125+ }
126+ void . forkIO $ runNodeController node
127+ void . forkIO $ handleIncomingMessages node
128+ return node
129+
130+-- | Force-close a local node
131+--
132+-- TODO: for now we just close the associated endpoint
133+closeLocalNode :: LocalNode -> IO ()
134+closeLocalNode node =
135+ NT.closeEndPoint (localEndPoint node)
136+
137+-- | Run a process on a local node and wait for it to finish
138+runProcess :: LocalNode -> Process () -> IO ()
139+runProcess node proc = do
140+ done <- newEmptyMVar
141+ void $ forkProcess node (proc >> liftIO (putMVar done ()))
142+ takeMVar done
143+
144+-- | Spawn a new process on a local node
145+forkProcess :: LocalNode -> Process () -> IO ProcessId
146+forkProcess node proc = modifyMVar (localState node) $ \st -> do
147+ let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
148+ , lpidUnique = st ^. localPidUnique
149+ }
150+ let pid = ProcessId { processNodeId = localNodeId node
151+ , processLocalId = lpid
152+ }
153+ pst <- newMVar LocalProcessState { _monitorCounter = 0
154+ , _spawnCounter = 0
155+ }
156+ queue <- newCQueue
157+ (_, lproc) <- fixIO $ \ ~(tid, _) -> do
158+ let lproc = LocalProcess { processQueue = queue
159+ , processId = pid
160+ , processState = pst
161+ , processThread = tid
162+ }
163+ tid' <- forkIO $ do
164+ reason <- Exception.catch
165+ (runLocalProcess node proc lproc >> return DiedNormal)
166+ (return . DiedException . (show :: SomeException -> String))
167+ -- [Unified: Table 4, rules termination and exiting]
168+ modifyMVar_ (localState node) $
169+ return . (localProcessWithId lpid ^= Nothing)
170+ writeChan (localCtrlChan node) NCMsg
171+ { ctrlMsgSender = Left pid
172+ , ctrlMsgSignal = Died (Left pid) reason
173+ }
174+ return (tid', lproc)
175+
176+ -- TODO: if the counter overflows we should pick a new unique
177+ return ( (localProcessWithId lpid ^= Just lproc)
178+ . (localPidCounter ^: (+ 1))
179+ $ st
180+ , pid
181+ )
182+
183+handleIncomingMessages :: LocalNode -> IO ()
184+handleIncomingMessages node = go [] Map.empty Set.empty
185+ where
186+ go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know
187+ -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes
188+ -> Set NT.ConnectionId -- ^ Connections to our controller
189+ -> IO ()
190+ go uninitConns procConns ctrlConns = do
191+ event <- NT.receive endpoint
192+ case event of
193+ NT.ConnectionOpened cid _rel _theirAddr ->
194+ go (cid : uninitConns) procConns ctrlConns
195+ NT.Received cid payload ->
196+ case Map.lookup cid procConns of
197+ Just proc -> do
198+ let msg = payloadToMessage payload
199+ enqueue (processQueue proc) msg
200+ go uninitConns procConns ctrlConns
201+ Nothing | cid `Set.member` ctrlConns -> do
202+ writeChan ctrlChan (decode . BSL.fromChunks $ payload)
203+ go uninitConns procConns ctrlConns
204+ Nothing | cid `elem` uninitConns ->
205+ case payloadToId payload of
206+ Just lpid -> do
207+ mProc <- withMVar state $ return . (^. localProcessWithId lpid)
208+ case mProc of
209+ Just proc ->
210+ go (List.delete cid uninitConns)
211+ (Map.insert cid proc procConns)
212+ ctrlConns
213+ Nothing ->
214+ fail "handleIncomingMessages: TODO 1"
215+ Nothing ->
216+ go (List.delete cid uninitConns)
217+ procConns
218+ (Set.insert cid ctrlConns)
219+ _ ->
220+ fail "handleIncomingMessages: TODO 2"
221+ NT.ConnectionClosed cid ->
222+ go (List.delete cid uninitConns)
223+ (Map.delete cid procConns)
224+ (Set.delete cid ctrlConns)
225+ NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do
226+ -- [Unified table 9, rule node_disconnect]
227+ let nid = Right $ NodeId theirAddr
228+ writeChan ctrlChan NCMsg
229+ { ctrlMsgSender = nid
230+ , ctrlMsgSignal = Died nid DiedDisconnect
231+ }
232+ NT.ErrorEvent _ ->
233+ fail "handleIncomingMessages: TODO 4"
234+ NT.EndPointClosed ->
235+ return ()
236+ NT.ReceivedMulticast _ _ ->
237+ fail "Unexpected multicast"
238+
239+ state = localState node
240+ endpoint = localEndPoint node
241+ ctrlChan = localCtrlChan node
242+
48243--------------------------------------------------------------------------------
49244-- Top-level access to the node controller --
50245--------------------------------------------------------------------------------
@@ -104,8 +299,8 @@ nodeController = do
104299 ncEffectNodeDied nid reason
105300 NCMsg _from (Died (Left pid) reason) ->
106301 ncEffectProcessDied pid reason
107− NCMsg _from (Spawn _ _) ->
108− fail $ show (localNodeId node) ++ ": spawn not implemented"
302+ NCMsg (Left from) (Spawn proc ref) ->
303+ ncEffectSpawn from proc ref
109304 unexpected ->
110305 error $ "Unexpected message " ++ show unexpected
111306
@@ -187,6 +382,16 @@ ncEffectProcessDied pid reason = do
187382 modify $ (linksForProcess pid ^= Set.empty)
188383 . (monitorsForProcess pid ^= Map.empty)
189384
385+-- [Unified: Table 13]
386+ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
387+ncEffectSpawn pid cProc ref = do
388+ mProc <- unClosure cProc
389+ -- TODO: what should we do when unClosure returns Nothing?
390+ forM_ mProc $ \proc -> do
391+ node <- lift getLocalNode
392+ pid' <- liftIO $ forkProcess node proc
393+ lift $ sendMessage (Left pid) (DidSpawn ref pid')
394+
190395--------------------------------------------------------------------------------
191396-- Auxiliary --
192397--------------------------------------------------------------------------------
@@ -225,6 +430,18 @@ destNid (Spawn _ _) = Nothing
225430isLocal :: LocalNode -> ProcessId -> Bool
226431isLocal nid pid = processNodeId pid == localNodeId nid
227432
433+-- | Lookup a local closure
434+--
435+-- TODO: Duplication with onClosure in Process
436+unClosure :: Typeable a => Closure a -> NC (Maybe a)
437+unClosure (Closure (Static label) env) = do
438+ rtable <- remoteTable <$> lift getLocalNode
439+ case Map.lookup label rtable of
440+ Nothing -> return Nothing
441+ Just dyn -> case fromDynamic dyn of
442+ Nothing -> return Nothing
443+ Just dec -> return (Just $ dec env)
444+
228445--------------------------------------------------------------------------------
229446-- Messages to local processes --
230447--------------------------------------------------------------------------------