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