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) $ \st -> case st of
327 LocalNodeValid vst -> do
328 forM_ (vst ^. localProcesses) $ \lproc ->
329 -- Semantics of 'throwTo' guarantee that target thread will get
330 -- delivered an exception. Therefore, target thread will be killed
331 -- eventually and that's as good as we can do. No need to wait for
332 -- thread to actually finish dying.
333 killThread (processThread lproc)
334 return LocalNodeClosed
335 LocalNodeClosed -> return LocalNodeClosed
336 NT.closeEndPoint (localEndPoint node)
338-- | Run a process on a local node and wait for it to finish
339runProcess :: LocalNode -> Process () -> IO ()
340runProcess node proc = do
343 void $ forkProcess node $ do
344 catch (proc `finally` liftIO (putMVar done ()))
345 (\(ex :: SomeException) -> liftIO $ throwTo tid ex)
348-- | Spawn a new process on a local node
349forkProcess :: LocalNode -> Process () -> IO ProcessId
350forkProcess node proc =
351 modifyMVarMasked (localState node) startProcess
353 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
354 startProcess (LocalNodeValid vst) = do
355 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
356 , lpidUnique = vst ^. localPidUnique
358 let pid = ProcessId { processNodeId = localNodeId node
359 , processLocalId = lpid
361 pst <- newMVar LocalProcessState { _monitorCounter = 0
363 , _channelCounter = 0
364 , _typedChannels = Map.empty
367 weakQueue <- mkWeakCQueue queue (return ())
368 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
369 let lproc = LocalProcess { processQueue = queue
370 , processWeakQ = weakQueue
373 , processThread = tid
376 tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do
377 reason <- Exception.catches
378 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
379 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
380 mMsg <- unwrapMessage msg :: IO (Maybe String)
382 Nothing -> return $ DiedException $ show ex
383 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
385 (return . DiedException . (show :: SomeException -> String)))]
387 -- [Unified: Table 4, rules termination and exiting]
388 modifyValidLocalState_ node (cleanupProcess pid)
389 writeChan (localCtrlChan node) NCMsg
390 { ctrlMsgSender = ProcessIdentifier pid
391 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
395 -- see note [tracer/forkProcess races]
396 trace node (MxSpawned pid)
398 if lpidCounter lpid == maxBound
400 -- TODO: this doesn't look right at all - how do we know
401 -- that newUnique represents a process id that is available!?
402 newUnique <- randomIO
403 return ( LocalNodeValid
404 $ (localProcessWithId lpid ^= Just lproc)
405 . (localPidCounter ^= firstNonReservedProcessId)
406 . (localPidUnique ^= newUnique)
411 return ( LocalNodeValid
412 $ (localProcessWithId lpid ^= Just lproc)
413 . (localPidCounter ^: (+ 1))
417 startProcess LocalNodeClosed = throwIO $ userError $ "Node closed " ++ show (localNodeId node)
419 cleanupProcess :: ProcessId -> ValidLocalNodeState -> IO ValidLocalNodeState
420 cleanupProcess pid vst = do
421 let pid' = ProcessIdentifier pid
422 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
423 mapM_ (NT.close . fst) (Map.elems affected)
424 return $ (localProcessWithId (processLocalId pid) ^= Nothing)
425 . (localConnections ^= unaffected)
428-- note [tracer/forkProcess races]
430-- Our startTracing function uses forkProcess to start the trace controller
431-- process, and of course forkProcess attempts to call traceEvent once the
432-- process has started. This is harmless, as the localEventBus is not updated
433-- until /after/ the initial forkProcess completes, so the first call to
434-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
437--------------------------------------------------------------------------------
438-- Handle incoming messages --
439--------------------------------------------------------------------------------
441type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
445 | ToProc ProcessId (Weak (CQueue Message))
446 | ToChan TypedChannel
449data ConnectionState = ConnectionState {
450 _incoming :: !(Map NT.ConnectionId IncomingConnection)
451 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
454initConnectionState :: ConnectionState
455initConnectionState = ConnectionState {
456 _incoming = Map.empty
457 , _incomingFrom = Map.empty
460incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
461incoming = accessor _incoming (\conns st -> st { _incoming = conns })
463incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
464incomingAt cid = incoming >>> DAC.mapMaybe cid
466incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
467incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
469 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
471handleIncomingMessages :: LocalNode -> IO ()
472handleIncomingMessages node = go initConnectionState
474 go :: ConnectionState -> IO ()
476 event <- NT.receive endpoint
478 NT.ConnectionOpened cid rel theirAddr ->
479 if rel == NT.ReliableOrdered
481 trace node (MxConnected cid theirAddr)
483 (incomingAt cid ^= Just (theirAddr, Uninit))
484 . (incomingFrom theirAddr ^: Set.insert cid)
487 else invalidRequest cid st
488 NT.Received cid payload ->
489 case st ^. incomingAt cid of
490 Just (_, ToProc pid weakQueue) -> do
491 mQueue <- deRefWeak weakQueue
492 forM_ mQueue $ \queue -> do
493 -- TODO: if we find that the queue is Nothing, should we remove
494 -- it from the NC state? (and same for channels, below)
495 let msg = payloadToMessage payload
496 enqueue queue msg -- 'enqueue' is strict
497 trace node (MxReceived pid msg)
499 Just (_, ToChan (TypedChannel chan')) -> do
500 mChan <- deRefWeak chan'
501 -- If mChan is Nothing, the process has given up the read end of
502 -- the channel and we simply ignore the incoming message
503 forM_ mChan $ \chan -> atomically $
504 -- We make sure the message is fully decoded when it is enqueued
505 writeTQueue chan $! decode (BSL.fromChunks payload)
507 Just (_, ToNode) -> do
508 let ctrlMsg = decode . BSL.fromChunks $ payload
509 writeChan ctrlChan $! ctrlMsg
511 Just (src, Uninit) ->
512 case decode (BSL.fromChunks payload) of
513 ProcessIdentifier pid -> do
514 let lpid = processLocalId pid
515 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
518 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
520 invalidRequest cid st
521 SendPortIdentifier chId -> do
522 let lcid = sendPortLocalId chId
523 lpid = processLocalId (sendPortProcessId chId)
524 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
527 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
530 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
532 invalidRequest cid st
534 invalidRequest cid st
535 NodeIdentifier nid ->
536 if nid == localNodeId node
537 then go (incomingAt cid ^= Just (src, ToNode) $ st)
538 else invalidRequest cid st
540 invalidRequest cid st
541 NT.ConnectionClosed cid ->
542 case st ^. incomingAt cid of
544 invalidRequest cid st
546 trace node (MxDisconnected cid src)
547 go ( (incomingAt cid ^= Nothing)
548 . (incomingFrom src ^: Set.delete cid)
551 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
552 -- [Unified table 9, rule node_disconnect]
553 let nid = NodeIdentifier $ NodeId theirAddr
554 writeChan ctrlChan NCMsg
555 { ctrlMsgSender = nid
556 , ctrlMsgSignal = Died nid DiedDisconnect
558 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
559 closeImplicitReconnections node nid
560 go ( (incomingFrom theirAddr ^= Set.empty)
561 . (incoming ^: Map.filterWithKey (const . notLost))
564 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
565 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
566 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
567 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
570 NT.ReceivedMulticast _ _ ->
571 -- If we received a multicast message, something went horribly wrong
572 -- and we just give up
573 fail "Cloud Haskell fatal error: received unexpected multicast"
575 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
576 invalidRequest cid st = do
577 -- TODO: We should treat this as a fatal error on the part of the remote
578 -- node. That is, we should report the remote node as having died, and we
579 -- should close incoming connections (this requires a Transport layer
581 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
583 go ( incomingAt cid ^= Nothing
587 state = localState node
588 endpoint = localEndPoint node
589 ctrlChan = localCtrlChan node
591--------------------------------------------------------------------------------
592-- Top-level access to the node controller --
593--------------------------------------------------------------------------------
595runNodeController :: LocalNode -> IO ()
597 runReaderT (evalStateT (unNC nodeController) initNCState)
599--------------------------------------------------------------------------------
600-- Internal data types --
601--------------------------------------------------------------------------------
603data NCState = NCState
604 { -- Mapping from remote processes to linked local processes
605 _links :: !(Map Identifier (Set ProcessId))
606 -- Mapping from remote processes to monitoring local processes
607 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
608 -- Process registry: names and where they live, mapped to the PIDs
609 , _registeredHere :: !(Map String ProcessId)
610 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
613newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
614 deriving ( Applicative
619 , MonadReader LocalNode
622initNCState :: NCState
623initNCState = NCState { _links = Map.empty
624 , _monitors = Map.empty
625 , _registeredHere = Map.empty
626 , _registeredOnNodes = Map.empty
629-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
630-- type is deliberately not exported so it cannot be caught explicitly.
631data ProcessKillException =
632 ProcessKillException !ProcessId !String
635instance Exception ProcessKillException
636instance Show ProcessKillException where
637 show (ProcessKillException pid reason) =
638 "killed-by=" ++ show pid ++ ",reason=" ++ reason
640--------------------------------------------------------------------------------
641-- Tracing/Debugging --
642--------------------------------------------------------------------------------
644-- [Issue #104 / DP-13]
646traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
647traceNotifyDied node ident reason =
648 -- TODO: sendPortDied notifications
649 liftIO $ withLocalTracer node $ \t ->
651 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
652 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
655traceEventFmtIO :: LocalNode
659traceEventFmtIO node fmt args =
660 withLocalTracer node $ \t -> traceLogFmt t fmt args
662trace :: LocalNode -> MxEvent -> IO ()
663trace node ev = withLocalTracer node $ \t -> traceEvent t ev
665withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
666withLocalTracer node act = act (localEventBus node)
668--------------------------------------------------------------------------------
669-- Core functionality --
670--------------------------------------------------------------------------------
673nodeController :: NC ()
677 msg <- liftIO $ readChan (localCtrlChan node)
679 -- [Unified: Table 7, rule nc_forward]
680 case destNid (ctrlMsgSignal msg) of
681 Just nid' | nid' /= localNodeId node ->
682 liftIO $ sendBinary node
684 (NodeIdentifier nid')
685 WithImplicitReconnect
691 NCMsg (ProcessIdentifier from) (Link them) ->
692 ncEffectMonitor from them Nothing
693 NCMsg (ProcessIdentifier from) (Monitor ref) ->
694 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
695 NCMsg (ProcessIdentifier from) (Unlink them) ->
696 ncEffectUnlink from them
697 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
698 ncEffectUnmonitor from ref
699 NCMsg _from (Died ident reason) ->
700 ncEffectDied ident reason
701 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
702 ncEffectSpawn from proc ref
703 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
704 ncEffectRegister from label atnode pid force
705 NCMsg (ProcessIdentifier from) (WhereIs label) ->
706 ncEffectWhereIs from label
707 NCMsg _ (NamedSend label msg') ->
708 ncEffectNamedSend label msg'
709 NCMsg _ (UnreliableSend lpid msg') ->
710 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
711 NCMsg _ (LocalSend to msg') ->
712 ncEffectLocalSend node to msg'
713 NCMsg _ (LocalPortSend to msg') ->
714 ncEffectLocalPortSend to msg'
715 NCMsg (ProcessIdentifier from) (Kill to reason) ->
716 ncEffectKill from to reason
717 NCMsg (ProcessIdentifier from) (Exit to reason) ->
718 ncEffectExit from to reason
719 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
720 ncEffectGetInfo from pid
721 NCMsg _ SigShutdown ->
723 NT.closeEndPoint (localEndPoint node)
724 `Exception.finally` throwIO ThreadKilled
725 -- ThreadKilled seems to make more sense than fail/error here
726 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
727 ncEffectGetNodeStats from nid
729 error $ "nodeController: unexpected message " ++ show unexpected
731-- [Unified: Table 10]
732ncEffectMonitor :: ProcessId -- ^ Who's watching?
733 -> Identifier -- ^ Who's being watched?
734 -> Maybe MonitorRef -- ^ 'Nothing' to link
736ncEffectMonitor from them mRef = do
739 if not (isLocal node them)
741 else isValidLocalIdentifier them
742 case (shouldLink, isLocal node (ProcessIdentifier from)) of
743 (True, _) -> -- [Unified: first rule]
745 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
746 Nothing -> modify' $ linksFor them ^: Set.insert from
747 (False, True) -> -- [Unified: second rule]
748 notifyDied from them DiedUnknownId mRef
749 (False, False) -> -- [Unified: third rule]
750 -- TODO: this is the right sender according to the Unified semantics,
751 -- but perhaps having 'them' as the sender would make more sense
752 -- (see also: notifyDied)
753 liftIO $ sendBinary node
754 (NodeIdentifier $ localNodeId node)
755 (NodeIdentifier $ processNodeId from)
756 WithImplicitReconnect
758 { ctrlMsgSender = NodeIdentifier (localNodeId node)
759 , ctrlMsgSignal = Died them DiedUnknownId
762-- [Unified: Table 11]
763ncEffectUnlink :: ProcessId -> Identifier -> NC ()
764ncEffectUnlink from them = do
766 when (isLocal node (ProcessIdentifier from)) $
768 ProcessIdentifier pid ->
769 postAsMessage from $ DidUnlinkProcess pid
770 NodeIdentifier nid ->
771 postAsMessage from $ DidUnlinkNode nid
772 SendPortIdentifier cid ->
773 postAsMessage from $ DidUnlinkPort cid
774 modify' $ linksFor them ^: Set.delete from
776-- [Unified: Table 11]
777ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
778ncEffectUnmonitor from ref = do
780 when (isLocal node (ProcessIdentifier from)) $
781 postAsMessage from $ DidUnmonitor ref
782 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
784-- [Unified: Table 12]
785ncEffectDied :: Identifier -> DiedReason -> NC ()
786ncEffectDied ident reason = do
788 traceNotifyDied node ident reason
789 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
790 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
792-- _registry :: !(Map (String,NodeId) ProcessId)
794 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
796 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
798 when (localOnly <= isLocal node (ProcessIdentifier us)) $
799 notifyDied us them reason Nothing
801 forM_ (Map.toList affectedMons) $ \(them, refs) ->
802 forM_ refs $ \(us, ref) ->
803 when (localOnly <= isLocal node (ProcessIdentifier us)) $
804 notifyDied us them reason (Just ref)
806 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
808 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
810 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
811 mapM (\(pid,nidlist) ->
812 case ident `impliesDeathOf` ProcessIdentifier pid of
814 do forM_ nidlist $ \(nid,_) ->
815 when (not $ isLocal node (NodeIdentifier nid))
816 (forwardNameDeath node nid)
818 False -> return $ Just (pid,nidlist) )
819 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
821 forwardNameDeath node nid =
822 liftIO $ sendBinary node
823 (NodeIdentifier $ localNodeId node)
824 (NodeIdentifier $ nid)
825 WithImplicitReconnect
827 { ctrlMsgSender = NodeIdentifier (localNodeId node)
828 , ctrlMsgSignal = Died ident reason
831-- [Unified: Table 13]
832ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
833ncEffectSpawn pid cProc ref = do
834 mProc <- unClosure cProc
835 -- If the closure does not exist, we spawn a process that throws an exception
836 -- This allows the remote node to find out what's happening
838 let proc = case mProc of
839 Left err -> fail $ "Error: Could not resolve closure: " ++ err
842 pid' <- liftIO $ forkProcess node proc
843 liftIO $ sendMessage node
844 (NodeIdentifier (localNodeId node))
845 (ProcessIdentifier pid)
846 WithImplicitReconnect
849-- Unified semantics does not explicitly describe how to implement 'register',
850-- but mentions it's "very similar to nsend" (Table 14)
851-- We send a response indicated if the operation is invalid
852ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
853ncEffectRegister from label atnode mPid reregistration = do
855 currentVal <- gets (^. registeredHereFor label)
858 Nothing -> -- unregister request
859 return $ isJust currentVal
860 Just thepid -> -- register request
861 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
862 return $ (isNothing currentVal /= reregistration) &&
863 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
864 if isLocal node (NodeIdentifier atnode)
865 then do when (isOk) $
866 do modify' $ registeredHereFor label ^= mPid
867 updateRemote node currentVal mPid
869 (Just p) -> liftIO $ trace node (MxRegistered p label)
870 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
871 liftIO $ sendMessage node
872 (NodeIdentifier (localNodeId node))
873 (ProcessIdentifier from)
874 WithImplicitReconnect
875 (RegisterReply label isOk)
877 case reregistration of
879 False -> flip incList
882 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
883 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
884 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
885 forward node (processNodeId newval) (Register label atnode (Just newval) False)
886 updateRemote node Nothing (Just newval) =
887 forward node (processNodeId newval) (Register label atnode (Just newval) False)
888 updateRemote node (Just oldval) Nothing =
889 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
890 updateRemote _ _ _ = return ()
891 maybeify f Nothing = unmaybeify $ f []
892 maybeify f (Just x) = unmaybeify $ f x
894 unmaybeify [] = Nothing
895 unmaybeify x = Just x
896 incList [] tag = [(tag,1)]
897 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
898 incList (x:xs) tag = x : incList xs tag
900 decList ((atag,1):xs) tag | atag == tag = xs
901 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
902 decList (x:xs) tag = x:decList xs tag
903 forward node to reg =
904 when (not $ isLocal node (NodeIdentifier to)) $
905 liftIO $ sendBinary node
906 (ProcessIdentifier from)
908 WithImplicitReconnect
910 { ctrlMsgSender = ProcessIdentifier from
911 , ctrlMsgSignal = reg
915-- Unified semantics does not explicitly describe 'whereis'
916ncEffectWhereIs :: ProcessId -> String -> NC ()
917ncEffectWhereIs from label = do
919 mPid <- gets (^. registeredHereFor label)
920 liftIO $ sendMessage node
921 (NodeIdentifier (localNodeId node))
922 (ProcessIdentifier from)
923 WithImplicitReconnect
924 (WhereIsReply label mPid)
926-- [Unified: Table 14]
927ncEffectNamedSend :: String -> Message -> NC ()
928ncEffectNamedSend label msg = do
929 mPid <- gets (^. registeredHereFor label)
930 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
931 forM_ mPid $ \pid -> postMessage pid msg
934ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
935ncEffectLocalSend node to msg =
936 liftIO $ withLocalProc node to $ \p -> do
937 enqueue (processQueue p) msg
938 trace node (MxReceived to msg)
941ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
942ncEffectLocalPortSend from msg = do
944 let pid = sendPortProcessId from
945 cid = sendPortLocalId from
946 liftIO $ withLocalProc node pid $ \proc -> do
947 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
949 -- in the unlikely event we know nothing about this channel id,
950 -- there's little to be done - perhaps some logging/tracing though...
952 Just (TypedChannel chan') -> do
953 -- If ch is Nothing, the process has given up the read end of
954 -- the channel and we simply ignore the incoming message - this
955 ch <- deRefWeak chan'
956 forM_ ch $ \chan -> deliverChan msg chan
957 where deliverChan :: forall a . Message -> TQueue a -> IO ()
958 deliverChan (UnencodedMessage _ raw) chan' =
959 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
960 deliverChan (EncodedMessage _ _) _ =
961 -- this will not happen unless someone screws with Primitives.hs
962 error "invalid local channel delivery"
965ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
966ncEffectKill from to reason = do
968 when (isLocal node (ProcessIdentifier to)) $
969 throwException to $ ProcessKillException from reason
972ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
973ncEffectExit from to reason = do
975 when (isLocal node (ProcessIdentifier to)) $
976 throwException to $ ProcessExitException from reason
979ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
980ncEffectGetInfo from pid =
981 let lpid = processLocalId pid
982 them = (ProcessIdentifier pid)
985 mProc <- liftIO $ withValidLocalState node
986 $ return . (^. localProcessWithId lpid)
988 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
989 from node (ProcessInfoNone DiedUnknownId)
991 itsLinks <- gets (^. linksFor them)
992 itsMons <- gets (^. monitorsFor them)
993 registered <- gets (^. registeredHere)
994 size <- liftIO $ queueSize $ processQueue $ proc
996 let reg = registeredNames registered
997 dispatch (isLocal node (ProcessIdentifier from))
1001 infoNode = (processNodeId pid)
1002 , infoRegisteredNames = reg
1003 , infoMessageQueueLength = size
1004 , infoMonitors = Set.toList itsMons
1005 , infoLinks = Set.toList itsLinks
1007 where dispatch :: (Serializable a, Show a)
1013 dispatch True dest _ pInfo = postAsMessage dest $ pInfo
1014 dispatch False dest node pInfo = do
1015 liftIO $ sendMessage node
1016 (NodeIdentifier (localNodeId node))
1017 (ProcessIdentifier dest)
1018 WithImplicitReconnect
1021 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1025ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1026ncEffectGetNodeStats from _nid = do
1028 ncState <- StateT.get
1029 nodeState <- liftIO $ withValidLocalState node return
1032 nodeStatsNode = localNodeId node
1033 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1034 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1035 , nodeStatsLinks = Map.size $ ncState ^. links
1036 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1038 postAsMessage from stats
1040--------------------------------------------------------------------------------
1042--------------------------------------------------------------------------------
1044notifyDied :: ProcessId -- ^ Who to notify?
1045 -> Identifier -- ^ Who died?
1046 -> DiedReason -- ^ How did they die?
1047 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1049notifyDied dest src reason mRef = do
1051 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1052 (True, Just ref, ProcessIdentifier pid) ->
1053 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1054 (True, Just ref, NodeIdentifier nid) ->
1055 postAsMessage dest $ NodeMonitorNotification ref nid reason
1056 (True, Just ref, SendPortIdentifier cid) ->
1057 postAsMessage dest $ PortMonitorNotification ref cid reason
1058 (True, Nothing, ProcessIdentifier pid) ->
1059 throwException dest $ ProcessLinkException pid reason
1060 (True, Nothing, NodeIdentifier pid) ->
1061 throwException dest $ NodeLinkException pid reason
1062 (True, Nothing, SendPortIdentifier pid) ->
1063 throwException dest $ PortLinkException pid reason
1065 -- The change in sender comes from [Unified: Table 10]
1066 liftIO $ sendBinary node
1067 (NodeIdentifier $ localNodeId node)
1068 (NodeIdentifier $ processNodeId dest)
1069 WithImplicitReconnect
1071 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1072 , ctrlMsgSignal = Died src reason
1075-- | [Unified: Table 8]
1076destNid :: ProcessSignal -> Maybe NodeId
1077destNid (Link ident) = Just $ nodeOf ident
1078destNid (Unlink ident) = Just $ nodeOf ident
1079destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1080destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1081destNid (Spawn _ _) = Nothing
1082destNid (Register _ _ _ _) = Nothing
1083destNid (WhereIs _) = Nothing
1084destNid (NamedSend _ _) = Nothing
1085destNid (UnreliableSend _ _) = Nothing
1086-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1087-- then when a local process dies the monitoring/linking machinery will take
1088-- care of notifying remote nodes
1089destNid (Died _ _) = Nothing
1090destNid (Kill pid _) = Just $ processNodeId pid
1091destNid (Exit pid _) = Just $ processNodeId pid
1092destNid (GetInfo pid) = Just $ processNodeId pid
1093destNid (GetNodeStats nid) = Just nid
1094destNid (LocalSend pid _) = Just $ processNodeId pid
1095destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1096destNid (SigShutdown) = Nothing
1098-- | Check if a process is local to our own node
1099isLocal :: LocalNode -> Identifier -> Bool
1100isLocal nid ident = nodeOf ident == localNodeId nid
1102-- | Lookup a local closure
1103unClosure :: Typeable a => Closure a -> NC (Either String a)
1104unClosure closure = do
1105 rtable <- remoteTable <$> ask
1106 return (Static.unclosure rtable closure)
1108-- | Check if an identifier refers to a valid local object
1109isValidLocalIdentifier :: Identifier -> NC Bool
1110isValidLocalIdentifier ident = do
1112 liftIO . withValidLocalState node $ \nSt ->
1114 NodeIdentifier nid ->
1115 return $ nid == localNodeId node
1116 ProcessIdentifier pid -> do
1117 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1118 return $ isJust mProc
1119 SendPortIdentifier cid -> do
1120 let pid = sendPortProcessId cid
1121 mProc = nSt ^. localProcessWithId (processLocalId pid)
1123 Nothing -> return False
1124 Just proc -> withMVar (processState proc) $ \pSt -> do
1125 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1128--------------------------------------------------------------------------------
1129-- Messages to local processes --
1130--------------------------------------------------------------------------------
1132postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1133postAsMessage pid = postMessage pid . createUnencodedMessage
1135postMessage :: ProcessId -> Message -> NC ()
1136postMessage pid msg = do
1138 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1140throwException :: Exception e => ProcessId -> e -> NC ()
1141throwException pid e = do
1143 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1145withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1146withLocalProc node pid p =
1147 -- By [Unified: table 6, rule missing_process] messages to dead processes
1148 -- can silently be dropped
1149 let lpid = processLocalId pid in do
1150 withValidLocalState node $ \vst ->
1151 forM_ (vst ^. localProcessWithId lpid) p
1153--------------------------------------------------------------------------------
1155--------------------------------------------------------------------------------
1157links :: Accessor NCState (Map Identifier (Set ProcessId))
1158links = accessor _links (\ls st -> st { _links = ls })
1160monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1161monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1163registeredHere :: Accessor NCState (Map String ProcessId)
1164registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1166registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1167registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1169linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1170linksFor ident = links >>> DAC.mapDefault Set.empty ident
1172monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1173monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1175registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1176registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1178registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1179registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1181-- | @splitNotif ident@ splits a notifications map into those
1182-- notifications that should trigger when 'ident' fails and those links that
1185-- There is a hierarchy between identifiers: failure of a node implies failure
1186-- of all processes on that node, and failure of a process implies failure of
1187-- all typed channels to that process. In other words, if 'ident' refers to a
1188-- node, then the /should trigger/ set will include
1190-- * the notifications for the node specifically
1191-- * the notifications for processes on that node, and
1192-- * the notifications for typed channels to processes on that node.
1194-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1197-- * the notifications for that process specifically and
1198-- * the notifications for typed channels to that process.
1200-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1201splitNotif :: Identifier
1203 -> (Map Identifier a, Map Identifier a)
1204splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1206--------------------------------------------------------------------------------
1208--------------------------------------------------------------------------------
1210-- | Modify and evaluate the state
1211modify' :: MonadState s m => (s -> s) -> m ()
1212modify' 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