ea1b3b8 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Similar to getProcessInfo, these give developers a means to query node

ea1b3b8Roman Gonzalez13 years ago

src/Control/Distributed/Process.hs+6−1

compare with master

unifiedsplit

@@ -82,6 +82,9 @@ module Control.Distributed.Process

82 , getSelfNode

83 , ProcessInfo(..)

84 , getProcessInfo

85+ , NodeStats(..)

86+ , getNodeStats

87+ , getLocalNodeStats

85 -- * Monitoring and linking

88 -- * Monitoring and linking

86 , link

89 , link

87 , linkNode

90 , linkNode

@@ -237,6 +240,9 @@ import Control.Distributed.Process.Internal.Primitives

237 , getSelfNode

240 , getSelfNode

238 , ProcessInfo(..)

241 , ProcessInfo(..)

239 , getProcessInfo

242 , getProcessInfo

243+ , NodeStats(..)

244+ , getNodeStats

245+ , getLocalNodeStats

240 -- Monitoring and linking

246 -- Monitoring and linking

241 , link

247 , link

242 , linkNode

248 , linkNode

@@ -383,4 +389,3 @@ spawnChannelLocal proc = do

383 liftIO $ putMVar mvar sport

389 liftIO $ putMVar mvar sport

384 proc rport

390 proc rport

385 takeMVar mvar

391 takeMVar mvar

386−

src/Control/Distributed/Process/Internal/Primitives.hs+31

@@ -58,6 +58,9 @@ module Control.Distributed.Process.Internal.Primitives

58 , getSelfNode

59 , ProcessInfo(..)

60 , getProcessInfo

61+ , NodeStats(..)

62+ , getNodeStats

63+ , getLocalNodeStats

61 -- * Monitoring and linking

64 -- * Monitoring and linking

62 , link

65 , link

63 , unlink

66 , unlink

@@ -160,6 +163,7 @@ import Control.Distributed.Process.Internal.Types

160 , MonitorRef(..)

163 , MonitorRef(..)

161 , SpawnRef(..)

164 , SpawnRef(..)

162 , ProcessSignal(..)

165 , ProcessSignal(..)

166+ , NodeMonitorNotification(..)

163 , monitorCounter

167 , monitorCounter

164 , spawnCounter

168 , spawnCounter

165 , SendPort(..)

169 , SendPort(..)

@@ -170,6 +174,7 @@ import Control.Distributed.Process.Internal.Types

170 , SendPortId(..)

174 , SendPortId(..)

171 , Identifier(..)

175 , Identifier(..)

172 , ProcessExitException(..)

176 , ProcessExitException(..)

177+ , DiedReason(..)

173 , DidUnmonitor(..)

178 , DidUnmonitor(..)

174 , DidUnlinkProcess(..)

179 , DidUnlinkProcess(..)

175 , DidUnlinkNode(..)

180 , DidUnlinkNode(..)

@@ -179,6 +184,7 @@ import Control.Distributed.Process.Internal.Types

179 , ProcessRegistrationException(..)

184 , ProcessRegistrationException(..)

180 , ProcessInfo(..)

185 , ProcessInfo(..)

181 , ProcessInfoNone(..)

186 , ProcessInfoNone(..)

187+ , NodeStats(..)

182 , isEncoded

188 , isEncoded

183 , createMessage

189 , createMessage

184 , createUnencodedMessage

190 , createUnencodedMessage

@@ -719,6 +725,31 @@ getSelfPid = processId <$> ask

719getSelfNode :: Process NodeId

725getSelfNode :: Process NodeId

720getSelfNode = localNodeId . processNode <$> ask

726getSelfNode = localNodeId . processNode <$> ask

721

727

728+

729+-- | Get statistics about the specified node

730+getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)

731+getNodeStats nid = do

732+ selfNode <- getSelfNode

733+ if nid == selfNode

734+ then Right `fmap` getLocalNodeStats -- optimisation

735+ else getNodeStatsRemote nid

736+ where

737+ getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)

738+ getNodeStatsRemote nid = do

739+ sendCtrlMsg (Just nid) $ GetNodeStats nid

740+ bracket (monitorNode nid) unmonitor $ \mRef ->

741+ receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)

742+ , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)

743+ (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)

744+ ]

745+

746+-- | Get statistics about our local node

