afa1451 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Keep a central mapping of monitor actions

afa1451Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+59−30

compare with master

unifiedsplit

@@ -96,7 +96,7 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString

96 )

97import Data.Binary (Binary, decode, encode, put, get, getWord8, putWord8)

98import Data.Map (Map)

99−import qualified Data.Map as Map (empty, lookup, insert, delete)

99+import qualified Data.Map as Map (empty, lookup, insert, delete, keys)

100import qualified Data.List as List (delete)

101import Data.Int (Int32)

102import Data.Typeable (Typeable)

@@ -106,8 +106,8 @@ import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

106import Control.Monad.IO.Class (MonadIO, liftIO)

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

108import Control.Category ((>>>))

109−import Control.Exception (Exception, throwIO, catch)

109+import Control.Exception (Exception, throwIO, catch, throwTo)

110−import Control.Concurrent (forkIO)

110+import Control.Concurrent (forkIO, myThreadId)

111import Control.Concurrent.MVar ( MVar

112 , newMVar

113 , withMVar

@@ -141,12 +141,14 @@ import qualified Network.Transport as NT ( Transport

141 , newEndPoint

142 , receive

143 , Event(..)

144+ , EventErrorCode(..)

145+ , TransportError(..)

144 , address

146 , address

145 , closeEndPoint

147 , closeEndPoint

146 )

148 )

147import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)

149import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)

148import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

150import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

149−import qualified Data.Accessor.Container as DAC (mapMaybe)

151+import qualified Data.Accessor.Container as DAC (mapMaybe, mapDefault)

150import System.Random (randomIO)

152import System.Random (randomIO)

151

153

152-- | A local process ID consists of a seed which distinguishes processes from

154-- | A local process ID consists of a seed which distinguishes processes from

@@ -173,15 +175,16 @@ data Message = Message

173

175

174-- | Local nodes

176-- | Local nodes

175data LocalNode = LocalNode

177data LocalNode = LocalNode

176− { localEndPoint :: NT.EndPoint

178+ { localEndPoint :: NT.EndPoint

177− , localState :: MVar LocalNodeState

179+ , localState :: MVar LocalNodeState

178 }

180 }

179

181

180-- | Local node state

182-- | Local node state

181data LocalNodeState = LocalNodeState

183data LocalNodeState = LocalNodeState

182− { _localProcesses :: Map LocalProcessId LocalProcess

184+ { _localProcesses :: Map LocalProcessId LocalProcess

183− , _localPidCounter :: Int32

185+ , _localPidCounter :: Int32

184− , _localPidUnique :: Int32

186+ , _localPidUnique :: Int32

187+ , _monitorActions :: Map ProcessId [IO ()]

185 }

188 }

186

189

187-- | Processes running on our local node

190-- | Processes running on our local node

@@ -194,8 +197,7 @@ data LocalProcess = LocalProcess

194

197

195-- | Local process state

198-- | Local process state

196data LocalProcessState = LocalProcessState

199data LocalProcessState = LocalProcessState

197− { _monitorActions :: Map ProcessId MonitorAction

198− , _connections :: Map ProcessId NT.Connection

200+ { _connections :: Map ProcessId NT.Connection

199 }

201 }

200

202

201-- The Cloud Haskell 'Process' type

203-- The Cloud Haskell 'Process' type

@@ -234,9 +236,22 @@ getSelfPid = processId <$> ask

