aafd733 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Improvements to the redirection of logging

aafd733Simon Marlow14 years ago

distributed-process-simplelocalnet/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs+78−32

compare with master

unifiedsplit

@@ -104,7 +104,7 @@ import Data.Foldable (forM_)

104import Data.Typeable (Typeable)

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

106import Control.Exception (throw)

107−import Control.Monad (forever, forM, replicateM)

107+import Control.Monad (forever, forM, replicateM, when, replicateM_)

108import Control.Monad.IO.Class (liftIO)

109import Control.Concurrent (forkIO, threadDelay, ThreadId)

110import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)

@@ -112,26 +112,36 @@ import Control.Distributed.Process

112 ( RemoteTable

113 , NodeId

114 , Process

115+ , ProcessId

115 , WhereIsReply(..)

116 , WhereIsReply(..)

116 , whereis

117 , whereis

117 , whereisRemoteAsync

118 , whereisRemoteAsync

119+ , registerRemoteAsync

118 , reregisterRemoteAsync

120 , reregisterRemoteAsync

119 , getSelfPid

121 , getSelfPid

120 , register

122 , register

123+ , reregister

121 , expect

124 , expect

122 , nsendRemote

125 , nsendRemote

123 , receiveWait

126 , receiveWait

127+ , receiveTimeout

124 , match

128 , match

125 , matchIf

129 , matchIf

126 , processNodeId

130 , processNodeId

127 , monitorNode

131 , monitorNode

132+ , monitor

128 , unmonitor

133 , unmonitor

129 , NodeMonitorNotification(..)

134 , NodeMonitorNotification(..)

135+ , ProcessRegistrationException

130 , finally

136 , finally

131 , newChan

137 , newChan

132 , receiveChan

138 , receiveChan

133 , nsend

139 , nsend

134 , SendPort

140 , SendPort

141+ , RegisterReply(..)

142+ , bracket

143+ , try

144+ , send

135 )

145 )

136import qualified Control.Distributed.Process.Node as Node

146import qualified Control.Distributed.Process.Node as Node

