51a1608 · distributed-process

git.iko.soy/distributed-process

on master help sign in

provide process stats (mailbox size, links, monitors, etc) via the NC

51a1608Tim Watson14 years ago

distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+11−1

compare with master

unifiedsplit

@@ -5,6 +5,7 @@ module Control.Distributed.Process.Internal.CQueue

5 , BlockSpec(..)

6 , MatchOn(..)

7 , newCQueue

8+ , length

8 , enqueue

9 , enqueue

9 , dequeue

10 , dequeue

10 , mkWeakCQueue

11 , mkWeakCQueue

@@ -35,6 +36,9 @@ import Control.Distributed.Process.Internal.StrictList

35 ( StrictList(..)

36 ( StrictList(..)

36 , append

37 , append

37 )

38 )

39+import qualified Control.Distributed.Process.Internal.StrictList as StrictList

40+ ( foldr

41+ )

38import Data.Maybe (fromJust)

42import Data.Maybe (fromJust)

39import GHC.MVar (MVar(MVar))

43import GHC.MVar (MVar(MVar))

40import GHC.IO (IO(IO))

44import GHC.IO (IO(IO))

@@ -48,11 +52,17 @@ data CQueue a = CQueue (StrictMVar (StrictList a)) -- Arrived

48newCQueue :: IO (CQueue a)

52newCQueue :: IO (CQueue a)

49newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan

53newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan

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+

51-- | Enqueue an element

61-- | Enqueue an element

52--

62--

53-- Enqueue is strict.

63-- Enqueue is strict.

54enqueue :: CQueue a -> a -> IO ()

64enqueue :: CQueue a -> a -> IO ()

55−enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a

65+enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a

57data BlockSpec =

67data BlockSpec =

58 NonBlocking

68 NonBlocking

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+21

@@ -32,6 +32,8 @@ module Control.Distributed.Process.Internal.Primitives

32 , ProcessExitException(..)

33 , getSelfPid

34 , getSelfNode

35+ , ProcessInfo(..)

36+ , getProcessInfo

35 -- * Monitoring and linking

37 -- * Monitoring and linking

36 , link

38 , link

37 , unlink

39 , unlink

@@ -97,6 +99,7 @@ import Control.Distributed.Process.Internal.StrictMVar

97 , modifyMVar

99 , modifyMVar

98 , modifyMVar_

100 , modifyMVar_

99 )

101 )

102+import Control.Concurrent (ThreadId)

100import Control.Concurrent.Chan (writeChan)

103import Control.Concurrent.Chan (writeChan)

101import Control.Concurrent.STM

104import Control.Concurrent.STM

102 ( STM

105 ( STM

@@ -144,6 +147,8 @@ import Control.Distributed.Process.Internal.Types

144 , WhereIsReply(..)

147 , WhereIsReply(..)

145 , RegisterReply(..)

148 , RegisterReply(..)

146 , ProcessRegistrationException(..)

149 , ProcessRegistrationException(..)

150+ , ProcessInfo(..)

151+ , ProcessInfoNone(..)

147 , createMessage

152 , createMessage

148 , runLocalProcess

153 , runLocalProcess

149 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

154 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

@@ -400,6 +405,22 @@ getSelfPid = processId <$> ask

400getSelfNode :: Process NodeId

405getSelfNode :: Process NodeId

401getSelfNode = localNodeId . processNode <$> ask

406getSelfNode = localNodeId . processNode <$> ask

402

407

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+

403--------------------------------------------------------------------------------

424--------------------------------------------------------------------------------

404-- Monitoring and linking --

425-- Monitoring and linking --

405--------------------------------------------------------------------------------

426--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+71

@@ -49,7 +49,14 @@ module Control.Distributed.Process.Internal.Types

49 , DidSpawn(..)

50 , WhereIsReply(..)

51 , RegisterReply(..)

52+ , ProcessInfo(..)

53+ , ProcessInfoKey

54+ , ProcessInfoNone(..)

55+ -- * Node controller entities required for primatives

56+ , toMarshalledPInfoKey

57+ , toPInfoKey

52 -- * Node controller internal data types

58 -- * Node controller internal data types

59+ , SerializedProcessInfoKey

53 , NCMsg(..)

60 , NCMsg(..)

54 , ProcessSignal(..)

61 , ProcessSignal(..)

55 -- * Accessors

62 -- * Accessors

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

69import System.Mem.Weak (Weak)

76import System.Mem.Weak (Weak)

70import Data.Map (Map)

77import Data.Map (Map)

78+import Data.Set (Set)

71import Data.Int (Int32)

79import Data.Int (Int32)

72import Data.Typeable (Typeable)

80import Data.Typeable (Typeable)

73import Data.Binary (Binary(put, get), putWord8, getWord8, encode)

81import Data.Binary (Binary(put, get), putWord8, getWord8, encode)

@@ -420,6 +428,66 @@ data WhereIsReply = WhereIsReply String (Maybe ProcessId)

420data RegisterReply = RegisterReply String Bool

428data RegisterReply = RegisterReply String Bool

421 deriving (Show, Typeable)

429 deriving (Show, Typeable)

422

430

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+

423--------------------------------------------------------------------------------

491--------------------------------------------------------------------------------

424-- Node controller internal data types --

492-- Node controller internal data types --

425--------------------------------------------------------------------------------

493--------------------------------------------------------------------------------

@@ -444,6 +512,7 @@ data ProcessSignal =

444 | NamedSend !String !Message

512 | NamedSend !String !Message

445 | Kill !ProcessId !String

513 | Kill !ProcessId !String

446 | Exit !ProcessId !Message

514 | Exit !ProcessId !Message

515+ | GetInfo !ProcessId

447 deriving Show

516 deriving Show

448

517

449--------------------------------------------------------------------------------

518--------------------------------------------------------------------------------

@@ -490,6 +559,7 @@ instance Binary ProcessSignal where

490 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

559 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

491 put (Kill pid reason) = putWord8 9 >> put pid >> put reason

560 put (Kill pid reason) = putWord8 9 >> put pid >> put reason

492 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

561 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

562+ put (GetInfo about) = putWord8 30 >> put about

493 get = do

563 get = do

494 header <- getWord8

564 header <- getWord8

495 case header of

565 case header of

@@ -504,6 +574,7 @@ instance Binary ProcessSignal where

504 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

574 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

505 9 -> Kill <$> get <*> get

575 9 -> Kill <$> get <*> get

506 10 -> Exit <$> get <*> (payloadToMessage <$> get)

576 10 -> Exit <$> get <*> (payloadToMessage <$> get)

577+ 30 -> GetInfo <$> get

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

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

508

579

509instance Binary DiedReason where

580instance Binary DiedReason where

distributed-process/src/Control/Distributed/Process/Node.hs+71−1

@@ -29,9 +29,11 @@ import qualified Data.Map as Map

29 , partitionWithKey

30 , elems

31 , filterWithKey

32+ , foldlWithKey

33+ , keys

32 )

34 )

