@@ -38,7 +38,7 @@ import Control.Applicative ((<$>))
3838import Control.Monad (void, when, forever)
3939import Control.Monad.IO.Class (MonadIO, liftIO)
4040import 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)
4242import Control.Exception (throwIO, SomeException, Exception, throwTo)
4343import qualified Control.Exception as Exception (catch)
4444import Control.Concurrent (forkIO)
@@ -101,7 +101,6 @@ import Control.Distributed.Process.Internal.Types
101101 , Closure(..)
102102 , Static(..)
103103 , Message
104− , MessageT
105104 , TypedChannel(..)
106105 , Identifier(..)
107106 , nodeOf
@@ -110,20 +109,18 @@ import Control.Distributed.Process.Internal.Types
110109 , WhereIsReply(..)
111110 , messageToPayload
112111 , RemoteTable(..)
112+ , payloadToMessage
113+ , createMessage
114+ , runLocalProcess
113115 )
114116import 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
116120 ( sendBinary
117121 , sendMessage
118122 , sendPayload
119− , getLocalNode
120− , runMessageT
121− , payloadToMessage
122− , createMessage
123123 )
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)
127124import Control.Distributed.Process.Internal.Primitives (expect, register)
128125import qualified Control.Distributed.Process.Internal.Closure.Static as Static (__remoteTable)
129126import qualified Control.Distributed.Process.Internal.Closure.CP as CP (__remoteTable)
@@ -156,9 +153,10 @@ createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
156153createBareLocalNode endPoint rtable = do
157154 unq <- randomIO
158155 state <- newMVar LocalNodeState
159− { _localProcesses = Map.empty
160− , _localPidCounter = 0
161− , _localPidUnique = unq
156+ { _localProcesses = Map.empty
157+ , _localPidCounter = 0
158+ , _localPidUnique = unq
159+ , _localConnections = Map.empty
162160 }
163161 ctrlChan <- newChan
164162 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
@@ -214,10 +212,11 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do
214212 , processId = pid
215213 , processState = pst
216214 , processThread = tid
215+ , processNode = node
217216 }
218217 tid' <- forkIO $ do
219218 reason <- Exception.catch
220− (runLocalProcess node proc lproc >> return DiedNormal)
219+ (runLocalProcess lproc proc >> return DiedNormal)
221220 (return . DiedException . (show :: SomeException -> String))
222221 -- [Unified: Table 4, rules termination and exiting]
223222 modifyMVar_ (localState node) $
@@ -287,7 +286,7 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty
287286 (Map.insert cid proc procs)
288287 chans
289288 ctrls
290− Nothing ->
289+ Nothing ->
291290 -- Request for an unknown process.
292291 --
293292 -- TODO: We should treat this as a fatal error on the
@@ -367,8 +366,8 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty
367366--------------------------------------------------------------------------------
368367
369368runNodeController :: LocalNode -> IO ()
370−runNodeController node =
371− runMessageT node (evalStateT (unNC nodeController) initNCState)
369+runNodeController =
370+ runReaderT (evalStateT (unNC nodeController) initNCState)
372371
373372--------------------------------------------------------------------------------
374373-- Internal data types --
@@ -383,11 +382,8 @@ data NCState = NCState
383382 , _registry :: Map String ProcessId
384383 }
385384
386−newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }
387− deriving (Functor, Monad, MonadIO, MonadState NCState)
388−
389−ncMsg :: MessageT IO a -> NC a
390−ncMsg = NC . Trans.lift
385+newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
386+ deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
391387
392388initNCState :: NCState
393389initNCState = NCState { _links = Map.empty
@@ -402,14 +398,17 @@ initNCState = NCState { _links = Map.empty
402398-- [Unified: Table 7]
403399nodeController :: NC ()
404400nodeController = do
405− node <- ncMsg getLocalNode
401+ node <- ask
406402 forever $ do
407403 msg <- liftIO $ readChan (localCtrlChan node)
408404
409405 -- [Unified: Table 7, rule nc_forward]
410406 case destNid (ctrlMsgSignal msg) of
411407 Just nid' | nid' /= localNodeId node ->
412− ncMsg $ sendBinary (NodeIdentifier nid') msg
408+ liftIO $ sendBinary node
409+ (ctrlMsgSender msg)
410+ (NodeIdentifier nid')
411+ msg
413412 _ ->
414413 return ()
415414
@@ -441,7 +440,7 @@ ncEffectMonitor :: ProcessId -- ^ Who's watching?
441440 -> Maybe MonitorRef -- ^ 'Nothing' to link
442441 -> NC ()
443442ncEffectMonitor from them mRef = do
444− node <- ncMsg getLocalNode
443+ node <- ask
445444 shouldLink <-
446445 if not (isLocal node them)
447446 then return True
@@ -454,15 +453,21 @@ ncEffectMonitor from them mRef = do
454453 (False, True) -> -- [Unified: second rule]
455454 notifyDied from them DiedUnknownId mRef
456455 (False, False) -> -- [Unified: third rule]
457− ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg
458− { ctrlMsgSender = NodeIdentifier (localNodeId node)
459− , ctrlMsgSignal = Died them DiedUnknownId
460− }
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)
461+ (NodeIdentifier $ processNodeId from)
462+ NCMsg
463+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
464+ , ctrlMsgSignal = Died them DiedUnknownId
465+ }
461466
462467-- [Unified: Table 11]
463468ncEffectUnlink :: ProcessId -> Identifier -> NC ()
464469ncEffectUnlink from them = do
465− node <- ncMsg getLocalNode
470+ node <- ask
466471 when (isLocal node (ProcessIdentifier from)) $
467472 case them of
468473 ProcessIdentifier pid ->
@@ -476,7 +481,7 @@ ncEffectUnlink from them = do
476481-- [Unified: Table 11]
477482ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
478483ncEffectUnmonitor from ref = do
479− node <- ncMsg getLocalNode
484+ node <- ask
480485 when (isLocal node (ProcessIdentifier from)) $
481486 postAsMessage from $ DidUnmonitor ref
482487 modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
@@ -484,7 +489,7 @@ ncEffectUnmonitor from ref = do
484489-- [Unified: Table 12]
485490ncEffectDied :: Identifier -> DiedReason -> NC ()
486491ncEffectDied ident reason = do
487− node <- ncMsg getLocalNode
492+ node <- ask
488493 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
489494 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
490495
@@ -509,9 +514,12 @@ ncEffectSpawn pid cProc ref = do
509514 -- If the closure does not exist, we spawn a process that throws an exception
510515 -- This allows the remote node to find out what's happening
511516 let proc = fromMaybe (fail $ "Error: unknown closure " ++ show cProc) mProc
512− node <- ncMsg getLocalNode
517+ node <- ask
513518 pid' <- liftIO $ forkProcess node proc
514− ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')
519+ liftIO $ sendMessage node
520+ (NodeIdentifier (localNodeId node))
521+ (ProcessIdentifier pid)
522+ (DidSpawn ref pid')
515523
516524-- Unified semantics does not explicitly describe how to implement 'register',
517525-- but mentions it's "very similar to nsend" (Table 14)
@@ -525,18 +533,24 @@ ncEffectRegister label mPid =
525533-- Unified semantics does not explicitly describe 'whereis'
526534ncEffectWhereIs :: ProcessId -> String -> NC ()
527535ncEffectWhereIs from label = do
536+ node <- ask
528537 mPid <- gets (^. registryFor label)
529− ncMsg $ sendMessage (ProcessIdentifier from) (WhereIsReply label mPid)
538+ liftIO $ sendMessage node
539+ (NodeIdentifier (localNodeId node))
540+ (ProcessIdentifier from)
541+ (WhereIsReply label mPid)
530542
531543-- [Unified: Table 14]
532544ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
533−ncEffectNamedSend _from label msg = do
545+ncEffectNamedSend from label msg = do
534546 mPid <- gets (^. registryFor label)
535547 -- 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
537− -- to be the 'from' of the original named send, not this node controller
548+ node <- ask
538549 forM_ mPid $ \pid ->
539− ncMsg $ sendPayload (ProcessIdentifier pid) (messageToPayload msg)
550+ liftIO $ sendPayload node
551+ from
552+ (ProcessIdentifier pid)
553+ (messageToPayload msg)
540554
541555--------------------------------------------------------------------------------
542556-- Auxiliary --
@@ -548,7 +562,7 @@ notifyDied :: ProcessId -- ^ Who to notify?
548562 -> Maybe MonitorRef -- ^ 'Nothing' for linking
549563 -> NC ()
550564notifyDied dest src reason mRef = do
551− node <- ncMsg getLocalNode
565+ node <- ask
552566 case (isLocal node (ProcessIdentifier dest), mRef, src) of
553567 (True, Just ref, ProcessIdentifier pid) ->
554568 postAsMessage dest $ ProcessMonitorNotification ref pid reason
@@ -565,10 +579,13 @@ notifyDied dest src reason mRef = do
565579 (False, _, _) ->
566580 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
567581 -- (see [Unified: Table 10]
568− ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg
569− { ctrlMsgSender = NodeIdentifier (localNodeId node)
570− , ctrlMsgSignal = Died src reason
571− }
582+ liftIO $ sendBinary node
583+ (NodeIdentifier $ localNodeId node)
584+ (NodeIdentifier $ processNodeId dest)
585+ NCMsg
586+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
587+ , ctrlMsgSignal = Died src reason
588+ }
572589
573590-- | [Unified: Table 8]
574591destNid :: ProcessSignal -> Maybe NodeId
@@ -592,13 +609,13 @@ isLocal nid ident = nodeOf ident == localNodeId nid
592609-- | Lookup a local closure
593610unClosure :: Typeable a => Closure a -> NC (Maybe a)
594611unClosure (Closure (Static label) env) = do
595− rtable <- remoteTable <$> ncMsg getLocalNode
612+ rtable <- remoteTable <$> ask
596613 return (resolveClosure rtable label env >>= fromDynamic)
597614
598615-- | Check if an identifier refers to a valid local object
599616isValidLocalIdentifier :: Identifier -> NC Bool
600617isValidLocalIdentifier ident = do
601− node <- ncMsg getLocalNode
618+ node <- ask
602619 liftIO . withMVar (localState node) $ \nSt ->
603620 case ident of
604621 NodeIdentifier nid ->
@@ -631,7 +648,7 @@ throwException pid e = withLocalProc pid $ \p ->
631648
632649withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
633650withLocalProc pid p = do
634− node <- ncMsg getLocalNode
651+ node <- ask
635652 liftIO $ do
636653 -- By [Unified: table 6, rule missing_process] messages to dead processes
637654 -- can silently be dropped