9377760 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Add finalisation support to the Management Agent API

9377760Tim Watson13 years ago

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

compare with master

unifiedsplit

@@ -47,6 +47,7 @@ module Control.Distributed.Process.Management

47 , MxAgentId(..)

48 , MxAgent()

49 , mxAgent

50+ , mxAgentWithFinalize

50 , MxSink

51 , MxSink

51 , mxSink

52 , mxSink

52 , mxGetId

53 , mxGetId

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

77 , receiveWait

78 , receiveWait

78 , matchChan

79 , matchChan

79 , unwrapMessage

80 , unwrapMessage

81+ , onException

80 )

82 )

81import Control.Distributed.Process.Internal.Types

83import Control.Distributed.Process.Internal.Types

82 ( Process

84 ( Process

@@ -197,17 +199,27 @@ mxSink act msg = do

197data MxPipeline s =

199data MxPipeline s =

198 MxPipeline

200 MxPipeline

199 {

201 {

200− current :: !(MxSink s)

202+ current :: !(MxSink s)

201− , next :: !(MxPipeline s)

203+ , next :: !(MxPipeline s)

202 } | MxStop

204 } | MxStop

203

205

204-- | Activates a new agent.

206-- | Activates a new agent.

205--

207--

206−mxAgent :: MxAgentId

208+mxAgent :: MxAgentId -> s -> [MxSink s] -> Process ProcessId

209+mxAgent mxId st hs = mxAgentWithFinalize mxId st hs $ return ()

210+

211+-- | Activates a new agent. This variant takes a /finalizer/ expression,

212+-- that is run once the agent shuts down (even in case of failure/exceptions).

213+-- The /finalizer/ expression runs in the mx monad - @MxAgent s ()@ - such

214+-- that the agent's internal state remains accessible to the shutdown/cleanup

215+-- code.

216+--

217+mxAgentWithFinalize :: MxAgentId

207 -> s

218 -> s

208 -> [MxSink s]

219 -> [MxSink s]

220+ -> MxAgent s ()

209 -> Process ProcessId

221 -> Process ProcessId

210−mxAgent mxId initState handlers = do

222+mxAgentWithFinalize mxId initState handlers dtor = do

211 node <- processNode <$> ask

223 node <- processNode <$> ask

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

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

213 return pid

225 return pid

@@ -220,15 +232,21 @@ mxAgent mxId initState handlers = do

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

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

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

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

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

234 -- liftIO $ putStrLn "starting agent listener..."

223− runAgent handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState

235+ runAgent dtor handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState

224−

236+

225− runAgent hs c s = do

237+ runAgent eh hs c s =

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

239+

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

226− msg <- (liftIO $ atomically $ readTChan c)

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

227− (action, state) <- runPipeline msg s $ pipeline hs

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

228− case action of

243+ case action of

229− MxAgentReady -> runAgent hs c state

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

230− MxAgentDeactivate _ -> {- TODO: log r -} return ()

245+ MxAgentDeactivate _ -> runAgentFinalizer eh' state

231−-- MxAgentBecome h' -> runAgent h' c state

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

247+

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

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

232

250

233 pipeline :: forall s . [MxSink s] -> MxPipeline s

251 pipeline :: forall s . [MxSink s] -> MxPipeline s

234 pipeline [] = MxStop

252 pipeline [] = MxStop