2{-# LANGUAGE DeriveDataTypeable #-}
3{-# LANGUAGE ScopedTypeVariables #-}
4{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE BangPatterns #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7{-# LANGUAGE RecordWildCards #-}
11module Control.Distributed.Process.Node
21-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
23#if ! MIN_VERSION_base(4,6,0)
24import Prelude hiding (catch)
27import System.IO (fixIO, hPutStrLn, stderr)
28import System.Mem.Weak (Weak, deRefWeak)
29import qualified Data.ByteString.Lazy as BSL (fromChunks)
30import Data.Binary (decode)
32import qualified Data.Map as Map
44import qualified Data.Set as Set
51import Data.Foldable (forM_)
52import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
53import Data.Typeable (Typeable)
54import Control.Category ((>>>))
55import Control.Applicative (Applicative, (<$>))
56import Control.Monad (void, when)
57import Control.Monad.IO.Class (MonadIO, liftIO)
58import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
59import qualified Control.Monad.State.Strict as StateT (get, put)
60import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
61import Control.Exception
63 , AsyncException(ThreadKilled)
67 , uninterruptibleMask_
69import qualified Control.Exception as Exception (Handler(..), catches, finally)
70import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId)
71import Control.Distributed.Process.Internal.StrictMVar
80import Control.Concurrent.Chan (newChan, writeChan, readChan)
81import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
82import Control.Concurrent.STM
85import Control.Distributed.Process.Internal.CQueue
92import qualified Network.Transport as NT
105 , Reliability(ReliableOrdered)
107import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
108import System.Random (randomIO)
109import Control.Distributed.Static (RemoteTable, Closure)
110import qualified Control.Distributed.Static as Static
114import Control.Distributed.Process.Internal.Types
121 , ValidLocalNodeState(..)
122 , withValidLocalState
123 , modifyValidLocalState_
125 , LocalProcessState(..)
137 , ProcessMonitorNotification(..)
138 , NodeMonitorNotification(..)
139 , PortMonitorNotification(..)
140 , ProcessExitException(..)
141 , ProcessLinkException(..)
142 , NodeLinkException(..)
143 , PortLinkException(..)
145 , DidUnlinkProcess(..)
155 , ProcessInfoNone(..)
162 , createUnencodedMessage
164 , firstNonReservedProcessId
165 , ImplicitReconnect(WithImplicitReconnect)
167import Control.Distributed.Process.Management.Internal.Agent
170import qualified Control.Distributed.Process.Management.Internal.Table as Table
172 , startTableCoordinator
174import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
177import Control.Distributed.Process.Management.Internal.Trace.Tracer
180import Control.Distributed.Process.Management.Internal.Trace.Types
186import Control.Distributed.Process.Management.Internal.Types
189import Control.Distributed.Process.Serializable (Serializable)
190import Control.Distributed.Process.Internal.Messaging
193 , closeImplicitReconnections
196import Control.Distributed.Process.Internal.Primitives
205import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
206import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
207import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
208import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
214--------------------------------------------------------------------------------
216--------------------------------------------------------------------------------
218initRemoteTable :: RemoteTable
219initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
221-- | Initialize a new local node.
222newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
223newLocalNode transport rtable = do
224 mEndPoint <- NT.newEndPoint transport
226 Left ex -> throwIO ex
228 localNode <- createBareLocalNode endPoint rtable
229 startServiceProcesses localNode
232-- | Create a new local node (without any service processes running)
233createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
234createBareLocalNode endPoint rtable = do
236 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
237 { _localProcesses = Map.empty
238 , _localPidCounter = firstNonReservedProcessId
239 , _localPidUnique = unq
240 , _localConnections = Map.empty
243 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
244 , localEndPoint = endPoint
246 , localCtrlChan = ctrlChan
247 , localEventBus = MxEventBusInitialising
248 , remoteTable = rtable
250 tracedNode <- startMxAgent node
252 -- Once the NC terminates, the endpoint isn't much use,
253 void $ forkIO $ Exception.finally (runNodeController tracedNode)
254 (NT.closeEndPoint (localEndPoint node))
256 -- whilst a closed/failing endpoint will terminate the NC
257 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
263 writeChan (localCtrlChan node) NCMsg
264 { ctrlMsgSender = NodeIdentifier (localNodeId node)
265 , ctrlMsgSignal = SigShutdown
268startMxAgent :: LocalNode -> IO LocalNode
269startMxAgent node = do
270 -- see note [tracer/forkProcess races]
271 let fork = forkProcess node
272 mv <- MVar.newEmptyMVar
273 pid <- fork $ mxAgentController fork mv
274 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
275 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
277startDefaultTracer :: LocalNode -> IO ()
278startDefaultTracer node' = do
279 let t = localEventBus node'
281 MxEventBus _ (Tracer pid _) _ _ -> do
282 runProcess node' $ register "trace.controller" pid
283 pid' <- forkProcess node' defaultTracer
284 enableTrace (localEventBus node') pid'
285 runProcess node' $ register "tracer.initial" pid'
288-- TODO: we need a better mechanism for defining and registering services
290-- | Start and register the service processes on a node
291startServiceProcesses :: LocalNode -> IO ()
292startServiceProcesses node = do
293 -- tracing /spawns/ relies on the tracer being enabled, but we start
294 -- the default tracer first, even though it might @nsend@ to the logger
295 -- before /that/ process has started - this is a totally harmless race
296 -- however, so we deliberably ignore it
297 startDefaultTracer node
298 tableCoordinatorPid <- fork $ Table.startTableCoordinator fork
299 runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid
300 logger <- forkProcess node loop
302 register "logger" logger
303 -- The trace.logger is used for tracing to the console to avoid feedback
304 -- loops during tracing if the user reregisters the "logger" with a custom
305 -- process which uses 'send' or other primitives which are traced.
306 register "trace.logger" logger
308 fork = forkProcess node
312 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
313 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
315 , match $ \((time, string) :: (String, String)) -> do
316 -- this is a 'trace' message from the local node tracer
317 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
319 , match $ \(ch :: SendPort ()) -> -- a shutdown request
323-- | Force-close a local node, killing all processes on that node.
324closeLocalNode :: LocalNode -> IO ()
325closeLocalNode node = do
326 modifyMVar_ (localState node) $ const $ return LocalNodeClosed
327 withValidLocalState (localState node) $ \vst ->
328 forM_ (vst ^. localProcesses) $ \lproc ->
329 -- Semantics of 'throwTo' guarantee that target thread will get delivered
330 -- an exception. Therefore, target thread will be killed eventually and
331 -- that's as good as we can do. No need to wait for thread to actually
333 killThread (processThread lproc)
334 NT.closeEndPoint (localEndPoint node)
336-- | Run a process on a local node and wait for it to finish
337runProcess :: LocalNode -> Process () -> IO ()
338runProcess node proc = do
341 void $ forkProcess node $ do
342 catch (proc `finally` liftIO (putMVar done ()))
343 (\(ex :: SomeException) -> liftIO $ throwTo tid ex)
346-- | Spawn a new process on a local node
347forkProcess :: LocalNode -> Process () -> IO ProcessId
348forkProcess node proc =
349 modifyMVarMasked (localState node) startProcess
351 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
352 startProcess (LocalNodeValid vst) = do
353 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
354 , lpidUnique = vst ^. localPidUnique
356 let pid = ProcessId { processNodeId = localNodeId node
357 , processLocalId = lpid
359 pst <- newMVar LocalProcessState { _monitorCounter = 0
361 , _channelCounter = 0
362 , _typedChannels = Map.empty
365 weakQueue <- mkWeakCQueue queue (return ())
366 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
367 let lproc = LocalProcess { processQueue = queue
368 , processWeakQ = weakQueue
371 , processThread = tid
374 tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do
375 reason <- Exception.catches
376 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
377 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
378 mMsg <- unwrapMessage msg :: IO (Maybe String)
380 Nothing -> return $ DiedException $ show ex
381 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
383 (return . DiedException . (show :: SomeException -> String)))]
385 -- [Unified: Table 4, rules termination and exiting]
386 modifyValidLocalState_ (localState node) (cleanupProcess pid)
387 writeChan (localCtrlChan node) NCMsg
388 { ctrlMsgSender = ProcessIdentifier pid
389 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
393 -- see note [tracer/forkProcess races]
394 trace node (MxSpawned pid)
396 if lpidCounter lpid == maxBound
398 -- TODO: this doesn't look right at all - how do we know
399 -- that newUnique represents a process id that is available!?
400 newUnique <- randomIO
401 return ( LocalNodeValid
402 $ (localProcessWithId lpid ^= Just lproc)
403 . (localPidCounter ^= firstNonReservedProcessId)
404 . (localPidUnique ^= newUnique)
409 return ( LocalNodeValid
410 $ (localProcessWithId lpid ^= Just lproc)
411 . (localPidCounter ^: (+ 1))
415 startProcess LocalNodeClosed = throwIO $ userError "LocalNode closed"
417 cleanupProcess :: ProcessId -> ValidLocalNodeState -> IO ValidLocalNodeState
418 cleanupProcess pid vst = do
419 let pid' = ProcessIdentifier pid
420 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
421 mapM_ (NT.close . fst) (Map.elems affected)
422 return $ (localProcessWithId (processLocalId pid) ^= Nothing)
423 . (localConnections ^= unaffected)
426-- note [tracer/forkProcess races]
428-- Our startTracing function uses forkProcess to start the trace controller
429-- process, and of course forkProcess attempts to call traceEvent once the
430-- process has started. This is harmless, as the localEventBus is not updated
431-- until /after/ the initial forkProcess completes, so the first call to
432-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
435--------------------------------------------------------------------------------
436-- Handle incoming messages --
437--------------------------------------------------------------------------------
439type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
443 | ToProc ProcessId (Weak (CQueue Message))
444 | ToChan TypedChannel
447data ConnectionState = ConnectionState {
448 _incoming :: !(Map NT.ConnectionId IncomingConnection)
449 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
452initConnectionState :: ConnectionState
453initConnectionState = ConnectionState {
454 _incoming = Map.empty
455 , _incomingFrom = Map.empty
458incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
459incoming = accessor _incoming (\conns st -> st { _incoming = conns })
461incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
462incomingAt cid = incoming >>> DAC.mapMaybe cid
464incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
465incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
467 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
469handleIncomingMessages :: LocalNode -> IO ()
470handleIncomingMessages node = go initConnectionState
472 go :: ConnectionState -> IO ()
474 event <- NT.receive endpoint
476 NT.ConnectionOpened cid rel theirAddr ->
477 if rel == NT.ReliableOrdered
479 trace node (MxConnected cid theirAddr)
481 (incomingAt cid ^= Just (theirAddr, Uninit))
482 . (incomingFrom theirAddr ^: Set.insert cid)
485 else invalidRequest cid st
486 NT.Received cid payload ->
487 case st ^. incomingAt cid of
488 Just (_, ToProc pid weakQueue) -> do
489 mQueue <- deRefWeak weakQueue
490 forM_ mQueue $ \queue -> do
491 -- TODO: if we find that the queue is Nothing, should we remove
492 -- it from the NC state? (and same for channels, below)
493 let msg = payloadToMessage payload
494 enqueue queue msg -- 'enqueue' is strict
495 trace node (MxReceived pid msg)
497 Just (_, ToChan (TypedChannel chan')) -> do
498 mChan <- deRefWeak chan'
499 -- If mChan is Nothing, the process has given up the read end of
500 -- the channel and we simply ignore the incoming message
501 forM_ mChan $ \chan -> atomically $
502 -- We make sure the message is fully decoded when it is enqueued
503 writeTQueue chan $! decode (BSL.fromChunks payload)
505 Just (_, ToNode) -> do
506 let ctrlMsg = decode . BSL.fromChunks $ payload
507 writeChan ctrlChan $! ctrlMsg
509 Just (src, Uninit) ->
510 case decode (BSL.fromChunks payload) of
511 ProcessIdentifier pid -> do
512 let lpid = processLocalId pid
513 mProc <- withValidLocalState state $ return . (^. localProcessWithId lpid)
516 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
518 invalidRequest cid st
519 SendPortIdentifier chId -> do
520 let lcid = sendPortLocalId chId
521 lpid = processLocalId (sendPortProcessId chId)
522 mProc <- withValidLocalState state $ return . (^. localProcessWithId lpid)
525 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
528 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
530 invalidRequest cid st
532 invalidRequest cid st
533 NodeIdentifier nid ->
534 if nid == localNodeId node
535 then go (incomingAt cid ^= Just (src, ToNode) $ st)
536 else invalidRequest cid st
538 invalidRequest cid st
539 NT.ConnectionClosed cid ->
540 case st ^. incomingAt cid of
542 invalidRequest cid st
544 trace node (MxDisconnected cid src)
545 go ( (incomingAt cid ^= Nothing)
546 . (incomingFrom src ^: Set.delete cid)
549 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
550 -- [Unified table 9, rule node_disconnect]
551 let nid = NodeIdentifier $ NodeId theirAddr
552 writeChan ctrlChan NCMsg
553 { ctrlMsgSender = nid
554 , ctrlMsgSignal = Died nid DiedDisconnect
556 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
557 closeImplicitReconnections node nid
558 go ( (incomingFrom theirAddr ^= Set.empty)
559 . (incoming ^: Map.filterWithKey (const . notLost))
562 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
563 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
564 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
565 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
568 NT.ReceivedMulticast _ _ ->
569 -- If we received a multicast message, something went horribly wrong
570 -- and we just give up
571 fail "Cloud Haskell fatal error: received unexpected multicast"
573 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
574 invalidRequest cid st = do
575 -- TODO: We should treat this as a fatal error on the part of the remote
576 -- node. That is, we should report the remote node as having died, and we
577 -- should close incoming connections (this requires a Transport layer
579 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
581 go ( incomingAt cid ^= Nothing
585 state = localState node
586 endpoint = localEndPoint node
587 ctrlChan = localCtrlChan node
589--------------------------------------------------------------------------------
590-- Top-level access to the node controller --
591--------------------------------------------------------------------------------
593runNodeController :: LocalNode -> IO ()
595 runReaderT (evalStateT (unNC nodeController) initNCState)
597--------------------------------------------------------------------------------
598-- Internal data types --
599--------------------------------------------------------------------------------
601data NCState = NCState
602 { -- Mapping from remote processes to linked local processes
603 _links :: !(Map Identifier (Set ProcessId))
604 -- Mapping from remote processes to monitoring local processes
605 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
606 -- Process registry: names and where they live, mapped to the PIDs
607 , _registeredHere :: !(Map String ProcessId)
608 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
611newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
612 deriving ( Applicative
617 , MonadReader LocalNode
620initNCState :: NCState
621initNCState = NCState { _links = Map.empty
622 , _monitors = Map.empty
623 , _registeredHere = Map.empty
624 , _registeredOnNodes = Map.empty
627-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
628-- type is deliberately not exported so it cannot be caught explicitly.
629data ProcessKillException =
630 ProcessKillException !ProcessId !String
633instance Exception ProcessKillException
634instance Show ProcessKillException where
635 show (ProcessKillException pid reason) =
636 "killed-by=" ++ show pid ++ ",reason=" ++ reason
638--------------------------------------------------------------------------------
639-- Tracing/Debugging --
640--------------------------------------------------------------------------------
642-- [Issue #104 / DP-13]
644traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
645traceNotifyDied node ident reason =
646 -- TODO: sendPortDied notifications
647 liftIO $ withLocalTracer node $ \t ->
649 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
650 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
653traceEventFmtIO :: LocalNode
657traceEventFmtIO node fmt args =
658 withLocalTracer node $ \t -> traceLogFmt t fmt args
660trace :: LocalNode -> MxEvent -> IO ()
661trace node ev = withLocalTracer node $ \t -> traceEvent t ev
663withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
664withLocalTracer node act = act (localEventBus node)
666--------------------------------------------------------------------------------
667-- Core functionality --
668--------------------------------------------------------------------------------
671nodeController :: NC ()
675 msg <- liftIO $ readChan (localCtrlChan node)
677 -- [Unified: Table 7, rule nc_forward]
678 case destNid (ctrlMsgSignal msg) of
679 Just nid' | nid' /= localNodeId node ->
680 liftIO $ sendBinary node
682 (NodeIdentifier nid')
683 WithImplicitReconnect
689 NCMsg (ProcessIdentifier from) (Link them) ->
690 ncEffectMonitor from them Nothing
691 NCMsg (ProcessIdentifier from) (Monitor ref) ->
692 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
693 NCMsg (ProcessIdentifier from) (Unlink them) ->
694 ncEffectUnlink from them
695 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
696 ncEffectUnmonitor from ref
697 NCMsg _from (Died ident reason) ->
698 ncEffectDied ident reason
699 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
700 ncEffectSpawn from proc ref
701 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
702 ncEffectRegister from label atnode pid force
703 NCMsg (ProcessIdentifier from) (WhereIs label) ->
704 ncEffectWhereIs from label
705 NCMsg _ (NamedSend label msg') ->
706 ncEffectNamedSend label msg'
707 NCMsg _ (UnreliableSend lpid msg') ->
708 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
709 NCMsg _ (LocalSend to msg') ->
710 ncEffectLocalSend node to msg'
711 NCMsg _ (LocalPortSend to msg') ->
712 ncEffectLocalPortSend to msg'
713 NCMsg (ProcessIdentifier from) (Kill to reason) ->
714 ncEffectKill from to reason
715 NCMsg (ProcessIdentifier from) (Exit to reason) ->
716 ncEffectExit from to reason
717 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
718 ncEffectGetInfo from pid
719 NCMsg _ SigShutdown ->
721 NT.closeEndPoint (localEndPoint node)
722 `Exception.finally` throwIO ThreadKilled
723 -- ThreadKilled seems to make more sense than fail/error here
724 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
725 ncEffectGetNodeStats from nid
727 error $ "nodeController: unexpected message " ++ show unexpected
729-- [Unified: Table 10]
730ncEffectMonitor :: ProcessId -- ^ Who's watching?
731 -> Identifier -- ^ Who's being watched?
732 -> Maybe MonitorRef -- ^ 'Nothing' to link
734ncEffectMonitor from them mRef = do
737 if not (isLocal node them)
739 else isValidLocalIdentifier them
740 case (shouldLink, isLocal node (ProcessIdentifier from)) of
741 (True, _) -> -- [Unified: first rule]
743 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
744 Nothing -> modify' $ linksFor them ^: Set.insert from
745 (False, True) -> -- [Unified: second rule]
746 notifyDied from them DiedUnknownId mRef
747 (False, False) -> -- [Unified: third rule]
748 -- TODO: this is the right sender according to the Unified semantics,
749 -- but perhaps having 'them' as the sender would make more sense
750 -- (see also: notifyDied)
751 liftIO $ sendBinary node
752 (NodeIdentifier $ localNodeId node)
753 (NodeIdentifier $ processNodeId from)
754 WithImplicitReconnect
756 { ctrlMsgSender = NodeIdentifier (localNodeId node)
757 , ctrlMsgSignal = Died them DiedUnknownId
760-- [Unified: Table 11]
761ncEffectUnlink :: ProcessId -> Identifier -> NC ()
762ncEffectUnlink from them = do
764 when (isLocal node (ProcessIdentifier from)) $
766 ProcessIdentifier pid ->
767 postAsMessage from $ DidUnlinkProcess pid
768 NodeIdentifier nid ->
769 postAsMessage from $ DidUnlinkNode nid
770 SendPortIdentifier cid ->
771 postAsMessage from $ DidUnlinkPort cid
772 modify' $ linksFor them ^: Set.delete from
774-- [Unified: Table 11]
775ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
776ncEffectUnmonitor from ref = do
778 when (isLocal node (ProcessIdentifier from)) $
779 postAsMessage from $ DidUnmonitor ref
780 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
782-- [Unified: Table 12]
783ncEffectDied :: Identifier -> DiedReason -> NC ()
784ncEffectDied ident reason = do
786 traceNotifyDied node ident reason
787 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
788 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
790-- _registry :: !(Map (String,NodeId) ProcessId)
792 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
794 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
796 when (localOnly <= isLocal node (ProcessIdentifier us)) $
797 notifyDied us them reason Nothing
799 forM_ (Map.toList affectedMons) $ \(them, refs) ->
800 forM_ refs $ \(us, ref) ->
801 when (localOnly <= isLocal node (ProcessIdentifier us)) $
802 notifyDied us them reason (Just ref)
804 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
806 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
808 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
809 mapM (\(pid,nidlist) ->
810 case ident `impliesDeathOf` ProcessIdentifier pid of
812 do forM_ nidlist $ \(nid,_) ->
813 when (not $ isLocal node (NodeIdentifier nid))
814 (forwardNameDeath node nid)
816 False -> return $ Just (pid,nidlist) )
817 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
819 forwardNameDeath node nid =
820 liftIO $ sendBinary node
821 (NodeIdentifier $ localNodeId node)
822 (NodeIdentifier $ nid)
823 WithImplicitReconnect
825 { ctrlMsgSender = NodeIdentifier (localNodeId node)
826 , ctrlMsgSignal = Died ident reason
829-- [Unified: Table 13]
830ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
831ncEffectSpawn pid cProc ref = do
832 mProc <- unClosure cProc
833 -- If the closure does not exist, we spawn a process that throws an exception
834 -- This allows the remote node to find out what's happening
836 let proc = case mProc of
837 Left err -> fail $ "Error: Could not resolve closure: " ++ err
840 pid' <- liftIO $ forkProcess node proc
841 liftIO $ sendMessage node
842 (NodeIdentifier (localNodeId node))
843 (ProcessIdentifier pid)
844 WithImplicitReconnect
847-- Unified semantics does not explicitly describe how to implement 'register',
848-- but mentions it's "very similar to nsend" (Table 14)
849-- We send a response indicated if the operation is invalid
850ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
851ncEffectRegister from label atnode mPid reregistration = do
853 currentVal <- gets (^. registeredHereFor label)
856 Nothing -> -- unregister request
857 return $ isJust currentVal
858 Just thepid -> -- register request
859 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
860 return $ (isNothing currentVal /= reregistration) &&
861 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
862 if isLocal node (NodeIdentifier atnode)
863 then do when (isOk) $
864 do modify' $ registeredHereFor label ^= mPid
865 updateRemote node currentVal mPid
867 (Just p) -> liftIO $ trace node (MxRegistered p label)
868 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
869 liftIO $ sendMessage node
870 (NodeIdentifier (localNodeId node))
871 (ProcessIdentifier from)
872 WithImplicitReconnect
873 (RegisterReply label isOk)
875 case reregistration of
877 False -> flip incList
880 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
881 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
882 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
883 forward node (processNodeId newval) (Register label atnode (Just newval) False)
884 updateRemote node Nothing (Just newval) =
885 forward node (processNodeId newval) (Register label atnode (Just newval) False)
886 updateRemote node (Just oldval) Nothing =
887 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
888 updateRemote _ _ _ = return ()
889 maybeify f Nothing = unmaybeify $ f []
890 maybeify f (Just x) = unmaybeify $ f x
892 unmaybeify [] = Nothing
893 unmaybeify x = Just x
894 incList [] tag = [(tag,1)]
895 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
896 incList (x:xs) tag = x : incList xs tag
898 decList ((atag,1):xs) tag | atag == tag = xs
899 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
900 decList (x:xs) tag = x:decList xs tag
901 forward node to reg =
902 when (not $ isLocal node (NodeIdentifier to)) $
903 liftIO $ sendBinary node
904 (ProcessIdentifier from)
906 WithImplicitReconnect
908 { ctrlMsgSender = ProcessIdentifier from
909 , ctrlMsgSignal = reg
913-- Unified semantics does not explicitly describe 'whereis'
914ncEffectWhereIs :: ProcessId -> String -> NC ()
915ncEffectWhereIs from label = do
917 mPid <- gets (^. registeredHereFor label)
918 liftIO $ sendMessage node
919 (NodeIdentifier (localNodeId node))
920 (ProcessIdentifier from)
921 WithImplicitReconnect
922 (WhereIsReply label mPid)
924-- [Unified: Table 14]
925ncEffectNamedSend :: String -> Message -> NC ()
926ncEffectNamedSend label msg = do
927 mPid <- gets (^. registeredHereFor label)
928 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
929 forM_ mPid $ \pid -> postMessage pid msg
932ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
933ncEffectLocalSend node to msg =
934 liftIO $ withLocalProc node to $ \p -> do
935 enqueue (processQueue p) msg
936 trace node (MxReceived to msg)
939ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
940ncEffectLocalPortSend from msg = do
942 let pid = sendPortProcessId from
943 cid = sendPortLocalId from
944 liftIO $ withLocalProc node pid $ \proc -> do
945 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
947 -- in the unlikely event we know nothing about this channel id,
948 -- there's little to be done - perhaps some logging/tracing though...
950 Just (TypedChannel chan') -> do
951 -- If ch is Nothing, the process has given up the read end of
952 -- the channel and we simply ignore the incoming message - this
953 ch <- deRefWeak chan'
954 forM_ ch $ \chan -> deliverChan msg chan
955 where deliverChan :: forall a . Message -> TQueue a -> IO ()
956 deliverChan (UnencodedMessage _ raw) chan' =
957 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
958 deliverChan (EncodedMessage _ _) _ =
959 -- this will not happen unless someone screws with Primitives.hs
960 error "invalid local channel delivery"
963ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
964ncEffectKill from to reason = do
966 when (isLocal node (ProcessIdentifier to)) $
967 throwException to $ ProcessKillException from reason
970ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
971ncEffectExit from to reason = do
973 when (isLocal node (ProcessIdentifier to)) $
974 throwException to $ ProcessExitException from reason
977ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
978ncEffectGetInfo from pid =
979 let lpid = processLocalId pid
980 them = (ProcessIdentifier pid)
983 mProc <- liftIO $ withValidLocalState (localState node)
984 $ return . (^. localProcessWithId lpid)
986 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
987 from node (ProcessInfoNone DiedUnknownId)
989 itsLinks <- gets (^. linksFor them)
990 itsMons <- gets (^. monitorsFor them)
991 registered <- gets (^. registeredHere)
992 size <- liftIO $ queueSize $ processQueue $ proc
994 let reg = registeredNames registered
995 dispatch (isLocal node (ProcessIdentifier from))
999 infoNode = (processNodeId pid)
1000 , infoRegisteredNames = reg
1001 , infoMessageQueueLength = size
1002 , infoMonitors = Set.toList itsMons
1003 , infoLinks = Set.toList itsLinks
1005 where dispatch :: (Serializable a, Show a)
1011 dispatch True dest _ pInfo = postAsMessage dest $ pInfo
1012 dispatch False dest node pInfo = do
1013 liftIO $ sendMessage node
1014 (NodeIdentifier (localNodeId node))
1015 (ProcessIdentifier dest)
1016 WithImplicitReconnect
1019 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1023ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1024ncEffectGetNodeStats from _nid = do
1026 ncState <- StateT.get
1027 nodeState <- liftIO $ withValidLocalState (localState node) return
1030 nodeStatsNode = localNodeId node
1031 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1032 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1033 , nodeStatsLinks = Map.size $ ncState ^. links
1034 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1036 postAsMessage from stats
1038--------------------------------------------------------------------------------
1040--------------------------------------------------------------------------------
1042notifyDied :: ProcessId -- ^ Who to notify?
1043 -> Identifier -- ^ Who died?
1044 -> DiedReason -- ^ How did they die?
1045 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1047notifyDied dest src reason mRef = do
1049 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1050 (True, Just ref, ProcessIdentifier pid) ->
1051 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1052 (True, Just ref, NodeIdentifier nid) ->
1053 postAsMessage dest $ NodeMonitorNotification ref nid reason
1054 (True, Just ref, SendPortIdentifier cid) ->
1055 postAsMessage dest $ PortMonitorNotification ref cid reason
1056 (True, Nothing, ProcessIdentifier pid) ->
1057 throwException dest $ ProcessLinkException pid reason
1058 (True, Nothing, NodeIdentifier pid) ->
1059 throwException dest $ NodeLinkException pid reason
1060 (True, Nothing, SendPortIdentifier pid) ->
1061 throwException dest $ PortLinkException pid reason
1063 -- The change in sender comes from [Unified: Table 10]
1064 liftIO $ sendBinary node
1065 (NodeIdentifier $ localNodeId node)
1066 (NodeIdentifier $ processNodeId dest)
1067 WithImplicitReconnect
1069 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1070 , ctrlMsgSignal = Died src reason
1073-- | [Unified: Table 8]
1074destNid :: ProcessSignal -> Maybe NodeId
1075destNid (Link ident) = Just $ nodeOf ident
1076destNid (Unlink ident) = Just $ nodeOf ident
1077destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1078destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1079destNid (Spawn _ _) = Nothing
1080destNid (Register _ _ _ _) = Nothing
1081destNid (WhereIs _) = Nothing
1082destNid (NamedSend _ _) = Nothing
1083destNid (UnreliableSend _ _) = Nothing
1084-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1085-- then when a local process dies the monitoring/linking machinery will take
1086-- care of notifying remote nodes
1087destNid (Died _ _) = Nothing
1088destNid (Kill pid _) = Just $ processNodeId pid
1089destNid (Exit pid _) = Just $ processNodeId pid
1090destNid (GetInfo pid) = Just $ processNodeId pid
1091destNid (GetNodeStats nid) = Just nid
1092destNid (LocalSend pid _) = Just $ processNodeId pid
1093destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1094destNid (SigShutdown) = Nothing
1096-- | Check if a process is local to our own node
1097isLocal :: LocalNode -> Identifier -> Bool
1098isLocal nid ident = nodeOf ident == localNodeId nid
1100-- | Lookup a local closure
1101unClosure :: Typeable a => Closure a -> NC (Either String a)
1102unClosure closure = do
1103 rtable <- remoteTable <$> ask
1104 return (Static.unclosure rtable closure)
1106-- | Check if an identifier refers to a valid local object
1107isValidLocalIdentifier :: Identifier -> NC Bool
1108isValidLocalIdentifier ident = do
1110 liftIO . withValidLocalState (localState node) $ \nSt ->
1112 NodeIdentifier nid ->
1113 return $ nid == localNodeId node
1114 ProcessIdentifier pid -> do
1115 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1116 return $ isJust mProc
1117 SendPortIdentifier cid -> do
1118 let pid = sendPortProcessId cid
1119 mProc = nSt ^. localProcessWithId (processLocalId pid)
1121 Nothing -> return False
1122 Just proc -> withMVar (processState proc) $ \pSt -> do
1123 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1126--------------------------------------------------------------------------------
1127-- Messages to local processes --
1128--------------------------------------------------------------------------------
1130postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1131postAsMessage pid = postMessage pid . createUnencodedMessage
1133postMessage :: ProcessId -> Message -> NC ()
1134postMessage pid msg = do
1136 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1138throwException :: Exception e => ProcessId -> e -> NC ()
1139throwException pid e = do
1141 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1143withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1144withLocalProc node pid p =
1145 -- By [Unified: table 6, rule missing_process] messages to dead processes
1146 -- can silently be dropped
1147 let lpid = processLocalId pid in do
1148 withValidLocalState (localState node) $ \vst ->
1149 forM_ (vst ^. localProcessWithId lpid) p
1151--------------------------------------------------------------------------------
1153--------------------------------------------------------------------------------
1155links :: Accessor NCState (Map Identifier (Set ProcessId))
1156links = accessor _links (\ls st -> st { _links = ls })
1158monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1159monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1161registeredHere :: Accessor NCState (Map String ProcessId)
1162registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1164registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1165registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1167linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1168linksFor ident = links >>> DAC.mapDefault Set.empty ident
1170monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1171monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1173registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1174registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1176registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1177registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1179-- | @splitNotif ident@ splits a notifications map into those
1180-- notifications that should trigger when 'ident' fails and those links that
1183-- There is a hierarchy between identifiers: failure of a node implies failure
1184-- of all processes on that node, and failure of a process implies failure of
1185-- all typed channels to that process. In other words, if 'ident' refers to a
1186-- node, then the /should trigger/ set will include
1188-- * the notifications for the node specifically
1189-- * the notifications for processes on that node, and
1190-- * the notifications for typed channels to processes on that node.
1192-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1195-- * the notifications for that process specifically and
1196-- * the notifications for typed channels to that process.
1198-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1199splitNotif :: Identifier
1201 -> (Map Identifier a, Map Identifier a)
1202splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1204--------------------------------------------------------------------------------
1206--------------------------------------------------------------------------------
1208-- | Modify and evaluate the state
1209modify' :: MonadState s m => (s -> s) -> m ()
1210modify' f = StateT.get >>= \s -> StateT.put $! f s
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list