33import Data.Set (Set)

35import 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')

35import Data.Foldable (forM_)

37import Data.Foldable (forM_)

36import Data.Maybe (isJust, isNothing, catMaybes)

38import Data.Maybe (isJust, isNothing, catMaybes)

37import Data.Typeable (Typeable)

39import Data.Typeable (Typeable)

@@ -62,6 +64,9 @@ import Control.Distributed.Process.Internal.CQueue

62 , newCQueue

64 , newCQueue

63 , mkWeakCQueue

65 , mkWeakCQueue

64 )

66 )

67+import qualified Control.Distributed.Process.Internal.CQueue as Mailbox

68+ ( length

69+ )

65import qualified Network.Transport as NT

70import qualified Network.Transport as NT

66 ( Transport

71 ( Transport

67 , EndPoint

72 , EndPoint

@@ -118,6 +123,7 @@ import Control.Distributed.Process.Internal.Types

118 , TypedChannel(..)

123 , TypedChannel(..)

119 , Identifier(..)

124 , Identifier(..)

120 , nodeOf

125 , nodeOf

126+ , ProcessInfo(..)

121 , SendPortId(..)

127 , SendPortId(..)

122 , typedChannelWithId

128 , typedChannelWithId

123 , RegisterReply(..)

129 , RegisterReply(..)

@@ -517,6 +523,8 @@ nodeController = do

517 ncEffectKill from to reason

523 ncEffectKill from to reason

518 NCMsg (ProcessIdentifier from) (Exit to reason) ->

524 NCMsg (ProcessIdentifier from) (Exit to reason) ->

519 ncEffectExit from to reason

525 ncEffectExit from to reason

526+ NCMsg (ProcessIdentifier from) (GetInfo pid) ->

527+ ncEffectGetInfo from pid

520 unexpected ->

528 unexpected ->

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

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

522

530

@@ -737,6 +745,67 @@ ncEffectExit from to reason = do

737 when (isLocal node (ProcessIdentifier to)) $

745 when (isLocal node (ProcessIdentifier to)) $

738 throwException to $ ProcessExitException from reason

746 throwException to $ ProcessExitException from reason

739

747

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+

740--------------------------------------------------------------------------------

809--------------------------------------------------------------------------------

741-- Auxiliary --

810-- Auxiliary --

742--------------------------------------------------------------------------------

811--------------------------------------------------------------------------------

@@ -788,6 +857,7 @@ destNid (NamedSend _ _) = Nothing

788destNid (Died _ _) = Nothing

857destNid (Died _ _) = Nothing

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

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

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

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

860+destNid (GetInfo pid) = Just $ processNodeId pid

791

861

792-- | Check if a process is local to our own node

862-- | Check if a process is local to our own node

793isLocal :: LocalNode -> Identifier -> Bool

863isLocal :: LocalNode -> Identifier -> Bool