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