747+getLocalNodeStats :: Process NodeStats

748+getLocalNodeStats = do

749+ self <- getSelfNode

750+ sendCtrlMsg Nothing $ GetNodeStats self

751+ receiveWait [ match (\(stats :: NodeStats) -> return stats) ]

752+

722-- | Get information about the specified process

753-- | Get information about the specified process

723getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)

754getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)

724getProcessInfo pid =

755getProcessInfo pid =

src/Control/Distributed/Process/Internal/Types.hs+21

@@ -59,6 +59,7 @@ module Control.Distributed.Process.Internal.Types

59 , RegisterReply(..)

60 , ProcessInfo(..)

61 , ProcessInfoNone(..)

62+ , NodeStats(..)

62 -- * Node controller internal data types

63 -- * Node controller internal data types

63 , NCMsg(..)

64 , NCMsg(..)

64 , ProcessSignal(..)

65 , ProcessSignal(..)

@@ -522,6 +523,15 @@ data WhereIsReply = WhereIsReply String (Maybe ProcessId)

522data RegisterReply = RegisterReply String Bool

523data RegisterReply = RegisterReply String Bool

523 deriving (Show, Typeable)

524 deriving (Show, Typeable)

524

525

526+data NodeStats = NodeStats {

527+ nodeStatsNode :: NodeId

528+ , nodeStatsRegisteredNames :: Int

529+ , nodeStatsMonitors :: Int

530+ , nodeStatsLinks :: Int

531+ , nodeStatsProcesses :: Int

532+ }

533+ deriving (Show, Eq, Typeable)

534+

525-- | Provide information about a running process

535-- | Provide information about a running process

526data ProcessInfo = ProcessInfo {

536data ProcessInfo = ProcessInfo {

527 infoNode :: NodeId

537 infoNode :: NodeId

@@ -562,6 +572,7 @@ data ProcessSignal =

562 | Exit !ProcessId !Message

572 | Exit !ProcessId !Message

563 | GetInfo !ProcessId

573 | GetInfo !ProcessId

564 | SigShutdown

574 | SigShutdown

575+ | GetNodeStats !NodeId

565 deriving Show

576 deriving Show

566

577

567--------------------------------------------------------------------------------

578--------------------------------------------------------------------------------

@@ -612,6 +623,7 @@ instance Binary ProcessSignal where

612 put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)

623 put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)

613 put (GetInfo about) = putWord8 30 >> put about

624 put (GetInfo about) = putWord8 30 >> put about

614 put (SigShutdown) = putWord8 31

625 put (SigShutdown) = putWord8 31

626+ put (GetNodeStats nid) = putWord8 32 >> put nid

615 get = do

627 get = do

616 header <- getWord8

628 header <- getWord8

617 case header of

629 case header of

@@ -630,6 +642,7 @@ instance Binary ProcessSignal where

630 12 -> LocalPortSend <$> get <*> (payloadToMessage <$> get)

642 12 -> LocalPortSend <$> get <*> (payloadToMessage <$> get)

631 30 -> GetInfo <$> get

643 30 -> GetInfo <$> get

632 31 -> return SigShutdown

644 31 -> return SigShutdown

645+ 32 -> GetNodeStats <$> get

633 _ -> fail "ProcessSignal.get: invalid"

646 _ -> fail "ProcessSignal.get: invalid"

634

647

635instance Binary DiedReason where

648instance Binary DiedReason where

@@ -687,6 +700,14 @@ instance Binary ProcessInfo where

687 >> put (infoMonitors pInfo)

700 >> put (infoMonitors pInfo)

688 >> put (infoLinks pInfo)

701 >> put (infoLinks pInfo)

689

702

703+instance Binary NodeStats where

704+ get = NodeStats <$> get <*> get <*> get <*> get <*> get

705+ put nStats = put (nodeStatsNode nStats)

706+ >> put (nodeStatsRegisteredNames nStats)

707+ >> put (nodeStatsMonitors nStats)

708+ >> put (nodeStatsLinks nStats)

709+ >> put (nodeStatsProcesses nStats)

710+

690instance Binary ProcessInfoNone where

711instance Binary ProcessInfoNone where

691 get = ProcessInfoNone <$> get

712 get = ProcessInfoNone <$> get

