@@ -19,16 +19,13 @@ import Data.Binary (decode)
19import Data.Map (Map)
19import Data.Map (Map)
20import qualified Data.Map as Map
20import qualified Data.Map as Map
26 , partitionWithKey
23 , partitionWithKey
30import Data.Set (Set)
27import Data.Set (Set)
31−import qualified Data.Set as Set (empty, insert, delete, member, filter)
28+import qualified Data.Set as Set (empty, insert, delete, member)
32import Data.Foldable (forM_)
29import Data.Foldable (forM_)
33import Data.Maybe (isJust)
30import Data.Maybe (isJust)
34import Data.Typeable (Typeable)
31import Data.Typeable (Typeable)
@@ -67,6 +64,8 @@ import qualified Network.Transport as NT
67 , ConnectionId
64 , ConnectionId
68 , Connection
65 , Connection
68+ , Reliability(ReliableOrdered)
71import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
70import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
72import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)
71import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)
@@ -257,59 +256,77 @@ forkProcess node proc = modifyMVar (localState node) startProcess
257 . (localConnections ^= unaffected)
256 . (localConnections ^= unaffected)
260−-- TODO: invariant: if P reconnects to Q, Q must consider its connection to be broken at that point
259+--------------------------------------------------------------------------------
260+-- Handle incoming messages --
261+--------------------------------------------------------------------------------
263+type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
265+data IncomingTarget =
267+ | ToProc LocalProcess
268+ | ToChan TypedChannel
271+data ConnectionState = ConnectionState {
272+ _incoming :: !(Map NT.ConnectionId IncomingConnection)
273+ , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
276+initConnectionState :: ConnectionState
277+initConnectionState = ConnectionState {
278+ _incoming = Map.empty
279+ , _incomingFrom = Map.empty
282+incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
283+incoming = accessor _incoming (\conns st -> st { _incoming = conns })
285+incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
286+incomingAt cid = incoming >>> DAC.mapMaybe cid
288+incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
289+incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
291+ aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
261handleIncomingMessages :: LocalNode -> IO ()
293handleIncomingMessages :: LocalNode -> IO ()
262−handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
294+handleIncomingMessages node = go initConnectionState
264− go :: Set NT.ConnectionId -- ^ Connections whose purpose we don't yet know
296+ go :: ConnectionState -> IO ()
265− -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes
297+ go !st = do
266− -> Map NT.ConnectionId TypedChannel -- ^ Connections to typed channels
267− -> Set NT.ConnectionId -- ^ Connections to our controller
269− go !uninitConns !procs !chans !ctrls = do
270 event <- NT.receive endpoint
298 event <- NT.receive endpoint
271 case event of
299 case event of
272− NT.ConnectionOpened cid _rel _theirAddr ->
300+ NT.ConnectionOpened cid rel theirAddr ->
273− -- TODO: Check if _rel is ReliableOrdered, and if not, treat as
301+ if rel == NT.ReliableOrdered
274− -- (**) below.
302+ then go ( (incomingAt cid ^= Just (theirAddr, Uninit))
275− go (Set.insert cid uninitConns) procs chans ctrls
303+ . (incomingFrom theirAddr ^: Set.insert cid)
306+ else invalidRequest cid st
276− NT.Received cid payload ->
307+ NT.Received cid payload ->
277− case ( Map.lookup cid procs
308+ case st ^. incomingAt cid of
278− , Map.lookup cid chans
279− , cid `Set.member` ctrls
280− , cid `Set.member` uninitConns
282− (Just proc, _, _, _) -> do
309+ Just (_, ToProc proc) -> do
283 let msg = payloadToMessage payload
310 let msg = payloadToMessage payload
284 enqueue (processQueue proc) msg
311 enqueue (processQueue proc) msg
285− go uninitConns procs chans ctrls
312+ go st
286− (_, Just (TypedChannel chan), _, _) -> do
313+ Just (_, ToChan (TypedChannel chan)) -> do
287 atomically $ writeTChan chan . decode . BSL.fromChunks $ payload
314 atomically $ writeTChan chan . decode . BSL.fromChunks $ payload
288− go uninitConns procs chans ctrls
315+ go st
289− (_, _, True, _) -> do
316+ Just (_, ToNode) -> do
290 let ctrlMsg = decode . BSL.fromChunks $ payload
317 let ctrlMsg = decode . BSL.fromChunks $ payload
291 writeChan ctrlChan ctrlMsg
318 writeChan ctrlChan ctrlMsg
292− go uninitConns procs chans ctrls
319+ go st
293− (_, _, _, True) ->
320+ Just (src, Uninit) ->
294 case decode (BSL.fromChunks payload) of
321 case decode (BSL.fromChunks payload) of
295 ProcessIdentifier pid -> do
322 ProcessIdentifier pid -> do
296 let lpid = processLocalId pid
323 let lpid = processLocalId pid
297 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
324 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
298 case mProc of
325 case mProc of
299 Just proc ->
326 Just proc ->
300− go (Set.delete cid uninitConns)
327+ go (incomingAt cid ^= Just (src, ToProc proc) $ st)
301− (Map.insert cid proc procs)
304 Nothing ->
328 Nothing ->
305− -- Request for an unknown process.
329+ invalidRequest cid st
307− -- TODO: We should treat this as a fatal error on the
308− -- part of the remote node. That is, we should report the
309− -- remote node as having died, and we should close
310− -- incoming connections (this requires a Transport layer
312− go (Set.delete cid uninitConns) procs chans ctrls
313 SendPortIdentifier chId -> do
330 SendPortIdentifier chId -> do
314 let lcid = sendPortLocalId chId
331 let lcid = sendPortLocalId chId
315 lpid = processLocalId (sendPortProcessId chId)
332 lpid = processLocalId (sendPortProcessId chId)
@@ -318,33 +335,27 @@ handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
318 Just proc -> do
335 Just proc -> do
319 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
336 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
320 case mChannel of
337 case mChannel of
321− Just channel ->
338+ Just channel ->
322− go (Set.delete cid uninitConns)
339+ go (incomingAt cid ^= Just (src, ToChan channel) $ st)
324− (Map.insert cid channel chans)
326 Nothing ->
340 Nothing ->
327− -- Unknown typed channel
341+ invalidRequest cid st
329− go (Set.delete cid uninitConns) procs chans ctrls
330 Nothing ->
342 Nothing ->
331− -- Unknown process
343+ invalidRequest cid st
333− go (Set.delete cid uninitConns) procs chans ctrls
334− NodeIdentifier _ ->
344+ NodeIdentifier nid ->
335− go (Set.delete cid uninitConns)
345+ if nid == localNodeId node
336− procs
346+ then go (incomingAt cid ^= Just (src, ToNode) $ st)
337− chans
347+ else invalidRequest cid st
338− (Set.insert cid ctrls)
348+ Nothing ->
339− _ ->
349+ invalidRequest cid st
340− -- Unexpected message
342− go uninitConns procs chans ctrls
343 NT.ConnectionClosed cid ->
350 NT.ConnectionClosed cid ->
344− go (Set.delete cid uninitConns)
351+ case st ^. incomingAt cid of
345− (Map.delete cid procs)
352+ Nothing ->
346− (Map.delete cid chans)
353+ invalidRequest cid st
355+ go ( (incomingAt cid ^= Nothing)
347− (Set.delete cid ctrls)
356+ . (incomingFrom src ^: Set.delete cid)
348 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
359 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
349 -- [Unified table 9, rule node_disconnect]
360 -- [Unified table 9, rule node_disconnect]
350 let nid = NodeIdentifier $ NodeId theirAddr
361 let nid = NodeIdentifier $ NodeId theirAddr
@@ -352,17 +363,13 @@ handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
352 { ctrlMsgSender = nid
363 { ctrlMsgSender = nid
353 , ctrlMsgSignal = Died nid DiedDisconnect
364 , ctrlMsgSignal = Died nid DiedDisconnect
355− -- TODO: we should remove the incoming transactions that were lost
366+ let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
367+ go ( (incomingFrom theirAddr ^= Set.empty)
368+ . (incoming ^: Map.filterWithKey (const . notLost))
356− -- TODO: and we should remove the outgoing connections for which
371+ -- TODO: we should remove the outgoing connections for which
357 -- we have an implicit reconnect
372 -- we have an implicit reconnect
359− let notRemoved k = k `notElem` cids
360− go (Set.filter notRemoved uninitConns)
361− (Map.filterWithKey (const . notRemoved) procs)
362− (Map.filterWithKey (const . notRemoved) chans)
363− (Set.filter notRemoved ctrls)
365− go uninitConns procs chans ctrls
366 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
373 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
367 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
374 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
368 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
375 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
@@ -373,7 +380,17 @@ handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
373 -- If we received a multicast message, something went horribly wrong
380 -- If we received a multicast message, something went horribly wrong
374 -- and we just give up
381 -- and we just give up
375 fail "Cloud Haskell fatal error: received unexpected multicast"
382 fail "Cloud Haskell fatal error: received unexpected multicast"
384+ invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
385+ invalidRequest cid st = do
386+ -- TODO: We should treat this as a fatal error on the part of the remote
387+ -- node. That is, we should report the remote node as having died, and we
388+ -- should close incoming connections (this requires a Transport layer
390+ go ( (incomingAt cid ^= Nothing)
377 state = localState node
394 state = localState node
378 endpoint = localEndPoint node
395 endpoint = localEndPoint node
379 ctrlChan = localCtrlChan node
396 ctrlChan = localCtrlChan node