@@ -98,14 +98,15 @@ import Data.Map (Map)
9898import qualified Data.Map as Map (empty, lookup, insert, delete, keys, toList)
9999import qualified Data.List as List (delete)
100100import Data.Set (Set)
101−import qualified Data.Set as Set (empty, insert)
101+import qualified Data.Set as Set (empty, insert, delete, member)
102102import Data.Int (Int32)
103103import Data.Typeable (Typeable)
104104import Data.Foldable (forM_)
105−import Data.Maybe (isNothing)
106−import Control.Monad (void, liftM2, liftM3, when)
105+import Data.Maybe (isJust)
106+import Control.Monad (void, liftM2, liftM3, when, unless, forever)
107107import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
108108import Control.Monad.IO.Class (MonadIO, liftIO)
109+import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
109110import Control.Applicative ((<$>))
110111import Control.Category ((>>>))
111112import Control.Exception (Exception, throwIO, catch)
@@ -304,10 +305,10 @@ data NodeCtrlMsg = NodeCtrlMsg
304305 }
305306
306307data NodeCtrlState = NodeCtrlState
307− { -- Mapping from remote processes to linked local processes
308+ { -- Mapping from remote processes to linked local processes
308309 _nodeCtrlLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
309310 -- Mapping from remote processes to monitoring local processes
310− , _nodeCtrlMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set (MonitorRef))))
311+ , _nodeCtrlMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
311312 -- The node controller maintains its own set of connections
312313 -- TODO: still not convinced that this is correct
313314 , _nodeCtrlConns :: Map Identifier NT.Connection
@@ -325,118 +326,147 @@ initNodeCtrlState = NodeCtrlState
325326 , _nodeCtrlConns = Map.empty
326327 }
327328
328−nodeController :: LocalNode -> IO ()
329−nodeController node = go initNodeCtrlState
329+newtype NodeCtrl a = NodeCtrl { unNodeCtrl :: ReaderT LocalNode (StateT NodeCtrlState IO) a }
330+ deriving (Functor, Monad, MonadIO, MonadState NodeCtrlState, MonadReader LocalNode)
331+
332+nodeController :: NodeCtrl ()
333+nodeController = forever $ do
334+ node <- ask
335+ msg <- liftIO $ readChan (localCtrlChan node)
336+
337+ -- Forward the message if appropriate
338+ case destNid (ctrlMsgSignal msg) of
339+ Just nid' | nid' /= localNodeId node ->
340+ ctrlSendTo (Right nid') . BSL.toChunks . encode $ msg
341+ _ ->
342+ return ()
343+
344+ processCtrlMsg msg
345+
346+ctrlSendTo :: Identifier -> [BSS.ByteString] -> NodeCtrl ()
347+ctrlSendTo them payload = do
348+ mConn <- ctrlConnTo them
349+ didSend <- case mConn of
350+ Just conn -> do
351+ didSend <- liftIO $ NT.send conn payload
352+ case didSend of
353+ Left _ -> return False
354+ Right () -> return True
355+ Nothing ->
356+ return False
357+ unless didSend $ do
358+ -- [Unified: Table 9, rule node_disconnect]
359+ node <- ask
360+ liftIO . writeChan (localCtrlChan node) $ NodeCtrlMsg
361+ { ctrlMsgSender = them
362+ , ctrlMsgSignal = Died them DiedDisconnect
363+ }
364+
365+ctrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)
366+ctrlConnTo them = do
367+ mConn <- gets (^. nodeCtrlConnFor them)
368+ case mConn of
369+ Just conn -> return (Just conn)
370+ Nothing -> createCtrlConnTo them
371+
372+createCtrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)
373+createCtrlConnTo them = do
374+ node <- ask
375+ mConn <- liftIO $ NT.connect (localEndPoint node)
376+ addr
377+ NT.ReliableOrdered
378+ NT.defaultConnectHints
379+ case mConn of
380+ Right conn -> do
381+ didSend <- liftIO $ NT.send conn firstMsg
382+ case didSend of
383+ Left _ ->
384+ return Nothing
385+ Right () -> do
386+ modify $ nodeCtrlConnFor them ^= Just conn
387+ return $ Just conn
388+ Left _ ->
389+ return Nothing
330390 where
331− -- [Unified: Table 7]
332− go :: NodeCtrlState -> IO ()
333− go st = do
334− msg <- readChan ourChan
335− st' <- case destNid (ctrlMsgSignal msg) of
336− Just nid' | nid' /= localNodeId node -> do
337− (st', mConn) <- connTo st (Right nid')
338− didSend <- case mConn of
339− Just conn -> do
340− didSend <- NT.send conn (BSL.toChunks . encode $ msg)
341− case didSend of
342− Left _ -> return False
343− Right () -> return True
344− Nothing ->
345− return False
346− if didSend
347− then
348− return st'
349− else do
350− writeChan ourChan $ NodeCtrlMsg
351− { ctrlMsgSender = Right nid'
352− , ctrlMsgSignal = Died (Right nid') DiedDisconnect
353− }
354− return st'
355− _ ->
356− -- No need to forward
357− return st
358− st'' <- processCtrlMsg node st' msg
359− go st''
360−
361− ourChan = localCtrlChan node
362− ourEndPoint = localEndPoint node
363−
364− connTo :: NodeCtrlState
365− -> Identifier
366− -> IO (NodeCtrlState, Maybe NT.Connection)
367− connTo st them = do
368− case st ^. nodeCtrlConnFor them of
369− Nothing -> do
370− (addr, firstMsg) <- return $ case them of
371− Left pid -> ( nodeAddress (processNodeId pid)
372− , idToPayload (Just $ processLocalId pid)
373− )
374− Right nid -> ( nodeAddress nid
375− , idToPayload Nothing
376− )
377− mConn <- NT.connect ourEndPoint
378− addr
379− NT.ReliableOrdered
380− NT.defaultConnectHints
381− case mConn of
382− Right conn -> do
383− didSend <- NT.send conn firstMsg
384− case didSend of
385− Left _ ->
386− return (st, Nothing)
387− Right () ->
388− return (nodeCtrlConnFor them ^= Just conn $ st, Just conn)
389− Left _ ->
390− return (st, Nothing)
391− Just conn ->
392− return (st, Just conn)
391+ (addr, firstMsg) = case them of
392+ Left pid -> ( nodeAddress (processNodeId pid)
393+ , idToPayload (Just $ processLocalId pid)
394+ )
395+ Right nid -> ( nodeAddress nid
396+ , idToPayload Nothing
397+ )
398+
393399
394400-- [Unified: ncEffect]
395−processCtrlMsg :: LocalNode
396− -> NodeCtrlState
397− -> NodeCtrlMsg
398− -> IO (NodeCtrlState)
401+processCtrlMsg :: NodeCtrlMsg -> NodeCtrl ()
402+
399403-- [Unified: Table 10]
400−processCtrlMsg node st (NodeCtrlMsg from (Monitor them ref)) = do
404+processCtrlMsg (NodeCtrlMsg from (Monitor them ref)) = do
405+ node <- ask
401406 shouldLink <-
402407 if processNodeId them /= localNodeId node
403408 then return True
404− else withMVar (localState node) $
405− return . isNothing . (^. localProcessWithId (processLocalId them))
409+ else liftIO . withMVar (localState node) $
410+ return . isJust . (^. localProcessWithId (processLocalId them))
406411 if shouldLink
407412 then
408− return (nodeCtrlMonsFor them (fromLeft from) ^: Set.insert ref $ st)
413+ modify $ nodeCtrlMonsFor them (fromLeft from) ^: Set.insert ref
409414 else
410415 fail "processCtrlMsg: TODO"
411416
412−-- [Unified Table 9 and Table 12]
413−processCtrlMsg node st (NodeCtrlMsg _from (Died (Right nid) reason)) = do
414− let localLinks :: Map LocalProcessId (Set ProcessId)
415− localLinks = st ^. nodeCtrlLinksForNode nid
417+-- [Unified: Table 12, bottom rule]
418+processCtrlMsg (NodeCtrlMsg _from (Died (Right nid) reason)) = do
419+ node <- ask
420+ links <- gets (^. nodeCtrlLinksForNode nid)
421+ mons <- gets (^. nodeCtrlMonsForNode nid)
416422
417− localMons :: Map LocalProcessId (Map ProcessId (Set MonitorRef))
418− localMons = st ^. nodeCtrlMonsForNode nid
423+ forM_ (Map.toList links) $ \(_them, _uss) -> do
424+ fail "processCtrlMsg: linking not implemented"
419425
420− forM_ (Map.toList localLinks) $ \(_them, _uss) -> do
421− fail "disconnected: linking not implemented"
422−
423− forM_ (Map.toList localMons) $ \(them, uss) -> do
426+ forM_ (Map.toList mons) $ \(them, uss) -> do
424427 forM_ (Map.toList uss) $ \(us, refs) -> do
425428 -- We only need to notify local processes
426− when (processNodeId us == localNodeId node) $ do
429+ when (processNodeId us == localNodeId node) . liftIO $ do
427430 let lpid = processLocalId us
428431 rpid = ProcessId nid them
429432 forM_ refs $ \ref -> do
430433 mProc <- withMVar (localState node) $
431434 return . (^. localProcessWithId lpid)
432− forM_ mProc $ \proc ->
435+ forM_ mProc $ \proc ->
433436 enqueue (processQueue proc) . createMessage $
434437 ProcessDied ref rpid reason
435438
436− return . (nodeCtrlLinksForNode nid ^= Map.empty)
439+ modify $ (nodeCtrlLinksForNode nid ^= Map.empty)
437440 . (nodeCtrlMonsForNode nid ^= Map.empty)
438− $ st
439441
442+-- [Unified: Table 12, top rule]
443+processCtrlMsg (NodeCtrlMsg _from (Died (Left pid) reason)) = do
444+ node <- ask
445+ links <- gets (^. nodeCtrlLinksForProcess pid)
446+ mons <- gets (^. nodeCtrlMonsForProcess pid)
447+
448+ forM_ links $ \_us ->
449+ fail "processCtrlMsg: linking not implemented"
450+
451+ forM_ (Map.toList mons) $ \(us, refs) ->
452+ forM_ refs $ \ref -> do
453+ let msg = createMessage $ ProcessDied ref pid reason
454+ if processNodeId us == localNodeId node
455+ then liftIO $ do
456+ let lpid = processLocalId us
457+ mProc <- withMVar (localState node) $
458+ return . (^. localProcessWithId lpid)
459+ forM_ mProc $ \proc ->
460+ enqueue (processQueue proc) msg
461+ else do
462+ ctrlSendTo (Right . processNodeId $ us) . BSL.toChunks . encode $
463+ NodeCtrlMsg
464+ { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
465+ , ctrlMsgSignal = Died (Left pid) reason
466+ }
467+
468+ modify $ (nodeCtrlLinksForProcess pid ^= Set.empty)
469+ . (nodeCtrlMonsForProcess pid ^= Map.empty)
440470
441471--------------------------------------------------------------------------------
442472-- Initialization --
@@ -462,7 +492,8 @@ newLocalNode transport = do
462492 , localState = state
463493 , localCtrlChan = ctrlChan
464494 }
465− void . forkIO $ nodeController node
495+ void . forkIO $ evalStateT (runReaderT (unNodeCtrl nodeController) node)
496+ initNodeCtrlState
466497 void . forkIO $ handleIncomingMessages node
467498 return node
468499
@@ -474,59 +505,77 @@ runProcess node proc = do
474505
475506forkProcess :: LocalNode -> Process () -> IO ProcessId
476507forkProcess node proc = do
477− queue <- newCQueue
478− lproc <- modifyMVar (localState node) $ \st -> do
479− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
480− , lpidUnique = st ^. localPidUnique
508+ queue <- newCQueue
509+ lproc <- modifyMVar (localState node) $ \st -> do
510+ let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
511+ , lpidUnique = st ^. localPidUnique
512+ }
513+ let pid = ProcessId { processNodeId = localNodeId node
514+ , processLocalId = lpid
515+ }
516+ pst <- newMVar LocalProcessState { _connections = Map.empty
517+ }
518+ let lproc = LocalProcess { processQueue = queue
519+ , processNode = node
520+ , processId = pid
521+ , processState = pst
481522 }
482− let pid = ProcessId { processNodeId = localNodeId node
483− , processLocalId = lpid
484− }
485− pst <- newMVar LocalProcessState { _connections = Map.empty
486− }
487− let lproc = LocalProcess { processQueue = queue
488− , processNode = node
489− , processId = pid
490− , processState = pst
491− }
492− -- TODO: if the counter overflows we should pick a new unique
493− return ( (localProcessWithId lpid ^= Just lproc)
494− . (localPidCounter ^: (+ 1))
495− $ st
496− , lproc
497− )
498− void . forkIO $ runLocalProcess lproc proc
499− return (processId lproc)
523+ -- TODO: if the counter overflows we should pick a new unique
524+ return ( (localProcessWithId lpid ^= Just lproc)
525+ . (localPidCounter ^: (+ 1))
526+ $ st
527+ , lproc
528+ )
529+ void . forkIO $ do
530+ runLocalProcess lproc proc
531+ -- [Unified: Table 4]
532+ let pid = processId lproc
533+ writeChan (localCtrlChan node) $ NodeCtrlMsg
534+ { ctrlMsgSender = Left pid
535+ , ctrlMsgSignal = Died (Left pid) DiedNormal
536+ }
537+ return (processId lproc)
500538
501539handleIncomingMessages :: LocalNode -> IO ()
502−handleIncomingMessages node = go [] Map.empty
540+handleIncomingMessages node = go [] Map.empty Set.empty
503541 where
504− go :: [NT.ConnectionId] -> Map NT.ConnectionId LocalProcess -> IO ()
505− go halfOpenConns openConns = do
542+ go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know
543+ -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes
544+ -> Set NT.ConnectionId -- ^ Connections to our controller
545+ -> IO ()
546+ go uninitConns procConns ctrlConns = do
506547 event <- NT.receive endpoint
507548 case event of
508549 NT.ConnectionOpened cid _rel _theirAddr ->
509− go (cid : halfOpenConns) openConns
550+ go (cid : uninitConns) procConns ctrlConns
510551 NT.Received cid payload ->
511− case Map.lookup cid openConns of
552+ case Map.lookup cid procConns of
512553 Just proc -> do
513554 let msg = payloadToMessage payload
514555 enqueue (processQueue proc) msg
515− go halfOpenConns openConns
516− Nothing -> if cid `elem` halfOpenConns
517− then do
518− let Just lpid = payloadToId payload
519− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
520− case mProc of
521− Just proc ->
522− go (List.delete cid halfOpenConns)
523− (Map.insert cid proc openConns)
556+ go uninitConns procConns ctrlConns
557+ Nothing -> if cid `Set.member` ctrlConns
558+ then writeChan ctrlChan (decode . BSL.fromChunks $ payload)
559+ else if cid `elem` uninitConns
560+ then case payloadToId payload of
561+ Just lpid -> do
562+ mProc <- withMVar state $ return . (^. localProcessWithId lpid)
563+ case mProc of
564+ Just proc ->
565+ go (List.delete cid uninitConns)
566+ (Map.insert cid proc procConns)
567+ ctrlConns
568+ Nothing ->
569+ fail "handleIncomingMessages: TODO 1"
524570 Nothing ->
525− fail "handleIncomingMessages: TODO 1"
526− else
527− fail "handleIncomingMessages: TODO 2"
571+ go (List.delete cid uninitConns)
572+ procConns
573+ (Set.insert cid ctrlConns)
574+ else fail "handleIncomingMessages: TODO 2"
528575 NT.ConnectionClosed cid ->
529− go (List.delete cid halfOpenConns) (Map.delete cid openConns)
576+ go (List.delete cid uninitConns)
577+ (Map.delete cid procConns)
578+ (Set.delete cid ctrlConns)
530579 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do
531580 -- TODO: we could cache this information
532581 pids <- withMVar state $ \st ->
@@ -543,6 +592,7 @@ handleIncomingMessages node = go [] Map.empty
543592
544593 state = localState node
545594 endpoint = localEndPoint node
595+ ctrlChan = localCtrlChan node
546596
547597--------------------------------------------------------------------------------
548598-- Auxiliary functions --
@@ -598,8 +648,10 @@ sendBinary them conn payload = do
598648 liftIO $ remoteProcessFailed node them
599649
600650sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()
601−sendMessage them conn (Message fp enc) =
602− sendBinary them conn $ encodeFingerprint fp : BSL.toChunks enc
651+sendMessage them conn = sendBinary them conn . messageToPayload
652+
653+messageToPayload :: Message -> [BSS.ByteString]
654+messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc
603655
604656payloadToMessage :: [BSS.ByteString] -> Message
605657payloadToMessage payload = Message fp msg
@@ -696,6 +748,7 @@ instance Binary ProcessSignal where
696748 header <- getWord8
697749 case header of
698750 0 -> liftM2 Monitor get get
751+ 1 -> liftM2 Died get get
699752 _ -> fail "ProcessSignal.get: invalid"
700753
701754instance Binary DiedReason where