234-- monitors (http://www.erlang.org/doc/reference_manual/processes.html#id82613)

236-- monitors (http://www.erlang.org/doc/reference_manual/processes.html#id82613)

235monitor :: ProcessId -> MonitorAction -> Process ()

237monitor :: ProcessId -> MonitorAction -> Process ()

236monitor them action = do

238monitor them action = do

239+ proc <- ask

240+ liftIO $ do

241+ tid <- myThreadId

242+ let err = ProcessMonitorException them SrNoPing

243+ act = case action of

244+ MaMonitor -> enqueue (processQueue proc) $ createMessage err

245+ MaLink -> throwTo tid err

246+ MaLinkError -> fail "monitor: MaLinkError not implemented"

247+ modifyMVar_ (localState . processNode $ proc) $

248+ return . (monitorActionsFor them ^: (act :))

249+

250+{-

237 ourState <- processState <$> ask

251 ourState <- processState <$> ask

238 liftIO $ modifyMVar_ ourState $ \st ->

252 liftIO $ modifyMVar_ ourState $ \st ->

239 return . (monitorActionFor them ^= Just action) $ st

253 return . (monitorActionFor them ^= Just action) $ st

254+-}

240

255

241-- | Have one process monitor another

256-- | Have one process monitor another

242--

257--

@@ -375,9 +390,10 @@ newLocalNode transport = do

375 Right endPoint -> do

390 Right endPoint -> do

376 unq <- randomIO

391 unq <- randomIO

377 state <- newMVar LocalNodeState

392 state <- newMVar LocalNodeState

378− { _localProcesses = Map.empty

393+ { _localProcesses = Map.empty

379− , _localPidCounter = 0

394+ , _localPidCounter = 0

380− , _localPidUnique = unq

395+ , _localPidUnique = unq

396+ , _monitorActions = Map.empty

381 }

397 }

382 let node = LocalNode { localEndPoint = endPoint

398 let node = LocalNode { localEndPoint = endPoint

383 , localState = state

399 , localState = state

@@ -401,8 +417,7 @@ forkProcess node proc = do

401 let pid = ProcessId { processAddress = NT.address (localEndPoint node)

417 let pid = ProcessId { processAddress = NT.address (localEndPoint node)

402 , processLocalId = lpid

418 , processLocalId = lpid

403 }

419 }

404− pst <- newMVar LocalProcessState { _connections = Map.empty

420+ pst <- newMVar LocalProcessState { _connections = Map.empty

405− , _monitorActions = Map.empty

406 }

421 }

407 let lproc = LocalProcess { processQueue = queue

422 let lproc = LocalProcess { processQueue = queue

408 , processNode = node

423 , processNode = node

@@ -446,9 +461,14 @@ handleIncomingMessages node = go [] Map.empty

446 fail "handleIncomingMessages: TODO 2"

461 fail "handleIncomingMessages: TODO 2"

447 NT.ConnectionClosed cid ->

462 NT.ConnectionClosed cid ->

448 go (List.delete cid halfOpenConns) (Map.delete cid openConns)

463 go (List.delete cid halfOpenConns) (Map.delete cid openConns)

464+ NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do

465+ -- TODO: we could cache this information

466+ pids <- withMVar state $ \st ->

467+ return $ filter ((== theirAddr) . processAddress) . Map.keys

468+ $ st ^. monitorActions

469+ forM_ pids $ remoteProcessFailed node

449 NT.ErrorEvent _ ->

470 NT.ErrorEvent _ ->

450− -- fail "handleIncomingMessages: TODO 3"

471+ fail "handleIncomingMessages: TODO 4"

451− go halfOpenConns openConns

452 NT.EndPointClosed ->

472 NT.EndPointClosed ->

453 return ()

473 return ()

454 NT.ReceivedMulticast _ _ ->

474 NT.ReceivedMulticast _ _ ->

@@ -496,16 +516,19 @@ createConnectionTo them = do

496 return . Just $ conn

516 return . Just $ conn

497 Left _err -> do

517 Left _err -> do

498 -- TODO: should probably pass this error to remoteProcessFailed

518 -- TODO: should probably pass this error to remoteProcessFailed

519+ node <- processNode <$> ask

499− remoteProcessFailed them

520+ liftIO $ remoteProcessFailed node them

500 return Nothing

521 return Nothing

501

522

502sendBinary :: ProcessId -> NT.Connection -> [BSS.ByteString] -> Process ()

523sendBinary :: ProcessId -> NT.Connection -> [BSS.ByteString] -> Process ()

503sendBinary them conn payload = do

524sendBinary them conn payload = do

504 result <- liftIO $ NT.send conn payload

525 result <- liftIO $ NT.send conn payload

505 case result of

526 case result of

527+ Right () ->

506− Right () -> return ()

528+ return ()

529+ Left _ -> do

530+ node <- processNode <$> ask

507− Left _ -> remoteProcessFailed them

531+ liftIO $ remoteProcessFailed node them

508−

509

532

510sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()

533sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()

511sendMessage them conn (Message fp enc) =

534sendMessage them conn (Message fp enc) =

@@ -529,8 +552,13 @@ payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss

529 , lpidUnique = NTI.decodeInt32 bs2

552 , lpidUnique = NTI.decodeInt32 bs2

530 }

553 }

531

554

532−remoteProcessFailed :: ProcessId -> Process ()

555+remoteProcessFailed :: LocalNode -> ProcessId -> IO ()

533−remoteProcessFailed them = do

556+remoteProcessFailed node them = do

557+ liftIO $ modifyMVar_ (localState node) $ \st -> do

558+ sequence_ (st ^. monitorActionsFor them)

559+ return (monitorActionsFor them ^= [] $ st)

560+

561+{-do

534 proc <- ask

562 proc <- ask

535 let ourState = processState proc

563 let ourState = processState proc

536 -- We only execute monitor actions once

564 -- We only execute monitor actions once

@@ -547,6 +575,7 @@ remoteProcessFailed them = do

547 throwIO err

575 throwIO err

548 Nothing ->

576 Nothing ->

549 return ()

577 return ()

578+ -}

550

579

551createMessage :: Serializable a => a -> Message

580createMessage :: Serializable a => a -> Message

552createMessage a = Message (fingerprint a) (encode a)

581createMessage a = Message (fingerprint a) (encode a)

@@ -580,17 +609,17 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =

580localPidUnique :: Accessor LocalNodeState Int32

609localPidUnique :: Accessor LocalNodeState Int32

581localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

610localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

582

611

612+monitorActions :: Accessor LocalNodeState (Map ProcessId [IO ()])

613+monitorActions = accessor _monitorActions (\as st -> st { _monitorActions = as })

614+

583localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

615localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

584localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

616localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

585

617

618+monitorActionsFor :: ProcessId -> Accessor LocalNodeState [IO ()]

619+monitorActionsFor pid = monitorActions >>> DAC.mapDefault [] pid

620+

586connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)

621connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)

587connections = accessor _connections (\conns st -> st { _connections = conns })

622connections = accessor _connections (\conns st -> st { _connections = conns })

588

623

589−monitorActions :: Accessor LocalProcessState (Map ProcessId MonitorAction)

590−monitorActions = accessor _monitorActions (\ms st -> st { _monitorActions = ms})

591−

592connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)

624connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)

