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