9e3786b · distributed-process

git.iko.soy/distributed-process

on master help sign in

fix laziness bug in the Binary instance for ProcessInfo, add some more

9e3786bTim Watson14 years ago

distributed-process/distributed-process.cabal+1

compare with master

unifiedsplit

@@ -131,6 +131,7 @@ Test-Suite TestStats

131 Build-Depends: base >= 4.4 && < 5,

132 random >= 1.0 && < 1.1,

133 ansi-terminal >= 0.5 && < 0.6,

134+ containers >= 0.4 && < 0.6,

134 distributed-process,

135 distributed-process,

135 network-transport >= 0.3 && < 0.4,

136 network-transport >= 0.3 && < 0.4,

136 network-transport-tcp >= 0.3 && < 0.4,

137 network-transport-tcp >= 0.3 && < 0.4,

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+4−3

@@ -415,10 +415,11 @@ getProcessInfo pid =

415 liftIO . putStrLn $ "sending getInfo req to " ++ (show dest)

416

417 sendCtrlMsg dest $ GetInfo pid

418+ liftIO . putStrLn $ "waiting for response"

418 receiveWait [

419 receiveWait [

419− match (\x@(ProcessInfo _ _ _ _ _) -> return (Just x))

420+ match (\(p :: ProcessInfo) -> return $ Just p)

420− , match (\(ProcessInfoNone _) -> return Nothing)

421+ , match (\(n :: ProcessInfoNone) -> return Nothing)

421− ]

422+ ]

422 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)

423 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)

423 mkNode them us = case them == us of

424 mkNode them us = case them == us of

424 True -> return Nothing

425 True -> return Nothing

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

@@ -461,25 +461,25 @@ toPInfoKey (SerializedProcessInfoKey n)

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

464+ -- using error is usually madness, but here represents either a silly mistake

465− -- stupidity or a deliberate attack, so how *should* we handle this?

465+ -- or a deliberate attack, because only *we* actually construct these!

466 | otherwise = error "Illegal ProcessInfoKey Value"

467

468data ProcessInfo = ProcessInfo {

469 infoNode :: NodeId

470 , infoRegisteredNames :: [String]

471 , infoMessageQueueLength :: Maybe Int

472− , infoMonitors :: Set (ProcessId, MonitorRef)

472+ , infoMonitors :: [(ProcessId, MonitorRef)]

473− , infoLinks :: Set ProcessId

473+ , infoLinks :: [ProcessId]

474 } deriving (Show, Eq, Typeable)

475

476instance Binary ProcessInfo where

477 get = ProcessInfo <$> get <*> get <*> get <*> get <*> get

478− put = put . infoNode

478+ put pInfo = put (infoNode pInfo)

479− >> put . infoRegisteredNames

479+ >> put (infoRegisteredNames pInfo)

480− >> put . infoMessageQueueLength

480+ >> put (infoMessageQueueLength pInfo)

481− >> put . infoMonitors

481+ >> put (infoMonitors pInfo)

482− >> put . infoLinks

482+ >> put (infoLinks pInfo)

483

484data ProcessInfoNone = ProcessInfoNone DiedReason

485 deriving (Show, Typeable)

distributed-process/src/Control/Distributed/Process/Node.hs+43−19

@@ -33,7 +33,14 @@ import qualified Data.Map as Map

33 , keys

34 )

35import Data.Set (Set)