692 put (ProcessInfoNone r) = put r

713 put (ProcessInfoNone r) = put r

src/Control/Distributed/Process/Node.hs+22

@@ -30,6 +30,7 @@ import qualified Data.Map as Map

30 , filter

31 , partitionWithKey

32 , elems

33+ , size

33 , filterWithKey

34 , filterWithKey

34 , foldlWithKey

35 , foldlWithKey

35 )

36 )

@@ -68,6 +69,7 @@ import Control.Distributed.Process.Internal.StrictMVar

68 , newEmptyMVar

69 , newEmptyMVar

69 , putMVar

70 , putMVar

70 , takeMVar

71 , takeMVar

72+ , readMVar

71 )

73 )

72import Control.Concurrent.Chan (newChan, writeChan, readChan)

74import Control.Concurrent.Chan (newChan, writeChan, readChan)

73import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)

75import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)

@@ -119,6 +121,7 @@ import Control.Distributed.Process.Internal.Types

119 , localPidCounter

121 , localPidCounter

120 , localPidUnique

122 , localPidUnique

121 , localProcessWithId

123 , localProcessWithId

124+ , localProcesses

122 , localConnections

125 , localConnections

123 , forever'

126 , forever'

124 , MonitorRef(..)

127 , MonitorRef(..)

@@ -141,6 +144,7 @@ import Control.Distributed.Process.Internal.Types

141 , nodeOf

144 , nodeOf

142 , ProcessInfo(..)

145 , ProcessInfo(..)

143 , ProcessInfoNone(..)

146 , ProcessInfoNone(..)

147+ , NodeStats(..)

144 , SendPortId(..)

148 , SendPortId(..)

145 , typedChannelWithId

149 , typedChannelWithId

146 , RegisterReply(..)

150 , RegisterReply(..)

@@ -686,6 +690,8 @@ nodeController = do

686 NT.closeEndPoint (localEndPoint node)

690 NT.closeEndPoint (localEndPoint node)

687 `Exception.finally` throwIO ThreadKilled

691 `Exception.finally` throwIO ThreadKilled

688 -- ThreadKilled seems to make more sense than fail/error here

692 -- ThreadKilled seems to make more sense than fail/error here

693+ NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->

694+ ncEffectGetNodeStats from nid

689 unexpected ->

695 unexpected ->

690 error $ "nodeController: unexpected message " ++ show unexpected

696 error $ "nodeController: unexpected message " ++ show unexpected

691

697

@@ -983,6 +989,21 @@ ncEffectGetInfo from pid =

983 then (k:ks)

989 then (k:ks)

984 else ks) []

990 else ks) []

985

991

992+ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()

993+ncEffectGetNodeStats from _nid = do

994+ node <- ask

995+ ncState <- StateT.get

996+ nodeState <- liftIO $ readMVar (localState node)

997+ let localProcesses' = nodeState ^. localProcesses

998+ stats =

999+ NodeStats {

1000+ nodeStatsNode = localNodeId node

1001+ , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere

1002+ , nodeStatsMonitors = Map.size $ ncState ^. monitors

1003+ , nodeStatsLinks = Map.size $ ncState ^. links

1004+ , nodeStatsProcesses = Map.size localProcesses'

1005+ }

1006+ postAsMessage from stats

986--------------------------------------------------------------------------------

1007--------------------------------------------------------------------------------

987-- Auxiliary --

1008-- Auxiliary --

988--------------------------------------------------------------------------------

1009--------------------------------------------------------------------------------

@@ -1035,6 +1056,7 @@ destNid (Died _ _) = Nothing

1035destNid (Kill pid _) = Just $ processNodeId pid

1056destNid (Kill pid _) = Just $ processNodeId pid

1036destNid (Exit pid _) = Just $ processNodeId pid

1057destNid (Exit pid _) = Just $ processNodeId pid

1037destNid (GetInfo pid) = Just $ processNodeId pid

1058destNid (GetInfo pid) = Just $ processNodeId pid

1059+destNid (GetNodeStats nid) = Just nid

1038destNid (LocalSend pid _) = Just $ processNodeId pid

1060destNid (LocalSend pid _) = Just $ processNodeId pid

1039destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)

1061destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)

1040destNid (SigShutdown) = Nothing

1062destNid (SigShutdown) = Nothing