24a0274 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Use new failure semantics in C.D.P.Node

24a0274Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs+63−25

compare with master

unifiedsplit

@@ -1,19 +1,20 @@

1module Control.Distributed.Process.Internal.Messaging

2− ( -- * Message sending

3− ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

4− , sendPayload

2+ ( sendPayload

5 , sendBinary

3 , sendBinary

6 , sendMessage

4 , sendMessage

7 , disconnect

5 , disconnect

6+ , closeImplicitReconnections

7+ , impliesDeathOf

8 ) where

10import Data.Accessor ((^.), (^=))

11import Data.Binary (Binary, encode)

12+import qualified Data.Map as Map (partitionWithKey, elems)

12import qualified Data.ByteString.Lazy as BSL (toChunks)

13import qualified Data.ByteString.Lazy as BSL (toChunks)

13import qualified Data.ByteString as BSS (ByteString)

14import qualified Data.ByteString as BSS (ByteString)

14import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)

15import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)

15import Control.Concurrent.Chan (writeChan)

16import Control.Concurrent.Chan (writeChan)

16−import Control.Monad (unless, when)

17+import Control.Monad (unless)

17import qualified Network.Transport as NT

18import qualified Network.Transport as NT

18 ( Connection

19 ( Connection

19 , send

20 , send

@@ -25,6 +26,7 @@ import qualified Network.Transport as NT

25import Control.Distributed.Process.Internal.Types

26import Control.Distributed.Process.Internal.Types

26 ( LocalNode(localState, localEndPoint, localCtrlChan)

27 ( LocalNode(localState, localEndPoint, localCtrlChan)

27 , Identifier

28 , Identifier

29+ , localConnections

28 , localConnectionBetween

30 , localConnectionBetween

29 , nodeAddress

31 , nodeAddress

30 , nodeOf

32 , nodeOf

@@ -33,6 +35,11 @@ import Control.Distributed.Process.Internal.Types

33 , NCMsg(..)

35 , NCMsg(..)

34 , ProcessSignal(Died)

36 , ProcessSignal(Died)

35 , DiedReason(DiedDisconnect)

37 , DiedReason(DiedDisconnect)

38+ , ImplicitReconnect(WithImplicitReconnect)

39+ , NodeId

40+ , ProcessId(processNodeId)

41+ , SendPortId(sendPortProcessId)

42+ , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)

36 )

43 )

37import Control.Distributed.Process.Serializable (Serializable)

44import Control.Distributed.Process.Serializable (Serializable)

@@ -40,9 +47,6 @@ import Control.Distributed.Process.Serializable (Serializable)

40-- Message sending --

47-- Message sending --

41--------------------------------------------------------------------------------

48--------------------------------------------------------------------------------

43−data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

44− deriving (Eq, Show)

45−

46sendPayload :: LocalNode

50sendPayload :: LocalNode

47 -> Identifier

51 -> Identifier

48 -> Identifier

52 -> Identifier

@@ -50,7 +54,7 @@ sendPayload :: LocalNode

50 -> [BSS.ByteString]

54 -> [BSS.ByteString]

51 -> IO ()

55 -> IO ()

52sendPayload node from to implicitReconnect payload = do

56sendPayload node from to implicitReconnect payload = do

53− mConn <- connBetween node from to

57+ mConn <- connBetween node from to implicitReconnect

54 didSend <- case mConn of

58 didSend <- case mConn of

55 Just conn -> do

59 Just conn -> do

56 didSend <- NT.send conn payload

60 didSend <- NT.send conn payload

@@ -63,11 +67,6 @@ sendPayload node from to implicitReconnect payload = do

63 { ctrlMsgSender = to

67 { ctrlMsgSender = to

64 , ctrlMsgSignal = Died to DiedDisconnect

68 , ctrlMsgSignal = Died to DiedDisconnect

65 }

69 }

66− when (implicitReconnect == WithImplicitReconnect) $ do

67− disconnect node from to

68− -- Try once more

69− -- (TODO: not quite sure this is the Right Thing)

70− sendPayload node from to NoImplicitReconnect payload

72sendBinary :: Binary a

71sendBinary :: Binary a

73 => LocalNode

72 => LocalNode

@@ -89,12 +88,16 @@ sendMessage :: Serializable a

89sendMessage node from to implicitReconnect =

88sendMessage node from to implicitReconnect =

90 sendPayload node from to implicitReconnect . messageToPayload . createMessage

89 sendPayload node from to implicitReconnect . messageToPayload . createMessage

92−setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)

