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