git.iko.soy/distributed-process
on 8aed555 help sign in
distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs
207 lines 6.9 kB history raw
1module Control.Distributed.Process.Internal.Messaging
2 ( sendPayload
3 , sendBinary
4 , sendMessage
5 , disconnect
6 , closeImplicitReconnections
7 , impliesDeathOf
8 , sendCtrlMsg
9 ) where
11import Data.Accessor ((^.), (^=))
12import Data.Binary (Binary, encode)
13import qualified Data.Map as Map (partitionWithKey, elems)
14import qualified Data.ByteString.Lazy as BSL (toChunks)
15import qualified Data.ByteString as BSS (ByteString)
16import Control.Distributed.Process.Serializable ()
18import Control.Concurrent (forkIO)
19import Control.Concurrent.Chan (writeChan)
20import Control.Exception (mask_)
21import Control.Monad (unless)
22import Control.Monad.IO.Class (liftIO)
23import Control.Monad.Reader (ask)
24import qualified Network.Transport as NT
25 ( Connection
26 , send
27 , defaultConnectHints
28 , connect
29 , Reliability(ReliableOrdered)
30 , close
31 )
32import Control.Distributed.Process.Internal.Types
33 ( LocalNode(localState, localEndPoint, localCtrlChan)
34 , withValidLocalState
35 , modifyValidLocalState
36 , modifyValidLocalState_
37 , Identifier
38 , localConnections
39 , localConnectionBetween
40 , nodeAddress
41 , nodeOf
42 , messageToPayload
43 , createMessage
44 , NCMsg(..)
45 , ProcessSignal(Died)
46 , DiedReason(DiedDisconnect)
47 , ImplicitReconnect(WithImplicitReconnect)
48 , NodeId(..)
49 , ProcessId(..)
50 , LocalNode(..)
51 , LocalProcess(..)
52 , Process(..)
53 , SendPortId(sendPortProcessId)
54 , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)
55 )
56import Control.Distributed.Process.Serializable (Serializable)
57import Data.Foldable (forM_)
59--------------------------------------------------------------------------------
60-- Message sending --
61--------------------------------------------------------------------------------
63sendPayload :: LocalNode
64 -> Identifier
65 -> Identifier
66 -> ImplicitReconnect
67 -> [BSS.ByteString]
68 -> IO ()
69sendPayload node from to implicitReconnect payload = do
70 mConn <- connBetween node from to implicitReconnect
71 didSend <- case mConn of
72 Just conn -> do
73 didSend <- NT.send conn payload
74 case didSend of
75 Left _err -> return False
76 Right () -> return True
77 Nothing -> return False
78 unless didSend $ do
79 writeChan (localCtrlChan node) NCMsg
80 { ctrlMsgSender = to
81 , ctrlMsgSignal = Died (NodeIdentifier $ nodeOf to) DiedDisconnect
82 }
84sendBinary :: Binary a
85 => LocalNode
86 -> Identifier
87 -> Identifier
88 -> ImplicitReconnect
89 -> a
90 -> IO ()
91sendBinary node from to implicitReconnect
92 = sendPayload node from to implicitReconnect . BSL.toChunks . encode
94sendMessage :: Serializable a
95 => LocalNode
96 -> Identifier
97 -> Identifier
98 -> ImplicitReconnect
99 -> a
100 -> IO ()
101sendMessage node from to implicitReconnect =
102 sendPayload node from to implicitReconnect . messageToPayload . createMessage
104setupConnBetween :: LocalNode
105 -> Identifier
106 -> Identifier
107 -> ImplicitReconnect
108 -> IO (Maybe NT.Connection)
109setupConnBetween node from to implicitReconnect = do
110 mConn <- NT.connect (localEndPoint node)
111 (nodeAddress . nodeOf $ to)
112 NT.ReliableOrdered
113 NT.defaultConnectHints
114 case mConn of
115 Right conn -> do
116 didSend <- NT.send conn (BSL.toChunks . encode $ to)
117 case didSend of
118 Left _ ->
119 return Nothing
120 Right () -> do
121 modifyValidLocalState_ node $
122 return . (localConnectionBetween from to ^= Just (conn, implicitReconnect))
123 return $ Just conn
124 Left _ ->
125 return Nothing
127connBetween :: LocalNode
128 -> Identifier
129 -> Identifier
130 -> ImplicitReconnect
131 -> IO (Maybe NT.Connection)
132connBetween node from to implicitReconnect = do
133 mConn <- withValidLocalState node $
134 return . (^. localConnectionBetween from to)
135 case mConn of
136 Just (conn, _) ->
137 return $ Just conn
138 Nothing ->
139 setupConnBetween node from to implicitReconnect
141disconnect :: LocalNode -> Identifier -> Identifier -> IO ()
142disconnect node from to = mask_ $ do
143 mio <- modifyValidLocalState node $ \vst ->
144 case vst ^. localConnectionBetween from to of
145 Nothing ->
146 return (vst, return ())
147 Just (conn, _) -> do
148 return ( localConnectionBetween from to ^= Nothing $ vst
149 , NT.close conn
150 )
151 forM_ mio forkIO
153closeImplicitReconnections :: LocalNode -> Identifier -> IO ()
154closeImplicitReconnections node to = mask_ $ do
155 mconns <- modifyValidLocalState node $ \vst -> do
156 let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'
157 shouldClose _ _ = False
158 let (affected, unaffected) =
159 Map.partitionWithKey shouldClose (vst ^. localConnections)
160 return ( localConnections ^= unaffected $ vst
161 , map fst $ Map.elems affected
162 )
163 forM_ mconns $ forkIO . mapM_ NT.close
165-- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node)
166-- implies the death of @b@ (for instance, a process on that node)
167impliesDeathOf :: Identifier
168 -> Identifier
169 -> Bool
170NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' =
171 nid' == nid
172NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =
173 processNodeId pid == nid
174NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =
175 processNodeId (sendPortProcessId cid) == nid
176ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =
177 pid' == pid
178ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =
179 sendPortProcessId cid == pid
180SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =
181 cid' == cid
182_ `impliesDeathOf` _ =
183 False
186-- Send a control message. Evaluates the message to HNF before sending it.
187--
188-- The message shouldn't produce more errors when further evaluated. If
189-- evaluation threw errors the node controller or the receiver would crash when
190-- inspecting it.
191sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
192 -> ProcessSignal -- ^ Message to send
193 -> Process ()
194sendCtrlMsg mNid signal = do
195 proc <- ask
196 let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)
197 , ctrlMsgSignal = signal
198 }
199 case mNid of
200 Nothing -> do
201 liftIO $ writeChan (localCtrlChan (processNode proc)) $! msg
202 Just nid ->
203 liftIO $ sendBinary (processNode proc)
204 (ProcessIdentifier (processId proc))
205 (NodeIdentifier nid)
206 WithImplicitReconnect
207 msg