git.iko.soy/distributed-process
on fix-spawnChan help sign in
distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs
169 lines 5.6 kB history raw
1module Control.Distributed.Process.Internal.Messaging
2 ( sendPayload
3 , sendBinary
4 , sendMessage
5 , disconnect
6 , closeImplicitReconnections
7 , impliesDeathOf
8 ) where
10import Data.Accessor ((^.), (^=))
11import Data.Binary (Binary, encode)
12import qualified Data.Map as Map (partitionWithKey, elems)
13import qualified Data.ByteString.Lazy as BSL (toChunks)
14import qualified Data.ByteString as BSS (ByteString)
15import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)
16import Control.Concurrent.Chan (writeChan)
17import Control.Monad (unless)
18import qualified Network.Transport as NT
19 ( Connection
20 , send
21 , defaultConnectHints
22 , connect
23 , Reliability(ReliableOrdered)
24 , close
25 )
26import Control.Distributed.Process.Internal.Types
27 ( LocalNode(localState, localEndPoint, localCtrlChan)
28 , Identifier
29 , localConnections
30 , localConnectionBetween
31 , nodeAddress
32 , nodeOf
33 , messageToPayload
34 , createMessage
35 , NCMsg(..)
36 , ProcessSignal(Died)
37 , DiedReason(DiedDisconnect)
38 , ImplicitReconnect(WithImplicitReconnect)
39 , NodeId
40 , ProcessId(processNodeId)
41 , SendPortId(sendPortProcessId)
42 , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)
43 )
44import Control.Distributed.Process.Serializable (Serializable)
46--------------------------------------------------------------------------------
47-- Message sending --
48--------------------------------------------------------------------------------
50sendPayload :: LocalNode
51 -> Identifier
52 -> Identifier
53 -> ImplicitReconnect
54 -> [BSS.ByteString]
55 -> IO ()
56sendPayload node from to implicitReconnect payload = do
57 mConn <- connBetween node from to implicitReconnect
58 didSend <- case mConn of
59 Just conn -> do
60 didSend <- NT.send conn payload
61 case didSend of
62 Left _err -> return False
63 Right () -> return True
64 Nothing -> return False
65 unless didSend $ do
66 writeChan (localCtrlChan node) NCMsg
67 { ctrlMsgSender = to
68 , ctrlMsgSignal = Died to DiedDisconnect
69 }
71sendBinary :: Binary a
72 => LocalNode
73 -> Identifier
74 -> Identifier
75 -> ImplicitReconnect
76 -> a
77 -> IO ()
78sendBinary node from to implicitReconnect
79 = sendPayload node from to implicitReconnect . BSL.toChunks . encode
81sendMessage :: Serializable a
82 => LocalNode
83 -> Identifier
84 -> Identifier
85 -> ImplicitReconnect
86 -> a
87 -> IO ()
88sendMessage node from to implicitReconnect =
89 sendPayload node from to implicitReconnect . messageToPayload . createMessage
91setupConnBetween :: LocalNode
92 -> Identifier
93 -> Identifier
94 -> ImplicitReconnect
95 -> IO (Maybe NT.Connection)
96setupConnBetween node from to implicitReconnect = do
97 mConn <- NT.connect endPoint
98 (nodeAddress . nodeOf $ to)
99 NT.ReliableOrdered
100 NT.defaultConnectHints
101 case mConn of
102 Right conn -> do
103 didSend <- NT.send conn (BSL.toChunks . encode $ to)
104 case didSend of
105 Left _ ->
106 return Nothing
107 Right () -> do
108 modifyMVar_ nodeState $ return .
109 (localConnectionBetween from to ^= Just (conn, implicitReconnect))
110 return $ Just conn
111 Left _ ->
112 return Nothing
113 where
114 endPoint = localEndPoint node
115 nodeState = localState node
117connBetween :: LocalNode
118 -> Identifier
119 -> Identifier
120 -> ImplicitReconnect
121 -> IO (Maybe NT.Connection)
122connBetween node from to implicitReconnect = do
123 mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)
124 case mConn of
125 Just (conn, _) ->
126 return $ Just conn
127 Nothing ->
128 setupConnBetween node from to implicitReconnect
129 where
130 nodeState = localState node
132disconnect :: LocalNode -> Identifier -> Identifier -> IO ()
133disconnect node from to =
134 modifyMVar_ (localState node) $ \st ->
135 case st ^. localConnectionBetween from to of
136 Nothing ->
137 return st
138 Just (conn, _) -> do
139 NT.close conn
140 return (localConnectionBetween from to ^= Nothing $ st)
142closeImplicitReconnections :: LocalNode -> Identifier -> IO ()
143closeImplicitReconnections 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)
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)
153impliesDeathOf :: Identifier
154 -> Identifier
155 -> Bool
156NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' =
157 nid' == nid
158NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =
159 processNodeId pid == nid
160NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =
161 processNodeId (sendPortProcessId cid) == nid
162ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =
163 pid' == pid
164ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =
165 sendPortProcessId cid == pid
166SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =
167 cid' == cid
168_ `impliesDeathOf` _ =
169 False