@@ -19,16 +19,13 @@ import Data.Binary (decode)
1919import Data.Map (Map)
2020import qualified Data.Map as Map
2121 ( empty
22− , lookup
23− , insert
24− , delete
2522 , toList
2623 , partitionWithKey
27− , filterWithKey
2824 , elems
25+ , filterWithKey
2926 )
3027import 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)
3229import Data.Foldable (forM_)
3330import Data.Maybe (isJust)
3431import Data.Typeable (Typeable)
@@ -67,6 +64,8 @@ import qualified Network.Transport as NT
6764 , ConnectionId
6865 , Connection
6966 , close
67+ , EndPointAddress
68+ , Reliability(ReliableOrdered)
7069 )
7170import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
7271import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)
@@ -257,59 +256,77 @@ forkProcess node proc = modifyMVar (localState node) startProcess
257256 . (localConnections ^= unaffected)
258257 $ st
259258
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+--------------------------------------------------------------------------------
262+
263+type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
264+
265+data IncomingTarget =
266+ Uninit
267+ | ToProc LocalProcess
268+ | ToChan TypedChannel
269+ | ToNode
270+
271+data ConnectionState = ConnectionState {
272+ _incoming :: !(Map NT.ConnectionId IncomingConnection)
273+ , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
274+ }
275+
276+initConnectionState :: ConnectionState
277+initConnectionState = ConnectionState {
278+ _incoming = Map.empty
279+ , _incomingFrom = Map.empty
280+ }
281+
282+incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
283+incoming = accessor _incoming (\conns st -> st { _incoming = conns })
284+
285+incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
286+incomingAt cid = incoming >>> DAC.mapMaybe cid
287+
288+incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
289+incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
290+ where
291+ aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
292+
261293handleIncomingMessages :: LocalNode -> IO ()
262−handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
294+handleIncomingMessages node = go initConnectionState
263295 where
264− go :: Set NT.ConnectionId -- ^ Connections whose purpose we don't yet know
265− -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes
266− -> Map NT.ConnectionId TypedChannel -- ^ Connections to typed channels
267− -> Set NT.ConnectionId -- ^ Connections to our controller
268− -> IO ()
269− go !uninitConns !procs !chans !ctrls = do
296+ go :: ConnectionState -> IO ()
297+ go !st = do
270298 event <- NT.receive endpoint
271299 case event of
272− NT.ConnectionOpened cid _rel _theirAddr ->
273− -- TODO: Check if _rel is ReliableOrdered, and if not, treat as
274− -- (**) below.
275− go (Set.insert cid uninitConns) procs chans ctrls
276− NT.Received cid payload ->
277− case ( Map.lookup cid procs
278− , Map.lookup cid chans
279− , cid `Set.member` ctrls
280− , cid `Set.member` uninitConns
281− ) of
282− (Just proc, _, _, _) -> do
300+ NT.ConnectionOpened cid rel theirAddr ->
301+ if rel == NT.ReliableOrdered
302+ then go ( (incomingAt cid ^= Just (theirAddr, Uninit))
303+ . (incomingFrom theirAddr ^: Set.insert cid)
304+ $ st
305+ )
306+ else invalidRequest cid st
307+ NT.Received cid payload ->
308+ case st ^. incomingAt cid of
309+ Just (_, ToProc proc) -> do
283310 let msg = payloadToMessage payload
284311 enqueue (processQueue proc) msg
285− go uninitConns procs chans ctrls
286− (_, Just (TypedChannel chan), _, _) -> do
312+ go st
313+ Just (_, ToChan (TypedChannel chan)) -> do
287314 atomically $ writeTChan chan . decode . BSL.fromChunks $ payload
288− go uninitConns procs chans ctrls
289− (_, _, True, _) -> do
315+ go st
316+ Just (_, ToNode) -> do
290317 let ctrlMsg = decode . BSL.fromChunks $ payload
291318 writeChan ctrlChan ctrlMsg
292− go uninitConns procs chans ctrls
293− (_, _, _, True) ->
319+ go st
320+ Just (src, Uninit) ->
294321 case decode (BSL.fromChunks payload) of
295322 ProcessIdentifier pid -> do
296323 let lpid = processLocalId pid
297324 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
298325 case mProc of
299326 Just proc ->
300− go (Set.delete cid uninitConns)
301− (Map.insert cid proc procs)
302− chans
303− ctrls
327+ go (incomingAt cid ^= Just (src, ToProc proc) $ st)
304328 Nothing ->
305− -- Request for an unknown process.
306− --
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
311− -- extension). (**)
312− go (Set.delete cid uninitConns) procs chans ctrls
329+ invalidRequest cid st
313330 SendPortIdentifier chId -> do
314331 let lcid = sendPortLocalId chId
315332 lpid = processLocalId (sendPortProcessId chId)
@@ -318,33 +335,27 @@ handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
318335 Just proc -> do
319336 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
320337 case mChannel of
321− Just channel ->
322− go (Set.delete cid uninitConns)
323− procs
324− (Map.insert cid channel chans)
325− ctrls
338+ Just channel ->
339+ go (incomingAt cid ^= Just (src, ToChan channel) $ st)
326340 Nothing ->
327− -- Unknown typed channel
328− -- TODO (**) above
329− go (Set.delete cid uninitConns) procs chans ctrls
341+ invalidRequest cid st
330342 Nothing ->
331− -- Unknown process
332− -- TODO (**) above
333− go (Set.delete cid uninitConns) procs chans ctrls
334− NodeIdentifier _ ->
335− go (Set.delete cid uninitConns)
336− procs
337− chans
338− (Set.insert cid ctrls)
339− _ ->
340− -- Unexpected message
341− -- TODO (**) above
342− go uninitConns procs chans ctrls
343+ invalidRequest cid st
344+ NodeIdentifier nid ->
345+ if nid == localNodeId node
346+ then go (incomingAt cid ^= Just (src, ToNode) $ st)
347+ else invalidRequest cid st
348+ Nothing ->
349+ invalidRequest cid st
343350 NT.ConnectionClosed cid ->
344− go (Set.delete cid uninitConns)
345− (Map.delete cid procs)
346− (Map.delete cid chans)
347− (Set.delete cid ctrls)
351+ case st ^. incomingAt cid of
352+ Nothing ->
353+ invalidRequest cid st
354+ Just (src, _) ->
355+ go ( (incomingAt cid ^= Nothing)
356+ . (incomingFrom src ^: Set.delete cid)
357+ $ st
358+ )
348359 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
349360 -- [Unified table 9, rule node_disconnect]
350361 let nid = NodeIdentifier $ NodeId theirAddr
@@ -352,17 +363,13 @@ handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
352363 { ctrlMsgSender = nid
353364 , ctrlMsgSignal = Died nid DiedDisconnect
354365 }
355− -- TODO: we should remove the incoming transactions that were lost
356− -- TODO: and we should remove the outgoing connections for which
366+ let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
367+ go ( (incomingFrom theirAddr ^= Set.empty)
368+ . (incoming ^: Map.filterWithKey (const . notLost))
369+ $ st
370+ )
371+ -- TODO: we should remove the outgoing connections for which
357372 -- we have an implicit reconnect
358− {-
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)
364− -}
365− go uninitConns procs chans ctrls
366373 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
367374 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
368375 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
@@ -373,7 +380,17 @@ handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
373380 -- If we received a multicast message, something went horribly wrong
374381 -- and we just give up
375382 fail "Cloud Haskell fatal error: received unexpected multicast"
376−
383+
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
389+ -- extension).
390+ go ( (incomingAt cid ^= Nothing)
391+ $ st
392+ )
393+
377394 state = localState node
378395 endpoint = localEndPoint node
379396 ctrlChan = localCtrlChan node