5ce2927 · distributed-process

git.iko.soy/distributed-process

on master help sign in

ignore mailbox size; remove printf noise; silence warnings

5ce2927Tim Watson14 years ago

distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs−10

compare with master

unifiedsplit

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

5 , BlockSpec(..)

6 , MatchOn(..)

7 , newCQueue

8− , length

9 , enqueue

8 , enqueue

10 , dequeue

9 , dequeue

11 , mkWeakCQueue

10 , mkWeakCQueue

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

36 ( StrictList(..)

35 ( StrictList(..)

37 , append

36 , append

38 )

37 )

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

40− ( foldr

41− )

42import Data.Maybe (fromJust)

38import Data.Maybe (fromJust)

43import GHC.MVar (MVar(MVar))

39import GHC.MVar (MVar(MVar))

44import GHC.IO (IO(IO))

40import GHC.IO (IO(IO))

@@ -52,12 +48,6 @@ data CQueue a = CQueue (StrictMVar (StrictList a)) -- Arrived

52newCQueue :: IO (CQueue a)

48newCQueue :: IO (CQueue a)

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

49newCQueue = 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−

61-- | Enqueue an element

51-- | Enqueue an element

62--

52--

63-- Enqueue is strict.

53-- Enqueue is strict.

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

@@ -412,10 +412,7 @@ getProcessInfo pid =

412 let them = processNodeId pid in do

413 us <- getSelfNode

414 dest <- mkNode them us

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

416−

417 sendCtrlMsg dest $ GetInfo pid

415 sendCtrlMsg dest $ GetInfo pid

418− liftIO . putStrLn $ "waiting for response"

419 receiveWait [

416 receiveWait [

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

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

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

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

distributed-process/src/Control/Distributed/Process/Node.hs+17−56

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

30 , elems

31 , filterWithKey

32 , foldlWithKey

33− , keys

34 )

33 )

35import Data.Set (Set)

34import Data.Set (Set)

36import qualified Data.Set as Set

35import qualified Data.Set as Set

@@ -38,7 +37,6 @@ import qualified Data.Set as Set

38 , insert

37 , insert

39 , delete

38 , delete

40 , member

39 , member

41− , foldl'

42 , toList

40 , toList

43 )

41 )

44import Data.Foldable (forM_)

42import Data.Foldable (forM_)

@@ -71,9 +69,6 @@ import Control.Distributed.Process.Internal.CQueue

71 , newCQueue

69 , newCQueue

72 , mkWeakCQueue

70 , mkWeakCQueue

73 )

71 )

74−import qualified Control.Distributed.Process.Internal.CQueue as Mailbox

75− ( length

76− )

77import qualified Network.Transport as NT

72import qualified Network.Transport as NT

78 ( Transport

73 ( Transport

79 , EndPoint

74 , EndPoint

@@ -759,76 +754,43 @@ ncEffectGetInfo from pid =

759 let lpid = processLocalId pid

754 let lpid = processLocalId pid

760 them = (ProcessIdentifier pid)

755 them = (ProcessIdentifier pid)

761 in do

756 in do

762− liftIO . putStrLn $ "processing get info"

763 node <- ask

757 node <- ask

758+ mProc <- liftIO $

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

759+ withMVar (localState node) $ return . (^. localProcessWithId lpid)

765 case mProc of

760 case mProc of

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

761 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

767 from node (ProcessInfoNone DiedUnknownId)

762 from node (ProcessInfoNone DiedUnknownId)

768− Just proc -> do

763+ Just _ -> do

769− liftIO . putStrLn $ "getting links"

770 itsLinks <- gets (^. linksFor them)

764 itsLinks <- gets (^. linksFor them)

771− liftIO . putStrLn $ "getting monitors"

772 itsMons <- gets (^. monitorsFor them)

765 itsMons <- gets (^. monitorsFor them)

773− liftIO . putStrLn $ "getting registered names"

774 registered <- gets (^. registeredHere)

766 registered <- gets (^. registeredHere)

775− sendInfo pid

776− (isLocal node (ProcessIdentifier from))

777− proc

778− node

779− (Set.toList itsLinks)

780− (Set.toList itsMons)

781− (registeredNames registered)

782−

783− where sendInfo :: ProcessId

784− -> Bool

785− -> LocalProcess

786− -> LocalNode

787− -> [ProcessId]

788− -> [(ProcessId, MonitorRef)]

789− -> [String]

790− -> NC ()

791− sendInfo d lc lp n l m r =

792− let n' = processNodeId pid

793− in do

794− qLen <- qLength lp

795− dispatch lc d n ProcessInfo {

796− infoNode = n'

797− , infoRegisteredNames = r

798− , infoMessageQueueLength = qLen

799− , infoMonitors = m

800− , infoLinks = l

801− }

802

767

768+ let reg = registeredNames registered

769+ dispatch (isLocal node (ProcessIdentifier from))

770+ pid

771+ node

772+ ProcessInfo {

773+ infoNode = (processNodeId pid)

774+ , infoRegisteredNames = reg

775+ -- we cannot populate this field

776+ , infoMessageQueueLength = Nothing

777+ , infoMonitors = Set.toList itsMons

778+ , infoLinks = Set.toList itsLinks

779+ }

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

780+ where dispatch :: (Serializable a, Show a)

804 => Bool

781 => Bool

805 -> ProcessId

782 -> ProcessId

806 -> LocalNode

783 -> LocalNode

807 -> a

784 -> a

808 -> NC ()

785 -> NC ()

809− dispatch True dest _ pInfo = do

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

811− postAsMessage dest $ pInfo

786+ dispatch True dest _ pInfo = postAsMessage dest $ pInfo

812 dispatch False dest node pInfo = do

787 dispatch False dest node pInfo = do

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

814 liftIO $ sendMessage node

788 liftIO $ sendMessage node

815 (NodeIdentifier (localNodeId node))

789 (NodeIdentifier (localNodeId node))

816 (ProcessIdentifier dest)

790 (ProcessIdentifier dest)

817 WithImplicitReconnect

791 WithImplicitReconnect

818 pInfo

792 pInfo

819

793

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

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

822− qLength lp = do

823− liftIO $ putStrLn "getting qLen"

824− qRef <- liftIO $ deRefWeak (processWeakQ lp)

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

826− case qRef of

827− Nothing -> return $ Nothing

828− Just q -> do

829− len <- liftIO $ Mailbox.length q

830− return $ Just len

831−

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

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

833 then (k:ks)

795 then (k:ks)

834 else ks) []

796 else ks) []

