git.iko.soy/distributed-process
on 0f0a5ec 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 Data.Set (Set)
41import qualified Data.Set as Set
42 ( empty
43 , insert
44 , delete
45 , map
46 , member
47 , toList
48 , union
49 )
50import Data.Foldable (forM_)
51import Data.List (foldl')
52import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
53import Data.Typeable (Typeable)
54import Control.Category ((>>>))
55import Control.Applicative
56import Control.Monad (void, when, join)
57import Control.Monad.IO.Class (MonadIO, liftIO)
58import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
59import qualified Control.Monad.State.Strict as StateT (get, put)
60import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
61import Control.Exception
62 ( throwIO
63 , SomeException
64 , Exception
65 , throwTo
66 , uninterruptibleMask_
67 , getMaskingState
68 , MaskingState(..)
69 )
70import qualified Control.Exception as Exception
71 ( Handler(..)
72 , catch
73 , catches
74 , finally
75 )
76import Control.Concurrent (forkIO, killThread)
77import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)
78import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap
79import Control.Distributed.Process.Internal.StrictMVar
80 ( newMVar
81 , withMVar
82 , modifyMVarMasked
83 , modifyMVar
84 , newEmptyMVar
85 , putMVar
86 , takeMVar
87 )
88import Control.Concurrent.Chan (newChan, writeChan, readChan)
89import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
90import Control.Concurrent.STM
91 ( atomically
92 )
93import Control.Distributed.Process.Internal.CQueue
94 ( CQueue
95 , enqueue
96 , newCQueue
97 , mkWeakCQueue
98 , queueSize
99 )
100import qualified Network.Transport as NT
101 ( Transport
102 , EndPoint
103 , newEndPoint
104 , receive
105 , Event(..)
106 , EventErrorCode(..)
107 , TransportError(..)
108 , address
109 , closeEndPoint
110 , Connection
111 , ConnectionId
112 , close
113 , EndPointAddress
114 , Reliability(ReliableOrdered)
115 )
116import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
117import System.Random (randomIO)
118import Control.Distributed.Static (RemoteTable, Closure)
119import qualified Control.Distributed.Static as Static
120 ( unclosure
121 , initRemoteTable
122 )
123import Control.Distributed.Process.Internal.Types
124 ( NodeId(..)
125 , LocalProcessId(..)
126 , ProcessId(..)
127 , LocalNode(..)
128 , MxEventBus(..)
129 , LocalNodeState(..)
130 , ValidLocalNodeState(..)
131 , withValidLocalState
132 , modifyValidLocalState
133 , LocalProcess(..)
134 , LocalProcessState(..)
135 , Process(..)
136 , DiedReason(..)
137 , NCMsg(..)
138 , ProcessSignal(..)
139 , localPidCounter
140 , localPidUnique
141 , localProcessWithId
142 , localProcesses
143 , localConnections
144 , forever'
145 , MonitorRef(..)
146 , NodeClosedException(..)
147 , ProcessMonitorNotification(..)
148 , NodeMonitorNotification(..)
149 , PortMonitorNotification(..)
150 , ProcessExitException(..)
151 , ProcessLinkException(..)
152 , NodeLinkException(..)
153 , PortLinkException(..)
154 , DidUnmonitor(..)
155 , DidUnlinkProcess(..)
156 , DidUnlinkNode(..)
157 , DidUnlinkPort(..)
158 , SpawnRef
159 , DidSpawn(..)
160 , Message(..)
161 , TypedChannel(..)
162 , Identifier(..)
163 , nodeOf
164 , ProcessInfo(..)
165 , ProcessInfoNone(..)
166 , NodeStats(..)
167 , SendPortId(..)
168 , typedChannelWithId
169 , RegisterReply(..)
170 , WhereIsReply(..)
171 , payloadToMessage
172 , createUnencodedMessage
173 , unsafeCreateUnencodedMessage
174 , runLocalProcess
175 , firstNonReservedProcessId
176 , ImplicitReconnect(WithImplicitReconnect)
177 )
178import Control.Distributed.Process.Management.Internal.Agent
179 ( mxAgentController
180 )
181import qualified Control.Distributed.Process.Management.Internal.Table as Table
182 ( mxTableCoordinator
183 , startTableCoordinator
184 )
185import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
186 ( remoteTable
187 )
188import Control.Distributed.Process.Management.Internal.Trace.Tracer
189 ( defaultTracer
190 )
191import Control.Distributed.Process.Management.Internal.Trace.Types
192 ( TraceArg(..)
193 , traceEvent
194 , traceLogFmt
195 , enableTrace
196 )
197import Control.Distributed.Process.Management.Internal.Types
198 ( MxEvent(..)
199 )
200import Control.Distributed.Process.Serializable (Serializable)
201import Control.Distributed.Process.Internal.Messaging
202 ( sendBinary
203 , closeImplicitReconnections
204 , impliesDeathOf
205 )
206import Control.Distributed.Process.Internal.Primitives
207 ( register
208 , receiveWait
209 , match
210 , sendChan
211 , unwrapMessage
212 )
213import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
214import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
215import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
216import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
217 ( mapMaybe
218 , mapDefault
219 )
220import Control.Monad.Catch (try)
221import GHC.IO (IO(..), unsafeUnmask)
222import GHC.Base ( maskAsyncExceptions# )
224import Unsafe.Coerce
225import Prelude
227-- Remove these definitions when the fix for
228-- https://ghc.haskell.org/trac/ghc/ticket/10149
229-- is included in all supported compilers:
230block :: IO a -> IO a
231block (IO io) = IO $ maskAsyncExceptions# io
232unblock :: IO a -> IO a
233unblock = unsafeUnmask
235--------------------------------------------------------------------------------
236-- Initialization --
237--------------------------------------------------------------------------------
239initRemoteTable :: RemoteTable
240initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
242-- | Initialize a new local node.
243newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
244newLocalNode transport rtable = do
245 mEndPoint <- NT.newEndPoint transport
246 case mEndPoint of
247 Left ex -> throwIO ex
248 Right endPoint -> do
249 localNode <- createBareLocalNode endPoint rtable
250 startServiceProcesses localNode
251 return localNode
253-- | Create a new local node (without any service processes running)
254createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
255createBareLocalNode endPoint rtable = do
256 unq <- randomIO
257 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
258 { _localProcesses = Map.empty
259 , _localPidCounter = firstNonReservedProcessId
260 , _localPidUnique = unq
261 , _localConnections = Map.empty
262 }
263 ctrlChan <- newChan
264 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
265 , localEndPoint = endPoint
266 , localState = state
267 , localCtrlChan = ctrlChan
268 , localEventBus = MxEventBusInitialising
269 , remoteTable = rtable
270 }
271 tracedNode <- startMxAgent node
273 -- Once the NC terminates, the endpoint isn't much use,
274 void $ forkIO $ Exception.finally (runNodeController tracedNode)
275 (NT.closeEndPoint (localEndPoint node))
277 -- whilst a closed/failing endpoint will terminate the NC
278 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
279 (stopNC node)
281 return tracedNode
282 where
283 stopNC node =
284 writeChan (localCtrlChan node) NCMsg
285 { ctrlMsgSender = NodeIdentifier (localNodeId node)
286 , ctrlMsgSignal = SigShutdown
287 }
289startMxAgent :: LocalNode -> IO LocalNode
290startMxAgent node = do
291 -- see note [tracer/forkProcess races]
292 let fork = forkProcess node
293 mv <- MVar.newEmptyMVar
294 pid <- fork $ mxAgentController fork mv
295 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
296 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
298startDefaultTracer :: LocalNode -> IO ()
299startDefaultTracer node' = do
300 let t = localEventBus node'
301 case t of
302 MxEventBus _ (Tracer pid _) _ _ -> do
303 runProcess node' $ register "trace.controller" pid
304 pid' <- forkProcess node' defaultTracer
305 enableTrace (localEventBus node') pid'
306 runProcess node' $ register "tracer.initial" pid'
307 _ -> return ()
309-- TODO: we need a better mechanism for defining and registering services
311-- | Start and register the service processes on a node
312startServiceProcesses :: LocalNode -> IO ()
313startServiceProcesses node = do
314 -- tracing /spawns/ relies on the tracer being enabled, but we start
315 -- the default tracer first, even though it might @nsend@ to the logger
316 -- before /that/ process has started - this is a totally harmless race
317 -- however, so we deliberably ignore it
318 startDefaultTracer node
319 tableCoordinatorPid <- fork $ Table.startTableCoordinator fork
320 runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid
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 fork = forkProcess node
331 loop = do
332 receiveWait
333 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
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