593connectionTo pid = connections >>> DAC.mapMaybe pid

625connectionTo pid = connections >>> DAC.mapMaybe pid

594−

595−monitorActionFor :: ProcessId -> Accessor LocalProcessState (Maybe MonitorAction)

596−monitorActionFor pid = monitorActions >>> DAC.mapMaybe pid

distributed-process/tests/TestCH.hs+35

@@ -149,6 +149,40 @@ testMonitor3 transport = do

149

150 takeMVar done

151

152+

153+-- Like testMonitor3, except without the second send (so we must detect the

154+-- failure elsewhere)

155+testMonitor4 :: NT.Transport -> IO ()

156+testMonitor4 transport = do

157+ firstSend <- newEmptyMVar

158+ serverAddr <- newEmptyMVar

159+ serverDead <- newEmptyMVar

160+ done <- newEmptyMVar

161+

162+ forkIO $ do

163+ localNode <- newLocalNode transport

164+ -- TODO: what happens when processes terminate?

165+ addr <- forkProcess localNode $ return ()

166+ putMVar serverAddr addr

167+ readMVar firstSend

168+ closeLocalNode localNode

169+ threadDelay 10000 -- Give the TCP layer a chance to actually close the socket

170+ putMVar serverDead ()

171+

172+ forkIO $ do

173+ localNode <- newLocalNode transport

174+ theirAddr <- readMVar serverAddr

175+ runProcess localNode $ do

176+ monitor theirAddr MaMonitor

177+ send theirAddr "Hi"

178+ liftIO $ putMVar firstSend () >> readMVar serverDead

179+ ProcessMonitorException pid SrNoPing <- expect

180+ True <- return $ pid == theirAddr

181+ return ()

182+ putMVar done ()

183+

184+ takeMVar done

185+

152-- TODO: test/specify normal process termination

186-- TODO: test/specify normal process termination

153

187

154main :: IO ()

188main :: IO ()

@@ -159,4 +193,5 @@ main = do

159 ("Monitor1", testMonitor1 transport)

193 ("Monitor1", testMonitor1 transport)

160 , ("Monitor2", testMonitor2 transport)

194 , ("Monitor2", testMonitor2 transport)

161 , ("Monitor3", testMonitor3 transport)

195 , ("Monitor3", testMonitor3 transport)

196+ , ("Monitor4", testMonitor4 transport)

162 ]

197 ]