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