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