91+setupConnBetween :: LocalNode

92+ -> Identifier

93+ -> Identifier

94+ -> ImplicitReconnect

95+ -> IO (Maybe NT.Connection)

93−setupConnBetween node from to = do

96+setupConnBetween node from to implicitReconnect = do

94− mConn <- NT.connect endPoint

97+ mConn <- NT.connect endPoint

95− (nodeAddress . nodeOf $ to)

98+ (nodeAddress . nodeOf $ to)

96− NT.ReliableOrdered

99+ NT.ReliableOrdered

97− NT.defaultConnectHints

100+ NT.defaultConnectHints

98 case mConn of

101 case mConn of

99 Right conn -> do

102 Right conn -> do

100 didSend <- NT.send conn (BSL.toChunks . encode $ to)

103 didSend <- NT.send conn (BSL.toChunks . encode $ to)

@@ -103,7 +106,7 @@ setupConnBetween node from to = do

103 return Nothing

106 return Nothing

104 Right () -> do

107 Right () -> do

105 modifyMVar_ nodeState $ return .

108 modifyMVar_ nodeState $ return .

106− (localConnectionBetween from to ^= Just conn)

109+ (localConnectionBetween from to ^= Just (conn, implicitReconnect))

107 return $ Just conn

110 return $ Just conn

108 Left _ ->

111 Left _ ->

109 return Nothing

112 return Nothing

@@ -111,12 +114,18 @@ setupConnBetween node from to = do

111 endPoint = localEndPoint node

114 endPoint = localEndPoint node

112 nodeState = localState node

115 nodeState = localState node

113

116

114−connBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)

117+connBetween :: LocalNode

118+ -> Identifier

119+ -> Identifier

120+ -> ImplicitReconnect

121+ -> IO (Maybe NT.Connection)

115−connBetween node from to = do

122+connBetween node from to implicitReconnect = do

116 mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)

123 mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)

117 case mConn of

124 case mConn of

125+ Just (conn, _) ->

118− Just conn -> return $ Just conn

126+ return $ Just conn

127+ Nothing ->

119− Nothing -> setupConnBetween node from to

128+ setupConnBetween node from to implicitReconnect

120 where

129 where

121 nodeState = localState node

130 nodeState = localState node

122

131

@@ -126,6 +135,35 @@ disconnect node from to =

126 case st ^. localConnectionBetween from to of

135 case st ^. localConnectionBetween from to of

127 Nothing ->

136 Nothing ->

128 return st

137 return st

129− Just conn -> do

138+ Just (conn, _) -> do

130 NT.close conn

139 NT.close conn

131 return (localConnectionBetween from to ^= Nothing $ st)

140 return (localConnectionBetween from to ^= Nothing $ st)

141+

142+closeImplicitReconnections :: LocalNode -> Identifier -> IO ()

143+closeImplicitReconnections node to =

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

145+ let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'

146+ shouldClose _ _ = False

147+ let (affected, unaffected) = Map.partitionWithKey shouldClose (st ^. localConnections)

148+ mapM_ (NT.close . fst) (Map.elems affected)

149+ return (localConnections ^= unaffected $ st)

150+

151+-- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node)

152+-- implies the death of @b@ (for instance, a process on that node)

153+impliesDeathOf :: Identifier

154+ -> Identifier

155+ -> Bool

156+NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' =

157+ nid' == nid

158+NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =

159+ processNodeId pid == nid

160+NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =

161+ processNodeId (sendPortProcessId cid) == nid

162+ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =

163+ pid' == pid

164+ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =

165+ sendPortProcessId cid == pid

166+SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =

167+ cid' == cid

168+_ `impliesDeathOf` _ =

169+ False

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

@@ -121,11 +121,11 @@ import Control.Distributed.Process.Internal.Types

121 , WhereIsReply(..)

122 , createMessage

123 , runLocalProcess

124+ , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

