git.iko.soy/distributed-process
on 02494f2 help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1345 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 void $ registryMonitorAgent
380 where
381 loop = do
382 receiveWait
383 [ match $ \(SayMessage time pid string) -> do
384 let time' = formatTime defaultTimeLocale "%c" time
385 liftIO . hPutStrLn stderr $ time' ++ " " ++ show pid ++ ": " ++ string
386 loop
387 , match $ \((time, string) :: (String, String)) -> do
388 -- this is a 'trace' message from the local node tracer
389 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
390 loop
391 , match $ \(ch :: SendPort ()) -> -- a shutdown request
392 sendChan ch ()
393 ]
395-- | Force-close a local node, killing all processes on that node.
396closeLocalNode :: LocalNode -> IO ()
397closeLocalNode node = do
398 -- Kill processes after refilling the mvar. Otherwise, there is potential for
399 -- deadlock as a dying process tries to get the mvar while masking exceptions
400 -- uninterruptibly.
401 join $ modifyMVar (localState node) $ \st -> case st of
402 LocalNodeValid vst -> do
403 return ( LocalNodeClosed
404 , forM_ (vst ^. localProcesses) $ \lproc ->
405 -- Semantics of 'throwTo' guarantee that target thread will get
406 -- delivered an exception. Therefore, target thread will be
407 -- killed eventually and that's as good as we can do. No need
408 -- to wait for thread to actually finish dying.
409 killThread (processThread lproc)
410 )
411 LocalNodeClosed -> return (LocalNodeClosed, return ())
412 -- This call will have the effect of shutting down the NC as well (see
413 -- 'createBareLocalNode').
414 NT.closeEndPoint (localEndPoint node)
416-- | Run a process on a local node and wait for it to finish
417runProcess :: LocalNode -> Process () -> IO ()
418runProcess node proc = do
419 done <- newEmptyMVar
420 -- TODO; When forkProcess inherits the masking state, protect the forked
421 -- thread against async exceptions that could occur before 'try' is evaluated.
422 void $ forkProcess node $ try proc >>= liftIO . putMVar done
423 takeMVar done >>= either (throwIO :: SomeException -> IO a) return
425-- | Spawn a new process on a local node
426forkProcess :: LocalNode -> Process () -> IO ProcessId
427forkProcess node proc = do
428 ms <- getMaskingState
429 modifyMVarMasked (localState node) (startProcess ms)
430 where
431 startProcess :: MaskingState
432 -> LocalNodeState
433 -> IO (LocalNodeState, ProcessId)
434 startProcess ms (LocalNodeValid vst) = do
435 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
436 , lpidUnique = vst ^. localPidUnique
437 }
438 let pid = ProcessId { processNodeId = localNodeId node
439 , processLocalId = lpid
440 }
441 pst <- newMVar LocalProcessState { _monitorCounter = 0
442 , _spawnCounter = 0
443 , _channelCounter = 0
444 , _typedChannels = Map.empty
445 }
446 queue <- newCQueue
447 weakQueue <- mkWeakCQueue queue (return ())
448 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
449 let lproc = LocalProcess { processQueue = queue
450 , processWeakQ = weakQueue
451 , processId = pid
452 , processState = pst
453 , processThread = tid
454 , processNode = node
455 }
456 -- Rewrite this code when this is fixed:
457 -- https://ghc.haskell.org/trac/ghc/ticket/10149
458 let unmask = case ms of
459 Unmasked -> unblock
460 MaskedInterruptible -> block
461 MaskedUninterruptible -> id
462 tid' <- uninterruptibleMask_ $ forkIO $ do
463 reason <- Exception.catches
464 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
465 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
466 mMsg <- unwrapMessage msg :: IO (Maybe String)
467 case mMsg of
468 Nothing -> return $ DiedException $ show ex
469 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
470 , (Exception.Handler
471 (return . DiedException . (show :: SomeException -> String)))]
473 -- [Unified: Table 4, rules termination and exiting]
474 mconns <- modifyValidLocalState node (cleanupProcess pid)
475 -- XXX: Revisit after agreeing on the bigger picture for the semantics
476 -- of transport operations.
477 -- https://github.com/haskell-distributed/distributed-process/issues/204
478 forM_ mconns $ forkIO . mapM_ NT.close
480 writeChan (localCtrlChan node) NCMsg
481 { ctrlMsgSender = ProcessIdentifier pid
482 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
483 }
484 return (tid', lproc)
486 -- see note [tracer/forkProcess races]
487 trace node (MxSpawned pid)
489 if lpidCounter lpid == maxBound
490 then do
491 -- TODO: this doesn't look right at all - how do we know
492 -- that newUnique represents a process id that is available!?
493 newUnique <- randomIO
494 return ( LocalNodeValid
495 $ (localProcessWithId lpid ^= Just lproc)
496 . (localPidCounter ^= firstNonReservedProcessId)
497 . (localPidUnique ^= newUnique)
498 $ vst
499 , pid
500 )
501 else
502 return ( LocalNodeValid
503 $ (localProcessWithId lpid ^= Just lproc)
504 . (localPidCounter ^: (+ 1))
505 $ vst
506 , pid
507 )
508 startProcess _ LocalNodeClosed =
509 throwIO $ NodeClosedException $ localNodeId node
511 cleanupProcess :: ProcessId
512 -> ValidLocalNodeState
513 -> IO (ValidLocalNodeState, [NT.Connection])
514 cleanupProcess pid vst = do
515 let pid' = ProcessIdentifier pid
516 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
517 return ( (localProcessWithId (processLocalId pid) ^= Nothing)
518 . (localConnections ^= unaffected)
519 $ vst
520 , map fst $ Map.elems affected
521 )
523-- note [tracer/forkProcess races]
524--
525-- Our startTracing function uses forkProcess to start the trace controller
526-- process, and of course forkProcess attempts to call traceEvent once the
527-- process has started. This is harmless, as the localEventBus is not updated
528-- until /after/ the initial forkProcess completes, so the first call to
529-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
530--
532--------------------------------------------------------------------------------
533-- Handle incoming messages --
534--------------------------------------------------------------------------------
536type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
538data IncomingTarget =
539 Uninit
540 | ToProc ProcessId (Weak (CQueue Message))
541 | ToChan TypedChannel
542 | ToNode
544data ConnectionState = ConnectionState {
545 _incoming :: !(Map NT.ConnectionId IncomingConnection)
546 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
547 }
549initConnectionState :: ConnectionState
550initConnectionState = ConnectionState {
551 _incoming = Map.empty
552 , _incomingFrom = Map.empty
553 }
555incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
556incoming = accessor _incoming (\conns st -> st { _incoming = conns })
558incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
559incomingAt cid = incoming >>> DAC.mapMaybe cid
561incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
562incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
563 where
564 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
566handleIncomingMessages :: LocalNode -> IO ()
567handleIncomingMessages node = go initConnectionState
568 `Exception.catch` \(NodeClosedException _) -> return ()
569 where
570 go :: ConnectionState -> IO ()
571 go !st = do
572 event <- NT.receive endpoint
573 case event of
574 NT.ConnectionOpened cid rel theirAddr ->
575 if rel == NT.ReliableOrdered
576 then
577 trace node (MxConnected cid theirAddr)
578 >> go (
579 (incomingAt cid ^= Just (theirAddr, Uninit))
580 . (incomingFrom theirAddr ^: Set.insert cid)
581 $ st
582 )
583 else invalidRequest cid st $
584 "attempt to connect with unsupported reliability " ++ show rel
585 NT.Received cid payload ->
586 case st ^. incomingAt cid of
587 Just (_, ToProc pid weakQueue) -> do
588 mQueue <- deRefWeak weakQueue
589 forM_ mQueue $ \queue -> do
590 -- TODO: if we find that the queue is Nothing, should we remove
591 -- it from the NC state? (and same for channels, below)
592 let msg = payloadToMessage payload
593 enqueue queue msg -- 'enqueue' is strict
594 trace node (MxReceived pid msg)
595 go st
596 Just (_, ToChan (TypedChannel chan')) -> do
597 mChan <- deRefWeak chan'
598 -- If mChan is Nothing, the process has given up the read end of
599 -- the channel and we simply ignore the incoming message
600 forM_ mChan $ \chan -> atomically $
601 -- We make sure the message is fully decoded when it is enqueued
602 writeTQueue chan $! decode (BSL.fromChunks payload)
603 go st
604 Just (_, ToNode) -> do
605 let ctrlMsg = decode . BSL.fromChunks $ payload
606 writeChan ctrlChan $! ctrlMsg
607 go st
608 Just (src, Uninit) ->
609 case decode (BSL.fromChunks payload) of
610 ProcessIdentifier pid -> do
611 let lpid = processLocalId pid
612 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
613 case mProc of
614 Just proc ->
615 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
616 Nothing ->
617 -- incoming attempt to connect to unknown process - might
618 -- be dead already
619 go (incomingAt cid ^= Nothing $ st)
620 SendPortIdentifier chId -> do
621 let lcid = sendPortLocalId chId
622 lpid = processLocalId (sendPortProcessId chId)
623 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
624 case mProc of
625 Just proc -> do
626 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
627 case mChannel of
628 Just channel ->
629 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
630 Nothing ->
631 invalidRequest cid st $
632 "incoming attempt to connect to unknown channel of"
633 ++ " process " ++ show (sendPortProcessId chId)
634 Nothing ->
635 -- incoming attempt to connect to channel of unknown
636 -- process - might be dead already
637 go (incomingAt cid ^= Nothing $ st)
638 NodeIdentifier nid ->
639 if nid == localNodeId node
640 then go (incomingAt cid ^= Just (src, ToNode) $ st)
641 else invalidRequest cid st $
642 "incoming attempt to connect to a different node -"
643 ++ " I'm " ++ show (localNodeId node)
644 ++ " but the remote peer wants to connect to "
645 ++ show nid
646 Nothing ->
647 invalidRequest cid st
648 "message received from an unknown connection"
649 NT.ConnectionClosed cid ->
650 case st ^. incomingAt cid of
651 Nothing ->
652 invalidRequest cid st "closed unknown connection"
653 Just (src, _) -> do
654 trace node (MxDisconnected cid src)
655 go ( (incomingAt cid ^= Nothing)
656 . (incomingFrom src ^: Set.delete cid)
657 $ st
658 )
659 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
660 -- [Unified table 9, rule node_disconnect]
661 let nid = NodeIdentifier $ NodeId theirAddr
662 writeChan ctrlChan NCMsg
663 { ctrlMsgSender = nid
664 , ctrlMsgSignal = Died nid DiedDisconnect
665 }
666 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
667 closeImplicitReconnections node nid
668 go ( (incomingFrom theirAddr ^= Set.empty)
669 . (incoming ^: Map.filterWithKey (const . notLost))
670 $ st
671 )
672 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
673 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
674 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
675 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
676 NT.EndPointClosed ->
677 return ()
678 NT.ReceivedMulticast _ _ ->
679 -- If we received a multicast message, something went horribly wrong
680 -- and we just give up
681 fail "Cloud Haskell fatal error: received unexpected multicast"
683 invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
684 invalidRequest cid st msg = do
685 -- TODO: We should treat this as a fatal error on the part of the remote
686 -- node. That is, we should report the remote node as having died, and we
687 -- should close incoming connections (this requires a Transport layer
688 -- extension).
689 traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"
690 ++ " (" ++ msg ++ "): "
691 , (Trace cid)
692 ]
693 go ( incomingAt cid ^= Nothing
694 $ st
695 )
697 endpoint = localEndPoint node
698 ctrlChan = localCtrlChan node
700--------------------------------------------------------------------------------
701-- Top-level access to the node controller --
702--------------------------------------------------------------------------------
704runNodeController :: LocalNode -> IO ()
705runNodeController node =
706 runReaderT (evalStateT (unNC nodeController) initNCState) node
707 `Exception.catch` \(NodeClosedException _) -> return ()
709--------------------------------------------------------------------------------
710-- Internal data types --
711--------------------------------------------------------------------------------
713data NCState = NCState
714 { -- Mapping from remote processes to linked local processes
715 _links :: !(BiMultiMap Identifier ProcessId ())
716 -- Mapping from remote processes to monitoring local processes
717 , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef)
718 -- Process registry: names and where they live, mapped to the PIDs
719 , _registeredHere :: !(Map String ProcessId)
720 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
721 }
723newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
724 deriving ( Applicative
725 , Functor
726 , Monad
727 , MonadIO
728 , MonadState NCState
729 , MonadReader LocalNode
730 )
732initNCState :: NCState
733initNCState = NCState { _links = BiMultiMap.empty
734 , _monitors = BiMultiMap.empty
735 , _registeredHere = Map.empty
736 , _registeredOnNodes = Map.empty
737 }
739-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
740-- type is deliberately not exported so it cannot be caught explicitly.
741data ProcessKillException =
742 ProcessKillException !ProcessId !String
743 deriving (Typeable)
745instance Exception ProcessKillException
746instance Show ProcessKillException where
747 show (ProcessKillException pid reason) =
748 "killed-by=" ++ show pid ++ ",reason=" ++ reason
750ncSendToProcess :: ProcessId -> Message -> NC ()
751ncSendToProcess = ncSendToProcessAndTrace True
753ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
754ncSendToProcessAndTrace shouldTrace pid msg = do
755 node <- ask
756 if processNodeId pid == localNodeId node
757 then ncEffectLocalSendAndTrace shouldTrace node pid msg
758 else liftIO $ sendBinary node
759 (NodeIdentifier $ localNodeId node)
760 (NodeIdentifier $ processNodeId pid)
761 WithImplicitReconnect
762 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
763 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
764 }
766ncSendToNode :: NodeId -> NCMsg -> NC ()
767ncSendToNode to msg = do
768 node <- ask
769 liftIO $ if to == localNodeId node
770 then writeChan (localCtrlChan node) $! msg
771 else sendBinary node
772 (NodeIdentifier $ localNodeId node)
773 (NodeIdentifier to)
774 WithImplicitReconnect
775 msg
777--------------------------------------------------------------------------------
778-- Tracing/Debugging --
779--------------------------------------------------------------------------------
781-- [Issue #104 / DP-13]
783traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
784traceNotifyDied node ident reason =
785 -- TODO: sendPortDied notifications
786 liftIO $ withLocalTracer node $ \t ->
787 case ident of
788 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
789 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
790 _ -> return ()
792traceEventFmtIO :: LocalNode
793 -> String
794 -> [TraceArg]
795 -> IO ()
796traceEventFmtIO node fmt args =
797 withLocalTracer node $ \t -> traceLogFmt t fmt args
799trace :: LocalNode -> MxEvent -> IO ()
800trace node ev = withLocalTracer node $ \t -> traceEvent t ev
802withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
803withLocalTracer node act = act (localEventBus node)
805--------------------------------------------------------------------------------
806-- Core functionality --
807--------------------------------------------------------------------------------
809-- [Unified: Table 7]
810nodeController :: NC ()
811nodeController = do
812 node <- ask
813 forever' $ do
814 msg <- liftIO $ readChan (localCtrlChan node)
816 -- [Unified: Table 7, rule nc_forward]
817 case destNid (ctrlMsgSignal msg) of
818 Just nid' | nid' /= localNodeId node ->
819 ncSendToNode nid' msg
820 _ ->
821 return ()
823 case msg of
824 NCMsg (ProcessIdentifier from) (Link them) ->
825 ncEffectMonitor from them Nothing
826 NCMsg (ProcessIdentifier from) (Monitor ref) ->
827 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
828 NCMsg (ProcessIdentifier from) (Unlink them) ->
829 ncEffectUnlink from them
830 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
831 ncEffectUnmonitor from ref
832 NCMsg _from (Died ident reason) ->
833 ncEffectDied ident reason
834 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
835 ncEffectSpawn from proc ref
836 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
837 ncEffectRegister from label atnode pid force
838 NCMsg (ProcessIdentifier from) (WhereIs label) ->
839 ncEffectWhereIs from label
840 NCMsg _ (NamedSend label msg') ->
841 ncEffectNamedSend label msg'
842 NCMsg _ (UnreliableSend lpid msg') ->
843 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
844 NCMsg _ (LocalSend to msg') ->
845 ncEffectLocalSend node to msg'
846 NCMsg _ (LocalPortSend to msg') ->
847 ncEffectLocalPortSend to msg'
848 NCMsg (ProcessIdentifier from) (Kill to reason) ->
849 ncEffectKill from to reason
850 NCMsg (ProcessIdentifier from) (Exit to reason) ->
851 ncEffectExit from to reason
852 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
853 ncEffectGetInfo from pid
854 NCMsg _ SigShutdown ->
855 liftIO $ do
856 NT.closeEndPoint (localEndPoint node)
857 `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
858 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
859 ncEffectGetNodeStats from nid
860 unexpected ->
861 error $ "nodeController: unexpected message " ++ show unexpected
863-- [Unified: Table 10]
864ncEffectMonitor :: ProcessId -- ^ Who's watching?
865 -> Identifier -- ^ Who's being watched?
866 -> Maybe MonitorRef -- ^ 'Nothing' to link
867 -> NC ()
868ncEffectMonitor from them mRef = do
869 node <- ask
870 shouldLink <-
871 if not (isLocal node them)
872 then return True
873 else isValidLocalIdentifier them
874 case (shouldLink, isLocal node (ProcessIdentifier from)) of
875 (True, _) -> -- [Unified: first rule]
876 case mRef of
877 Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref
878 Nothing -> modify' $ links ^: BiMultiMap.insert them from ()
879 (False, True) -> -- [Unified: second rule]
880 notifyDied from them DiedUnknownId mRef
881 (False, False) -> -- [Unified: third rule]
882 -- TODO: this is the right sender according to the Unified semantics,
883 -- but perhaps having 'them' as the sender would make more sense
884 -- (see also: notifyDied)
885 ncSendToNode (processNodeId from) $ NCMsg
886 { ctrlMsgSender = NodeIdentifier (localNodeId node)
887 , ctrlMsgSignal = Died them DiedUnknownId
888 }
890-- [Unified: Table 11]
891ncEffectUnlink :: ProcessId -> Identifier -> NC ()
892ncEffectUnlink from them = do
893 node <- ask
894 when (isLocal node (ProcessIdentifier from)) $
895 case them of
896 ProcessIdentifier pid ->
897 postAsMessage from $ DidUnlinkProcess pid
898 NodeIdentifier nid ->
899 postAsMessage from $ DidUnlinkNode nid
900 SendPortIdentifier cid ->
901 postAsMessage from $ DidUnlinkPort cid
902 modify' $ links ^: BiMultiMap.delete them from ()
904-- [Unified: Table 11]
905ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
906ncEffectUnmonitor from ref = do
907 node <- ask
908 when (isLocal node (ProcessIdentifier from)) $
909 postAsMessage from $ DidUnmonitor ref
910 modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref
912-- [Unified: Table 12]
913ncEffectDied :: Identifier -> DiedReason -> NC ()
914ncEffectDied ident reason = do
915 node <- ask
916 traceNotifyDied node ident reason
917 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
918 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
920-- _registry :: !(Map (String,NodeId) ProcessId)
922 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
924 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
925 forM_ uss $ \(us, _) ->
926 when (localOnly <= isLocal node (ProcessIdentifier us)) $
927 notifyDied us them reason Nothing
929 forM_ (Map.toList affectedMons) $ \(them, refs) ->
930 forM_ refs $ \(us, ref) ->
931 when (localOnly <= isLocal node (ProcessIdentifier us)) $
932 notifyDied us them reason (Just ref)
934 -- Notify remote nodes that the process died so it can be removed from monitor
935 -- lists.
936 mapM_ (forwardDeath node) $
937 [ nid | ProcessIdentifier pid <- [ident]
938 , i <- Set.toList $ Set.union
939 (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedLinks)
940 (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedMons)
941 , let nid = nodeOf i
942 , nid /= localNodeId node
943 ]
945 -- Delete monitors in the local node.
946 let deleteDeads :: (Ord a, Ord v)
947 => BiMultiMap a ProcessId v -> BiMultiMap a ProcessId v
948 deleteDeads = case ident of
949 -- deleteAllBy2nd is faster than partitionWithKeyBy2nd
950 ProcessIdentifier pid -> BiMultiMap.deleteAllBy2nd pid
951 _ -> snd . BiMultiMap.partitionWithKeyBy2nd
952 (\pid _ -> ident `impliesDeathOf` ProcessIdentifier pid)
953 unaffectedLinks' = deleteDeads unaffectedLinks
954 unaffectedMons' = deleteDeads unaffectedMons
956 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
958 -- we now consider all labels for this identifier unregistered
959 let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid
960 (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)
961 mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)
962 modify' $ registeredHere ^= keepNames
964 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
965 mapM (\(pid,nidlist) ->
966 case ident `impliesDeathOf` ProcessIdentifier pid of
967 True ->
968 do forM_ nidlist $ \(nid,_) ->
969 when (not $ isLocal node (NodeIdentifier nid))
970 (forwardDeath node nid)
971 return Nothing
972 False -> return $ Just (pid,nidlist) )
973 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
974 where
975 forwardDeath node nid = ncSendToNode nid
976 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
977 , ctrlMsgSignal = Died ident reason
978 }
980-- [Unified: Table 13]
981ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
982ncEffectSpawn pid cProc ref = do
983 mProc <- unClosure cProc
984 -- If the closure does not exist, we spawn a process that throws an exception
985 -- This allows the remote node to find out what's happening
986 -- TODO:
987 let proc = case mProc of
988 Left err -> fail $ "Error: Could not resolve closure: " ++ err
989 Right p -> p
990 node <- ask
991 pid' <- liftIO $ forkProcess node proc
992 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
994-- Unified semantics does not explicitly describe how to implement 'register',
995-- but mentions it's "very similar to nsend" (Table 14)
996-- We send a response indicated if the operation is invalid
997ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
998ncEffectRegister from label atnode mPid reregistration = do
999 node <- ask
1000 currentVal <- gets (^. registeredHereFor label)
1001 isOk <-
1002 case mPid of
1003 Nothing -> -- unregister request
1004 return $ isJust currentVal
1005 Just thepid -> -- register request
1006 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
1007 return $ (isNothing currentVal /= reregistration) &&
1008 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
1009 if isLocal node (NodeIdentifier atnode)
1010 then do when isOk $
1011 do modify' $ registeredHereFor label ^= mPid
1012 updateRemote node currentVal mPid
1013 case mPid of
1014 (Just p) -> do
1015 if reregistration
1016 then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
1017 else return ()
1018 liftIO $ trace node (MxRegistered p label)
1019 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
1020 newVal <- gets (^. registeredHereFor label)
1021 ncSendToProcess from $ unsafeCreateUnencodedMessage $
1022 RegisterReply label isOk newVal
1023 else let operation =
1024 case reregistration of
1025 True -> flip decList
1026 False -> flip incList
1027 in case mPid of
1028 Nothing -> return ()
1029 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
1030 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
1031 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
1032 forward node (processNodeId newval) (Register label atnode (Just newval) False)
1033 updateRemote node Nothing (Just newval) =
1034 forward node (processNodeId newval) (Register label atnode (Just newval) False)
1035 updateRemote node (Just oldval) Nothing =
1036 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
1037 updateRemote _ _ _ = return ()
1038 maybeify f Nothing = unmaybeify $ f []
1039 maybeify f (Just x) = unmaybeify $ f x
1041 unmaybeify [] = Nothing
1042 unmaybeify x = Just x
1043 incList [] tag = [(tag,1)]
1044 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
1045 incList (x:xs) tag = x : incList xs tag
1046 decList [] _ = []
1047 decList ((atag,1):xs) tag | atag == tag = xs
1048 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
1049 decList (x:xs) tag = x:decList xs tag
1050 forward node to reg =
1051 when (not $ isLocal node (NodeIdentifier to)) $
1052 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
1053 , ctrlMsgSignal = reg
1057-- Unified semantics does not explicitly describe 'whereis'
1058ncEffectWhereIs :: ProcessId -> String -> NC ()
1059ncEffectWhereIs from label = do
1060 mPid <- gets (^. registeredHereFor label)
1061 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
1063-- [Unified: Table 14]
1064ncEffectNamedSend :: String -> Message -> NC ()
1065ncEffectNamedSend label msg = do
1066 mPid <- gets (^. registeredHereFor label)
1067 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
1068 forM_ mPid $ \to ->
1069 -- If this is a trace message we don't trace it to avoid entering a loop
1070 -- where trace messages produce more trace messages.
1071 ncSendToProcessAndTrace (label /= "trace.logger") to msg
1073-- [Issue #DP-20]
1074ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
1075ncEffectLocalSend = ncEffectLocalSendAndTrace True
1077ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
1078ncEffectLocalSendAndTrace shouldTrace node to msg =
1079 liftIO $ withLocalProc node to $ \p -> do
1080 enqueue (processQueue p) msg
1081 when shouldTrace $ trace node (MxReceived to msg)
1083-- [Issue #DP-20]
1084ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
1085ncEffectLocalPortSend from msg = do
1086 node <- ask
1087 let pid = sendPortProcessId from
1088 cid = sendPortLocalId from
1089 liftIO $ withLocalProc node pid $ \proc -> do
1090 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
1091 case mChan of
1092 -- in the unlikely event we know nothing about this channel id,
1093 -- there's little to be done - perhaps some logging/tracing though...
1094 Nothing -> return ()
1095 Just (TypedChannel chan') -> do
1096 -- If ch is Nothing, the process has given up the read end of
1097 -- the channel and we simply ignore the incoming message - this
1098 ch <- deRefWeak chan'
1099 forM_ ch $ \chan -> deliverChan msg chan
1100 where deliverChan :: forall a . Message -> TQueue a -> IO ()
1101 deliverChan (UnencodedMessage _ raw) chan' =
1102 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1103 deliverChan (EncodedMessage _ _) _ =
1104 -- this will not happen unless someone screws with Primitives.hs
1105 error "invalid local channel delivery"
1107-- [Issue #69]
1108ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
1109ncEffectKill from to reason = do
1110 node <- ask
1111 when (isLocal node (ProcessIdentifier to)) $
1112 throwException to $ ProcessKillException from reason
1114-- [Issue #69]
1115ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
1116ncEffectExit from to reason = do
1117 node <- ask
1118 when (isLocal node (ProcessIdentifier to)) $
1119 throwException to $ ProcessExitException from reason
1121-- [Issue #89]
1122ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
1123ncEffectGetInfo from pid =
1124 let lpid = processLocalId pid
1125 them = (ProcessIdentifier pid)
1126 in do
1127 node <- ask
1128 mProc <- liftIO $ withValidLocalState node
1129 $ return . (^. localProcessWithId lpid)
1130 case mProc of
1131 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1132 from (ProcessInfoNone DiedUnknownId)
1133 Just proc -> do
1134 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>
1135 gets (^. links)
1136 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)
1137 registered <- gets (^. registeredHere)
1138 size <- liftIO $ queueSize $ processQueue $ proc
1140 let reg = registeredNames registered
1141 dispatch (isLocal node (ProcessIdentifier from))
1142 from
1143 ProcessInfo {
1144 infoNode = (processNodeId pid)
1145 , infoRegisteredNames = reg
1146 , infoMessageQueueLength = size
1147 , infoMonitors = Set.toList itsMons
1148 , infoLinks = Set.toList itsLinks
1150 where dispatch :: (Serializable a)
1151 => Bool
1152 -> ProcessId
1153 -> a
1154 -> NC ()
1155 dispatch True dest pInfo = postAsMessage dest $ pInfo
1156 dispatch False dest pInfo =
1157 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1159 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1160 then (k:ks)
1161 else ks) []
1163ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1164ncEffectGetNodeStats from _nid = do
1165 node <- ask
1166 ncState <- StateT.get
1167 nodeState <- liftIO $ withValidLocalState node return
1168 let stats =
1169 NodeStats {
1170 nodeStatsNode = localNodeId node
1171 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1172 , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors
1173 , nodeStatsLinks = BiMultiMap.size $ ncState ^. links
1174 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1176 postAsMessage from stats
1178--------------------------------------------------------------------------------
1179-- Auxiliary --
1180--------------------------------------------------------------------------------
1182notifyDied :: ProcessId -- ^ Who to notify?
1183 -> Identifier -- ^ Who died?
1184 -> DiedReason -- ^ How did they die?
1185 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1186 -> NC ()
1187notifyDied dest src reason mRef = do
1188 node <- ask
1189 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1190 (True, Just ref, ProcessIdentifier pid) ->
1191 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1192 (True, Just ref, NodeIdentifier nid) ->
1193 postAsMessage dest $ NodeMonitorNotification ref nid reason
1194 (True, Just ref, SendPortIdentifier cid) ->
1195 postAsMessage dest $ PortMonitorNotification ref cid reason
1196 (True, Nothing, ProcessIdentifier pid) ->
1197 throwException dest $ ProcessLinkException pid reason
1198 (True, Nothing, NodeIdentifier pid) ->
1199 throwException dest $ NodeLinkException pid reason
1200 (True, Nothing, SendPortIdentifier pid) ->
1201 throwException dest $ PortLinkException pid reason
1202 (False, _, _) ->
1203 -- The change in sender comes from [Unified: Table 10]
1204 ncSendToNode (processNodeId dest) $ NCMsg
1205 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1206 , ctrlMsgSignal = Died src reason
1209-- | [Unified: Table 8]
1210destNid :: ProcessSignal -> Maybe NodeId
1211destNid (Link ident) = Just $ nodeOf ident
1212destNid (Unlink ident) = Just $ nodeOf ident
1213destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1214destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1215destNid (Spawn _ _) = Nothing
1216destNid (Register _ _ _ _) = Nothing
1217destNid (WhereIs _) = Nothing
1218destNid (NamedSend _ _) = Nothing
1219destNid (UnreliableSend _ _) = Nothing
1220-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1221-- then when a local process dies the monitoring/linking machinery will take
1222-- care of notifying remote nodes
1223destNid (Died _ _) = Nothing
1224destNid (Kill pid _) = Just $ processNodeId pid
1225destNid (Exit pid _) = Just $ processNodeId pid
1226destNid (GetInfo pid) = Just $ processNodeId pid
1227destNid (GetNodeStats nid) = Just nid
1228destNid (LocalSend pid _) = Just $ processNodeId pid
1229destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1230destNid (SigShutdown) = Nothing
1232-- | Check if a process is local to our own node
1233isLocal :: LocalNode -> Identifier -> Bool
1234isLocal nid ident = nodeOf ident == localNodeId nid
1236-- | Lookup a local closure
1237unClosure :: Typeable a => Closure a -> NC (Either String a)
1238unClosure closure = do
1239 rtable <- remoteTable <$> ask
1240 return (Static.unclosure rtable closure)
1242-- | Check if an identifier refers to a valid local object
1243isValidLocalIdentifier :: Identifier -> NC Bool
1244isValidLocalIdentifier ident = do
1245 node <- ask
1246 liftIO . withValidLocalState node $ \nSt ->
1247 case ident of
1248 NodeIdentifier nid ->
1249 return $ nid == localNodeId node
1250 ProcessIdentifier pid -> do
1251 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1252 return $ isJust mProc
1253 SendPortIdentifier cid -> do
1254 let pid = sendPortProcessId cid
1255 mProc = nSt ^. localProcessWithId (processLocalId pid)
1256 case mProc of
1257 Nothing -> return False
1258 Just proc -> withMVar (processState proc) $ \pSt -> do
1259 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1260 return $ isJust mCh
1262--------------------------------------------------------------------------------
1263-- Messages to local processes --
1264--------------------------------------------------------------------------------
1266postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1267postAsMessage pid = postMessage pid . createUnencodedMessage
1269postMessage :: ProcessId -> Message -> NC ()
1270postMessage pid msg = do
1271 node <- ask
1272 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1274throwException :: Exception e => ProcessId -> e -> NC ()
1275throwException pid e = do
1276 node <- ask
1277 -- throwTo blocks until the exception is received by the target thread.
1278 -- We cannot easily make it happen asynchronpusly because then 'unlink'
1279 -- semantics would break.
1280 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1282withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1283withLocalProc node pid p =
1284 -- By [Unified: table 6, rule missing_process] messages to dead processes
1285 -- can silently be dropped
1286 let lpid = processLocalId pid in do
1287 join $ withValidLocalState node $ \vst ->
1288 return $ forM_ (vst ^. localProcessWithId lpid) p
1290--------------------------------------------------------------------------------
1291-- Accessors --
1292--------------------------------------------------------------------------------
1294links :: Accessor NCState (BiMultiMap Identifier ProcessId ())
1295links = accessor _links (\ls st -> st { _links = ls })
1297monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)
1298monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1300registeredHere :: Accessor NCState (Map String ProcessId)
1301registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1303registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1304registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1306registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1307registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1309registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1310registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1312-- | @splitNotif ident@ splits a notifications map into those
1313-- notifications that should trigger when 'ident' fails and those links that
1314-- should not.
1316-- There is a hierarchy between identifiers: failure of a node implies failure
1317-- of all processes on that node, and failure of a process implies failure of
1318-- all typed channels to that process. In other words, if 'ident' refers to a
1319-- node, then the /should trigger/ set will include
1321-- * the notifications for the node specifically
1322-- * the notifications for processes on that node, and
1323-- * the notifications for typed channels to processes on that node.
1325-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1326-- include
1328-- * the notifications for that process specifically and
1329-- * the notifications for typed channels to that process.
1331-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1332splitNotif :: (Ord a, Ord v)
1333 => Identifier
1334 -> BiMultiMap Identifier a v
1335 -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)
1336splitNotif ident =
1337 BiMultiMap.partitionWithKeyBy1st (\k !_v -> ident `impliesDeathOf` k)
1339--------------------------------------------------------------------------------
1340-- Auxiliary --
1341--------------------------------------------------------------------------------
1343-- | Modify and evaluate the state
1344modify' :: MonadState s m => (s -> s) -> m ()
1345modify' f = StateT.get >>= \s -> StateT.put $! f s