137 ( LocalNode

147 ( LocalNode

@@ -157,7 +167,7 @@ data Backend = Backend {

157 , findPeers :: Int -> IO [NodeId]

167 , findPeers :: Int -> IO [NodeId]

158 -- | Make sure that all log messages are printed by the logger on the

168 -- | Make sure that all log messages are printed by the logger on the

159 -- current node

169 -- current node

160− , redirectLogsHere :: Process ()

170+ , redirectLogsHere :: [ProcessId] -> Process ()

161 }

171 }

162

172

163data BackendState = BackendState {

173data BackendState = BackendState {

@@ -244,12 +254,27 @@ peerDiscoveryDaemon backendState recv send = forever go

244--------------------------------------------------------------------------------

254--------------------------------------------------------------------------------

245

255

246-- | Make sure that all log messages are printed by the logger on this node

256-- | Make sure that all log messages are printed by the logger on this node

247−apiRedirectLogsHere :: Backend -> Process ()

257+apiRedirectLogsHere :: Backend -> [ProcessId] -> Process ()

248−apiRedirectLogsHere backend = do

258+apiRedirectLogsHere _backend slavecontrollers = do

249 mLogger <- whereis "logger"

259 mLogger <- whereis "logger"

260+ myPid <- getSelfPid

261+

250 forM_ mLogger $ \logger -> do

262 forM_ mLogger $ \logger -> do

251− nids <- liftIO $ findPeers backend 1000000

263+

252− forM_ nids $ \nid -> reregisterRemoteAsync nid "logger" logger -- ignore async response

264+ bracket

265+ (mapM monitor slavecontrollers)

266+ (mapM unmonitor)

267+ $ \_ -> do

268+

269+ -- fire off redirect requests

270+ forM_ slavecontrollers $ \pid -> send pid (RedirectLogsTo logger myPid)

271+

272+ -- Wait for the replies

273+ replicateM_ (length slavecontrollers) $ do

274+ receiveWait

275+ [ match (\(RedirectLogsReply from ok) -> return ())

276+ , match (\m@(NodeMonitorNotification {}) -> return ())

277+ ]

253

278

254--------------------------------------------------------------------------------

279--------------------------------------------------------------------------------

255-- Slaves --

280-- Slaves --

@@ -259,18 +284,31 @@ apiRedirectLogsHere backend = do

259--

284--

260-- This datatype is not exposed; instead, we expose primitives for dealing

285-- This datatype is not exposed; instead, we expose primitives for dealing

261-- with slaves.

286-- with slaves.

262−data SlaveControllerMsg =

287+data SlaveControllerMsg

263− SlaveTerminate

288+ = SlaveTerminate

289+ | RedirectLogsTo ProcessId ProcessId

264 deriving (Typeable, Show)

290 deriving (Typeable, Show)

265

291

266instance Binary SlaveControllerMsg where

292instance Binary SlaveControllerMsg where

267 put SlaveTerminate = putWord8 0

293 put SlaveTerminate = putWord8 0

294+ put (RedirectLogsTo a b) = do putWord8 1; put (a,b)

268 get = do

295 get = do

269 header <- getWord8

296 header <- getWord8

270 case header of

297 case header of

271 0 -> return SlaveTerminate

298 0 -> return SlaveTerminate

299+ 1 -> do (a,b) <- get; return (RedirectLogsTo a b)

272 _ -> fail "SlaveControllerMsg.get: invalid"

300 _ -> fail "SlaveControllerMsg.get: invalid"

273

301

302+data RedirectLogsReply

303+ = RedirectLogsReply ProcessId Bool

304+ deriving (Typeable, Show)

305+

306+instance Binary RedirectLogsReply where

307+ put (RedirectLogsReply from ok) = put (from,ok)

308+ get = do

309+ (from,ok) <- get

310+ return (RedirectLogsReply from ok)

311+

274-- | Calling 'slave' sets up a new local node and then waits. You start

312-- | Calling 'slave' sets up a new local node and then waits. You start

275-- processes on the slave by calling 'spawn' from other nodes.

313-- processes on the slave by calling 'spawn' from other nodes.

276--

314--

@@ -292,41 +330,49 @@ slaveController = do

292 msg <- expect

330 msg <- expect

293 case msg of

331 case msg of

294 SlaveTerminate -> return ()

332 SlaveTerminate -> return ()

333+ RedirectLogsTo loggerPid from -> do

334+ r <- try (reregister "logger" loggerPid)

335+ ok <- case (r :: Either ProcessRegistrationException ()) of

336+ Right _ -> return True

337+ Left _ -> do

338+ r <- try (register "logger" loggerPid)

339+ case (r :: Either ProcessRegistrationException ()) of

340+ Right _ -> return True

341+ Left _ -> return False

342+ pid <- getSelfPid

343+ send from (RedirectLogsReply pid ok)

344+ go

295

345

296-- | Terminate the slave at the given node ID

346-- | Terminate the slave at the given node ID

297terminateSlave :: NodeId -> Process ()

347terminateSlave :: NodeId -> Process ()

298terminateSlave nid = nsendRemote nid "slaveController" SlaveTerminate

348terminateSlave nid = nsendRemote nid "slaveController" SlaveTerminate

299

349

300-- | Find slave nodes

350-- | Find slave nodes

301−findSlaves :: Backend -> Process [NodeId]

351+findSlaves :: Backend -> Process [ProcessId]

302findSlaves backend = do

352findSlaves backend = do

303 nodes <- liftIO $ findPeers backend 1000000

353 nodes <- liftIO $ findPeers backend 1000000

304− -- Fire of asynchronous requests for the slave controller

354+ -- Fire off asynchronous requests for the slave controller

305− refs <- forM nodes $ \nid -> do

355+

356+ bracket

357+ (mapM monitorNode nodes)

358+ (mapM unmonitor)

359+ $ \_ -> do

360+

361+ -- fire off whereis requests

306− whereisRemoteAsync nid "slaveController"

362+ forM nodes $ \nid -> whereisRemoteAsync nid "slaveController"

307− ref <- monitorNode nid

363+

308− return (nid, ref)

309− -- Wait for the replies

364+ -- Wait for the replies

310− catMaybes <$> replicateM (length nodes) (

365+ catMaybes <$> replicateM (length nodes) (

311− receiveWait

366+ receiveWait

312− [ matchIf (\(WhereIsReply label _) -> label == "slaveController")

367+ [ match (\(WhereIsReply "slaveController" mPid) -> return mPid)

313− (\(WhereIsReply _ mPid) ->

314− case mPid of

315− Nothing ->

316− return Nothing

317− Just pid -> do

318− let nid = processNodeId pid

319− Just ref = lookup nid refs

320− unmonitor ref

321− return (Just nid))

322− , match (\(NodeMonitorNotification {}) -> return Nothing)

368+ , match (\(NodeMonitorNotification {}) -> return Nothing)

323− ])

369+ ])

324

370

325-- | Terminate all slaves

371-- | Terminate all slaves

326terminateAllSlaves :: Backend -> Process ()

372terminateAllSlaves :: Backend -> Process ()

327terminateAllSlaves backend = do

373terminateAllSlaves backend = do

328 slaves <- findSlaves backend

374 slaves <- findSlaves backend

329− forM_ slaves terminateSlave

375+ forM_ slaves $ \pid -> send pid SlaveTerminate

330 liftIO $ threadDelay 1000000

376 liftIO $ threadDelay 1000000

331

377

332--------------------------------------------------------------------------------

378--------------------------------------------------------------------------------

@@ -353,8 +399,8 @@ startMaster backend proc = do

353 node <- newLocalNode backend

399 node <- newLocalNode backend

354 Node.runProcess node $ do

400 Node.runProcess node $ do

355 slaves <- findSlaves backend

401 slaves <- findSlaves backend

356− redirectLogsHere backend

402+ redirectLogsHere backend slaves

357− proc slaves `finally` shutdownLogger

403+ proc (map processNodeId slaves) `finally` shutdownLogger

358

404

359--

405--

360-- | shut down the logger process. This ensures that any pending

406-- | shut down the logger process. This ensures that any pending