124 )

125 )

125import Control.Distributed.Process.Internal.Messaging

126import Control.Distributed.Process.Internal.Messaging

126 ( sendMessage

127 ( sendMessage

127 , sendBinary

128 , sendBinary

128− , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

129 , disconnect

130 )

131

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+14−4

@@ -19,6 +19,7 @@ module Control.Distributed.Process.Internal.Types

19 , LocalProcessState(..)

20 , Process(..)

21 , runLocalProcess

22+ , ImplicitReconnect(..)

22 -- * Typed channels

23 -- * Typed channels

23 , LocalSendPortId

24 , LocalSendPortId

24 , SendPortId(..)

25 , SendPortId(..)

@@ -181,12 +182,21 @@ data LocalNode = LocalNode

181 , remoteTable :: RemoteTable

182 , remoteTable :: RemoteTable

182 }

183 }

183

184

185+data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

186+ deriving (Eq, Show)

187+

184-- | Local node state

188-- | Local node state

185data LocalNodeState = LocalNodeState

189data LocalNodeState = LocalNodeState

190+ { -- | Processes running on this node

186− { _localProcesses :: !(Map LocalProcessId LocalProcess)

191+ _localProcesses :: !(Map LocalProcessId LocalProcess)

192+ -- | Counter to assign PIDs

187 , _localPidCounter :: !Int32

193 , _localPidCounter :: !Int32

194+ -- | The 'unique' value used to create PIDs (so that processes on

195+ -- restarted nodes have new PIDs)

188 , _localPidUnique :: !Int32

196 , _localPidUnique :: !Int32

197+ -- | Outgoing connections

189− , _localConnections :: !(Map (Identifier, Identifier) NT.Connection)

198+ , _localConnections :: !(Map (Identifier, Identifier)

199+ (NT.Connection, ImplicitReconnect))

190 }

200 }

191

201

192-- | Processes running on our local node

202-- | Processes running on our local node

@@ -509,13 +519,13 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =

509localPidUnique :: Accessor LocalNodeState Int32

519localPidUnique :: Accessor LocalNodeState Int32

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

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

511

521

512−localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) NT.Connection)

522+localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect))

513localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

523localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

514

524

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

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

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

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

517

527

518−localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe NT.Connection)

528+localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect))

519localConnectionBetween from to = localConnections >>> DAC.mapMaybe (from, to)

529localConnectionBetween from to = localConnections >>> DAC.mapMaybe (from, to)

520

530

521monitorCounter :: Accessor LocalProcessState Int32

531monitorCounter :: Accessor LocalProcessState Int32

distributed-process/src/Control/Distributed/Process/Node.hs+6−25

@@ -116,13 +116,15 @@ import Control.Distributed.Process.Internal.Types

116 , createMessage

117 , runLocalProcess

118 , firstNonReservedProcessId

119+ , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

119 )

120 )

120import Control.Distributed.Process.Serializable (Serializable)

121import Control.Distributed.Process.Serializable (Serializable)

121import Control.Distributed.Process.Internal.Messaging

122import Control.Distributed.Process.Internal.Messaging

122 ( sendBinary

123 ( sendBinary

123 , sendMessage

124 , sendMessage

124 , sendPayload

125 , sendPayload

125− , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

126+ , closeImplicitReconnections

127+ , impliesDeathOf

126 )

128 )

127import Control.Distributed.Process.Internal.Primitives (expect, register, finally)

129import Control.Distributed.Process.Internal.Primitives (expect, register, finally)

128import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

130import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

@@ -251,7 +253,7 @@ forkProcess node proc = modifyMVar (localState node) startProcess

251 cleanupProcess pid st = do

253 cleanupProcess pid st = do

252 let pid' = ProcessIdentifier pid

254 let pid' = ProcessIdentifier pid

253 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)

255 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)

254− mapM_ NT.close (Map.elems affected)

256+ mapM_ (NT.close . fst) (Map.elems affected)

255 return $ (localProcessWithId (processLocalId pid) ^= Nothing)

257 return $ (localProcessWithId (processLocalId pid) ^= Nothing)

256 . (localConnections ^= unaffected)

258 . (localConnections ^= unaffected)

