2ffa2bc · distributed-process

git.iko.soy/distributed-process

on master help sign in

Allow Management Agents to utilise their mailbox

2ffa2bcTim Watson13 years ago

src/Control/Distributed/Process/Management.hs+16−2

compare with master

unifiedsplit

@@ -69,16 +69,19 @@ module Control.Distributed.Process.Management

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

70import Control.Concurrent.STM (atomically)

71import Control.Concurrent.STM.TChan

72− ( readTChan

72+ ( tryReadTChan

73 , writeTChan

74 )

75import Control.Distributed.Process.Internal.Primitives

76 ( newChan

77 , nsend

78 , receiveWait

79+ , receiveTimeout

79 , matchChan

80 , matchChan

81+ , matchAny

80 , unwrapMessage

82 , unwrapMessage

81 , onException

83 , onException

84+ , register

82 )

85 )

83import Control.Distributed.Process.Internal.Types

86import Control.Distributed.Process.Internal.Types

84 ( Process

87 ( Process

@@ -222,6 +225,7 @@ mxAgentWithFinalize :: MxAgentId

222mxAgentWithFinalize mxId initState handlers dtor = do

225mxAgentWithFinalize mxId initState handlers dtor = do

223 node <- processNode <$> ask

226 node <- processNode <$> ask

224 pid <- liftIO $ mxNew (localEventBus node) $ start

227 pid <- liftIO $ mxNew (localEventBus node) $ start

228+ register (agentId mxId) pid

225 return pid

229 return pid

226 where

230 where

227 start (sendTChan, recvTChan) = do

231 start (sendTChan, recvTChan) = do

@@ -238,13 +242,23 @@ mxAgentWithFinalize mxId initState handlers dtor = do

238 runAgentWithFinalizer eh hs c s `onException` runAgentFinalizer eh s

242 runAgentWithFinalizer eh hs c s `onException` runAgentFinalizer eh s

239

243

240 runAgentWithFinalizer eh' hs' c' s' = do

244 runAgentWithFinalizer eh' hs' c' s' = do

241− msg <- (liftIO $ atomically $ readTChan c')

245+ msg <- getNextMessage c'

242 (action, state) <- runPipeline msg s' $ pipeline hs'

246 (action, state) <- runPipeline msg s' $ pipeline hs'

243 case action of

247 case action of

244 MxAgentReady -> runAgent eh' hs' c' state

248 MxAgentReady -> runAgent eh' hs' c' state

245 MxAgentDeactivate _ -> runAgentFinalizer eh' state

249 MxAgentDeactivate _ -> runAgentFinalizer eh' state

246-- MxAgentBecome h' -> runAgent h' c state

250-- MxAgentBecome h' -> runAgent h' c state

247

251

252+ getNextMessage tch = do

253+ inputs <- liftIO $ atomically $ tryReadTChan tch

254+ case inputs of

255+ Nothing -> do

256+ m <- receiveTimeout 0 [ matchAny return ]

257+ case m of

258+ Nothing -> getNextMessage tch

259+ Just msg -> return msg

260+ Just m -> return m

261+

248 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()

262 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()

249 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst

263 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst

250

264