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