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