905b487 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Improve the management data/properties API

905b487Tim Watson13 years ago

src/Control/Distributed/Process/Internal/Primitives.hs+1−1

compare with master

unifiedsplit

@@ -553,7 +553,7 @@ delegate pid p = do

553 ]

554 delegate pid p

555

556−-- | A straight relay that simply forwards all messages to the supplied pid.

556+-- | A straight relay that forwards all messages to the supplied pid.

557relay :: ProcessId -> Process ()

558relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid

559

src/Control/Distributed/Process/Management.hs+27−13

@@ -55,11 +55,14 @@ module Control.Distributed.Process.Management

55 , mxBroadcast

56 , mxSetLocal

57 , mxGetLocal

58− , mxLift

58+ , liftMX

59 -- * Mx Data API

60 , mxPublish

61 , mxSet

62 , mxGet

63+ , mxClear

64+ , mxPurgeTable

65+ , mxDropTable

63 ) where

66 ) where

65import Control.Applicative ((<$>))

68import Control.Applicative ((<$>))

@@ -74,7 +77,6 @@ import Control.Distributed.Process.Internal.Primitives

74 , receiveWait

77 , receiveWait

75 , matchChan

78 , matchChan

76 , unwrapMessage

79 , unwrapMessage

77− , whereis

78 )

80 )

79import Control.Distributed.Process.Internal.Types

81import Control.Distributed.Process.Internal.Types

80 ( Process

82 ( Process

@@ -133,14 +135,27 @@ mxPublish a k v = Table.set k v (Table.MxForAgent a)

133-- and attempts to force the value later on.

135-- and attempts to force the value later on.

134--

136--

135mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()

137mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()

136−mxSet mxId key msg =

138+mxSet mxId key msg = do

137 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)

139 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)

138

140

139-- | Fetches a property from the management database for the given key.

141-- | Fetches a property from the management database for the given key.

140-- If the property is not set, or does not match the expected type when

142-- If the property is not set, or does not match the expected type when

141-- typechecked (at runtime), returns @Nothing@.

143-- typechecked (at runtime), returns @Nothing@.

142mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)

144mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)

145+mxGet = Table.fetch . Table.MxForAgent

146+

147+-- | Clears a property from the management database using the given key.

148+-- If the key does not exist in the database, this is a noop.

149+mxClear :: MxAgentId -> String -> Process ()

143−mxGet mxId = Table.fetch (Table.MxForAgent mxId)

150+mxClear mxId key = Table.clear key (Table.MxForAgent mxId)

151+

152+-- | Purges a table in the management database of all its stored properties.

153+mxPurgeTable :: MxAgentId -> Process ()

154+mxPurgeTable = Table.purge . Table.MxForAgent

155+

156+-- | Deletes a table from the management database.

157+mxDropTable :: MxAgentId -> Process ()

158+mxDropTable = Table.delete . Table.MxForAgent

144

159

145--------------------------------------------------------------------------------

160--------------------------------------------------------------------------------

146-- API for writing user defined management extensions (i.e., agents) --

161-- API for writing user defined management extensions (i.e., agents) --

@@ -152,7 +167,7 @@ mxGetId = ST.get >>= return . mxAgentId

152mxBroadcast :: (Serializable m) => m -> MxAgent s ()

167mxBroadcast :: (Serializable m) => m -> MxAgent s ()

153mxBroadcast msg = do

168mxBroadcast msg = do

154 state <- ST.get

169 state <- ST.get

155− mxLift $ liftIO $ atomically $ do

170+ liftMX $ liftIO $ atomically $ do

156 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)

171 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)

157

172

158mxDeactivate :: forall s. String -> MxAgent s MxAction

173mxDeactivate :: forall s. String -> MxAgent s MxAction

@@ -161,8 +176,8 @@ mxDeactivate = return . MxAgentDeactivate

161mxReady :: forall s. MxAgent s MxAction

176mxReady :: forall s. MxAgent s MxAction

162mxReady = return MxAgentReady

177mxReady = return MxAgentReady

163

178

164−mxLift :: Process a -> MxAgent s a

179+liftMX :: Process a -> MxAgent s a

165−mxLift p = MxAgent $ ST.lift p

180+liftMX p = MxAgent $ ST.lift p

166

181

167mxSetLocal :: s -> MxAgent s ()

182mxSetLocal :: s -> MxAgent s ()

168mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }

183mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }

@@ -174,7 +189,7 @@ mxSink :: forall s m . (Serializable m)

174 => (m -> MxAgent s MxAction)

