git.iko.soy/distributed-process
on master help sign in
First implementation of spawn
50bb4c1Edsko de Vries
distributed-process/src/Control/Distributed/Process/Internal/MessageT.hsnew file+116
compare with master
unifiedsplit
@@ -0,0 +1,116 @@
1+-- Add message sending capability to a monad
2+--
3+-- NOTE: Not thread-safe (you should not do concurrent sends within the same
4+-- monad).
5+module Control.Distributed.Process.Internal.MessageT
6+ ( MessageT
7+ , sendPayload
8+ , sendBinary
9+ , sendMessage
10+ , getLocalNode
11+ ) where
12+
13+import Data.Binary (Binary, encode)
14+import Data.ByteString (ByteString)
15+import qualified Data.ByteString.Lazy as BSL (toChunks)
16+import Data.Map (Map)
17+import Data.Accessor (Accessor, accessor, (^=), (^.))
18+import qualified Data.Accessor.Container as DAC (mapMaybe)
19+import Control.Category ((>>>))
20+import Control.Monad (unless, liftM)
21+import Control.Monad.State (gets, modify)
22+import Control.Monad.IO.Class (MonadIO, liftIO)
23+import Control.Concurrent.Chan (writeChan)
24+import Control.Distributed.Process.Internal.Types
25+ ( Identifier
26+ , NodeId(nodeAddress)
27+ , ProcessId(processNodeId, processLocalId)
28+ , idToPayload
29+ , LocalNode(localCtrlChan, localEndPoint)
30+ , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)
31+ , DiedReason(DiedDisconnect)
32+ , ProcessSignal(Died)
33+ , MessageT(..)
34+ , MessageState(..)
35+ , createMessage
36+ , messageToPayload
37+ )
38+import Control.Distributed.Process.Serializable (Serializable)
39+import qualified Network.Transport as NT
40+ ( EndPoint
41+ , Connection
42+ , connect
43+ , send
44+ , Reliability(ReliableOrdered)
45+ , defaultConnectHints
46+ )
47+
48+getLocalNode :: Monad m => MessageT m LocalNode
49+getLocalNode = gets messageLocalNode
50+
51+sendPayload :: MonadIO m => Identifier -> [ByteString] -> MessageT m ()
52+sendPayload them payload = do
53+ mConn <- connTo them
54+ didSend <- case mConn of
55+ Just conn -> do
56+ didSend <- liftIO $ NT.send conn payload
57+ case didSend of
58+ Left _ -> return False
59+ Right _ -> return True
60+ Nothing -> return False
61+ unless didSend $ do
62+ node <- getLocalNode
63+ liftIO . writeChan (localCtrlChan node) $ NCMsg
64+ { ctrlMsgSender = them
65+ , ctrlMsgSignal = Died them DiedDisconnect
66+ }
67+
68+sendBinary :: (MonadIO m, Binary a) => Identifier -> a -> MessageT m ()
69+sendBinary them = sendPayload them . BSL.toChunks . encode
70+
71+sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()
72+sendMessage them = sendPayload them . messageToPayload . createMessage
73+
74+setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
75+setupConnTo them = do
76+ endPoint <- localEndPoint `liftM` getLocalNode
77+ mConn <- liftIO $ NT.connect endPoint
78+ addr
79+ NT.ReliableOrdered
80+ NT.defaultConnectHints
81+ case mConn of
82+ Right conn -> do
83+ didSend <- liftIO $ NT.send conn firstMsg
84+ case didSend of
85+ Left _ ->
86+ return Nothing
87+ Right () -> do
88+ modify $ messageConnectionTo them ^= Just conn
89+ return $ Just conn
90+ Left _ ->
91+ return Nothing
92+ where
93+ (addr, firstMsg) = case them of
94+ Left pid -> ( nodeAddress (processNodeId pid)
95+ , idToPayload (Just $ processLocalId pid)
96+ )
97+ Right nid -> ( nodeAddress nid
98+ , idToPayload Nothing
99+ )
100+
101+connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
102+connTo them = do
103+ mConn <- gets (^. messageConnectionTo them)
104+ case mConn of
105+ Just conn -> return $ Just conn
106+ Nothing -> setupConnTo them
107+
108+--------------------------------------------------------------------------------
109+-- Accessors --
110+--------------------------------------------------------------------------------
111+
112+messageConnections :: Accessor MessageState (Map Identifier NT.Connection)
113+messageConnections = accessor _messageConnections (\conns st -> st { _messageConnections = conns })
114+
115+messageConnectionTo :: Identifier -> Accessor MessageState (Maybe NT.Connection)
116+messageConnectionTo them = messageConnections >>> DAC.mapMaybe them
distributed-process/src/Control/Distributed/Process/Internal/NodeController.hs → distributed-process/src/Control/Distributed/Process/Node.hsrenamed+238−21
@@ -1,50 +1,245 @@
1+-- | Local nodes
1−module Control.Distributed.Process.Internal.NodeController
2+module Control.Distributed.Process.Node
2− ( runNodeController
3+ ( newLocalNode
3− , NCMsg(..)
4+ , closeLocalNode
5+ , forkProcess
6+ , runProcess
7+ , initRemoteTable
8+ , localNodeId
4 ) where
9 ) where
5
10
11+import Prelude hiding (catch)
12+import System.IO (fixIO)
13+import qualified Data.ByteString.Lazy as BSL (fromChunks)
14+import Data.Binary (decode)
6import Data.Map (Map)
15import 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)
8import Data.Set (Set)
18import Data.Set (Set)
9−import qualified Data.Set as Set (empty, insert, delete)
19+import qualified Data.Set as Set (empty, insert, delete, member)
10import Data.Foldable (forM_)
20import Data.Foldable (forM_)
11import Data.Maybe (isJust)
21import Data.Maybe (isJust)
12−import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
22+import Data.Typeable (Typeable)
13−import qualified Data.Accessor.Container as DAC (mapDefault)
23+import Data.Dynamic (fromDynamic)
14import Control.Category ((>>>))
24import Control.Category ((>>>))
25+import Control.Applicative ((<$>))
15−import Control.Monad (when, forever)
26+import Control.Monad (void, when, forever)
16import Control.Monad.IO.Class (MonadIO, liftIO)
27import Control.Monad.IO.Class (MonadIO, liftIO)
17import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
28import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
18import qualified Control.Monad.Trans.Class as Trans (lift)
29import qualified Control.Monad.Trans.Class as Trans (lift)
30+import Control.Exception (throwIO, SomeException, Exception, throwTo)
31+import qualified Control.Exception as Exception (catch)
32+import Control.Concurrent (forkIO)
19−import Control.Concurrent.MVar (withMVar)
33+import Control.Concurrent.MVar
34+ ( newMVar
35+ , withMVar
36+ , modifyMVar
37+ , modifyMVar_
38+ , newEmptyMVar
39+ , putMVar
40+ , takeMVar
41+ )
20−import Control.Concurrent.Chan (readChan)
42+import Control.Concurrent.Chan (newChan, writeChan, readChan)
21−import Control.Exception (Exception, throwTo)
22−import Control.Distributed.Process.Serializable (Serializable)
23−import Control.Distributed.Process.Internal.CQueue (enqueue)
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)
24−import Control.Distributed.Process.Internal.Types
59+import Control.Distributed.Process.Internal.Types
60+ ( RemoteTable
25− ( NodeId(..)
61+ , NodeId(..)
26 , LocalProcessId(..)
62 , LocalProcessId(..)
27 , ProcessId(..)
63 , ProcessId(..)
28 , LocalNode(..)
64 , LocalNode(..)
65+ , LocalNodeState(..)
29 , LocalProcess(..)
66 , 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
30 , MonitorRef(..)
79 , MonitorRef(..)
31 , MonitorNotification(..)
80 , MonitorNotification(..)
32 , LinkException(..)
81 , LinkException(..)
33 , DidUnmonitor(..)
82 , DidUnmonitor(..)
34 , DidUnlink(..)
83 , DidUnlink(..)
35− , DiedReason(..)
84+ , SpawnRef
36− , NCMsg(..)
85+ , DidSpawn(..)
37− , ProcessSignal(..)
86+ , Closure(..)
87+ , Static(..)
38 , createMessage
88 , createMessage
39− , localProcessWithId
40 , MessageT
89 , MessageT
41 , runMessageT
90 , runMessageT
42 )
91 )
92+import Control.Distributed.Process.Serializable (Serializable)
43import Control.Distributed.Process.Internal.MessageT
93import Control.Distributed.Process.Internal.MessageT
44 ( sendBinary
94 ( sendBinary
95+ , sendMessage
45 , getLocalNode
96 , getLocalNode
46 )
97 )
47
98
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+
48--------------------------------------------------------------------------------
243--------------------------------------------------------------------------------
49-- Top-level access to the node controller --
244-- Top-level access to the node controller --
50--------------------------------------------------------------------------------
245--------------------------------------------------------------------------------
@@ -104,8 +299,8 @@ nodeController = do
104 ncEffectNodeDied nid reason
299 ncEffectNodeDied nid reason
105 NCMsg _from (Died (Left pid) reason) ->
300 NCMsg _from (Died (Left pid) reason) ->
106 ncEffectProcessDied pid reason
301 ncEffectProcessDied pid reason
107− NCMsg _from (Spawn _ _) ->
302+ NCMsg (Left from) (Spawn proc ref) ->
108− fail $ show (localNodeId node) ++ ": spawn not implemented"
303+ ncEffectSpawn from proc ref
109 unexpected ->
304 unexpected ->
110 error $ "Unexpected message " ++ show unexpected
305 error $ "Unexpected message " ++ show unexpected
111
306
@@ -187,6 +382,16 @@ ncEffectProcessDied pid reason = do
187 modify $ (linksForProcess pid ^= Set.empty)
382 modify $ (linksForProcess pid ^= Set.empty)
188 . (monitorsForProcess pid ^= Map.empty)
383 . (monitorsForProcess pid ^= Map.empty)
189
384
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+
190--------------------------------------------------------------------------------
395--------------------------------------------------------------------------------
191-- Auxiliary --
396-- Auxiliary --
192--------------------------------------------------------------------------------
397--------------------------------------------------------------------------------
@@ -225,6 +430,18 @@ destNid (Spawn _ _) = Nothing
225isLocal :: LocalNode -> ProcessId -> Bool
430isLocal :: LocalNode -> ProcessId -> Bool
226isLocal nid pid = processNodeId pid == localNodeId nid
431isLocal nid pid = processNodeId pid == localNodeId nid
227
432
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+
228--------------------------------------------------------------------------------
445--------------------------------------------------------------------------------
229-- Messages to local processes --
446-- Messages to local processes --
230--------------------------------------------------------------------------------
447--------------------------------------------------------------------------------
distributed-process/tests/TestClosure.hs−1
@@ -102,7 +102,6 @@ testSpawn transport rtable = do
102
102
103 forkIO $ do
103 forkIO $ do
104 node <- newLocalNode transport rtable
104 node <- newLocalNode transport rtable
105− print (localNodeId node)
106 putMVar serverNodeAddr (localNodeId node)
105 putMVar serverNodeAddr (localNodeId node)
107
106
108 forkIO $ do
107 forkIO $ do