master...fd46c61 · distributed-process

git.iko.soy/distributed-process

on fd46c61 help sign in

distributed-process.cabal+1

since splitsnapshots

unifiedsplit

@@ -68,6 +68,7 @@ Library

68 Control.Distributed.Process.Internal.StrictContainerAccessors,

69 Control.Distributed.Process.Internal.StrictList,

70 Control.Distributed.Process.Internal.StrictMVar,

71+ Control.Distributed.Process.Internal.ThreadPool,

71 Control.Distributed.Process.Internal.Types,

72 Control.Distributed.Process.Internal.Types,

72 Control.Distributed.Process.Internal.WeakTQueue,

73 Control.Distributed.Process.Internal.WeakTQueue,

73 Control.Distributed.Process.Management,

74 Control.Distributed.Process.Management,

src/Control/Distributed/Process/Internal/ThreadPool.hsnew file+78

@@ -0,0 +1,78 @@

1+-- | An implementation of a pool of threads

2+--

3+{-# LANGUAGE RecursiveDo #-}

4+module Control.Distributed.Process.Internal.ThreadPool

5+ ( newThreadPool

6+ , submitTask

7+ , lookupWorker

8+ , ThreadPool

9+ ) where

10+

11+import Control.Exception

12+import Control.Monad

13+import Data.IORef

14+import qualified Data.Map as Map

15+

16+

17+-- | A pool of worker threads that execute tasks.

18+--

19+-- Each worker thread is named with a key @k@. Tasks are submitted to a

20+-- specific worker using its key. While the worker is busy the tasks are queued.

21+-- When there are no more queued tasks the worker ceases to exist.

22+--

23+-- The next time a task is submitted the worker will be respawned.

24+--

25+newtype ThreadPool k w = ThreadPool (IORef (Map.Map k (Maybe (IO ()), w)))

26+

27+-- Each worker has an entry in the map with a closure that contains all

28+-- queued actions fot it.

29+--

30+-- No entry in the map is kept for defunct workers.

31+

32+-- | Creates a pool with no workers.

33+newThreadPool :: IO (ThreadPool k w)

34+newThreadPool = fmap ThreadPool $ newIORef Map.empty

35+

36+-- | @submitTask pool fork k task@ submits a task for the worker @k@.

37+--

38+-- If worker @k@ is busy, then the task is queued until the worker is available.

39+--

40+-- If worker @k@ does not exist, then the given @fork@ operation is used to

41+-- spawn the worker. @fork@ returns whatever information is deemed useful for

42+-- later retrieval via 'lookupWorker'.

43+--

44+submitTask :: Ord k

45+ => ThreadPool k w

46+ -> (IO () -> IO w)

47+ -> k -> IO () -> IO ()

48+submitTask (ThreadPool mapRef) fork k task = mdo

49+ m' <- join $ atomicModifyIORef mapRef $ \m ->

50+ case Map.lookup k m of

51+ -- There is no worker for this key, create one.

52+ Nothing -> ( m'

53+ , do w <- fork $ flip onException terminateWorker $ do

54+ task

55+ continue

56+ return $ Map.insert k (Nothing, w) m

57+ )

58+ -- Queue an action for the existing worker.

59+ Just (mp, w) ->

60+ (m', return $ Map.insert k (Just $ maybe task (>> task) mp, w) m)

61+ return ()

62+ where

63+ continue = join $ atomicModifyIORef mapRef $ \m ->

64+ case Map.lookup k m of

65+ -- Execute the next batch of queued actions.

66+ Just (Just p, w) -> (Map.insert k (Nothing, w) m, p >> continue)

67+ -- There are no more queued actions. Terminate the worker.

68+ Just (Nothing, w) -> (Map.delete k m, return ())

69+ -- The worker key was removed already (?)

70+ Nothing -> (m, return ())

71+ -- Remove the worker key regardless of whether there are more queued

72+ -- actions.

73+ terminateWorker = atomicModifyIORef mapRef $ \m -> (Map.delete k m, ())

74+

75+-- | Looks up a worker with the given key.

76+lookupWorker :: Ord k => ThreadPool k w -> k -> IO (Maybe w)

77+lookupWorker (ThreadPool mapRef) k =

78+ atomicModifyIORef mapRef $ \m -> (m, fmap snd $ Map.lookup k m)

src/Control/Distributed/Process/Internal/Types.hs+3

@@ -137,6 +137,7 @@ import Control.Distributed.Process.Internal.StrictMVar

137 , modifyMVar

138 , modifyMVar_

139 )

140+import Control.Distributed.Process.Internal.ThreadPool

140import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

141import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

141import Control.Distributed.Static (RemoteTable, Closure)

142import Control.Distributed.Static (RemoteTable, Closure)

142import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

143import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

@@ -267,6 +268,8 @@ data LocalNode = LocalNode

267 -- | Runtime lookup table for supporting closures

268 -- | Runtime lookup table for supporting closures

268 -- TODO: this should be part of the CH state, not the local endpoint state

269 -- TODO: this should be part of the CH state, not the local endpoint state

269 , remoteTable :: !RemoteTable

270 , remoteTable :: !RemoteTable

271+ -- The pool of threads and queues to send messages

272+ , localSendPool :: ThreadPool NodeId ThreadId

270 }

