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