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