273 }

271

274

272data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

275data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

src/Control/Distributed/Process/Node.hs+126−31

@@ -25,6 +25,7 @@ import System.IO (fixIO, hPutStrLn, stderr)

25import System.Mem.Weak (Weak, deRefWeak)

26import qualified Data.ByteString.Lazy as BSL (fromChunks)

27import Data.Binary (decode)

28+import Data.Function (fix)

28import Data.Map (Map)

29import Data.Map (Map)

29import qualified Data.Map as Map

30import qualified Data.Map as Map

30 ( empty

31 ( empty

@@ -78,6 +79,8 @@ import qualified Control.Exception as Exception

78 , catch

79 , catch

79 , catches

80 , catches

80 , finally

81 , finally

82+ , bracket

83+ , mask_

81 )

84 )

82import Control.Concurrent (forkIO, killThread)

85import Control.Concurrent (forkIO, killThread)

83import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)

86import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)

@@ -91,8 +94,18 @@ import Control.Distributed.Process.Internal.StrictMVar

91 , putMVar

94 , putMVar

92 , takeMVar

95 , takeMVar

93 )

96 )

97+import Control.Distributed.Process.Internal.ThreadPool

98+ ( newThreadPool

99+ , submitTask

100+ , lookupWorker

101+ , ThreadPool

102+ )

94import Control.Concurrent.Chan (newChan, writeChan, readChan)

103import Control.Concurrent.Chan (newChan, writeChan, readChan)

95−import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)

104+import qualified Control.Concurrent.MVar as MVar

105+ ( newEmptyMVar

106+ , takeMVar

107+ , putMVar

108+ )

96import Control.Concurrent.STM

109import Control.Concurrent.STM

97 ( atomically

110 ( atomically

98 )

111 )

@@ -224,6 +237,7 @@ import Control.Monad.Catch (try)

224import GHC.IO (IO(..), unsafeUnmask)

237import GHC.IO (IO(..), unsafeUnmask)

225import GHC.Base ( maskAsyncExceptions# )

238import GHC.Base ( maskAsyncExceptions# )

226

239

240+import System.Timeout

227import Unsafe.Coerce

241import Unsafe.Coerce

228import Prelude

242import Prelude

229

243

@@ -264,12 +278,14 @@ createBareLocalNode endPoint rtable = do

264 , _localConnections = Map.empty

278 , _localConnections = Map.empty

265 }

279 }

266 ctrlChan <- newChan

280 ctrlChan <- newChan

281+ sendPool <- newThreadPool

267 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

282 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

268 , localEndPoint = endPoint

283 , localEndPoint = endPoint

269 , localState = state

284 , localState = state

270 , localCtrlChan = ctrlChan

285 , localCtrlChan = ctrlChan

271 , localEventBus = MxEventBusInitialising

286 , localEventBus = MxEventBusInitialising

272 , remoteTable = rtable

