git.iko.soy/distributed-process
on fd46c61 help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1381 lines 55 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE DeriveDataTypeable #-}
3{-# LANGUAGE ScopedTypeVariables #-}
4{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE BangPatterns #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7{-# LANGUAGE RecordWildCards #-}
8{-# LANGUAGE MagicHash #-}
10-- | Local nodes
11--
12module Control.Distributed.Process.Node
13 ( LocalNode
14 , newLocalNode
15 , closeLocalNode
16 , forkProcess
17 , runProcess
18 , initRemoteTable
19 , localNodeId
20 ) where
22-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
24import System.IO (fixIO, hPutStrLn, stderr)
25import System.Mem.Weak (Weak, deRefWeak)
26import qualified Data.ByteString.Lazy as BSL (fromChunks)
27import Data.Binary (decode)
28import Data.Function (fix)
29import Data.Map (Map)
30import qualified Data.Map as Map
31 ( empty
32 , toList
33 , fromList
34 , filter
35 , partitionWithKey
36 , elems
37 , size
38 , filterWithKey
39 , foldlWithKey
40 )
41import qualified Data.Set as Set
42import Data.Time.Format (formatTime)
43#if MIN_VERSION_time(1,5,0)
44import Data.Time.Format (defaultTimeLocale)
45#else
46import System.Locale (defaultTimeLocale)
47#endif
48import Data.Set (Set)
49import qualified Data.Set as Set
50 ( empty
51 , insert
52 , delete
53 , map
54 , member
55 , toList
56 )
57import Data.Foldable (forM_)
58import Data.List (foldl')
59import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
60import Data.Typeable (Typeable)
61import Control.Category ((>>>))
62import Control.Applicative
63import Control.Monad (void, when, join)
64import Control.Monad.IO.Class (MonadIO, liftIO)
65import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
66import qualified Control.Monad.State.Strict as StateT (get, put)
67import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
68import Control.Exception
69 ( throwIO
70 , SomeException
71 , Exception
72 , throwTo
73 , uninterruptibleMask_
74 , getMaskingState
75 , MaskingState(..)
76 )
77import qualified Control.Exception as Exception
78 ( Handler(..)
79 , catch
80 , catches
81 , finally
82 , bracket
83 , mask_
84 )
85import Control.Concurrent (forkIO, killThread)
86import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)
87import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap
88import Control.Distributed.Process.Internal.StrictMVar
89 ( newMVar
90 , withMVar
91 , modifyMVarMasked
92 , modifyMVar
93 , newEmptyMVar
94 , putMVar
95 , takeMVar
96 )
97import Control.Distributed.Process.Internal.ThreadPool
98 ( newThreadPool
99 , submitTask
100 , lookupWorker
101 , ThreadPool
102 )
103import Control.Concurrent.Chan (newChan, writeChan, readChan)
104import qualified Control.Concurrent.MVar as MVar
105 ( newEmptyMVar
106 , takeMVar
107 , putMVar
108 )
109import Control.Concurrent.STM
110 ( atomically
111 )
112import Control.Distributed.Process.Internal.CQueue
113 ( CQueue
114 , enqueue
115 , newCQueue
116 , mkWeakCQueue
117 , queueSize
118 )
119import qualified Network.Transport as NT
120 ( Transport
121 , EndPoint
122 , newEndPoint
123 , receive
124 , Event(..)
125 , EventErrorCode(..)
126 , TransportError(..)
127 , address
128 , closeEndPoint
129 , Connection
130 , ConnectionId
131 , close
132 , EndPointAddress
133 , Reliability(ReliableOrdered)
134 )
135import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
136import System.Random (randomIO)
137import Control.Distributed.Static (RemoteTable, Closure)
138import qualified Control.Distributed.Static as Static
139 ( unclosure
140 , initRemoteTable
141 )
142import Control.Distributed.Process.Internal.Types
143 ( NodeId(..)
144 , LocalProcessId(..)
145 , ProcessId(..)
146 , LocalNode(..)
147 , MxEventBus(..)
148 , LocalNodeState(..)
149 , ValidLocalNodeState(..)
150 , withValidLocalState
151 , modifyValidLocalState
152 , LocalProcess(..)
153 , LocalProcessState(..)
154 , Process(..)
155 , DiedReason(..)
156 , NCMsg(..)
157 , ProcessSignal(..)
158 , localPidCounter
159 , localPidUnique
160 , localProcessWithId
161 , localProcesses
162 , localConnections
163 , forever'
164 , MonitorRef(..)
165 , NodeClosedException(..)
166 , ProcessMonitorNotification(..)
167 , NodeMonitorNotification(..)
168 , PortMonitorNotification(..)
169 , ProcessExitException(..)
170 , ProcessLinkException(..)
171 , NodeLinkException(..)
172 , PortLinkException(..)
173 , DidUnmonitor(..)
174 , DidUnlinkProcess(..)
175 , DidUnlinkNode(..)
176 , DidUnlinkPort(..)
177 , SpawnRef
178 , DidSpawn(..)
179 , Message(..)
180 , TypedChannel(..)
181 , Identifier(..)
182 , nodeOf
183 , ProcessInfo(..)
184 , ProcessInfoNone(..)
185 , NodeStats(..)
186 , SendPortId(..)
187 , typedChannelWithId
188 , RegisterReply(..)
189 , WhereIsReply(..)
190 , payloadToMessage
191 , createUnencodedMessage
192 , unsafeCreateUnencodedMessage
193 , runLocalProcess
194 , firstNonReservedProcessId
195 , ImplicitReconnect(WithImplicitReconnect)
196 )
197import Control.Distributed.Process.Management.Internal.Agent
198 ( mxAgentController
199 )
200import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
201 ( remoteTable
202 )
203import Control.Distributed.Process.Management.Internal.Trace.Tracer
204 ( defaultTracer
205 )
206import Control.Distributed.Process.Management.Internal.Trace.Types
207 ( TraceArg(..)
208 , traceEvent
209 , traceLogFmt
210 , enableTrace
211 )
212import Control.Distributed.Process.Management.Internal.Types
213 ( MxEvent(..)
214 )
215import Control.Distributed.Process.Serializable (Serializable)
216import Control.Distributed.Process.Internal.Messaging
217 ( sendBinary
218 , closeImplicitReconnections
219 , impliesDeathOf
220 )
221import Control.Distributed.Process.Internal.Primitives
222 ( register
223 , receiveWait
224 , match
225 , sendChan
226 , unwrapMessage
227 , SayMessage(..)
228 )
229import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
230import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
231import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
232import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
233 ( mapMaybe
234 , mapDefault
235 )
236import Control.Monad.Catch (try)
237import GHC.IO (IO(..), unsafeUnmask)
238import GHC.Base ( maskAsyncExceptions# )
240import System.Timeout
241import Unsafe.Coerce
242import Prelude
244-- Remove these definitions when the fix for
245-- https://ghc.haskell.org/trac/ghc/ticket/10149
246-- is included in all supported compilers:
247block :: IO a -> IO a
248block (IO io) = IO $ maskAsyncExceptions# io
249unblock :: IO a -> IO a
250unblock = unsafeUnmask
252--------------------------------------------------------------------------------
253-- Initialization --
254--------------------------------------------------------------------------------
256initRemoteTable :: RemoteTable
257initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
259-- | Initialize a new local node.
260newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
261newLocalNode transport rtable = do
262 mEndPoint <- NT.newEndPoint transport
263 case mEndPoint of
264 Left ex -> throwIO ex
265 Right endPoint -> do
266 localNode <- createBareLocalNode endPoint rtable
267 startServiceProcesses localNode
268 return localNode
270-- | Create a new local node (without any service processes running)
271createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
272createBareLocalNode endPoint rtable = do
273 unq <- randomIO
274 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
275 { _localProcesses = Map.empty
276 , _localPidCounter = firstNonReservedProcessId
277 , _localPidUnique = unq
278 , _localConnections = Map.empty
279 }
280 ctrlChan <- newChan
281 sendPool <- newThreadPool
282 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
283 , localEndPoint = endPoint
284 , localState = state
285 , localCtrlChan = ctrlChan
286 , localEventBus = MxEventBusInitialising
287 , remoteTable = rtable
288 , localSendPool = sendPool
289 }
290 tracedNode <- startMxAgent node
292 -- Once the NC terminates, the endpoint isn't much use,
293 void $ forkIO $ Exception.finally (runNodeController tracedNode)
294 (NT.closeEndPoint (localEndPoint node))
296 -- whilst a closed/failing endpoint will terminate the NC
297 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
298 (stopNC node)
300 return tracedNode
301 where
302 stopNC node =
303 writeChan (localCtrlChan node) NCMsg
304 { ctrlMsgSender = NodeIdentifier (localNodeId node)
305 , ctrlMsgSignal = SigShutdown
306 }
308startMxAgent :: LocalNode -> IO LocalNode
309startMxAgent node = do
310 -- see note [tracer/forkProcess races]
311 let fork = forkProcess node
312 mv <- MVar.newEmptyMVar
313 pid <- fork $ mxAgentController fork mv
314 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
315 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
317startDefaultTracer :: LocalNode -> IO ()
318startDefaultTracer node' = do
319 let t = localEventBus node'
320 case t of
321 MxEventBus _ (Tracer pid _) _ _ -> do
322 runProcess node' $ register "trace.controller" pid
323 pid' <- forkProcess node' defaultTracer
324 enableTrace (localEventBus node') pid'
325 runProcess node' $ register "tracer.initial" pid'
326 _ -> return ()
328-- TODO: we need a better mechanism for defining and registering services
330-- | Start and register the service processes on a node
331startServiceProcesses :: LocalNode -> IO ()
332startServiceProcesses node = do
333 -- tracing /spawns/ relies on the tracer being enabled, but we start
334 -- the default tracer first, even though it might @nsend@ to the logger
335 -- before /that/ process has started - this is a totally harmless race
336 -- however, so we deliberably ignore it
337 startDefaultTracer node
338 logger <- forkProcess node loop
339 runProcess node $ do
340 register "logger" logger
341 -- The trace.logger is used for tracing to the console to avoid feedback
342 -- loops during tracing if the user reregisters the "logger" with a custom
343 -- process which uses 'send' or other primitives which are traced.
344 register "trace.logger" logger
345 where
346 loop = do
347 receiveWait
348 [ match $ \(SayMessage time pid string) -> do
349 let time' = formatTime defaultTimeLocale "%c" time
350 liftIO . hPutStrLn stderr $ time' ++ " " ++ show pid ++ ": " ++ string
351 loop
352 , match $ \((time, string) :: (String, String)) -> do
353 -- this is a 'trace' message from the local node tracer
354 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
355 loop
356 , match $ \(ch :: SendPort ()) -> -- a shutdown request
357 sendChan ch ()
358 ]
360-- | Force-close a local node, killing all processes on that node.
361closeLocalNode :: LocalNode -> IO ()
362closeLocalNode node = do
363 -- Kill processes after refilling the mvar. Otherwise, there is potential for
364 -- deadlock as a dying process tries to get the mvar while masking exceptions
365 -- uninterruptibly.
366 join $ modifyMVar (localState node) $ \st -> case st of
367 LocalNodeValid vst -> do
368 return ( LocalNodeClosed
369 , forM_ (vst ^. localProcesses) $ \lproc ->
370 -- Semantics of 'throwTo' guarantee that target thread will get
371 -- delivered an exception. Therefore, target thread will be
372 -- killed eventually and that's as good as we can do. No need
373 -- to wait for thread to actually finish dying.
374 killThread (processThread lproc)
375 )
376 LocalNodeClosed -> return (LocalNodeClosed, return ())
377 -- This call will have the effect of shutting down the NC as well (see
378 -- 'createBareLocalNode').
379 NT.closeEndPoint (localEndPoint node)
381-- | Run a process on a local node and wait for it to finish
382runProcess :: LocalNode -> Process () -> IO ()
383runProcess node proc = do
384 done <- newEmptyMVar
385 -- TODO; When forkProcess inherits the masking state, protect the forked
386 -- thread against async exceptions that could occur before 'try' is evaluated.
387 void $ forkProcess node $ try proc >>= liftIO . putMVar done
388 takeMVar done >>= either (throwIO :: SomeException -> IO a) return
390-- | Spawn a new process on a local node
391forkProcess :: LocalNode -> Process () -> IO ProcessId
392forkProcess node proc = do
393 ms <- getMaskingState
394 modifyMVarMasked (localState node) (startProcess ms)
395 where
396 startProcess :: MaskingState
397 -> LocalNodeState
398 -> IO (LocalNodeState, ProcessId)
399 startProcess ms (LocalNodeValid vst) = do
400 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
401 , lpidUnique = vst ^. localPidUnique
402 }
403 let pid = ProcessId { processNodeId = localNodeId node
404 , processLocalId = lpid
405 }
406 pst <- newMVar LocalProcessState { _monitorCounter = 0
407 , _spawnCounter = 0
408 , _channelCounter = 0
409 , _typedChannels = Map.empty
410 }
411 queue <- newCQueue
412 weakQueue <- mkWeakCQueue queue (return ())
413 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
414 let lproc = LocalProcess { processQueue = queue
415 , processWeakQ = weakQueue
416 , processId = pid
417 , processState = pst
418 , processThread = tid
419 , processNode = node
420 }
421 -- Rewrite this code when this is fixed:
422 -- https://ghc.haskell.org/trac/ghc/ticket/10149
423 let unmask = case ms of
424 Unmasked -> unblock
425 MaskedInterruptible -> block
426 MaskedUninterruptible -> id
427 tid' <- uninterruptibleMask_ $ forkIO $ do
428 reason <- Exception.catches
429 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
430 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
431 mMsg <- unwrapMessage msg :: IO (Maybe String)
432 case mMsg of
433 Nothing -> return $ DiedException $ show ex
434 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
435 , (Exception.Handler
436 (return . DiedException . (show :: SomeException -> String)))]
438 -- [Unified: Table 4, rules termination and exiting]
439 mconns <- modifyValidLocalState node (cleanupProcess pid)
440 -- XXX: Revisit after agreeing on the bigger picture for the semantics
441 -- of transport operations.
442 -- https://github.com/haskell-distributed/distributed-process/issues/204
443 forM_ mconns $ forkIO . mapM_ NT.close
445 writeChan (localCtrlChan node) NCMsg
446 { ctrlMsgSender = ProcessIdentifier pid
447 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
448 }
449 return (tid', lproc)
451 -- see note [tracer/forkProcess races]
452 trace node (MxSpawned pid)
454 if lpidCounter lpid == maxBound
455 then do
456 -- TODO: this doesn't look right at all - how do we know
457 -- that newUnique represents a process id that is available!?
458 newUnique <- randomIO
459 return ( LocalNodeValid
460 $ (localProcessWithId lpid ^= Just lproc)
461 . (localPidCounter ^= firstNonReservedProcessId)
462 . (localPidUnique ^= newUnique)
463 $ vst
464 , pid
465 )
466 else
467 return ( LocalNodeValid
468 $ (localProcessWithId lpid ^= Just lproc)
469 . (localPidCounter ^: (+ 1))
470 $ vst
471 , pid
472 )
473 startProcess _ LocalNodeClosed =
474 throwIO $ NodeClosedException $ localNodeId node
476 cleanupProcess :: ProcessId
477 -> ValidLocalNodeState
478 -> IO (ValidLocalNodeState, [NT.Connection])
479 cleanupProcess pid vst = do
480 let pid' = ProcessIdentifier pid
481 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
482 return ( (localProcessWithId (processLocalId pid) ^= Nothing)
483 . (localConnections ^= unaffected)
484 $ vst
485 , map fst $ Map.elems affected
486 )
488-- note [tracer/forkProcess races]
489--
490-- Our startTracing function uses forkProcess to start the trace controller
491-- process, and of course forkProcess attempts to call traceEvent once the
492-- process has started. This is harmless, as the localEventBus is not updated
493-- until /after/ the initial forkProcess completes, so the first call to
494-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
495--
497--------------------------------------------------------------------------------
498-- Handle incoming messages --
499--------------------------------------------------------------------------------
501type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
503data IncomingTarget =
504 Uninit
505 | ToProc ProcessId (Weak (CQueue Message))
506 | ToChan TypedChannel
507 | ToNode
509data ConnectionState = ConnectionState {
510 _incoming :: !(Map NT.ConnectionId IncomingConnection)
511 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
512 }
514initConnectionState :: ConnectionState
515initConnectionState = ConnectionState {
516 _incoming = Map.empty
517 , _incomingFrom = Map.empty
518 }
520incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
521incoming = accessor _incoming (\conns st -> st { _incoming = conns })
523incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
524incomingAt cid = incoming >>> DAC.mapMaybe cid
526incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
527incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
528 where
529 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
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.
533data 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 }
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--
551monitorThread :: Int -> (a -> IO ()) -> IO (EventMonitor a)
552monitorThread 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 }
570handleIncomingMessages :: LocalNode -> IO ()
571handleIncomingMessages 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
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
586 where
587 go :: EventMonitor NT.Event -> ConnectionState -> IO ()
588 go eventMon !st = do
589 eventMonitorHandledEvent eventMon
590 event <- NT.receive endpoint
591 eventMonitorGotEvent eventMon event
592 case event of
593 NT.ConnectionOpened cid rel theirAddr ->
594 if rel == NT.ReliableOrdered
595 then
596 trace node (MxConnected cid theirAddr)
597 >> go eventMon (
598 (incomingAt cid ^= Just (theirAddr, Uninit))
599 . (incomingFrom theirAddr ^: Set.insert cid)
600 $ st
601 )
602 else invalidRequest eventMon cid st $
603 "attempt to connect with unsupported reliability " ++ show rel
604 NT.Received cid payload ->
605 case st ^. incomingAt cid of
606 Just (_, ToProc pid weakQueue) -> do
607 mQueue <- deRefWeak weakQueue
608 forM_ mQueue $ \queue -> do
609 -- TODO: if we find that the queue is Nothing, should we remove
610 -- it from the NC state? (and same for channels, below)
611 let msg = payloadToMessage payload
612 enqueue queue msg -- 'enqueue' is strict
613 trace node (MxReceived pid msg)
614 go eventMon st
615 Just (_, ToChan (TypedChannel chan')) -> do
616 mChan <- deRefWeak chan'
617 -- If mChan is Nothing, the process has given up the read end of
618 -- the channel and we simply ignore the incoming message
619 forM_ mChan $ \chan -> atomically $
620 -- We make sure the message is fully decoded when it is enqueued
621 writeTQueue chan $! decode (BSL.fromChunks payload)
622 go eventMon st
623 Just (_, ToNode) -> do
624 let ctrlMsg = decode . BSL.fromChunks $ payload
625 writeChan ctrlChan $! ctrlMsg
626 go eventMon st
627 Just (src, Uninit) ->
628 case decode (BSL.fromChunks payload) of
629 ProcessIdentifier pid -> do
630 let lpid = processLocalId pid
631 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
632 case mProc of
633 Just proc ->
634 go eventMon (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
635 Nothing ->
636 -- incoming attempt to connect to unknown process - might
637 -- be dead already
638 go eventMon (incomingAt cid ^= Nothing $ st)
639 SendPortIdentifier chId -> do
640 let lcid = sendPortLocalId chId
641 lpid = processLocalId (sendPortProcessId chId)
642 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
643 case mProc of
644 Just proc -> do
645 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
646 case mChannel of
647 Just channel ->
648 go eventMon
649 (incomingAt cid ^= Just (src, ToChan channel) $ st)
650 Nothing ->
651 invalidRequest eventMon cid st $
652 "incoming attempt to connect to unknown channel of"
653 ++ " process " ++ show (sendPortProcessId chId)
654 Nothing ->
655 -- incoming attempt to connect to channel of unknown
656 -- process - might be dead already
657 go eventMon (incomingAt cid ^= Nothing $ st)
658 NodeIdentifier nid ->
659 if nid == localNodeId node
660 then go eventMon (incomingAt cid ^= Just (src, ToNode) $ st)
661 else invalidRequest eventMon cid st $
662 "incoming attempt to connect to a different node -"
663 ++ " I'm " ++ show (localNodeId node)
664 ++ " but the remote peer wants to connect to "
665 ++ show nid
666 Nothing ->
667 invalidRequest eventMon cid st
668 "message received from an unknown connection"
669 NT.ConnectionClosed cid ->
670 case st ^. incomingAt cid of
671 Nothing ->
672 invalidRequest eventMon cid st "closed unknown connection"
673 Just (src, _) -> do
674 trace node (MxDisconnected cid src)
675 go eventMon ( (incomingAt cid ^= Nothing)
676 . (incomingFrom src ^: Set.delete cid)
677 $ st
678 )
679 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
680 -- [Unified table 9, rule node_disconnect]
681 let nid = NodeIdentifier $ NodeId theirAddr
682 writeChan ctrlChan NCMsg
683 { ctrlMsgSender = nid
684 , ctrlMsgSignal = Died nid DiedDisconnect
685 }
686 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
687 liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)
688 >>= maybe (return ()) killThread
689 closeImplicitReconnections node nid
690 go eventMon ( (incomingFrom theirAddr ^= Set.empty)
691 . (incoming ^: Map.filterWithKey (const . notLost))
692 $ st
693 )
694 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
695 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
696 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
697 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
698 NT.EndPointClosed ->
699 return ()
700 NT.ReceivedMulticast _ _ ->
701 -- If we received a multicast message, something went horribly wrong
702 -- and we just give up
703 fail "Cloud Haskell fatal error: received unexpected multicast"
705 invalidRequest :: EventMonitor NT.Event
706 -> NT.ConnectionId
707 -> ConnectionState -> String -> IO ()
708 invalidRequest eventMon cid st msg = do
709 -- TODO: We should treat this as a fatal error on the part of the remote
710 -- node. That is, we should report the remote node as having died, and we
711 -- should close incoming connections (this requires a Transport layer
712 -- extension).
713 traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"
714 ++ " (" ++ msg ++ "): "
715 , (Trace cid)
716 ]
717 go eventMon ( incomingAt cid ^= Nothing
718 $ st
719 )
721 endpoint = localEndPoint node
722 ctrlChan = localCtrlChan node
724--------------------------------------------------------------------------------
725-- Top-level access to the node controller --
726--------------------------------------------------------------------------------
728runNodeController :: LocalNode -> IO ()
729runNodeController 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 ->
737 (runReaderT (evalStateT (unNC (nodeController eventMon)) initNCState) node
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
745--------------------------------------------------------------------------------
746-- Internal data types --
747--------------------------------------------------------------------------------
749data NCState = NCState
750 { -- Mapping from remote processes to linked local processes
751 _links :: !(BiMultiMap Identifier ProcessId ())
752 -- Mapping from remote processes to monitoring local processes
753 , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef)
754 -- Process registry: names and where they live, mapped to the PIDs
755 , _registeredHere :: !(Map String ProcessId)
756 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
757 }
759submitSendPool :: LocalNode -> NodeId -> IO () -> IO ()
760submitSendPool node nid task = submitTask (localSendPool node) fork nid task
761 where
762 fork io =
763 forkIO $ io `Exception.catch` \(NodeClosedException _) -> return ()
765newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
766 deriving ( Applicative
767 , Functor
768 , Monad
769 , MonadIO
770 , MonadState NCState
771 , MonadReader LocalNode
772 )
774initNCState :: NCState
775initNCState = NCState { _links = BiMultiMap.empty
776 , _monitors = BiMultiMap.empty
777 , _registeredHere = Map.empty
778 , _registeredOnNodes = Map.empty
779 }
781-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
782-- type is deliberately not exported so it cannot be caught explicitly.
783data ProcessKillException =
784 ProcessKillException !ProcessId !String
785 deriving (Typeable)
787instance Exception ProcessKillException
788instance Show ProcessKillException where
789 show (ProcessKillException pid reason) =
790 "killed-by=" ++ show pid ++ ",reason=" ++ reason
792ncSendToProcess :: ProcessId -> Message -> NC ()
793ncSendToProcess = ncSendToProcessAndTrace True
795ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
796ncSendToProcessAndTrace shouldTrace pid msg = do
797 node <- ask
798 if processNodeId pid == localNodeId node
799 then ncEffectLocalSendAndTrace shouldTrace node pid msg
800 else liftIO $ submitSendPool node (processNodeId pid) $ sendBinary node
801 (NodeIdentifier $ localNodeId node)
802 (NodeIdentifier $ processNodeId pid)
803 WithImplicitReconnect
804 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
805 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
806 }
808ncSendToNode :: NodeId -> NCMsg -> NC ()
809ncSendToNode to msg = do
810 node <- ask
811 liftIO $ if to == localNodeId node
812 then writeChan (localCtrlChan node) $! msg
813 else submitSendPool node to $ sendBinary node
814 (NodeIdentifier $ localNodeId node)
815 (NodeIdentifier to)
816 WithImplicitReconnect
817 msg
819--------------------------------------------------------------------------------
820-- Tracing/Debugging --
821--------------------------------------------------------------------------------
823-- [Issue #104 / DP-13]
825traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
826traceNotifyDied node ident reason =
827 -- TODO: sendPortDied notifications
828 liftIO $ withLocalTracer node $ \t ->
829 case ident of
830 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
831 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
832 _ -> return ()
834traceEventFmtIO :: LocalNode
835 -> String
836 -> [TraceArg]
837 -> IO ()
838traceEventFmtIO node fmt args =
839 withLocalTracer node $ \t -> traceLogFmt t fmt args
841trace :: LocalNode -> MxEvent -> IO ()
842trace node ev = withLocalTracer node $ \t -> traceEvent t ev
844withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
845withLocalTracer node act = act (localEventBus node)
847--------------------------------------------------------------------------------
848-- Core functionality --
849--------------------------------------------------------------------------------
851-- [Unified: Table 7]
852nodeController :: EventMonitor NCMsg -> NC ()
853nodeController eventMon = do
854 node <- ask
855 forever' $ do
856 liftIO $ eventMonitorHandledEvent eventMon
857 msg <- liftIO $ readChan (localCtrlChan node)
858 liftIO $ eventMonitorGotEvent eventMon msg
860 -- [Unified: Table 7, rule nc_forward]
861 case destNid (ctrlMsgSignal msg) of
862 Just nid' | nid' /= localNodeId node ->
863 ncSendToNode nid' msg
864 _ ->
865 return ()
867 case msg of
868 NCMsg (ProcessIdentifier from) (Link them) ->
869 ncEffectMonitor from them Nothing
870 NCMsg (ProcessIdentifier from) (Monitor ref) ->
871 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
872 NCMsg (ProcessIdentifier from) (Unlink them) ->
873 ncEffectUnlink from them
874 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
875 ncEffectUnmonitor from ref
876 NCMsg _from (Died ident reason) ->
877 ncEffectDied ident reason
878 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
879 ncEffectSpawn from proc ref
880 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
881 ncEffectRegister from label atnode pid force
882 NCMsg (ProcessIdentifier from) (WhereIs label) ->
883 ncEffectWhereIs from label
884 NCMsg _ (NamedSend label msg') ->
885 ncEffectNamedSend label msg'
886 NCMsg _ (UnreliableSend lpid msg') ->
887 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
888 NCMsg _ (LocalSend to msg') ->
889 ncEffectLocalSend node to msg'
890 NCMsg _ (LocalPortSend to msg') ->
891 ncEffectLocalPortSend to msg'
892 NCMsg (ProcessIdentifier from) (Kill to reason) ->
893 ncEffectKill from to reason
894 NCMsg (ProcessIdentifier from) (Exit to reason) ->
895 ncEffectExit from to reason
896 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
897 ncEffectGetInfo from pid
898 NCMsg _ SigShutdown ->
899 liftIO $ do
900 NT.closeEndPoint (localEndPoint node)
901 `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
902 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
903 ncEffectGetNodeStats from nid
904 unexpected ->
905 error $ "nodeController: unexpected message " ++ show unexpected
907-- [Unified: Table 10]
908ncEffectMonitor :: ProcessId -- ^ Who's watching?
909 -> Identifier -- ^ Who's being watched?
910 -> Maybe MonitorRef -- ^ 'Nothing' to link
911 -> NC ()
912ncEffectMonitor from them mRef = do
913 node <- ask
914 shouldLink <-
915 if not (isLocal node them)
916 then return True
917 else isValidLocalIdentifier them
918 case (shouldLink, isLocal node (ProcessIdentifier from)) of
919 (True, _) -> -- [Unified: first rule]
920 case mRef of
921 Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref
922 Nothing -> modify' $ links ^: BiMultiMap.insert them from ()
923 (False, True) -> -- [Unified: second rule]
924 notifyDied from them DiedUnknownId mRef
925 (False, False) -> -- [Unified: third rule]
926 -- TODO: this is the right sender according to the Unified semantics,
927 -- but perhaps having 'them' as the sender would make more sense
928 -- (see also: notifyDied)
929 ncSendToNode (processNodeId from) $ NCMsg
930 { ctrlMsgSender = NodeIdentifier (localNodeId node)
931 , ctrlMsgSignal = Died them DiedUnknownId
932 }
934-- [Unified: Table 11]
935ncEffectUnlink :: ProcessId -> Identifier -> NC ()
936ncEffectUnlink from them = do
937 node <- ask
938 when (isLocal node (ProcessIdentifier from)) $
939 case them of
940 ProcessIdentifier pid ->
941 postAsMessage from $ DidUnlinkProcess pid
942 NodeIdentifier nid ->
943 postAsMessage from $ DidUnlinkNode nid
944 SendPortIdentifier cid ->
945 postAsMessage from $ DidUnlinkPort cid
946 modify' $ links ^: BiMultiMap.delete them from ()
948-- [Unified: Table 11]
949ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
950ncEffectUnmonitor from ref = do
951 node <- ask
952 when (isLocal node (ProcessIdentifier from)) $
953 postAsMessage from $ DidUnmonitor ref
954 modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref
956-- [Unified: Table 12]
957ncEffectDied :: Identifier -> DiedReason -> NC ()
958ncEffectDied ident reason = do
959 node <- ask
960 traceNotifyDied node ident reason
961 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
962 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
964-- _registry :: !(Map (String,NodeId) ProcessId)
966 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
968 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
969 forM_ uss $ \(us, _) ->
970 when (localOnly <= isLocal node (ProcessIdentifier us)) $
971 notifyDied us them reason Nothing
973 forM_ (Map.toList affectedMons) $ \(them, refs) ->
974 forM_ refs $ \(us, ref) ->
975 when (localOnly <= isLocal node (ProcessIdentifier us)) $
976 notifyDied us them reason (Just ref)
978 -- Notify remote nodes that the process died so it can be removed from monitor
979 -- lists.
980 mapM_ (forwardDeath node) $
981 [ nid | ProcessIdentifier pid <- [ident]
982 , i <- Set.toList $ Set.union
983 (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedLinks)
984 (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedMons)
985 , let nid = nodeOf i
986 , nid /= localNodeId node
987 ]
989 -- Delete monitors in the local node.
990 let deleteDeads :: (Ord a, Ord v)
991 => BiMultiMap a ProcessId v -> BiMultiMap a ProcessId v
992 deleteDeads = case ident of
993 -- deleteAllBy2nd is faster than partitionWithKeyBy2nd
994 ProcessIdentifier pid -> BiMultiMap.deleteAllBy2nd pid
995 _ -> snd . BiMultiMap.partitionWithKeyBy2nd
996 (\pid _ -> ident `impliesDeathOf` ProcessIdentifier pid)
997 unaffectedLinks' = deleteDeads unaffectedLinks
998 unaffectedMons' = deleteDeads unaffectedMons
1000 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
1002 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
1004 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
1005 mapM (\(pid,nidlist) ->
1006 case ident `impliesDeathOf` ProcessIdentifier pid of
1007 True ->
1008 do forM_ nidlist $ \(nid,_) ->
1009 when (not $ isLocal node (NodeIdentifier nid))
1010 (forwardDeath node nid)
1011 return Nothing
1012 False -> return $ Just (pid,nidlist) )
1013 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
1014 where
1015 forwardDeath node nid = ncSendToNode nid
1016 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
1017 , ctrlMsgSignal = Died ident reason
1020-- [Unified: Table 13]
1021ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
1022ncEffectSpawn pid cProc ref = do
1023 mProc <- unClosure cProc
1024 -- If the closure does not exist, we spawn a process that throws an exception
1025 -- This allows the remote node to find out what's happening
1026 -- TODO:
1027 let proc = case mProc of
1028 Left err -> fail $ "Error: Could not resolve closure: " ++ err
1029 Right p -> p
1030 node <- ask
1031 pid' <- liftIO $ forkProcess node proc
1032 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
1034-- Unified semantics does not explicitly describe how to implement 'register',
1035-- but mentions it's "very similar to nsend" (Table 14)
1036-- We send a response indicated if the operation is invalid
1037ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
1038ncEffectRegister from label atnode mPid reregistration = do
1039 node <- ask
1040 currentVal <- gets (^. registeredHereFor label)
1041 isOk <-
1042 case mPid of
1043 Nothing -> -- unregister request
1044 return $ isJust currentVal
1045 Just thepid -> -- register request
1046 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
1047 return $ (isNothing currentVal /= reregistration) &&
1048 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
1049 if isLocal node (NodeIdentifier atnode)
1050 then do when isOk $
1051 do modify' $ registeredHereFor label ^= mPid
1052 updateRemote node currentVal mPid
1053 case mPid of
1054 (Just p) -> liftIO $ trace node (MxRegistered p label)
1055 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
1056 newVal <- gets (^. registeredHereFor label)
1057 ncSendToProcess from $ unsafeCreateUnencodedMessage $
1058 RegisterReply label isOk newVal
1059 else let operation =
1060 case reregistration of
1061 True -> flip decList
1062 False -> flip incList
1063 in case mPid of
1064 Nothing -> return ()
1065 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
1066 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
1067 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
1068 forward node (processNodeId newval) (Register label atnode (Just newval) False)
1069 updateRemote node Nothing (Just newval) =
1070 forward node (processNodeId newval) (Register label atnode (Just newval) False)
1071 updateRemote node (Just oldval) Nothing =
1072 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
1073 updateRemote _ _ _ = return ()
1074 maybeify f Nothing = unmaybeify $ f []
1075 maybeify f (Just x) = unmaybeify $ f x
1077 unmaybeify [] = Nothing
1078 unmaybeify x = Just x
1079 incList [] tag = [(tag,1)]
1080 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
1081 incList (x:xs) tag = x : incList xs tag
1082 decList [] _ = []
1083 decList ((atag,1):xs) tag | atag == tag = xs
1084 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
1085 decList (x:xs) tag = x:decList xs tag
1086 forward node to reg =
1087 when (not $ isLocal node (NodeIdentifier to)) $
1088 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
1089 , ctrlMsgSignal = reg
1093-- Unified semantics does not explicitly describe 'whereis'
1094ncEffectWhereIs :: ProcessId -> String -> NC ()
1095ncEffectWhereIs from label = do
1096 mPid <- gets (^. registeredHereFor label)
1097 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
1099-- [Unified: Table 14]
1100ncEffectNamedSend :: String -> Message -> NC ()
1101ncEffectNamedSend label msg = do
1102 mPid <- gets (^. registeredHereFor label)
1103 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
1104 forM_ mPid $ \to ->
1105 -- If this is a trace message we don't trace it to avoid entering a loop
1106 -- where trace messages produce more trace messages.
1107 ncSendToProcessAndTrace (label /= "trace.logger") to msg
1109-- [Issue #DP-20]
1110ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
1111ncEffectLocalSend = ncEffectLocalSendAndTrace True
1113ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
1114ncEffectLocalSendAndTrace shouldTrace node to msg =
1115 liftIO $ withLocalProc node to $ \p -> do
1116 enqueue (processQueue p) msg
1117 when shouldTrace $ trace node (MxReceived to msg)
1119-- [Issue #DP-20]
1120ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
1121ncEffectLocalPortSend from msg = do
1122 node <- ask
1123 let pid = sendPortProcessId from
1124 cid = sendPortLocalId from
1125 liftIO $ withLocalProc node pid $ \proc -> do
1126 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
1127 case mChan of
1128 -- in the unlikely event we know nothing about this channel id,
1129 -- there's little to be done - perhaps some logging/tracing though...
1130 Nothing -> return ()
1131 Just (TypedChannel chan') -> do
1132 -- If ch is Nothing, the process has given up the read end of
1133 -- the channel and we simply ignore the incoming message - this
1134 ch <- deRefWeak chan'
1135 forM_ ch $ \chan -> deliverChan msg chan
1136 where deliverChan :: forall a . Message -> TQueue a -> IO ()
1137 deliverChan (UnencodedMessage _ raw) chan' =
1138 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1139 deliverChan (EncodedMessage _ _) _ =
1140 -- this will not happen unless someone screws with Primitives.hs
1141 error "invalid local channel delivery"
1143-- [Issue #69]
1144ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
1145ncEffectKill from to reason = do
1146 node <- ask
1147 when (isLocal node (ProcessIdentifier to)) $
1148 throwException to $ ProcessKillException from reason
1150-- [Issue #69]
1151ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
1152ncEffectExit from to reason = do
1153 node <- ask
1154 when (isLocal node (ProcessIdentifier to)) $
1155 throwException to $ ProcessExitException from reason
1157-- [Issue #89]
1158ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
1159ncEffectGetInfo from pid =
1160 let lpid = processLocalId pid
1161 them = (ProcessIdentifier pid)
1162 in do
1163 node <- ask
1164 mProc <- liftIO $ withValidLocalState node
1165 $ return . (^. localProcessWithId lpid)
1166 case mProc of
1167 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1168 from (ProcessInfoNone DiedUnknownId)
1169 Just proc -> do
1170 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>
1171 gets (^. links)
1172 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)
1173 registered <- gets (^. registeredHere)
1174 size <- liftIO $ queueSize $ processQueue $ proc
1176 let reg = registeredNames registered
1177 dispatch (isLocal node (ProcessIdentifier from))
1178 from
1179 ProcessInfo {
1180 infoNode = (processNodeId pid)
1181 , infoRegisteredNames = reg
1182 , infoMessageQueueLength = size
1183 , infoMonitors = Set.toList itsMons
1184 , infoLinks = Set.toList itsLinks
1186 where dispatch :: (Serializable a)
1187 => Bool
1188 -> ProcessId
1189 -> a
1190 -> NC ()
1191 dispatch True dest pInfo = postAsMessage dest $ pInfo
1192 dispatch False dest pInfo =
1193 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1195 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1196 then (k:ks)
1197 else ks) []
1199ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1200ncEffectGetNodeStats from _nid = do
1201 node <- ask
1202 ncState <- StateT.get
1203 nodeState <- liftIO $ withValidLocalState node return
1204 let stats =
1205 NodeStats {
1206 nodeStatsNode = localNodeId node
1207 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1208 , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors
1209 , nodeStatsLinks = BiMultiMap.size $ ncState ^. links
1210 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1212 postAsMessage from stats
1214--------------------------------------------------------------------------------
1215-- Auxiliary --
1216--------------------------------------------------------------------------------
1218notifyDied :: ProcessId -- ^ Who to notify?
1219 -> Identifier -- ^ Who died?
1220 -> DiedReason -- ^ How did they die?
1221 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1222 -> NC ()
1223notifyDied dest src reason mRef = do
1224 node <- ask
1225 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1226 (True, Just ref, ProcessIdentifier pid) ->
1227 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1228 (True, Just ref, NodeIdentifier nid) ->
1229 postAsMessage dest $ NodeMonitorNotification ref nid reason
1230 (True, Just ref, SendPortIdentifier cid) ->
1231 postAsMessage dest $ PortMonitorNotification ref cid reason
1232 (True, Nothing, ProcessIdentifier pid) ->
1233 throwException dest $ ProcessLinkException pid reason
1234 (True, Nothing, NodeIdentifier pid) ->
1235 throwException dest $ NodeLinkException pid reason
1236 (True, Nothing, SendPortIdentifier pid) ->
1237 throwException dest $ PortLinkException pid reason
1238 (False, _, _) ->
1239 -- The change in sender comes from [Unified: Table 10]
1240 ncSendToNode (processNodeId dest) $ NCMsg
1241 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1242 , ctrlMsgSignal = Died src reason
1245-- | [Unified: Table 8]
1246destNid :: ProcessSignal -> Maybe NodeId
1247destNid (Link ident) = Just $ nodeOf ident
1248destNid (Unlink ident) = Just $ nodeOf ident
1249destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1250destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1251destNid (Spawn _ _) = Nothing
1252destNid (Register _ _ _ _) = Nothing
1253destNid (WhereIs _) = Nothing
1254destNid (NamedSend _ _) = Nothing
1255destNid (UnreliableSend _ _) = Nothing
1256-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1257-- then when a local process dies the monitoring/linking machinery will take
1258-- care of notifying remote nodes
1259destNid (Died _ _) = Nothing
1260destNid (Kill pid _) = Just $ processNodeId pid
1261destNid (Exit pid _) = Just $ processNodeId pid
1262destNid (GetInfo pid) = Just $ processNodeId pid
1263destNid (GetNodeStats nid) = Just nid
1264destNid (LocalSend pid _) = Just $ processNodeId pid
1265destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1266destNid (SigShutdown) = Nothing
1268-- | Check if a process is local to our own node
1269isLocal :: LocalNode -> Identifier -> Bool
1270isLocal nid ident = nodeOf ident == localNodeId nid
1272-- | Lookup a local closure
1273unClosure :: Typeable a => Closure a -> NC (Either String a)
1274unClosure closure = do
1275 rtable <- remoteTable <$> ask
1276 return (Static.unclosure rtable closure)
1278-- | Check if an identifier refers to a valid local object
1279isValidLocalIdentifier :: Identifier -> NC Bool
1280isValidLocalIdentifier ident = do
1281 node <- ask
1282 liftIO . withValidLocalState node $ \nSt ->
1283 case ident of
1284 NodeIdentifier nid ->
1285 return $ nid == localNodeId node
1286 ProcessIdentifier pid -> do
1287 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1288 return $ isJust mProc
1289 SendPortIdentifier cid -> do
1290 let pid = sendPortProcessId cid
1291 mProc = nSt ^. localProcessWithId (processLocalId pid)
1292 case mProc of
1293 Nothing -> return False
1294 Just proc -> withMVar (processState proc) $ \pSt -> do
1295 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1296 return $ isJust mCh
1298--------------------------------------------------------------------------------
1299-- Messages to local processes --
1300--------------------------------------------------------------------------------
1302postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1303postAsMessage pid = postMessage pid . createUnencodedMessage
1305postMessage :: ProcessId -> Message -> NC ()
1306postMessage pid msg = do
1307 node <- ask
1308 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1310throwException :: Exception e => ProcessId -> e -> NC ()
1311throwException pid e = do
1312 node <- ask
1313 -- throwTo blocks until the exception is received by the target thread.
1314 -- We cannot easily make it happen asynchronpusly because then 'unlink'
1315 -- semantics would break.
1316 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1318withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1319withLocalProc node pid p =
1320 -- By [Unified: table 6, rule missing_process] messages to dead processes
1321 -- can silently be dropped
1322 let lpid = processLocalId pid in do
1323 join $ withValidLocalState node $ \vst ->
1324 return $ forM_ (vst ^. localProcessWithId lpid) p
1326--------------------------------------------------------------------------------
1327-- Accessors --
1328--------------------------------------------------------------------------------
1330links :: Accessor NCState (BiMultiMap Identifier ProcessId ())
1331links = accessor _links (\ls st -> st { _links = ls })
1333monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)
1334monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1336registeredHere :: Accessor NCState (Map String ProcessId)
1337registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1339registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1340registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1342registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1343registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1345registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1346registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1348-- | @splitNotif ident@ splits a notifications map into those
1349-- notifications that should trigger when 'ident' fails and those links that
1350-- should not.
1352-- There is a hierarchy between identifiers: failure of a node implies failure
1353-- of all processes on that node, and failure of a process implies failure of
1354-- all typed channels to that process. In other words, if 'ident' refers to a
1355-- node, then the /should trigger/ set will include
1357-- * the notifications for the node specifically
1358-- * the notifications for processes on that node, and
1359-- * the notifications for typed channels to processes on that node.
1361-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1362-- include
1364-- * the notifications for that process specifically and
1365-- * the notifications for typed channels to that process.
1367-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1368splitNotif :: (Ord a, Ord v)
1369 => Identifier
1370 -> BiMultiMap Identifier a v
1371 -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)
1372splitNotif ident =
1373 BiMultiMap.partitionWithKeyBy1st (\k !_v -> ident `impliesDeathOf` k)
1375--------------------------------------------------------------------------------
1376-- Auxiliary --
1377--------------------------------------------------------------------------------
1379-- | Modify and evaluate the state
1380modify' :: MonadState s m => (s -> s) -> m ()
1381modify' f = StateT.get >>= \s -> StateT.put $! f s