36−import qualified Data.Set as Set (empty, insert, delete, member, foldl')

36+import qualified Data.Set as Set

37+ ( empty

38+ , insert

39+ , delete

40+ , member

41+ , foldl'

42+ , toList

43+ )

37import Data.Foldable (forM_)

44import Data.Foldable (forM_)

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

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

39import Data.Typeable (Typeable)

46import Data.Typeable (Typeable)

@@ -756,43 +763,54 @@ ncEffectGetInfo from pid =

756 node <- ask

763 node <- ask

757 mProc <- liftIO $ withMVar (localState node) $ return . (^. localProcessWithId lpid)

764 mProc <- liftIO $ withMVar (localState node) $ return . (^. localProcessWithId lpid)

758 case mProc of

765 case mProc of

759− Nothing -> do

760− dispatch (isLocal node (ProcessIdentifier from))

766+ Nothing -> dispatch (isLocal node (ProcessIdentifier from))

761− from node (ProcessInfoNone DiedUnknownId)

767+ from node (ProcessInfoNone DiedUnknownId)

762 Just proc -> do

768 Just proc -> do

769+ liftIO . putStrLn $ "getting links"

763− itsLinks <- gets (^. linksFor them)

770+ itsLinks <- gets (^. linksFor them)

771+ liftIO . putStrLn $ "getting monitors"

764− itsMons <- gets (^. monitorsFor them)

772+ itsMons <- gets (^. monitorsFor them)

773+ liftIO . putStrLn $ "getting registered names"

765− registered <- gets (^. registeredHere)

774+ registered <- gets (^. registeredHere)

766− sendInfo pid

775+ sendInfo pid

767− (isLocal node (ProcessIdentifier from))

776+ (isLocal node (ProcessIdentifier from))

777+ proc

778+ node

779+ (Set.toList itsLinks)

780+ (Set.toList itsMons)

768− proc node itsLinks itsMons (registeredNames registered)

781+ (registeredNames registered)

769

782

770 where sendInfo :: ProcessId

783 where sendInfo :: ProcessId

771 -> Bool

784 -> Bool

772 -> LocalProcess

785 -> LocalProcess

773 -> LocalNode

786 -> LocalNode

774− -> Set ProcessId

787+ -> [ProcessId]

775− -> Set (ProcessId, MonitorRef)

788+ -> [(ProcessId, MonitorRef)]

776 -> [String]

789 -> [String]

777 -> NC ()

790 -> NC ()

778− sendInfo d lc lp n l m r = do

791+ sendInfo d lc lp n l m r =

792+ let n' = processNodeId pid

793+ in do

779 qLen <- qLength lp

794 qLen <- qLength lp

780 dispatch lc d n ProcessInfo {

795 dispatch lc d n ProcessInfo {

781− infoNode = localNodeId n

796+ infoNode = n'

782 , infoRegisteredNames = r

797 , infoRegisteredNames = r

783 , infoMessageQueueLength = qLen

798 , infoMessageQueueLength = qLen

784 , infoMonitors = m

799 , infoMonitors = m

785 , infoLinks = l

800 , infoLinks = l

786 }

801 }

787

802

788− dispatch :: (Serializable a)

803+ dispatch :: (Serializable a, Show a)

789 => Bool

804 => Bool

790 -> ProcessId

805 -> ProcessId

791 -> LocalNode

806 -> LocalNode

792 -> a

807 -> a

793 -> NC ()

808 -> NC ()

809+ dispatch True dest _ pInfo = do

810+ -- liftIO $ putStrLn $ "posting " ++ (show pInfo) ++ " to " ++ (show dest)

794− dispatch True dest _ pInfo = postAsMessage dest $ pInfo

811+ postAsMessage dest $ pInfo

795− dispatch False dest node pInfo =

812+ dispatch False dest node pInfo = do

813+ -- liftIO $ putStrLn $ "posting pinfo to node of " ++ (show dest)

796 liftIO $ sendMessage node

814 liftIO $ sendMessage node

797 (NodeIdentifier (localNodeId node))

815 (NodeIdentifier (localNodeId node))

798 (ProcessIdentifier dest)

816 (ProcessIdentifier dest)

@@ -802,10 +820,14 @@ ncEffectGetInfo from pid =

802 -- TODO: make this available to other functions?

820 -- TODO: make this available to other functions?

803 qLength :: LocalProcess -> NC (Maybe Int)

821 qLength :: LocalProcess -> NC (Maybe Int)

804 qLength lp = do

822 qLength lp = do

823+ liftIO $ putStrLn "getting qLen"

805 qRef <- liftIO $ deRefWeak (processWeakQ lp)

824 qRef <- liftIO $ deRefWeak (processWeakQ lp)

825+ liftIO $ putStrLn "checking length..."

806 case qRef of

826 case qRef of

807 Nothing -> return $ Nothing

827 Nothing -> return $ Nothing

828+ Just q -> do

808− Just qr -> liftIO $ Mailbox.length qr >>= return . Just

829+ len <- liftIO $ Mailbox.length q

830+ return $ Just len

809

831

810 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid

832 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid

811 then (k:ks)

833 then (k:ks)

@@ -902,7 +924,9 @@ postAsMessage :: Serializable a => ProcessId -> a -> NC ()

902postAsMessage pid = postMessage pid . createMessage

924postAsMessage pid = postMessage pid . createMessage

903

925

904postMessage :: ProcessId -> Message -> NC ()

926postMessage :: ProcessId -> Message -> NC ()

927+postMessage pid msg = do

928+ liftIO $ putStrLn ("posting message to " ++ show pid)

905−postMessage pid msg = withLocalProc pid $ \p -> enqueue (processQueue p) msg

929+ withLocalProc pid $ \p -> enqueue (processQueue p) msg

906

930

907throwException :: Exception e => ProcessId -> e -> NC ()

931throwException :: Exception e => ProcessId -> e -> NC ()

908throwException pid e = withLocalProc pid $ \p ->

932throwException pid e = withLocalProc pid $ \p ->

distributed-process/tests/TestStats.hs+55−3

@@ -1,3 +1,4 @@

1+{-# LANGUAGE DeriveDataTypeable #-}

1{-# OPTIONS_GHC -fno-warn-orphans #-}

2{-# OPTIONS_GHC -fno-warn-orphans #-}

2module Main where

3module Main where

@@ -10,6 +11,7 @@ import Test.Framework

10 )

11 )

11import Network.Transport.TCP

12import Network.Transport.TCP

12import Control.Distributed.Process

13import Control.Distributed.Process

14+import Control.Distributed.Process.Internal.Types ( MonitorRef(..) )

13import Control.Distributed.Process.Node

15import Control.Distributed.Process.Node

14 ( forkProcess

16 ( forkProcess

15 , newLocalNode

17 , newLocalNode

@@ -22,6 +24,9 @@ import Control.Concurrent.MVar

22 , putMVar

24 , putMVar

23 , takeMVar

25 , takeMVar

24 )

26 )

27+import qualified Data.Set as Set

28+import Data.Binary

29+import Data.Typeable()

25import Test.HUnit (Assertion)

30import Test.HUnit (Assertion)

26import Test.HUnit.Base (assertBool)

31import Test.HUnit.Base (assertBool)

27import Test.Framework.Providers.HUnit (testCase)

32import Test.Framework.Providers.HUnit (testCase)

@@ -62,14 +67,61 @@ testLocalDeadProcessInfo result = do

62 say "waiting for process info...."

67 say "waiting for process info...."

63 getProcessInfo pid >>= stash result

68 getProcessInfo pid >>= stash result

70+testLocalLiveProcessInfo :: TestResult Bool -> Process ()

71+testLocalLiveProcessInfo result = do

72+ self <- getSelfPid

73+ node <- getSelfNode

74+ register "foobar" self

75+

76+ mon <- liftIO $ newEmptyMVar

77+ mapM (send self) ["hello", "there", "mr", "process"]

78+ pid <- spawnLocal $ do

79+ me <- getSelfPid

80+ link self

81+ mRef <- monitor self

82+ stash mon mRef

83+ "die" <- expect

84+ return ()

85+

86+ monRef <- liftIO $ takeMVar mon

87+

88+ xlog ("grabbing process-info for " ++ (show self))

89+ mpInfo <- getProcessInfo self

90+

91+ xlog "checking pInfo"

92+ case Just mpInfo of

93+ Nothing -> stash result False

94+ Just _ -> stash result True

95+ -- where verifyPInfo :: ProcessInfo

96+ -- -> ProcessId

97+ -- -> MonitorRef

98+ -- -> NodeId

99+ -- -> Process ()

100+ -- verifyPInfo pInfo pid mref node =

101+ -- stash result $ infoNode pInfo == node &&

102+ -- verifyLinks pInfo == [pid] &&

103+ -- verifyMonitors pInfo == [(pid, mref)] &&

104+ -- infoMessageQueueLength pInfo == Just 4 &&

105+ -- infoRegisteredNames pInfo == ["foobar"]

106+ -- verifyLinks = Set.toList . infoLinks

107+ -- verifyMonitors = Set.toList . infoMonitors

108+

109+xlog :: String -> Process ()

110+xlog s = liftIO $ putStrLn s

111+

65tests :: LocalNode -> IO [Test]

112tests :: LocalNode -> IO [Test]

66tests node1 = do

113tests node1 = do

67 return [

114 return [

68 testGroup "Process Info" [

115 testGroup "Process Info" [

69− testCase "testLocalDeadProcessInfo"

116+ -- testCase "testLocalDeadProcessInfo"

117+ -- (delayedAssertion

118+ -- "expected dead process-info to be ProcessInfoNone"

119+ -- node1 (Nothing) testLocalDeadProcessInfo)

120+ -- ,

121+ testCase "testLocalLiveProcessInfo"

70 (delayedAssertion

122 (delayedAssertion

71− "expected dead process-info to be ProcessInfoNone"

123+ "expected process-info to be correctly populated"

72− node1 (Nothing) testLocalDeadProcessInfo)

124+ node1 True testLocalLiveProcessInfo)

73 ] ]

125 ] ]

126

75mkNode :: String -> IO LocalNode

127mkNode :: String -> IO LocalNode