287 , remoteTable = rtable

288+ , localSendPool = sendPool

273 }

289 }

274 tracedNode <- startMxAgent node

290 tracedNode <- startMxAgent node

275

291

@@ -512,24 +528,78 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr

512 where

528 where

513 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })

529 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })

514

530

531+-- | And 'EventMonitor' monitors the handling of events reporting when an event

532+-- is not handled promptly. It monitors events which are handled in sequence.

533+data EventMonitor a = EventMonitor

534+ { -- | Notifies that an event was received. Takes the event as argument.

535+ eventMonitorGotEvent :: a -> IO ()

536+ -- | Notifies that the last received event was handled.

537+ --

538+ -- Might be called even when there is no event, in which case it will have

539+ -- no effect.

540+ , eventMonitorHandledEvent :: IO ()

541+ -- | Stops the event monitor.

542+ , eventMonitorStop :: IO ()

543+ }

544+

545+-- | `eventMon <- monitorThread t action`

546+--

547+-- Creates an event monitor that executes `action` when an event hasn't been

548+-- handled withing `t` microseconds. `action` takes as argument the event that

549+-- wasn't handled timely.

550+--

551+monitorThread :: Int -> (a -> IO ()) -> IO (EventMonitor a)

552+monitorThread tmicrosecs action = do

553+ mv <- MVar.newEmptyMVar

554+ tid <- forkIO $ forever' $ do

555+ -- loop until some event is reported

556+ ev <- fix $ \loop -> MVar.takeMVar mv >>= maybe loop return

557+ -- wait until the event is handled

558+ Exception.mask_ (timeout tmicrosecs $ MVar.takeMVar mv) >>=

559+ maybe (do action ev -- report the unhandled event

560+ -- wait for the slow event handler

561+ void $ MVar.takeMVar mv

562+ )

563+ (const $ return ())

564+ return EventMonitor

565+ { eventMonitorGotEvent = MVar.putMVar mv . Just

566+ , eventMonitorHandledEvent = MVar.putMVar mv Nothing

567+ , eventMonitorStop = killThread tid

568+ }

569+

515handleIncomingMessages :: LocalNode -> IO ()

570handleIncomingMessages :: LocalNode -> IO ()

516−handleIncomingMessages node = go initConnectionState

571+handleIncomingMessages node =

572+ Exception.bracket

573+ (monitorThread 1000000 $ \ev ->

574+ hPutStrLn stderr $

575+ "handleIncomingMessages blocked " ++ show (localNodeId node) ++

576+ " on " ++ show ev

577+ )

578+ eventMonitorStop $ \eventMon ->

579+ (go eventMon initConnectionState

517− `Exception.catch` \(NodeClosedException _) -> return ()

580+ `Exception.catch` \(NodeClosedException _) -> return ()

581+ ) `Exception.catch` \e -> do

582+ hPutStrLn stderr $

583+ "handleIncomingMessages failed: " ++

584+ show (localNodeId node, e :: SomeException)

585+ throwIO e

518 where

586 where

519− go :: ConnectionState -> IO ()

587+ go :: EventMonitor NT.Event -> ConnectionState -> IO ()

520− go !st = do

588+ go eventMon !st = do

589+ eventMonitorHandledEvent eventMon

521 event <- NT.receive endpoint

590 event <- NT.receive endpoint

591+ eventMonitorGotEvent eventMon event

522 case event of

592 case event of

523 NT.ConnectionOpened cid rel theirAddr ->

593 NT.ConnectionOpened cid rel theirAddr ->

524 if rel == NT.ReliableOrdered

594 if rel == NT.ReliableOrdered

525 then

595 then

526 trace node (MxConnected cid theirAddr)

596 trace node (MxConnected cid theirAddr)

527− >> go (

597+ >> go eventMon (

528 (incomingAt cid ^= Just (theirAddr, Uninit))

598 (incomingAt cid ^= Just (theirAddr, Uninit))

529 . (incomingFrom theirAddr ^: Set.insert cid)

599 . (incomingFrom theirAddr ^: Set.insert cid)

530 $ st

600 $ st

531 )

601 )