189 => (m -> MxAgent s MxAction)

175 -> MxSink s

190 -> MxSink s

176mxSink act msg = do

191mxSink act msg = do

177− msg' <- mxLift $ (unwrapMessage msg :: Process (Maybe m))

192+ msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))

178 case msg' of

193 case msg' of

179 Nothing -> return Nothing

194 Nothing -> return Nothing

180 Just m -> act m >>= return . Just

195 Just m -> act m >>= return . Just

@@ -200,7 +215,11 @@ mxAgent mxId initState handlers = do

200 start (sendTChan, recvTChan) = do

215 start (sendTChan, recvTChan) = do

201 (sp, rp) <- newChan

216 (sp, rp) <- newChan

202 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)

217 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)

218+ -- liftIO $ putStrLn $ "waiting on table coordinator " ++ Table.mxTableCoordinator

219+ -- p <- whereis Table.mxTableCoordinator

220+ -- liftIO $ putStrLn $ "registration == " ++ (show p)

203 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]

221 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]

222+ -- liftIO $ putStrLn "starting agent listener..."

204 runAgent handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState

223 runAgent handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState

205

224

206 runAgent hs c s = do

225 runAgent hs c s = do

@@ -228,8 +247,3 @@ mxAgent mxId initState handlers = do

228 Nothing -> runPipeline msg state next

247 Nothing -> runPipeline msg state next

