@@ -38,7 +38,7 @@ import Control.Applicative ((<$>))
38import Control.Monad (void, when, forever)
38import Control.Monad (void, when, forever)
39import Control.Monad.IO.Class (MonadIO, liftIO)
39import Control.Monad.IO.Class (MonadIO, liftIO)
40import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
40import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
41−import qualified Control.Monad.Trans.Class as Trans (lift)
41+import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
42import Control.Exception (throwIO, SomeException, Exception, throwTo)
42import Control.Exception (throwIO, SomeException, Exception, throwTo)
43import qualified Control.Exception as Exception (catch)
43import qualified Control.Exception as Exception (catch)
44import Control.Concurrent (forkIO)
44import Control.Concurrent (forkIO)
@@ -101,7 +101,6 @@ import Control.Distributed.Process.Internal.Types
101 , Closure(..)
101 , Closure(..)
102 , Static(..)
102 , Static(..)
103 , Message
103 , Message
105 , TypedChannel(..)
104 , TypedChannel(..)
106 , Identifier(..)
105 , Identifier(..)
@@ -110,20 +109,18 @@ import Control.Distributed.Process.Internal.Types
110 , WhereIsReply(..)
109 , WhereIsReply(..)
111 , messageToPayload
110 , messageToPayload
112 , RemoteTable(..)
111 , RemoteTable(..)
114import Control.Distributed.Process.Serializable (Serializable)
116import Control.Distributed.Process.Serializable (Serializable)
115−import Control.Distributed.Process.Internal.MessageT
117+import Control.Distributed.Process.Internal.Dynamic (fromDynamic)
118+import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)
119+import Control.Distributed.Process.Internal.Node
116 ( sendBinary
120 ( sendBinary
117 , sendMessage
121 , sendMessage
118 , sendPayload
122 , sendPayload
124−import Control.Distributed.Process.Internal.Dynamic (fromDynamic)
125−import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)
126−import Control.Distributed.Process.Internal.Node (runLocalProcess)
127import Control.Distributed.Process.Internal.Primitives (expect, register)
124import Control.Distributed.Process.Internal.Primitives (expect, register)
128import qualified Control.Distributed.Process.Internal.Closure.Static as Static (__remoteTable)
125import qualified Control.Distributed.Process.Internal.Closure.Static as Static (__remoteTable)
129import qualified Control.Distributed.Process.Internal.Closure.CP as CP (__remoteTable)
126import qualified Control.Distributed.Process.Internal.Closure.CP as CP (__remoteTable)
@@ -156,9 +153,10 @@ createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
156createBareLocalNode endPoint rtable = do
153createBareLocalNode endPoint rtable = do
157 unq <- randomIO
154 unq <- randomIO
158 state <- newMVar LocalNodeState
155 state <- newMVar LocalNodeState
159− { _localProcesses = Map.empty
156+ { _localProcesses = Map.empty
160− , _localPidCounter = 0
157+ , _localPidCounter = 0
161− , _localPidUnique = unq
158+ , _localPidUnique = unq
159+ , _localConnections = Map.empty
163 ctrlChan <- newChan
161 ctrlChan <- newChan
164 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
162 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
@@ -214,10 +212,11 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do
214 , processId = pid
212 , processId = pid
215 , processState = pst
213 , processState = pst
216 , processThread = tid
214 , processThread = tid
215+ , processNode = node
218 tid' <- forkIO $ do
217 tid' <- forkIO $ do
219 reason <- Exception.catch
218 reason <- Exception.catch
220− (runLocalProcess node proc lproc >> return DiedNormal)
219+ (runLocalProcess lproc proc >> return DiedNormal)
221 (return . DiedException . (show :: SomeException -> String))
220 (return . DiedException . (show :: SomeException -> String))
222 -- [Unified: Table 4, rules termination and exiting]
221 -- [Unified: Table 4, rules termination and exiting]
223 modifyMVar_ (localState node) $
222 modifyMVar_ (localState node) $
@@ -287,7 +286,7 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty
287 (Map.insert cid proc procs)
286 (Map.insert cid proc procs)
290− Nothing ->
289+ Nothing ->
291 -- Request for an unknown process.
290 -- Request for an unknown process.
293 -- TODO: We should treat this as a fatal error on the
292 -- TODO: We should treat this as a fatal error on the
@@ -367,8 +366,8 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty
367--------------------------------------------------------------------------------
366--------------------------------------------------------------------------------
369runNodeController :: LocalNode -> IO ()
368runNodeController :: LocalNode -> IO ()
370−runNodeController node =
369+runNodeController =
371− runMessageT node (evalStateT (unNC nodeController) initNCState)
370+ runReaderT (evalStateT (unNC nodeController) initNCState)
373--------------------------------------------------------------------------------
372--------------------------------------------------------------------------------
374-- Internal data types --
373-- Internal data types --
@@ -383,11 +382,8 @@ data NCState = NCState
383 , _registry :: Map String ProcessId
382 , _registry :: Map String ProcessId
386−newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }
385+newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
387− deriving (Functor, Monad, MonadIO, MonadState NCState)
386+ deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
389−ncMsg :: MessageT IO a -> NC a
390−ncMsg = NC . Trans.lift
392initNCState :: NCState
388initNCState :: NCState
393initNCState = NCState { _links = Map.empty
389initNCState = NCState { _links = Map.empty
@@ -402,14 +398,17 @@ initNCState = NCState { _links = Map.empty
402-- [Unified: Table 7]
398-- [Unified: Table 7]
403nodeController :: NC ()
399nodeController :: NC ()
404nodeController = do
400nodeController = do
405− node <- ncMsg getLocalNode
401+ node <- ask
406 forever $ do
402 forever $ do
407 msg <- liftIO $ readChan (localCtrlChan node)
403 msg <- liftIO $ readChan (localCtrlChan node)
409 -- [Unified: Table 7, rule nc_forward]
405 -- [Unified: Table 7, rule nc_forward]
410 case destNid (ctrlMsgSignal msg) of
406 case destNid (ctrlMsgSignal msg) of
411 Just nid' | nid' /= localNodeId node ->
407 Just nid' | nid' /= localNodeId node ->
408+ liftIO $ sendBinary node
412− ncMsg $ sendBinary (NodeIdentifier nid') msg
410+ (NodeIdentifier nid')
414 return ()
413 return ()
@@ -441,7 +440,7 @@ ncEffectMonitor :: ProcessId -- ^ Who's watching?
441 -> Maybe MonitorRef -- ^ 'Nothing' to link
440 -> Maybe MonitorRef -- ^ 'Nothing' to link
443ncEffectMonitor from them mRef = do
442ncEffectMonitor from them mRef = do
444− node <- ncMsg getLocalNode
443+ node <- ask
445 shouldLink <-
444 shouldLink <-
446 if not (isLocal node them)
445 if not (isLocal node them)
447 then return True
446 then return True
@@ -454,15 +453,21 @@ ncEffectMonitor from them mRef = do
454 (False, True) -> -- [Unified: second rule]
453 (False, True) -> -- [Unified: second rule]
455 notifyDied from them DiedUnknownId mRef
454 notifyDied from them DiedUnknownId mRef
456 (False, False) -> -- [Unified: third rule]
455 (False, False) -> -- [Unified: third rule]
456+ -- TODO: this is the right sender according to the Unified semantics,
457+ -- but perhaps having 'them' as the sender would make more sense
458+ -- (see also: notifyDied)
459+ liftIO $ sendBinary node
460+ (NodeIdentifier $ localNodeId node)
457− ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg
461+ (NodeIdentifier $ processNodeId from)
458− { ctrlMsgSender = NodeIdentifier (localNodeId node)
463+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
459− , ctrlMsgSignal = Died them DiedUnknownId
464+ , ctrlMsgSignal = Died them DiedUnknownId
462-- [Unified: Table 11]
467-- [Unified: Table 11]
463ncEffectUnlink :: ProcessId -> Identifier -> NC ()
468ncEffectUnlink :: ProcessId -> Identifier -> NC ()
464ncEffectUnlink from them = do
469ncEffectUnlink from them = do
465− node <- ncMsg getLocalNode
470+ node <- ask
466 when (isLocal node (ProcessIdentifier from)) $
471 when (isLocal node (ProcessIdentifier from)) $
467 case them of
472 case them of
468 ProcessIdentifier pid ->
473 ProcessIdentifier pid ->
@@ -476,7 +481,7 @@ ncEffectUnlink from them = do
476-- [Unified: Table 11]
481-- [Unified: Table 11]
477ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
482ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
478ncEffectUnmonitor from ref = do
483ncEffectUnmonitor from ref = do
479− node <- ncMsg getLocalNode
484+ node <- ask
480 when (isLocal node (ProcessIdentifier from)) $
485 when (isLocal node (ProcessIdentifier from)) $
481 postAsMessage from $ DidUnmonitor ref
486 postAsMessage from $ DidUnmonitor ref
482 modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
487 modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
@@ -484,7 +489,7 @@ ncEffectUnmonitor from ref = do
484-- [Unified: Table 12]
489-- [Unified: Table 12]
485ncEffectDied :: Identifier -> DiedReason -> NC ()
490ncEffectDied :: Identifier -> DiedReason -> NC ()
486ncEffectDied ident reason = do
491ncEffectDied ident reason = do
487− node <- ncMsg getLocalNode
492+ node <- ask
488 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
493 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
489 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
494 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
@@ -509,9 +514,12 @@ ncEffectSpawn pid cProc ref = do
509 -- If the closure does not exist, we spawn a process that throws an exception
514 -- If the closure does not exist, we spawn a process that throws an exception
510 -- This allows the remote node to find out what's happening
515 -- This allows the remote node to find out what's happening
511 let proc = fromMaybe (fail $ "Error: unknown closure " ++ show cProc) mProc
516 let proc = fromMaybe (fail $ "Error: unknown closure " ++ show cProc) mProc
512− node <- ncMsg getLocalNode
517+ node <- ask
513 pid' <- liftIO $ forkProcess node proc
518 pid' <- liftIO $ forkProcess node proc
519+ liftIO $ sendMessage node
520+ (NodeIdentifier (localNodeId node))
514− ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')
521+ (ProcessIdentifier pid)
516-- Unified semantics does not explicitly describe how to implement 'register',
524-- Unified semantics does not explicitly describe how to implement 'register',
517-- but mentions it's "very similar to nsend" (Table 14)
525-- but mentions it's "very similar to nsend" (Table 14)
@@ -525,18 +533,24 @@ ncEffectRegister label mPid =
525-- Unified semantics does not explicitly describe 'whereis'
533-- Unified semantics does not explicitly describe 'whereis'
526ncEffectWhereIs :: ProcessId -> String -> NC ()
534ncEffectWhereIs :: ProcessId -> String -> NC ()
527ncEffectWhereIs from label = do
535ncEffectWhereIs from label = do
528 mPid <- gets (^. registryFor label)
537 mPid <- gets (^. registryFor label)
538+ liftIO $ sendMessage node
539+ (NodeIdentifier (localNodeId node))
529− ncMsg $ sendMessage (ProcessIdentifier from) (WhereIsReply label mPid)
540+ (ProcessIdentifier from)
541+ (WhereIsReply label mPid)
531-- [Unified: Table 14]
543-- [Unified: Table 14]
532ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
544ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
533−ncEffectNamedSend _from label msg = do
545+ncEffectNamedSend from label msg = do
534 mPid <- gets (^. registryFor label)
546 mPid <- gets (^. registryFor label)
535 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
547 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
536− -- TODO: messages don't carry a "from", but if they do, we should set it
548+ node <- ask
537− -- to be the 'from' of the original named send, not this node controller
538 forM_ mPid $ \pid ->
549 forM_ mPid $ \pid ->
550+ liftIO $ sendPayload node
539− ncMsg $ sendPayload (ProcessIdentifier pid) (messageToPayload msg)
552+ (ProcessIdentifier pid)
553+ (messageToPayload msg)
541--------------------------------------------------------------------------------
555--------------------------------------------------------------------------------
542-- Auxiliary --
556-- Auxiliary --
@@ -548,7 +562,7 @@ notifyDied :: ProcessId -- ^ Who to notify?
548 -> Maybe MonitorRef -- ^ 'Nothing' for linking
562 -> Maybe MonitorRef -- ^ 'Nothing' for linking
550notifyDied dest src reason mRef = do
564notifyDied dest src reason mRef = do
551− node <- ncMsg getLocalNode
565+ node <- ask
552 case (isLocal node (ProcessIdentifier dest), mRef, src) of
566 case (isLocal node (ProcessIdentifier dest), mRef, src) of
553 (True, Just ref, ProcessIdentifier pid) ->
567 (True, Just ref, ProcessIdentifier pid) ->
554 postAsMessage dest $ ProcessMonitorNotification ref pid reason
568 postAsMessage dest $ ProcessMonitorNotification ref pid reason
@@ -565,10 +579,13 @@ notifyDied dest src reason mRef = do
565 (False, _, _) ->
579 (False, _, _) ->
566 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
580 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
567 -- (see [Unified: Table 10]
581 -- (see [Unified: Table 10]
582+ liftIO $ sendBinary node
583+ (NodeIdentifier $ localNodeId node)
568− ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg
584+ (NodeIdentifier $ processNodeId dest)
569− { ctrlMsgSender = NodeIdentifier (localNodeId node)
586+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
570− , ctrlMsgSignal = Died src reason
587+ , ctrlMsgSignal = Died src reason
573-- | [Unified: Table 8]
590-- | [Unified: Table 8]
574destNid :: ProcessSignal -> Maybe NodeId
591destNid :: ProcessSignal -> Maybe NodeId
@@ -592,13 +609,13 @@ isLocal nid ident = nodeOf ident == localNodeId nid
592-- | Lookup a local closure
609-- | Lookup a local closure
593unClosure :: Typeable a => Closure a -> NC (Maybe a)
610unClosure :: Typeable a => Closure a -> NC (Maybe a)
594unClosure (Closure (Static label) env) = do
611unClosure (Closure (Static label) env) = do
595− rtable <- remoteTable <$> ncMsg getLocalNode
612+ rtable <- remoteTable <$> ask
596 return (resolveClosure rtable label env >>= fromDynamic)
613 return (resolveClosure rtable label env >>= fromDynamic)
598-- | Check if an identifier refers to a valid local object
615-- | Check if an identifier refers to a valid local object
599isValidLocalIdentifier :: Identifier -> NC Bool
616isValidLocalIdentifier :: Identifier -> NC Bool
600isValidLocalIdentifier ident = do
617isValidLocalIdentifier ident = do
601− node <- ncMsg getLocalNode
618+ node <- ask
602 liftIO . withMVar (localState node) $ \nSt ->
619 liftIO . withMVar (localState node) $ \nSt ->
603 case ident of
620 case ident of
604 NodeIdentifier nid ->
621 NodeIdentifier nid ->
@@ -631,7 +648,7 @@ throwException pid e = withLocalProc pid $ \p ->
632withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
649withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
633withLocalProc pid p = do
650withLocalProc pid p = do
634− node <- ncMsg getLocalNode
651+ node <- ask
635 liftIO $ do
652 liftIO $ do
636 -- By [Unified: table 6, rule missing_process] messages to dead processes
653 -- By [Unified: table 6, rule missing_process] messages to dead processes
637 -- can silently be dropped
654 -- can silently be dropped