532− else invalidRequest cid st $

602+ else invalidRequest eventMon cid st $

533 "attempt to connect with unsupported reliability " ++ show rel

603 "attempt to connect with unsupported reliability " ++ show rel

534 NT.Received cid payload ->

604 NT.Received cid payload ->

535 case st ^. incomingAt cid of

605 case st ^. incomingAt cid of

@@ -541,7 +611,7 @@ handleIncomingMessages node = go initConnectionState

541 let msg = payloadToMessage payload

611 let msg = payloadToMessage payload

542 enqueue queue msg -- 'enqueue' is strict

612 enqueue queue msg -- 'enqueue' is strict

543 trace node (MxReceived pid msg)

613 trace node (MxReceived pid msg)

544− go st

614+ go eventMon st

545 Just (_, ToChan (TypedChannel chan')) -> do

615 Just (_, ToChan (TypedChannel chan')) -> do

546 mChan <- deRefWeak chan'

616 mChan <- deRefWeak chan'

547 -- If mChan is Nothing, the process has given up the read end of

617 -- If mChan is Nothing, the process has given up the read end of

@@ -549,11 +619,11 @@ handleIncomingMessages node = go initConnectionState

549 forM_ mChan $ \chan -> atomically $

619 forM_ mChan $ \chan -> atomically $

550 -- We make sure the message is fully decoded when it is enqueued

620 -- We make sure the message is fully decoded when it is enqueued

551 writeTQueue chan $! decode (BSL.fromChunks payload)

621 writeTQueue chan $! decode (BSL.fromChunks payload)

552− go st

622+ go eventMon st

553 Just (_, ToNode) -> do

623 Just (_, ToNode) -> do

554 let ctrlMsg = decode . BSL.fromChunks $ payload

624 let ctrlMsg = decode . BSL.fromChunks $ payload

555 writeChan ctrlChan $! ctrlMsg

625 writeChan ctrlChan $! ctrlMsg

556− go st

626+ go eventMon st

557 Just (src, Uninit) ->

627 Just (src, Uninit) ->

558 case decode (BSL.fromChunks payload) of

628 case decode (BSL.fromChunks payload) of

559 ProcessIdentifier pid -> do

629 ProcessIdentifier pid -> do

@@ -561,11 +631,11 @@ handleIncomingMessages node = go initConnectionState

561 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)

631 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)

562 case mProc of

632 case mProc of

563 Just proc ->

633 Just proc ->

564− go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)

634+ go eventMon (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)

565 Nothing ->

635 Nothing ->

566 -- incoming attempt to connect to unknown process - might

636 -- incoming attempt to connect to unknown process - might

567 -- be dead already

637 -- be dead already

568− go (incomingAt cid ^= Nothing $ st)

638+ go eventMon (incomingAt cid ^= Nothing $ st)

569 SendPortIdentifier chId -> do

639 SendPortIdentifier chId -> do

570 let lcid = sendPortLocalId chId

640 let lcid = sendPortLocalId chId

571 lpid = processLocalId (sendPortProcessId chId)

641 lpid = processLocalId (sendPortProcessId chId)

@@ -575,33 +645,34 @@ handleIncomingMessages node = go initConnectionState

575 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

645 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

576 case mChannel of

646 case mChannel of

577 Just channel ->

647 Just channel ->

648+ go eventMon

578− go (incomingAt cid ^= Just (src, ToChan channel) $ st)

649+ (incomingAt cid ^= Just (src, ToChan channel) $ st)

579 Nothing ->

650 Nothing ->

580− invalidRequest cid st $

651+ invalidRequest eventMon cid st $

581 "incoming attempt to connect to unknown channel of"

652 "incoming attempt to connect to unknown channel of"

582 ++ " process " ++ show (sendPortProcessId chId)

653 ++ " process " ++ show (sendPortProcessId chId)

583 Nothing ->

654 Nothing ->

584 -- incoming attempt to connect to channel of unknown

655 -- incoming attempt to connect to channel of unknown

585 -- process - might be dead already

656 -- process - might be dead already

586− go (incomingAt cid ^= Nothing $ st)

657+ go eventMon (incomingAt cid ^= Nothing $ st)

587 NodeIdentifier nid ->

658 NodeIdentifier nid ->

588 if nid == localNodeId node

659 if nid == localNodeId node

589− then go (incomingAt cid ^= Just (src, ToNode) $ st)

660+ then go eventMon (incomingAt cid ^= Just (src, ToNode) $ st)

590− else invalidRequest cid st $

661+ else invalidRequest eventMon cid st $

591 "incoming attempt to connect to a different node -"

662 "incoming attempt to connect to a different node -"

592 ++ " I'm " ++ show (localNodeId node)

663 ++ " I'm " ++ show (localNodeId node)

593 ++ " but the remote peer wants to connect to "

664 ++ " but the remote peer wants to connect to "

594 ++ show nid

665 ++ show nid

595 Nothing ->

666 Nothing ->

596− invalidRequest cid st

667+ invalidRequest eventMon cid st

597 "message received from an unknown connection"

668 "message received from an unknown connection"

598 NT.ConnectionClosed cid ->

669 NT.ConnectionClosed cid ->

599 case st ^. incomingAt cid of

670 case st ^. incomingAt cid of

600 Nothing ->

671 Nothing ->

601− invalidRequest cid st "closed unknown connection"

672+ invalidRequest eventMon cid st "closed unknown connection"

602 Just (src, _) -> do

673 Just (src, _) -> do

603 trace node (MxDisconnected cid src)

674 trace node (MxDisconnected cid src)

604− go ( (incomingAt cid ^= Nothing)

675+ go eventMon ( (incomingAt cid ^= Nothing)

605 . (incomingFrom src ^: Set.delete cid)

676 . (incomingFrom src ^: Set.delete cid)

606 $ st

677 $ st

607 )

678 )

@@ -613,8 +684,10 @@ handleIncomingMessages node = go initConnectionState

613 , ctrlMsgSignal = Died nid DiedDisconnect

684 , ctrlMsgSignal = Died nid DiedDisconnect

614 }