257 $ st

259 $ st

@@ -364,12 +366,11 @@ handleIncomingMessages node = go initConnectionState

364 , ctrlMsgSignal = Died nid DiedDisconnect

366 , ctrlMsgSignal = Died nid DiedDisconnect

365 }

367 }

366 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))

368 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))

369+ closeImplicitReconnections node nid

367 go ( (incomingFrom theirAddr ^= Set.empty)

370 go ( (incomingFrom theirAddr ^= Set.empty)

368 . (incoming ^: Map.filterWithKey (const . notLost))

371 . (incoming ^: Map.filterWithKey (const . notLost))

369 $ st

372 $ st

370 )

373 )

371− -- TODO: we should remove the outgoing connections for which

372− -- we have an implicit reconnect

373 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->

374 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->

374 fail $ "Cloud Haskell fatal error: end point failed: " ++ str

375 fail $ "Cloud Haskell fatal error: end point failed: " ++ str

375 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->

376 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->

@@ -742,27 +743,7 @@ registryFor ident = registry >>> DAC.mapMaybe ident

742splitNotif :: Identifier

743splitNotif :: Identifier

743 -> Map Identifier a

744 -> Map Identifier a

744 -> (Map Identifier a, Map Identifier a)

745 -> (Map Identifier a, Map Identifier a)

745−splitNotif ident = Map.partitionWithKey (\k !_v -> impliesDeathOf ident k)

746+splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)

746−

747−-- | Does the death of one entity (node, project, channel) imply the death

748−-- of another?

749−impliesDeathOf :: Identifier -- ^ Who died

750− -> Identifier -- ^ Who's being watched

751− -> Bool -- ^ Does this death implies the death of the watchee?

752−NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' =

753− nid' == nid

754−NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =

755− processNodeId pid == nid

756−NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =

757− processNodeId (sendPortProcessId cid) == nid

758−ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =

759− pid' == pid

760−ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =

761− sendPortProcessId cid == pid

762−SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =

763− cid' == cid

764−_ `impliesDeathOf` _ =

765− False

766

747

767--------------------------------------------------------------------------------

748--------------------------------------------------------------------------------

768-- Strict evaluation of the state --

749-- Strict evaluation of the state --

distributed-process/tests/TestCH.hs+1

@@ -621,6 +621,7 @@ testReconnect transport transportInternals = do

621 liftIO $ do

622 sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)

623 sClose sock

624+ threadDelay 10000

624

625

625 -- Should not arrive

626 -- Should not arrive

626 send them "message 2"

627 send them "message 2"

distributed-process/tests/TestClosure.hs+9−12

@@ -355,6 +355,12 @@ testTDict transport rtable = do

355 liftIO $ putMVar done ()

356 takeMVar done

357

358+simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process ()

359+simulateNetworkFailure transportInternals fr to = liftIO $ do

360+ threadDelay 10000

361+ sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to)

362+ sClose sock

363+ threadDelay 10000

358

364

359testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO ()

365testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO ()

360testSpawnReconnect transport rtable transportInternals = do

366testSpawnReconnect transport rtable transportInternals = do

@@ -370,23 +376,14 @@ testSpawnReconnect transport rtable transportInternals = do

370

376

371 forkProcess node2 $ do

377 forkProcess node2 $ do

372 _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)

378 _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)

373−

374− -- Simulate network failure

375− liftIO $ do

376− sock <- socketBetween transportInternals (nodeAddress nid2) (nodeAddress nid1)

379+ simulateNetworkFailure transportInternals nid2 nid1

377− sClose sock

378−

379− -- Implicit reconnect: spawn happens; but since we also receive the

380− -- monitor signal, the process might not actually be started

381 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)

380 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)

382−

383− -- As does this one

384 _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)

381 _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)

385

382

386− liftIO $ threadDelay 10000

383+ liftIO $ threadDelay 100000

387

384

388 count <- liftIO $ takeMVar iv

385 count <- liftIO $ takeMVar iv

389− True <- return $ count == 2 || count == 3 -- Depending on whether we get the monitor notification first or not

386+ True <- return $ count == 3

390

387

391 liftIO $ putMVar done ()

388 liftIO $ putMVar done ()

392

389