git.iko.soy/distributed-process
on eca0b26 help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1271 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 (unaffectedLinks', unaffectedMons') =
883 case ident of
884 ProcessIdentifier pid ->
885 ( BiMultiMap.deleteAllBy2nd pid unaffectedLinks
886 , BiMultiMap.deleteAllBy2nd pid unaffectedMons
887 )
888 _ -> (unaffectedLinks, unaffectedMons)
890 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
892 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
894 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
895 mapM (\(pid,nidlist) ->
896 case ident `impliesDeathOf` ProcessIdentifier pid of
897 True ->
898 do forM_ nidlist $ \(nid,_) ->
899 when (not $ isLocal node (NodeIdentifier nid))
900 (forwardNameDeath node nid)
901 return Nothing
902 False -> return $ Just (pid,nidlist) )
903 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
904 where
905 forwardNameDeath node nid = ncSendToNode nid
906 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
907 , ctrlMsgSignal = Died ident reason
908 }
910-- [Unified: Table 13]
911ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
912ncEffectSpawn pid cProc ref = do
913 mProc <- unClosure cProc
914 -- If the closure does not exist, we spawn a process that throws an exception
915 -- This allows the remote node to find out what's happening
916 -- TODO:
917 let proc = case mProc of
918 Left err -> fail $ "Error: Could not resolve closure: " ++ err
919 Right p -> p
920 node <- ask
921 pid' <- liftIO $ forkProcess node proc
922 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
924-- Unified semantics does not explicitly describe how to implement 'register',
925-- but mentions it's "very similar to nsend" (Table 14)
926-- We send a response indicated if the operation is invalid
927ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
928ncEffectRegister from label atnode mPid reregistration = do
929 node <- ask
930 currentVal <- gets (^. registeredHereFor label)
931 isOk <-
932 case mPid of
933 Nothing -> -- unregister request
934 return $ isJust currentVal
935 Just thepid -> -- register request
936 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
937 return $ (isNothing currentVal /= reregistration) &&
938 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
939 if isLocal node (NodeIdentifier atnode)
940 then do when isOk $
941 do modify' $ registeredHereFor label ^= mPid
942 updateRemote node currentVal mPid
943 case mPid of
944 (Just p) -> liftIO $ trace node (MxRegistered p label)
945 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
946 newVal <- gets (^. registeredHereFor label)
947 ncSendToProcess from $ unsafeCreateUnencodedMessage $
948 RegisterReply label isOk newVal
949 else let operation =
950 case reregistration of
951 True -> flip decList
952 False -> flip incList
953 in case mPid of
954 Nothing -> return ()
955 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
956 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
957 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
958 forward node (processNodeId newval) (Register label atnode (Just newval) False)
959 updateRemote node Nothing (Just newval) =
960 forward node (processNodeId newval) (Register label atnode (Just newval) False)
961 updateRemote node (Just oldval) Nothing =
962 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
963 updateRemote _ _ _ = return ()
964 maybeify f Nothing = unmaybeify $ f []
965 maybeify f (Just x) = unmaybeify $ f x
967 unmaybeify [] = Nothing
968 unmaybeify x = Just x
969 incList [] tag = [(tag,1)]
970 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
971 incList (x:xs) tag = x : incList xs tag
972 decList [] _ = []
973 decList ((atag,1):xs) tag | atag == tag = xs
974 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
975 decList (x:xs) tag = x:decList xs tag
976 forward node to reg =
977 when (not $ isLocal node (NodeIdentifier to)) $
978 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
979 , ctrlMsgSignal = reg
980 }
983-- Unified semantics does not explicitly describe 'whereis'
984ncEffectWhereIs :: ProcessId -> String -> NC ()
985ncEffectWhereIs from label = do
986 mPid <- gets (^. registeredHereFor label)
987 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
989-- [Unified: Table 14]
990ncEffectNamedSend :: String -> Message -> NC ()
991ncEffectNamedSend label msg = do
992 mPid <- gets (^. registeredHereFor label)
993 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
994 forM_ mPid $ \to ->
995 -- If this is a trace message we don't trace it to avoid entering a loop
996 -- where trace messages produce more trace messages.
997 ncSendToProcessAndTrace (label /= "trace.logger") to msg
999-- [Issue #DP-20]
1000ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
1001ncEffectLocalSend = ncEffectLocalSendAndTrace True
1003ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
1004ncEffectLocalSendAndTrace shouldTrace node to msg =
1005 liftIO $ withLocalProc node to $ \p -> do
1006 enqueue (processQueue p) msg
1007 when shouldTrace $ trace node (MxReceived to msg)
1009-- [Issue #DP-20]
1010ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
1011ncEffectLocalPortSend from msg = do
1012 node <- ask
1013 let pid = sendPortProcessId from
1014 cid = sendPortLocalId from
1015 liftIO $ withLocalProc node pid $ \proc -> do
1016 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
1017 case mChan of
1018 -- in the unlikely event we know nothing about this channel id,
1019 -- there's little to be done - perhaps some logging/tracing though...
1020 Nothing -> return ()
1021 Just (TypedChannel chan') -> do
1022 -- If ch is Nothing, the process has given up the read end of
1023 -- the channel and we simply ignore the incoming message - this
1024 ch <- deRefWeak chan'
1025 forM_ ch $ \chan -> deliverChan msg chan
1026 where deliverChan :: forall a . Message -> TQueue a -> IO ()
1027 deliverChan (UnencodedMessage _ raw) chan' =
1028 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1029 deliverChan (EncodedMessage _ _) _ =
1030 -- this will not happen unless someone screws with Primitives.hs
1031 error "invalid local channel delivery"
1033-- [Issue #69]
1034ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
1035ncEffectKill from to reason = do
1036 node <- ask
1037 when (isLocal node (ProcessIdentifier to)) $
1038 throwException to $ ProcessKillException from reason
1040-- [Issue #69]
1041ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
1042ncEffectExit from to reason = do
1043 node <- ask
1044 when (isLocal node (ProcessIdentifier to)) $
1045 throwException to $ ProcessExitException from reason
1047-- [Issue #89]
1048ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
1049ncEffectGetInfo from pid =
1050 let lpid = processLocalId pid
1051 them = (ProcessIdentifier pid)
1052 in do
1053 node <- ask
1054 mProc <- liftIO $ withValidLocalState node
1055 $ return . (^. localProcessWithId lpid)
1056 case mProc of
1057 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1058 from (ProcessInfoNone DiedUnknownId)
1059 Just proc -> do
1060 itsLinks <- Set.map fst . BiMultiMap.lookup them <$>
1061 gets (^. links)
1062 itsMons <- BiMultiMap.lookup them <$> gets (^. monitors)
1063 registered <- gets (^. registeredHere)
1064 size <- liftIO $ queueSize $ processQueue $ proc
1066 let reg = registeredNames registered
1067 dispatch (isLocal node (ProcessIdentifier from))
1068 from
1069 ProcessInfo {
1070 infoNode = (processNodeId pid)
1071 , infoRegisteredNames = reg
1072 , infoMessageQueueLength = size
1073 , infoMonitors = Set.toList itsMons
1074 , infoLinks = Set.toList itsLinks
1076 where dispatch :: (Serializable a)
1077 => Bool
1078 -> ProcessId
1079 -> a
1080 -> NC ()
1081 dispatch True dest pInfo = postAsMessage dest $ pInfo
1082 dispatch False dest pInfo =
1083 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1085 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1086 then (k:ks)
1087 else ks) []
1089ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1090ncEffectGetNodeStats from _nid = do
1091 node <- ask
1092 ncState <- StateT.get
1093 nodeState <- liftIO $ withValidLocalState node return
1094 let stats =
1095 NodeStats {
1096 nodeStatsNode = localNodeId node
1097 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1098 , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors
1099 , nodeStatsLinks = BiMultiMap.size $ ncState ^. links
1100 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1102 postAsMessage from stats
1104--------------------------------------------------------------------------------
1105-- Auxiliary --
1106--------------------------------------------------------------------------------
1108notifyDied :: ProcessId -- ^ Who to notify?
1109 -> Identifier -- ^ Who died?
1110 -> DiedReason -- ^ How did they die?
1111 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1112 -> NC ()
1113notifyDied dest src reason mRef = do
1114 node <- ask
1115 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1116 (True, Just ref, ProcessIdentifier pid) ->
1117 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1118 (True, Just ref, NodeIdentifier nid) ->
1119 postAsMessage dest $ NodeMonitorNotification ref nid reason
1120 (True, Just ref, SendPortIdentifier cid) ->
1121 postAsMessage dest $ PortMonitorNotification ref cid reason
1122 (True, Nothing, ProcessIdentifier pid) ->
1123 throwException dest $ ProcessLinkException pid reason
1124 (True, Nothing, NodeIdentifier pid) ->
1125 throwException dest $ NodeLinkException pid reason
1126 (True, Nothing, SendPortIdentifier pid) ->
1127 throwException dest $ PortLinkException pid reason
1128 (False, _, _) ->
1129 -- The change in sender comes from [Unified: Table 10]
1130 ncSendToNode (processNodeId dest) $ NCMsg
1131 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1132 , ctrlMsgSignal = Died src reason
1135-- | [Unified: Table 8]
1136destNid :: ProcessSignal -> Maybe NodeId
1137destNid (Link ident) = Just $ nodeOf ident
1138destNid (Unlink ident) = Just $ nodeOf ident
1139destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1140destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1141destNid (Spawn _ _) = Nothing
1142destNid (Register _ _ _ _) = Nothing
1143destNid (WhereIs _) = Nothing
1144destNid (NamedSend _ _) = Nothing
1145destNid (UnreliableSend _ _) = Nothing
1146-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1147-- then when a local process dies the monitoring/linking machinery will take
1148-- care of notifying remote nodes
1149destNid (Died _ _) = Nothing
1150destNid (Kill pid _) = Just $ processNodeId pid
1151destNid (Exit pid _) = Just $ processNodeId pid
1152destNid (GetInfo pid) = Just $ processNodeId pid
1153destNid (GetNodeStats nid) = Just nid
1154destNid (LocalSend pid _) = Just $ processNodeId pid
1155destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1156destNid (SigShutdown) = Nothing
1158-- | Check if a process is local to our own node
1159isLocal :: LocalNode -> Identifier -> Bool
1160isLocal nid ident = nodeOf ident == localNodeId nid
1162-- | Lookup a local closure
1163unClosure :: Typeable a => Closure a -> NC (Either String a)
1164unClosure closure = do
1165 rtable <- remoteTable <$> ask
1166 return (Static.unclosure rtable closure)
1168-- | Check if an identifier refers to a valid local object
1169isValidLocalIdentifier :: Identifier -> NC Bool
1170isValidLocalIdentifier ident = do
1171 node <- ask
1172 liftIO . withValidLocalState node $ \nSt ->
1173 case ident of
1174 NodeIdentifier nid ->
1175 return $ nid == localNodeId node
1176 ProcessIdentifier pid -> do
1177 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1178 return $ isJust mProc
1179 SendPortIdentifier cid -> do
1180 let pid = sendPortProcessId cid
1181 mProc = nSt ^. localProcessWithId (processLocalId pid)
1182 case mProc of
1183 Nothing -> return False
1184 Just proc -> withMVar (processState proc) $ \pSt -> do
1185 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1186 return $ isJust mCh
1188--------------------------------------------------------------------------------
1189-- Messages to local processes --
1190--------------------------------------------------------------------------------
1192postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1193postAsMessage pid = postMessage pid . createUnencodedMessage
1195postMessage :: ProcessId -> Message -> NC ()
1196postMessage pid msg = do
1197 node <- ask
1198 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1200throwException :: Exception e => ProcessId -> e -> NC ()
1201throwException pid e = do
1202 node <- ask
1203 -- throwTo blocks until the exception is received by the target thread.
1204 -- We cannot easily make it happen asynchronpusly because then 'unlink'
1205 -- semantics would break.
1206 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1208withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1209withLocalProc node pid p =
1210 -- By [Unified: table 6, rule missing_process] messages to dead processes
1211 -- can silently be dropped
1212 let lpid = processLocalId pid in do
1213 join $ withValidLocalState node $ \vst ->
1214 return $ forM_ (vst ^. localProcessWithId lpid) p
1216--------------------------------------------------------------------------------
1217-- Accessors --
1218--------------------------------------------------------------------------------
1220links :: Accessor NCState (BiMultiMap Identifier ProcessId ())
1221links = accessor _links (\ls st -> st { _links = ls })
1223monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)
1224monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1226registeredHere :: Accessor NCState (Map String ProcessId)
1227registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1229registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1230registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1232registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1233registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1235registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1236registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1238-- | @splitNotif ident@ splits a notifications map into those
1239-- notifications that should trigger when 'ident' fails and those links that
1240-- should not.
1242-- There is a hierarchy between identifiers: failure of a node implies failure
1243-- of all processes on that node, and failure of a process implies failure of
1244-- all typed channels to that process. In other words, if 'ident' refers to a
1245-- node, then the /should trigger/ set will include
1247-- * the notifications for the node specifically
1248-- * the notifications for processes on that node, and
1249-- * the notifications for typed channels to processes on that node.
1251-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1252-- include
1254-- * the notifications for that process specifically and
1255-- * the notifications for typed channels to that process.
1257-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1258splitNotif :: (Ord a, Ord v)
1259 => Identifier
1260 -> BiMultiMap Identifier a v
1261 -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)
1262splitNotif ident =
1263 BiMultiMap.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1265--------------------------------------------------------------------------------
1266-- Auxiliary --
1267--------------------------------------------------------------------------------
1269-- | Modify and evaluate the state
1270modify' :: MonadState s m => (s -> s) -> m ()
1271modify' f = StateT.get >>= \s -> StateT.put $! f s