685 }

615 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))

686 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))

687+ liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)

688+ >>= maybe (return ()) killThread

616 closeImplicitReconnections node nid

689 closeImplicitReconnections node nid

617− go ( (incomingFrom theirAddr ^= Set.empty)

690+ go eventMon ( (incomingFrom theirAddr ^= Set.empty)

618 . (incoming ^: Map.filterWithKey (const . notLost))

691 . (incoming ^: Map.filterWithKey (const . notLost))

619 $ st

692 $ st

620 )

693 )

@@ -629,8 +702,10 @@ handleIncomingMessages node = go initConnectionState

629 -- and we just give up

702 -- and we just give up

630 fail "Cloud Haskell fatal error: received unexpected multicast"

703 fail "Cloud Haskell fatal error: received unexpected multicast"

631

704

705+ invalidRequest :: EventMonitor NT.Event

706+ -> NT.ConnectionId

632− invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()

707+ -> ConnectionState -> String -> IO ()

633− invalidRequest cid st msg = do

708+ invalidRequest eventMon cid st msg = do

634 -- TODO: We should treat this as a fatal error on the part of the remote

709 -- TODO: We should treat this as a fatal error on the part of the remote

635 -- node. That is, we should report the remote node as having died, and we

710 -- node. That is, we should report the remote node as having died, and we

636 -- should close incoming connections (this requires a Transport layer

711 -- should close incoming connections (this requires a Transport layer

@@ -639,7 +714,7 @@ handleIncomingMessages node = go initConnectionState

639 ++ " (" ++ msg ++ "): "

714 ++ " (" ++ msg ++ "): "

640 , (Trace cid)

715 , (Trace cid)

641 ]

716 ]

642− go ( incomingAt cid ^= Nothing

717+ go eventMon ( incomingAt cid ^= Nothing

643 $ st

718 $ st

644 )

719 )

645

720

@@ -651,9 +726,21 @@ handleIncomingMessages node = go initConnectionState

651--------------------------------------------------------------------------------

