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