229 Just result -> return (result, state')

248 Just result -> return (result, state')

230

249

231− runAgentST :: MxAgentState s

232− -> MxAgent s MxAction

233− -> Process (MxAction, MxAgentState s)

234− runAgentST state proc = ST.runStateT (unAgent proc) state

235−

src/Control/Distributed/Process/Management/Table.hs+26−23

@@ -141,33 +141,33 @@ resolve (MxForPid pid) = \msg -> send pid msg

141startTableCoordinator :: Fork -> Process ()

142startTableCoordinator fork = run Map.empty

143 where

144−

145− -- TODO: rethink this design. routing via the

146− -- coordinator will be a publishing bottle neck,

147− -- plus the code in 'fetch' really requires a

148− -- pid to send to, so that'll not work via an

149− -- intermediary anyway

150−

151 run :: MxTables -> Process ()

144 run :: MxTables -> Process ()

152 run tables =

145 run tables =

153 receiveWait [

146 receiveWait [

154− match (\(MxAgentStart ch agent) ->

147+ -- note that this state change can race with MxAgentStart requests

155− lookupAgent tables agent >>= \(p, t) -> do

148+ match (\(ProcessMonitorNotification _ pid _) -> do

156− sendChan ch p >> return t)

149+ return $ Map.filter (/= pid) tables)

157− , match (\(agent, Get k sp) -> do

150+ , match (\(MxAgentStart ch agent) -> do

158 lookupAgent tables agent >>= \(p, t) -> do

151 lookupAgent tables agent >>= \(p, t) -> do

152+ sendChan ch p >> return t)

153+ , match (\req@(agent, tReq :: MxTableRequest) -> do

154+ case tReq of

155+ Get k sp -> do

156+ lookupAgent tables agent >>= \(p, t) -> do

159− safeFetch p k >>= sendChan sp >> return t)

157+ safeFetch p k >>= sendChan sp >> return t

158+ _ -> do

160− , match $ handleRequest tables

159+ handleRequest tables req)

160+ , matchAny (\_ -> return tables) -- unrecognised messages are dropped

161 ] >>= run

162

163 handleRequest :: MxTables

164 -> (MxAgentId, MxTableRequest)

165 -> Process MxTables

166− handleRequest tables' (agent, req) =

166+ handleRequest tables' (agent, req) = do

167− lookupAgent tables' agent >>= \(p, t) -> do send p req >> return t

167+ lookupAgent tables' agent >>= \(p, t) -> send p req >> return t

168

169 lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables)

170− lookupAgent tables' agentId' =

170+ lookupAgent tables' agentId' = do

171 case Map.lookup agentId' tables' of

172 Nothing -> launchNew agentId' tables'

173 Just p -> return (p, tables')

@@ -188,28 +188,31 @@ startTableCoordinator fork = run Map.empty

188 -- break an import cycle with Node.hs via the management

189 -- agent, management API and the tracing modules

190 them <- liftIO $ fork $ link us >> proc

191− -- them <- spawnLocal nid $ link us >> proc

192 ref <- monitor them

191 ref <- monitor them

193 return (them, ref)

192 return (them, ref)

194

193

195tableHandler :: MxTableState -> Process ()

194tableHandler :: MxTableState -> Process ()

196tableHandler state = do

195tableHandler state = do

197 ns <- receiveWait [

196 ns <- receiveWait [

198− match (\Delete -> return Nothing)

197+ match (handleTableRequest state)

199− , match (\Purge -> return $ Just $ (entries ^= Map.empty) $ state)

200− , match (\(Clear k) -> return $ Just $ (entries ^: Map.delete k) $ state)

201− , match (\(Set k v) -> return $ Just $ (entries ^: Map.insert k v) state)

202− , match (\(Get k c) -> getEntry k c state >> return (Just state))

198+ , matchAny (\_ -> return (Just state))

203 ]

199 ]

204 case ns of

200 case ns of

205 Nothing -> return ()

201 Nothing -> return ()

206 Just s' -> tableHandler s'

202 Just s' -> tableHandler s'

203+ where

204+ handleTableRequest _ Delete = return Nothing

205+ handleTableRequest st Purge = return $ Just $ (entries ^= Map.empty) $ st

206+ handleTableRequest st (Clear k) = return $ Just $ (entries ^: Map.delete k) $ st

207+ handleTableRequest st (Set k v) = return $ Just $ (entries ^: Map.insert k v) st

208+ handleTableRequest st (Get k c) = getEntry k c st >> return (Just st)

207

209

208getEntry :: String

210getEntry :: String

209 -> SendPort (Maybe Message)

211 -> SendPort (Maybe Message)

210 -> MxTableState

212 -> MxTableState

211 -> Process ()

213 -> Process ()

214+getEntry k m MxTableState{..} = do

212−getEntry k m MxTableState{..} = sendChan m =<< return (Map.lookup k _entries)

215+ sendChan m =<< return (Map.lookup k _entries)

213

216

214entries :: Accessor MxTableState (Map String Message)

217entries :: Accessor MxTableState (Map String Message)

215entries = accessor _entries (\ls st -> st { _entries = ls })

218entries = accessor _entries (\ls st -> st { _entries = ls })

tests/TestMx.hs+137−31

@@ -1,11 +1,7 @@

1+{-# LANGUAGE DeriveGeneric #-}

1module Main where

2module Main where

3import Control.Concurrent (threadDelay)

4import Control.Concurrent (threadDelay)

4−import Control.Concurrent.MVar

5− ( newEmptyMVar

6− , tryPutMVar

7− , takeMVar

8− )

9import Control.Distributed.Process

5import Control.Distributed.Process

10import Control.Distributed.Process.Management

6import Control.Distributed.Process.Management

11 ( MxEvent(..)

7 ( MxEvent(..)

@@ -13,18 +9,27 @@ import Control.Distributed.Process.Management

13 , mxAgent

9 , mxAgent

14 , mxSink

10 , mxSink

15 , mxReady

11 , mxReady

16− , mxLift

12+ , liftMX

17 , mxGetLocal

13 , mxGetLocal

18 , mxSetLocal

14 , mxSetLocal

19 , mxNotify

15 , mxNotify

16+ , mxBroadcast

17+ , mxGetId

18+ , mxGet

19+ , mxSet

20+ , mxClear

21+ , mxPurgeTable

22+ , mxDropTable

20 )

23 )

21import Control.Distributed.Process.Node

24import Control.Distributed.Process.Node

22 ( LocalNode

25 ( LocalNode

23 )

26 )

27+import Data.Binary

24import Data.List (find)

28import Data.List (find)

25import Data.Maybe (isJust)

29import Data.Maybe (isJust)

30+import Data.Typeable

31+import GHC.Generics

26import qualified Network.Transport as NT

32import qualified Network.Transport as NT

27−

28#if ! MIN_VERSION_base(4,6,0)

33#if ! MIN_VERSION_base(4,6,0)

29import Prelude hiding (catch, log)

34import Prelude hiding (catch, log)

30#endif

35#endif

@@ -36,41 +41,71 @@ import Test.Framework

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

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

37import TestUtils

42import TestUtils

44+data Publish = Publish

45+ deriving (Typeable, Generic, Eq)

46+

47+instance Binary Publish where

48+

49+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()

50+testAgentBroadcast result = do

51+ (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())

52+

53+ publisher <- mxAgent (MxAgentId "publisher-agent") () [

54+ mxSink $ \() -> mxBroadcast Publish >> mxReady

55+ ]

56+

57+ consumer <- mxAgent (MxAgentId "consumer-agent") () [

58+ mxSink $ \Publish -> (liftMX $ sendChan resultSP ()) >> mxReady

59+ ]

60+

61+ mxNotify ()

62+ -- once the publisher has seen our message, it will broadcast the Publish

63+ stash result =<< receiveChanTimeout 1000000 resultRP

64+

65+ kill publisher "finished"

66+ kill consumer "finished"

67+

39testAgentEventHandling :: TestResult Bool -> Process ()

68testAgentEventHandling :: TestResult Bool -> Process ()

40testAgentEventHandling result = do

69testAgentEventHandling result = do

41 let initState = [] :: [MxEvent]

70 let initState = [] :: [MxEvent]

42− agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

71+ agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

43 (mxSink $ \ev -> do

72 (mxSink $ \ev -> do

44 st <- mxGetLocal

73 st <- mxGetLocal

74+ let act =

45− let act = case ev of

75+ case ev of

46− (MxSpawned _) -> mxSetLocal (ev:st)

76+ (MxSpawned _) -> mxSetLocal (ev:st)

47− (MxProcessDied _ _) -> mxSetLocal (ev:st)

77+ (MxProcessDied _ _) -> mxSetLocal (ev:st)

48− _ -> return ()

78+ _ -> return ()

49 act >> mxReady),

79 act >> mxReady),

50 (mxSink $ \(ev, sp :: SendPort Bool) -> do

80 (mxSink $ \(ev, sp :: SendPort Bool) -> do

51 st <- mxGetLocal

81 st <- mxGetLocal

52− let found = case ev of

82+ let found =

83+ case ev of

53− MxSpawned p ->

84+ MxSpawned p ->

54− isJust $ find (\ev' ->

85+ isJust $ find (\ev' ->

55− case ev' of

86+ case ev' of

56− (MxSpawned p') -> p' == p

87+ (MxSpawned p') -> p' == p

57− _ -> False) st

88+ _ -> False) st

58− MxProcessDied p r ->

89+ MxProcessDied p r ->

59− isJust $ find (\ev' ->

90+ isJust $ find (\ev' ->

60− case ev' of

91+ case ev' of

61− (MxProcessDied p' r') -> p' == p && r == r'

92+ (MxProcessDied p' r') -> p' == p && r == r'

62− _ -> False) st

93+ _ -> False) st

63− _ -> False

94+ _ -> False

64− mxLift $ sendChan sp found

95+ liftMX $ sendChan sp found

65− liftIO $ putStrLn $ "replied to " ++ (show sp)

66 mxReady)

96 mxReady)

67 ]

97 ]

69− mref <- monitor agentPid

99+ _ <- monitor agentPid

70 (sp, rp) <- newChan

100 (sp, rp) <- newChan

71 pid <- spawnLocal $ sendChan sp ()

101 pid <- spawnLocal $ sendChan sp ()

72 () <- receiveChan rp

102 () <- receiveChan rp

103

104+ -- By waiting for a monitor notification, we have a

105+ -- higher probably that the agent has seen

106+ monitor pid

107+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]

108+

74 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

109 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

75 mxNotify (MxSpawned pid, replyTo)

110 mxNotify (MxSpawned pid, replyTo)

76 mxNotify (MxProcessDied pid DiedNormal, replyTo)

111 mxNotify (MxProcessDied pid DiedNormal, replyTo)

@@ -80,21 +115,92 @@ testAgentEventHandling result = do

115

81 stash result $ seenAlive && seenDead

116 stash result $ seenAlive && seenDead

117

118+testAgentPublication :: TestResult (Maybe Publish) -> Process ()

119+testAgentPublication result = do

120+ (syncChan, rp) <- newChan

121+ let ourId = MxAgentId "publication-agent"

122+ agentPid <- mxAgent ourId () [

123+ mxSink $ \() -> do

124+ selfId <- mxGetId

125+ liftMX $ mxSet selfId "publish" Publish >> sendChan syncChan ()

126+ mxReady

127+ ]

128+

129+ mxNotify ()

130+ () <- receiveChan rp

131+

132+ stash result =<< mxGet ourId "publish"

133+

134+ mref <- monitor agentPid

83 kill agentPid "finished"

135 kill agentPid "finished"

84 receiveWait [

136 receiveWait [

85 matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)

137 matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)

86 ((\_ -> return ()))

138 ((\_ -> return ()))

87 ]

139 ]

140

141+testAgentTableClear :: Int -> TestResult (Maybe Int, Maybe Int) -> Process ()

142+testAgentTableClear val result =

143+ let tId = (MxAgentId "agent-1")

144+ tKey = "key-1" in do

145+ mxSet tId tKey val

146+ get1 <- mxGet tId tKey

147+ mxClear tId tKey

148+ get2 <- mxGet tId tKey

149+ stash result (get1, get2)

150+

151+testAgentTablePurge :: TestResult (Maybe Int) -> Process ()

152+testAgentTablePurge result =

153+ let tId = MxAgentId "agent-2"

154+ tKey = "key-2" in do

155+ mxSet tId tKey (12345 :: Int)

156+ mxPurgeTable tId

157+ stash result =<< mxGet tId tKey

158+

159+testAgentTableDelete :: Int

160+ -> TestResult (Maybe Int, Maybe Int, Maybe Int)

161+ -> Process ()

162+testAgentTableDelete val result =

163+ let tId = (MxAgentId "agent-3")

164+ tKey = "key-3" in do

165+ mxSet tId tKey val

166+ get1 <- mxGet tId tKey

167+ mxDropTable tId

168+ get2 <- mxGet tId tKey

169+ mxSet tId tKey val

170+ get3 <- mxGet tId tKey

171+ stash result (get1, get2, get3)

172+

89tests :: LocalNode -> IO [Test]

173tests :: LocalNode -> IO [Test]

90tests node1 = do

174tests node1 = do

91 return [

175 return [

92− testGroup "Basic Agent Handling" [

176+ testGroup "Mx Agents" [

93− testCase "testAgentEventHandling"

177+ testCase "Event Handling"

94 (delayedAssertion

178 (delayedAssertion

95− "expected "

179+ "expected True, but events where not as expected"

96 node1 True testAgentEventHandling)

180 node1 True testAgentEventHandling)

181+ , testCase "Inter-Agent Broadcast"

182+ (delayedAssertion

183+ "expected (Just ()), but no broadcast was received"

184+ node1 (Just ()) testAgentBroadcast)

97− ] ]

185+ ],

186+ testGroup "Mx Global Properties" [

187+ testCase "Global Property Publication"

188+ (delayedAssertion

189+ "expected (Just Publish), but no table entry was found"

190+ node1 (Just Publish) testAgentPublication)

191+ , testCase "Clearing Global Properties"

192+ (delayedAssertion

193+ "expected (Just 1024, Nothing): invalid table entry found!"

194+ node1 (Just 1024, Nothing) (testAgentTableClear 1024))

195+ , testCase "Purging Global Tables"

196+ (delayedAssertion

197+ "expected Nothing, but a table entry was found"

198+ node1 Nothing testAgentTablePurge)

199+ , testCase "Deleting and (Re)Creating Global Tables"

200+ (delayedAssertion

201+ "expected (Just 15, Nothing, Just 15): invalid table entry found!"

202+ node1 (Just 15, Nothing, Just 15) (testAgentTableDelete 15))

203+ ]]

204

99statsTests :: NT.Transport -> IO [Test]

205statsTests :: NT.Transport -> IO [Test]

100statsTests _ = do

206statsTests _ = do

tests/TestTracing.hs+1−7

@@ -17,11 +17,8 @@ import Control.Distributed.Process.Management

17 )

18import Control.Distributed.Process.Node

19 ( forkProcess

20− , newLocalNode

21− , initRemoteTable

22 , closeLocalNode

20 , closeLocalNode

23 , LocalNode)

21 , LocalNode)

24−import Network.Transport.TCP

25import qualified Network.Transport as NT

22import qualified Network.Transport as NT

27#if ! MIN_VERSION_base(4,6,0)

24#if ! MIN_VERSION_base(4,6,0)

@@ -30,11 +27,8 @@ import Prelude hiding (catch, log)

31import Test.Framework

28import Test.Framework

32 ( Test

29 ( Test

33− , defaultMain

34 , testGroup

30 , testGroup

35 )

31 )

36−import Test.HUnit (Assertion)

37−import Test.HUnit.Base (assertBool)

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

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

39import TestUtils

33import TestUtils

@@ -333,7 +327,7 @@ tests node1 = do

333 ] ]

327 ] ]

334

328

335timerTests :: NT.Transport -> IO [Test]

329timerTests :: NT.Transport -> IO [Test]

336−timerTests transport = do

330+timerTests _ = do

337 mkNode "8080" >>= tests >>= return

331 mkNode "8080" >>= tests >>= return

338

332

339main :: IO ()

333main :: IO ()