726--------------------------------------------------------------------------------

652

727

653runNodeController :: LocalNode -> IO ()

728runNodeController :: LocalNode -> IO ()

654−runNodeController node =

729+runNodeController node = do

730+ Exception.bracket

731+ (monitorThread 1000000 $ \ev ->

732+ hPutStrLn stderr $

733+ "nodeController blocked " ++ show (localNodeId node) ++

734+ " on " ++ show ev

735+ )

736+ eventMonitorStop $ \eventMon ->

655− runReaderT (evalStateT (unNC nodeController) initNCState) node

737+ (runReaderT (evalStateT (unNC (nodeController eventMon)) initNCState) node

656− `Exception.catch` \(NodeClosedException _) -> return ()

738+ `Exception.catch` \(NodeClosedException _) -> return ()

739+ ) `Exception.catch` \e -> do

740+ hPutStrLn stderr $

741+ "nodeController failed: " ++

742+ show (localNodeId node, e :: SomeException)

743+ throwIO e

657

744

658--------------------------------------------------------------------------------

745--------------------------------------------------------------------------------

659-- Internal data types --

746-- Internal data types --

@@ -669,6 +756,12 @@ data NCState = NCState

669 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])

756 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])

670 }

757 }

671

758

759+submitSendPool :: LocalNode -> NodeId -> IO () -> IO ()

760+submitSendPool node nid task = submitTask (localSendPool node) fork nid task

761+ where

762+ fork io =

763+ forkIO $ io `Exception.catch` \(NodeClosedException _) -> return ()

764+

672newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

765newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

673 deriving ( Applicative

766 deriving ( Applicative

674 , Functor

767 , Functor

@@ -704,7 +797,7 @@ ncSendToProcessAndTrace shouldTrace pid msg = do

704 node <- ask

797 node <- ask

705 if processNodeId pid == localNodeId node

798 if processNodeId pid == localNodeId node

706 then ncEffectLocalSendAndTrace shouldTrace node pid msg

799 then ncEffectLocalSendAndTrace shouldTrace node pid msg

707− else liftIO $ sendBinary node

800+ else liftIO $ submitSendPool node (processNodeId pid) $ sendBinary node

708 (NodeIdentifier $ localNodeId node)

801 (NodeIdentifier $ localNodeId node)

709 (NodeIdentifier $ processNodeId pid)

802 (NodeIdentifier $ processNodeId pid)

710 WithImplicitReconnect

803 WithImplicitReconnect

@@ -717,7 +810,7 @@ ncSendToNode to msg = do

717 node <- ask

810 node <- ask

718 liftIO $ if to == localNodeId node

811 liftIO $ if to == localNodeId node

719 then writeChan (localCtrlChan node) $! msg

812 then writeChan (localCtrlChan node) $! msg

720− else sendBinary node

813+ else submitSendPool node to $ sendBinary node

721 (NodeIdentifier $ localNodeId node)

814 (NodeIdentifier $ localNodeId node)

722 (NodeIdentifier to)

815 (NodeIdentifier to)

723 WithImplicitReconnect

816 WithImplicitReconnect

@@ -756,11 +849,13 @@ withLocalTracer node act = act (localEventBus node)

756--------------------------------------------------------------------------------

849--------------------------------------------------------------------------------

757

850

758-- [Unified: Table 7]

851-- [Unified: Table 7]

759−nodeController :: NC ()

852+nodeController :: EventMonitor NCMsg -> NC ()

760−nodeController = do

853+nodeController eventMon = do

761 node <- ask

854 node <- ask

762 forever' $ do

855 forever' $ do

856+ liftIO $ eventMonitorHandledEvent eventMon

763 msg <- liftIO $ readChan (localCtrlChan node)

857 msg <- liftIO $ readChan (localCtrlChan node)

858+ liftIO $ eventMonitorGotEvent eventMon msg

764

859

765 -- [Unified: Table 7, rule nc_forward]

860 -- [Unified: Table 7, rule nc_forward]

766 case destNid (ctrlMsgSignal msg) of

861 case destNid (ctrlMsgSignal msg) of