git.iko.soy/distributed-process
on master help sign in
provide process stats (mailbox size, links, monitors, etc) via the NC
commit51a1608f63bb7059ca18fce2722e469958bba98f authorTim Watson <tim@rabbitmq.com> dateSun 16 Dec 2012, 13:09 (14 years ago) parent3fa8671
4 files changed+174−2
compare with master
unifiedsplit
distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+11−1 distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+21 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+71 distributed-process/src/Control/Distributed/Process/Node.hs+71−1
distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+11−1
@@ -5,6 +5,7 @@ module Control.Distributed.Process.Internal.CQueue
55 , BlockSpec(..)
66 , MatchOn(..)
77 , newCQueue
8+ , length
89 , enqueue
910 , dequeue
1011 , mkWeakCQueue
@@ -35,6 +36,9 @@ import Control.Distributed.Process.Internal.StrictList
3536 ( StrictList(..)
3637 , append
3738 )
39+import qualified Control.Distributed.Process.Internal.StrictList as StrictList
40+ ( foldr
41+ )
3842import Data.Maybe (fromJust)
3943import GHC.MVar (MVar(MVar))
4044import GHC.IO (IO(IO))
@@ -48,11 +52,17 @@ data CQueue a = CQueue (StrictMVar (StrictList a)) -- Arrived
4852newCQueue :: IO (CQueue a)
4953newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan
5054
55+length :: CQueue a -> IO Int
56+length (CQueue arrived _) = do
57+ arr <- takeMVar arrived
58+ let len = StrictList.foldr (\_ i -> i+1) 0 arr
59+ return len
60+
5161-- | Enqueue an element
5262--
5363-- Enqueue is strict.
5464enqueue :: CQueue a -> a -> IO ()
55−enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a
65+enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a
5666
5767data BlockSpec =
5868 NonBlocking
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+21
@@ -32,6 +32,8 @@ module Control.Distributed.Process.Internal.Primitives
3232 , ProcessExitException(..)
3333 , getSelfPid
3434 , getSelfNode
35+ , ProcessInfo(..)
36+ , getProcessInfo
3537 -- * Monitoring and linking
3638 , link
3739 , unlink
@@ -97,6 +99,7 @@ import Control.Distributed.Process.Internal.StrictMVar
9799 , modifyMVar
98100 , modifyMVar_
99101 )
102+import Control.Concurrent (ThreadId)
100103import Control.Concurrent.Chan (writeChan)
101104import Control.Concurrent.STM
102105 ( STM
@@ -144,6 +147,8 @@ import Control.Distributed.Process.Internal.Types
144147 , WhereIsReply(..)
145148 , RegisterReply(..)
146149 , ProcessRegistrationException(..)
150+ , ProcessInfo(..)
151+ , ProcessInfoNone(..)
147152 , createMessage
148153 , runLocalProcess
149154 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
@@ -400,6 +405,22 @@ getSelfPid = processId <$> ask
400405getSelfNode :: Process NodeId
401406getSelfNode = localNodeId . processNode <$> ask
402407
408+-- | Get information about the specified process
409+getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
410+getProcessInfo pid =
411+ let them = processNodeId pid in do
412+ us <- getSelfNode
413+ dest <- mkNode them us
414+ sendCtrlMsg dest $ GetInfo pid
415+ receiveWait [
416+ match (\x@(ProcessInfo _ _ _ _ _) -> return (Just x))
417+ , match (\(ProcessInfoNone _) -> return Nothing)
418+ ]
419+ where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
420+ mkNode them us = case them == us of
421+ True -> return Nothing
422+ _ -> return $ Just them
423+
403424--------------------------------------------------------------------------------
404425-- Monitoring and linking --
405426--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+71
@@ -49,7 +49,14 @@ module Control.Distributed.Process.Internal.Types
4949 , DidSpawn(..)
5050 , WhereIsReply(..)
5151 , RegisterReply(..)
52+ , ProcessInfo(..)
53+ , ProcessInfoKey
54+ , ProcessInfoNone(..)
55+ -- * Node controller entities required for primatives
56+ , toMarshalledPInfoKey
57+ , toPInfoKey
5258 -- * Node controller internal data types
59+ , SerializedProcessInfoKey
5360 , NCMsg(..)
5461 , ProcessSignal(..)
5562 -- * Accessors
@@ -68,6 +75,7 @@ module Control.Distributed.Process.Internal.Types
6875
6976import System.Mem.Weak (Weak)
7077import Data.Map (Map)
78+import Data.Set (Set)
7179import Data.Int (Int32)
7280import Data.Typeable (Typeable)
7381import Data.Binary (Binary(put, get), putWord8, getWord8, encode)
@@ -420,6 +428,66 @@ data WhereIsReply = WhereIsReply String (Maybe ProcessId)
420428data RegisterReply = RegisterReply String Bool
421429 deriving (Show, Typeable)
422430
431+data ProcessInfoKey =
432+ ProcessInfoAll
433+ | ProcessInfoNode
434+ | ProcessInfoRegisteredNames
435+ | ProcessInfoMessageQueueLength
436+ | ProcessInfoMonitors
437+ | ProcessInfoLinks
438+ deriving (Show, Typeable, Eq)
439+
440+data SerializedProcessInfoKey = SerializedProcessInfoKey Int
441+
442+instance Binary SerializedProcessInfoKey where
443+ put (SerializedProcessInfoKey n) = put n
444+ get = SerializedProcessInfoKey <$> get
445+
446+-- | Convert 'ProcessInfoKey' to/from 'SerializedProcessInfoKey'
447+toMarshalledPInfoKey :: ProcessInfoKey -> SerializedProcessInfoKey
448+toMarshalledPInfoKey ProcessInfoAll = SerializedProcessInfoKey 1
449+toMarshalledPInfoKey ProcessInfoNode = SerializedProcessInfoKey 2
450+toMarshalledPInfoKey ProcessInfoRegisteredNames = SerializedProcessInfoKey 3
451+toMarshalledPInfoKey ProcessInfoMessageQueueLength = SerializedProcessInfoKey 4
452+toMarshalledPInfoKey ProcessInfoMonitors = SerializedProcessInfoKey 5
453+toMarshalledPInfoKey ProcessInfoLinks = SerializedProcessInfoKey 6
454+
455+-- | Convert 'SerializedProcessInfoKey' back into 'ProcessInfoKey'
456+toPInfoKey :: SerializedProcessInfoKey -> ProcessInfoKey
457+toPInfoKey (SerializedProcessInfoKey n)
458+ | n == 1 = ProcessInfoAll
459+ | n == 2 = ProcessInfoNode
460+ | n == 3 = ProcessInfoRegisteredNames
461+ | n == 4 = ProcessInfoMessageQueueLength
462+ | n == 5 = ProcessInfoMonitors
463+ | n == 6 = ProcessInfoLinks
464+ -- using error is usually madness, but here represents a bit of complete
465+ -- stupidity or a deliberate attack, so how *should* we handle this?
466+ | otherwise = error "Illegal ProcessInfoKey Value"
467+
468+data ProcessInfo = ProcessInfo {
469+ infoNode :: NodeId
470+ , infoRegisteredNames :: [String]
471+ , infoMessageQueueLength :: Maybe Int
472+ , infoMonitors :: Set (ProcessId, MonitorRef)
473+ , infoLinks :: Set ProcessId
474+ } deriving (Show, Typeable)
475+
476+instance Binary ProcessInfo where
477+ get = ProcessInfo <$> get <*> get <*> get <*> get <*> get
478+ put = put . infoNode
479+ >> put . infoRegisteredNames
480+ >> put . infoMessageQueueLength
481+ >> put . infoMonitors
482+ >> put . infoLinks
483+
484+data ProcessInfoNone = ProcessInfoNone DiedReason
485+ deriving (Show, Typeable)
486+
487+instance Binary ProcessInfoNone where
488+ get = ProcessInfoNone <$> get
489+ put (ProcessInfoNone r) = put r
490+
423491--------------------------------------------------------------------------------
424492-- Node controller internal data types --
425493--------------------------------------------------------------------------------
@@ -444,6 +512,7 @@ data ProcessSignal =
444512 | NamedSend !String !Message
445513 | Kill !ProcessId !String
446514 | Exit !ProcessId !Message
515+ | GetInfo !ProcessId
447516 deriving Show
448517
449518--------------------------------------------------------------------------------
@@ -490,6 +559,7 @@ instance Binary ProcessSignal where
490559 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)
491560 put (Kill pid reason) = putWord8 9 >> put pid >> put reason
492561 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)
562+ put (GetInfo about) = putWord8 30 >> put about
493563 get = do
494564 header <- getWord8
495565 case header of
@@ -504,6 +574,7 @@ instance Binary ProcessSignal where
504574 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)
505575 9 -> Kill <$> get <*> get
506576 10 -> Exit <$> get <*> (payloadToMessage <$> get)
577+ 30 -> GetInfo <$> get
507578 _ -> fail "ProcessSignal.get: invalid"
508579
509580instance Binary DiedReason where
distributed-process/src/Control/Distributed/Process/Node.hs+71−1
@@ -29,9 +29,11 @@ import qualified Data.Map as Map
2929 , partitionWithKey
3030 , elems
3131 , filterWithKey
32+ , foldlWithKey
33+ , keys
3234 )
3335import Data.Set (Set)
34−import qualified Data.Set as Set (empty, insert, delete, member)
36+import qualified Data.Set as Set (empty, insert, delete, member, foldl')
3537import Data.Foldable (forM_)
3638import Data.Maybe (isJust, isNothing, catMaybes)
3739import Data.Typeable (Typeable)
@@ -62,6 +64,9 @@ import Control.Distributed.Process.Internal.CQueue
6264 , newCQueue
6365 , mkWeakCQueue
6466 )
67+import qualified Control.Distributed.Process.Internal.CQueue as Mailbox
68+ ( length
69+ )
6570import qualified Network.Transport as NT
6671 ( Transport
6772 , EndPoint
@@ -118,6 +123,7 @@ import Control.Distributed.Process.Internal.Types
118123 , TypedChannel(..)
119124 , Identifier(..)
120125 , nodeOf
126+ , ProcessInfo(..)
121127 , SendPortId(..)
122128 , typedChannelWithId
123129 , RegisterReply(..)
@@ -517,6 +523,8 @@ nodeController = do
517523 ncEffectKill from to reason
518524 NCMsg (ProcessIdentifier from) (Exit to reason) ->
519525 ncEffectExit from to reason
526+ NCMsg (ProcessIdentifier from) (GetInfo pid) ->
527+ ncEffectGetInfo from pid
520528 unexpected ->
521529 error $ "nodeController: unexpected message " ++ show unexpected
522530
@@ -737,6 +745,67 @@ ncEffectExit from to reason = do
737745 when (isLocal node (ProcessIdentifier to)) $
738746 throwException to $ ProcessExitException from reason
739747
748+-- [Issue #89]
749+ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
750+ncEffectGetInfo from pid =
751+ let lpid = processLocalId pid
752+ them = (ProcessIdentifier pid)
753+ in do
754+ node <- ask
755+ mProc <- liftIO $ withMVar (localState node) $ return . (^. localProcessWithId lpid)
756+ case mProc of
757+ Nothing -> return () -- send a response to the caller
758+ Just proc -> do
759+ itsLinks <- gets (^. linksFor them)
760+ itsMons <- gets (^. monitorsFor them)
761+ registered <- gets (^. registeredHere)
762+ sendInfo pid
763+ (isLocal node (ProcessIdentifier from))
764+ proc node itsLinks itsMons (registeredNames registered)
765+
766+ where sendInfo :: ProcessId
767+ -> Bool
768+ -> LocalProcess
769+ -> LocalNode
770+ -> Set ProcessId
771+ -> Set (ProcessId, MonitorRef)
772+ -> [String]
773+ -> NC ()
774+ sendInfo d lc lp n l m r = do
775+ qLen <- qLength lp
776+ dispatch lc d n ProcessInfo {
777+ infoNode = localNodeId n
778+ , infoRegisteredNames = r
779+ , infoMessageQueueLength = qLen
780+ , infoMonitors = m
781+ , infoLinks = l
782+ }
783+
784+ dispatch :: Bool
785+ -> ProcessId
786+ -> LocalNode
787+ -> ProcessInfo
788+ -> NC ()
789+ dispatch True dest _ pInfo = postAsMessage dest $ pInfo
790+ dispatch False dest node pInfo =
791+ liftIO $ sendMessage node
792+ (NodeIdentifier (localNodeId node))
793+ (ProcessIdentifier dest)
794+ WithImplicitReconnect
795+ pInfo
796+
797+ -- TODO: make this available to other functions?
798+ qLength :: LocalProcess -> NC (Maybe Int)
799+ qLength lp = do
800+ qRef <- liftIO $ deRefWeak (processWeakQ lp)
801+ case qRef of
802+ Nothing -> return $ Nothing
803+ Just qr -> liftIO $ Mailbox.length qr >>= return . Just
804+
805+ registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
806+ then (k:ks)
807+ else ks) []
808+
740809--------------------------------------------------------------------------------
741810-- Auxiliary --
742811--------------------------------------------------------------------------------
@@ -788,6 +857,7 @@ destNid (NamedSend _ _) = Nothing
788857destNid (Died _ _) = Nothing
789858destNid (Kill pid _) = Just $ processNodeId pid
790859destNid (Exit pid _) = Just $ processNodeId pid
860+destNid (GetInfo pid) = Just $ processNodeId pid
791861
792862-- | Check if a process is local to our own node
793863isLocal :: LocalNode -> Identifier -> Bool