@@ -925,7 +887,6 @@ postAsMessage pid = postMessage pid . createMessage

925

887

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

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

927postMessage pid msg = do

889postMessage pid msg = do

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

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

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

930

891

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

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

distributed-process/tests/TestStats.hs+21−34

@@ -3,7 +3,6 @@

3module Main where

5import Prelude hiding (catch)

6−import qualified Network.Transport as NT

7import Test.Framework

6import Test.Framework

8 ( Test

7 ( Test

9 , defaultMain

8 , defaultMain

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

11 )

10 )

12import Network.Transport.TCP

11import Network.Transport.TCP

13import Control.Distributed.Process

12import Control.Distributed.Process

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

15import Control.Distributed.Process.Node

13import Control.Distributed.Process.Node

16 ( forkProcess

14 ( forkProcess

17 , newLocalNode

15 , newLocalNode

@@ -24,8 +22,7 @@ import Control.Concurrent.MVar

24 , putMVar

22 , putMVar

25 , takeMVar

23 , takeMVar

26 )

24 )

27−import qualified Data.Set as Set

28−import Data.Binary

25+import Data.Binary()

29import Data.Typeable()

26import Data.Typeable()

30import Test.HUnit (Assertion)

27import Test.HUnit (Assertion)

31import Test.HUnit.Base (assertBool)

28import Test.HUnit.Base (assertBool)

@@ -64,7 +61,6 @@ testLocalDeadProcessInfo result = do

64 ref' == mref && pid' == pid && r == DiedNormal)

61 ref' == mref && pid' == pid && r == DiedNormal)

65 (\p -> return p)

62 (\p -> return p)

66 ]

63 ]

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

68 getProcessInfo pid >>= stash result

64 getProcessInfo pid >>= stash result

70testLocalLiveProcessInfo :: TestResult Bool -> Process ()

66testLocalLiveProcessInfo :: TestResult Bool -> Process ()

@@ -74,9 +70,9 @@ testLocalLiveProcessInfo result = do

74 register "foobar" self

70 register "foobar" self

76 mon <- liftIO $ newEmptyMVar

72 mon <- liftIO $ newEmptyMVar

73+ -- TODO: we can't get the mailbox's length

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

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

78 pid <- spawnLocal $ do

75 pid <- spawnLocal $ do

79− me <- getSelfPid

80 link self

76 link self

81 mRef <- monitor self

77 mRef <- monitor self

82 stash mon mRef

78 stash mon mRef

@@ -85,40 +81,31 @@ testLocalLiveProcessInfo result = do

86 monRef <- liftIO $ takeMVar mon

82 monRef <- liftIO $ takeMVar mon

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

89 mpInfo <- getProcessInfo self

84 mpInfo <- getProcessInfo self

90−

91− xlog "checking pInfo"

92− case Just mpInfo of

85+ case mpInfo of

93 Nothing -> stash result False

86 Nothing -> stash result False

94− Just _ -> stash result True

87+ Just p -> verifyPInfo p pid monRef node

95− -- where verifyPInfo :: ProcessInfo

88+ where verifyPInfo :: ProcessInfo

96− -- -> ProcessId

89+ -> ProcessId

97− -- -> MonitorRef

90+ -> MonitorRef

98− -- -> NodeId

91+ -> NodeId

99− -- -> Process ()

92+ -> Process ()

100− -- verifyPInfo pInfo pid mref node =

93+ verifyPInfo pInfo pid mref node =

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

94+ stash result $ infoNode pInfo == node &&

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

95+ infoLinks pInfo == [pid] &&

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

96+ infoMonitors pInfo == [(pid, mref)] &&

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

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

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

98+ 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

112tests :: LocalNode -> IO [Test]

100tests :: LocalNode -> IO [Test]

113tests node1 = do

101tests node1 = do

114 return [

102 return [

115 testGroup "Process Info" [

103 testGroup "Process Info" [

116− -- testCase "testLocalDeadProcessInfo"

104+ testCase "testLocalDeadProcessInfo"

117− -- (delayedAssertion

105+ (delayedAssertion

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

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

119− -- node1 (Nothing) testLocalDeadProcessInfo)

107+ node1 (Nothing) testLocalDeadProcessInfo)

120− -- ,

121− testCase "testLocalLiveProcessInfo"

108+ , testCase "testLocalLiveProcessInfo"

122 (delayedAssertion

109 (delayedAssertion

123 "expected process-info to be correctly populated"

110 "expected process-info to be correctly populated"

124 node1 True testLocalLiveProcessInfo)

111 node1 True testLocalLiveProcessInfo)