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