@@ -1,19 +1,20 @@
11module Control.Distributed.Process.Internal.Messaging
2− ( -- * Message sending
3− ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
4− , sendPayload
2+ ( sendPayload
53 , sendBinary
64 , sendMessage
75 , disconnect
6+ , closeImplicitReconnections
7+ , impliesDeathOf
88 ) where
99
1010import Data.Accessor ((^.), (^=))
1111import Data.Binary (Binary, encode)
12+import qualified Data.Map as Map (partitionWithKey, elems)
1213import qualified Data.ByteString.Lazy as BSL (toChunks)
1314import qualified Data.ByteString as BSS (ByteString)
1415import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)
1516import Control.Concurrent.Chan (writeChan)
16−import Control.Monad (unless, when)
17+import Control.Monad (unless)
1718import qualified Network.Transport as NT
1819 ( Connection
1920 , send
@@ -25,6 +26,7 @@ import qualified Network.Transport as NT
2526import Control.Distributed.Process.Internal.Types
2627 ( LocalNode(localState, localEndPoint, localCtrlChan)
2728 , Identifier
29+ , localConnections
2830 , localConnectionBetween
2931 , nodeAddress
3032 , nodeOf
@@ -33,6 +35,11 @@ import Control.Distributed.Process.Internal.Types
3335 , NCMsg(..)
3436 , ProcessSignal(Died)
3537 , DiedReason(DiedDisconnect)
38+ , ImplicitReconnect(WithImplicitReconnect)
39+ , NodeId
40+ , ProcessId(processNodeId)
41+ , SendPortId(sendPortProcessId)
42+ , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)
3643 )
3744import Control.Distributed.Process.Serializable (Serializable)
3845
@@ -40,9 +47,6 @@ import Control.Distributed.Process.Serializable (Serializable)
4047-- Message sending --
4148--------------------------------------------------------------------------------
4249
43−data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect
44− deriving (Eq, Show)
45−
4650sendPayload :: LocalNode
4751 -> Identifier
4852 -> Identifier
@@ -50,7 +54,7 @@ sendPayload :: LocalNode
5054 -> [BSS.ByteString]
5155 -> IO ()
5256sendPayload node from to implicitReconnect payload = do
53− mConn <- connBetween node from to
57+ mConn <- connBetween node from to implicitReconnect
5458 didSend <- case mConn of
5559 Just conn -> do
5660 didSend <- NT.send conn payload
@@ -63,11 +67,6 @@ sendPayload node from to implicitReconnect payload = do
6367 { ctrlMsgSender = to
6468 , ctrlMsgSignal = Died to DiedDisconnect
6569 }
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
7170
7271sendBinary :: Binary a
7372 => LocalNode
@@ -89,12 +88,16 @@ sendMessage :: Serializable a
8988sendMessage node from to implicitReconnect =
9089 sendPayload node from to implicitReconnect . messageToPayload . createMessage
9190
92−setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)
93−setupConnBetween node from to = do
94− mConn <- NT.connect endPoint
95− (nodeAddress . nodeOf $ to)
96− NT.ReliableOrdered
97− NT.defaultConnectHints
91+setupConnBetween :: LocalNode
92+ -> Identifier
93+ -> Identifier
94+ -> ImplicitReconnect
95+ -> IO (Maybe NT.Connection)
96+setupConnBetween node from to implicitReconnect = do
97+ mConn <- NT.connect endPoint
98+ (nodeAddress . nodeOf $ to)
99+ NT.ReliableOrdered
100+ NT.defaultConnectHints
98101 case mConn of
99102 Right conn -> do
100103 didSend <- NT.send conn (BSL.toChunks . encode $ to)
@@ -103,7 +106,7 @@ setupConnBetween node from to = do
103106 return Nothing
104107 Right () -> do
105108 modifyMVar_ nodeState $ return .
106− (localConnectionBetween from to ^= Just conn)
109+ (localConnectionBetween from to ^= Just (conn, implicitReconnect))
107110 return $ Just conn
108111 Left _ ->
109112 return Nothing
@@ -111,12 +114,18 @@ setupConnBetween node from to = do
111114 endPoint = localEndPoint node
112115 nodeState = localState node
113116
114−connBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)
115−connBetween node from to = do
117+connBetween :: LocalNode
118+ -> Identifier
119+ -> Identifier
120+ -> ImplicitReconnect
121+ -> IO (Maybe NT.Connection)
122+connBetween node from to implicitReconnect = do
116123 mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)
117124 case mConn of
118− Just conn -> return $ Just conn
119− Nothing -> setupConnBetween node from to
125+ Just (conn, _) ->
126+ return $ Just conn
127+ Nothing ->
128+ setupConnBetween node from to implicitReconnect
120129 where
121130 nodeState = localState node
122131
@@ -126,6 +135,35 @@ disconnect node from to =
126135 case st ^. localConnectionBetween from to of
127136 Nothing ->
128137 return st
129− Just conn -> do
138+ Just (conn, _) -> do
130139 NT.close conn
131140 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