git.iko.soy/distributed-process
on ced056e help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1299 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 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 (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 -> atomically $
555 -- We make sure the message is fully decoded when it is enqueued
556 writeTQueue chan $! decode (BSL.fromChunks payload)
557 go st
558 Just (_, ToNode) -> do
559 let ctrlMsg = decode . BSL.fromChunks $ payload
560 writeChan ctrlChan $! ctrlMsg
561 go st
562 Just (src, Uninit) ->
563 case decode (BSL.fromChunks payload) of
564 ProcessIdentifier pid -> do
565 let lpid = processLocalId pid
566 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
567 case mProc of
568 Just proc ->
569 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
570 Nothing ->
571 -- incoming attempt to connect to unknown process - might
572 -- be dead already
573 go (incomingAt cid ^= Nothing $ st)
574 SendPortIdentifier chId -> do
575 let lcid = sendPortLocalId chId
576 lpid = processLocalId (sendPortProcessId chId)
577 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
578 case mProc of
579 Just proc -> do
580 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
581 case mChannel of
582 Just channel ->
583 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
584 Nothing ->
585 invalidRequest cid st $
586 "incoming attempt to connect to unknown channel of"
587 ++ " process " ++ show (sendPortProcessId chId)
588 Nothing ->
589 -- incoming attempt to connect to channel of unknown
590 -- process - might be dead already
591 go (incomingAt cid ^= Nothing $ st)
592 NodeIdentifier nid ->
593 if nid == localNodeId node
594 then go (incomingAt cid ^= Just (src, ToNode) $ st)
595 else invalidRequest cid st $
596 "incoming attempt to connect to a different node -"
597 ++ " I'm " ++ show (localNodeId node)
598 ++ " but the remote peer wants to connect to "
599 ++ show nid
600 Nothing ->
601 invalidRequest cid st
602 "message received from an unknown connection"
603 NT.ConnectionClosed cid ->
604 case st ^. incomingAt cid of
605 Nothing ->
606 invalidRequest cid st "closed unknown connection"
607 Just (src, _) -> do
608 trace node (MxDisconnected cid src)
609 go ( (incomingAt cid ^= Nothing)
610 . (incomingFrom src ^: Set.delete cid)
611 $ st
612 )
613 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
614 -- [Unified table 9, rule node_disconnect]
615 let nid = NodeIdentifier $ NodeId theirAddr
616 writeChan ctrlChan NCMsg
617 { ctrlMsgSender = nid
618 , ctrlMsgSignal = Died nid DiedDisconnect
619 }
620 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
621 closeImplicitReconnections node nid
622 go ( (incomingFrom theirAddr ^= Set.empty)
623 . (incoming ^: Map.filterWithKey (const . notLost))
624 $ st
625 )
626 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
627 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
628 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
629 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
630 NT.EndPointClosed ->
631 return ()
632 NT.ReceivedMulticast _ _ ->
633 -- If we received a multicast message, something went horribly wrong
634 -- and we just give up
635 fail "Cloud Haskell fatal error: received unexpected multicast"
637 invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
638 invalidRequest cid st msg = do
639 -- TODO: We should treat this as a fatal error on the part of the remote
640 -- node. That is, we should report the remote node as having died, and we
641 -- should close incoming connections (this requires a Transport layer
642 -- extension).
643 traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"
644 ++ " (" ++ msg ++ "): "
645 , (Trace cid)
646 ]
647 go ( incomingAt cid ^= Nothing
648 $ st
649 )
651 endpoint = localEndPoint node
652 ctrlChan = localCtrlChan node
654--------------------------------------------------------------------------------
655-- Top-level access to the node controller --
656--------------------------------------------------------------------------------
658runNodeController :: LocalNode -> IO ()
659runNodeController node =
660 runReaderT (evalStateT (unNC nodeController) initNCState) node
661 `Exception.catch` \(NodeClosedException _) -> return ()
663--------------------------------------------------------------------------------
664-- Internal data types --
665--------------------------------------------------------------------------------
667data NCState = NCState
668 { -- Mapping from remote processes to linked local processes
669 _links :: !(BiMultiMap Identifier ProcessId ())
670 -- Mapping from remote processes to monitoring local processes
671 , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef)
672 -- Process registry: names and where they live, mapped to the PIDs
673 , _registeredHere :: !(Map String ProcessId)
674 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
675 }
677newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
678 deriving ( Applicative
679 , Functor
680 , Monad
681 , MonadIO
682 , MonadState NCState
683 , MonadReader LocalNode
684 )
686initNCState :: NCState
687initNCState = NCState { _links = BiMultiMap.empty
688 , _monitors = BiMultiMap.empty
689 , _registeredHere = Map.empty
690 , _registeredOnNodes = Map.empty
691 }
693-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
694-- type is deliberately not exported so it cannot be caught explicitly.
695data ProcessKillException =
696 ProcessKillException !ProcessId !String
697 deriving (Typeable)
699instance Exception ProcessKillException
700instance Show ProcessKillException where
701 show (ProcessKillException pid reason) =
702 "killed-by=" ++ show pid ++ ",reason=" ++ reason
704ncSendToProcess :: ProcessId -> Message -> NC ()
705ncSendToProcess = ncSendToProcessAndTrace True
707ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
708ncSendToProcessAndTrace shouldTrace pid msg = do
709 node <- ask
710 if processNodeId pid == localNodeId node
711 then ncEffectLocalSendAndTrace shouldTrace node pid msg
712 else liftIO $ sendBinary node
713 (NodeIdentifier $ localNodeId node)
714 (NodeIdentifier $ processNodeId pid)
715 WithImplicitReconnect
716 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
717 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
718 }
720ncSendToNode :: NodeId -> NCMsg -> NC ()
721ncSendToNode to msg = do
722 node <- ask
723 liftIO $ if to == localNodeId node
724 then writeChan (localCtrlChan node) $! msg
725 else sendBinary node
726 (NodeIdentifier $ localNodeId node)
727 (NodeIdentifier to)
728 WithImplicitReconnect
729 msg
731--------------------------------------------------------------------------------
732-- Tracing/Debugging --
733--------------------------------------------------------------------------------
735-- [Issue #104 / DP-13]
737traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
738traceNotifyDied node ident reason =
739 -- TODO: sendPortDied notifications
740 liftIO $ withLocalTracer node $ \t ->
741 case ident of
742 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
743 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
744 _ -> return ()
746traceEventFmtIO :: LocalNode
747 -> String
748 -> [TraceArg]
749 -> IO ()
750traceEventFmtIO node fmt args =
751 withLocalTracer node $ \t -> traceLogFmt t fmt args
753trace :: LocalNode -> MxEvent -> IO ()
754trace node ev = withLocalTracer node $ \t -> traceEvent t ev
756withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
757withLocalTracer node act = act (localEventBus node)
759--------------------------------------------------------------------------------
760-- Core functionality --
761--------------------------------------------------------------------------------
763-- [Unified: Table 7]
764nodeController :: NC ()
765nodeController = do
766 node <- ask
767 forever' $ do
768 msg <- liftIO $ readChan (localCtrlChan node)
770 -- [Unified: Table 7, rule nc_forward]
771 case destNid (ctrlMsgSignal msg) of
772 Just nid' | nid' /= localNodeId node ->
773 ncSendToNode nid' msg
774 _ ->
775 return ()
777 case msg of
778 NCMsg (ProcessIdentifier from) (Link them) ->
779 ncEffectMonitor from them Nothing
780 NCMsg (ProcessIdentifier from) (Monitor ref) ->
781 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
782 NCMsg (ProcessIdentifier from) (Unlink them) ->
783 ncEffectUnlink from them
784 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
785 ncEffectUnmonitor from ref
786 NCMsg _from (Died ident reason) ->
787 ncEffectDied ident reason
788 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
789 ncEffectSpawn from proc ref
790 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
791 ncEffectRegister from label atnode pid force
792 NCMsg (ProcessIdentifier from) (WhereIs label) ->
793 ncEffectWhereIs from label
794 NCMsg _ (NamedSend label msg') ->
795 ncEffectNamedSend label msg'
796 NCMsg _ (UnreliableSend lpid msg') ->
797 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
798 NCMsg _ (LocalSend to msg') ->
799 ncEffectLocalSend node to msg'
800 NCMsg _ (LocalPortSend to msg') ->
801 ncEffectLocalPortSend to msg'
802 NCMsg (ProcessIdentifier from) (Kill to reason) ->
803 ncEffectKill from to reason
804 NCMsg (ProcessIdentifier from) (Exit to reason) ->
805 ncEffectExit from to reason
806 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
807 ncEffectGetInfo from pid
808 NCMsg _ SigShutdown ->
809 liftIO $ do
810 NT.closeEndPoint (localEndPoint node)
811 `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
812 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
813 ncEffectGetNodeStats from nid
814 unexpected ->
815 error $ "nodeController: unexpected message " ++ show unexpected
817-- [Unified: Table 10]
818ncEffectMonitor :: ProcessId -- ^ Who's watching?
819 -> Identifier -- ^ Who's being watched?
820 -> Maybe MonitorRef -- ^ 'Nothing' to link
821 -> NC ()
822ncEffectMonitor from them mRef = do
823 node <- ask
824 shouldLink <-
825 if not (isLocal node them)
826 then return True
827 else isValidLocalIdentifier them
828 case (shouldLink, isLocal node (ProcessIdentifier from)) of
829 (True, _) -> -- [Unified: first rule]
830 case mRef of
831 Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref
832 Nothing -> modify' $ links ^: BiMultiMap.insert them from ()
833 (False, True) -> -- [Unified: second rule]
834 notifyDied from them DiedUnknownId mRef
835 (False, False) -> -- [Unified: third rule]
836 -- TODO: this is the right sender according to the Unified semantics,
837 -- but perhaps having 'them' as the sender would make more sense
838 -- (see also: notifyDied)
839 ncSendToNode (processNodeId from) $ NCMsg
840 { ctrlMsgSender = NodeIdentifier (localNodeId node)
841 , ctrlMsgSignal = Died them DiedUnknownId
842 }
844-- [Unified: Table 11]
845ncEffectUnlink :: ProcessId -> Identifier -> NC ()
846ncEffectUnlink from them = do
847 node <- ask
848 when (isLocal node (ProcessIdentifier from)) $
849 case them of
850 ProcessIdentifier pid ->
851 postAsMessage from $ DidUnlinkProcess pid
852 NodeIdentifier nid ->
853 postAsMessage from $ DidUnlinkNode nid
854 SendPortIdentifier cid ->
855 postAsMessage from $ DidUnlinkPort cid
856 modify' $ links ^: BiMultiMap.delete them from ()
858-- [Unified: Table 11]
859ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
860ncEffectUnmonitor from ref = do
861 node <- ask
862 when (isLocal node (ProcessIdentifier from)) $
863 postAsMessage from $ DidUnmonitor ref
864 modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref
866-- [Unified: Table 12]
867ncEffectDied :: Identifier -> DiedReason -> NC ()
868ncEffectDied ident reason = do
869 node <- ask
870 traceNotifyDied node ident reason
871 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
872 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
874-- _registry :: !(Map (String,NodeId) ProcessId)
876 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
878 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
879 forM_ uss $ \(us, _) ->
880 when (localOnly <= isLocal node (ProcessIdentifier us)) $
881 notifyDied us them reason Nothing
883 forM_ (Map.toList affectedMons) $ \(them, refs) ->
884 forM_ refs $ \(us, ref) ->
885 when (localOnly <= isLocal node (ProcessIdentifier us)) $
886 notifyDied us them reason (Just ref)
888 -- Notify remote nodes that the process died so it can be removed from monitor
889 -- lists.
890 mapM_ (forwardDeath node) $
891 [ nid | ProcessIdentifier pid <- [ident]
892 , i <- Set.toList $ Set.union
893 (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedLinks)
894 (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedMons)
895 , let nid = nodeOf i
896 , nid /= localNodeId node
897 ]
899 -- Delete monitors in the local node.
900 let deleteDeads :: (Ord a, Ord v)
901 => BiMultiMap a ProcessId v -> BiMultiMap a ProcessId v
902 deleteDeads = case ident of
903 -- deleteAllBy2nd is faster than partitionWithKeyBy2nd
904 ProcessIdentifier pid -> BiMultiMap.deleteAllBy2nd pid
905 _ -> snd . BiMultiMap.partitionWithKeyBy2nd
906 (\pid _ -> ident `impliesDeathOf` ProcessIdentifier pid)
907 unaffectedLinks' = deleteDeads unaffectedLinks
908 unaffectedMons' = deleteDeads unaffectedMons
910 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
912 -- we now consider all labels for this identifier unregistered
913 let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid
914 (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)
915 mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)
916 modify' $ registeredHere ^= keepNames
918 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
919 mapM (\(pid,nidlist) ->
920 case ident `impliesDeathOf` ProcessIdentifier pid of
921 True ->
922 do forM_ nidlist $ \(nid,_) ->
923 when (not $ isLocal node (NodeIdentifier nid))
924 (forwardDeath node nid)
925 return Nothing
926 False -> return $ Just (pid,nidlist) )
927 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
928 where
929 forwardDeath node nid = ncSendToNode nid
930 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
931 , ctrlMsgSignal = Died ident reason
932 }
934-- [Unified: Table 13]
935ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
936ncEffectSpawn pid cProc ref = do
937 mProc <- unClosure cProc
938 -- If the closure does not exist, we spawn a process that throws an exception
939 -- This allows the remote node to find out what's happening
940 -- TODO:
941 let proc = case mProc of
942 Left err -> fail $ "Error: Could not resolve closure: " ++ err
943 Right p -> p
944 node <- ask
945 pid' <- liftIO $ forkProcess node proc
946 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
948-- Unified semantics does not explicitly describe how to implement 'register',
949-- but mentions it's "very similar to nsend" (Table 14)
950-- We send a response indicated if the operation is invalid
951ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
952ncEffectRegister from label atnode mPid reregistration = do
953 node <- ask
954 currentVal <- gets (^. registeredHereFor label)
955 isOk <-
956 case mPid of
957 Nothing -> -- unregister request
958 return $ isJust currentVal
959 Just thepid -> -- register request
960 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
961 return $ (isNothing currentVal /= reregistration) &&
962 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
963 if isLocal node (NodeIdentifier atnode)
964 then do when isOk $
965 do modify' $ registeredHereFor label ^= mPid
966 updateRemote node currentVal mPid
967 case mPid of
968 (Just p) -> do
969 if reregistration
970 then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
971 else return ()
972 liftIO $ trace node (MxRegistered p label)
973 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
974 newVal <- gets (^. registeredHereFor label)
975 ncSendToProcess from $ unsafeCreateUnencodedMessage $
976 RegisterReply label isOk newVal
977 else let operation =
978 case reregistration of
979 True -> flip decList
980 False -> flip incList
981 in case mPid of
982 Nothing -> return ()
983 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
984 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
985 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
986 forward node (processNodeId newval) (Register label atnode (Just newval) False)
987 updateRemote node Nothing (Just newval) =
988 forward node (processNodeId newval) (Register label atnode (Just newval) False)
989 updateRemote node (Just oldval) Nothing =
990 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
991 updateRemote _ _ _ = return ()
992 maybeify f Nothing = unmaybeify $ f []
993 maybeify f (Just x) = unmaybeify $ f x
995 unmaybeify [] = Nothing
996 unmaybeify x = Just x
997 incList [] tag = [(tag,1)]
998 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
999 incList (x:xs) tag = x : incList xs tag
1000 decList [] _ = []
1001 decList ((atag,1):xs) tag | atag == tag = xs
1002 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
1003 decList (x:xs) tag = x:decList xs tag
1004 forward node to reg =
1005 when (not $ isLocal node (NodeIdentifier to)) $
1006 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
1007 , ctrlMsgSignal = reg
1011-- Unified semantics does not explicitly describe 'whereis'
1012ncEffectWhereIs :: ProcessId -> String -> NC ()
1013ncEffectWhereIs from label = do
1014 mPid <- gets (^. registeredHereFor label)
1015 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
1017-- [Unified: Table 14]
1018ncEffectNamedSend :: String -> Message -> NC ()
1019ncEffectNamedSend label msg = do
1020 mPid <- gets (^. registeredHereFor label)
1021 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
1022 forM_ mPid $ \to ->
1023 -- If this is a trace message we don't trace it to avoid entering a loop
1024 -- where trace messages produce more trace messages.
1025 ncSendToProcessAndTrace (label /= "trace.logger") to msg
1027-- [Issue #DP-20]
1028ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
1029ncEffectLocalSend = ncEffectLocalSendAndTrace True
1031ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
1032ncEffectLocalSendAndTrace shouldTrace node to msg =
1033 liftIO $ withLocalProc node to $ \p -> do
1034 enqueue (processQueue p) msg
1035 when shouldTrace $ trace node (MxReceived to msg)
1037-- [Issue #DP-20]
1038ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
1039ncEffectLocalPortSend from msg = do
1040 node <- ask
1041 let pid = sendPortProcessId from
1042 cid = sendPortLocalId from
1043 liftIO $ withLocalProc node pid $ \proc -> do
1044 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
1045 case mChan of
1046 -- in the unlikely event we know nothing about this channel id,
1047 -- there's little to be done - perhaps some logging/tracing though...
1048 Nothing -> return ()
1049 Just (TypedChannel chan') -> do
1050 -- If ch is Nothing, the process has given up the read end of
1051 -- the channel and we simply ignore the incoming message - this
1052 ch <- deRefWeak chan'
1053 forM_ ch $ \chan -> deliverChan msg chan
1054 where deliverChan :: forall a . Message -> TQueue a -> IO ()
1055 deliverChan (UnencodedMessage _ raw) chan' =
1056 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1057 deliverChan (EncodedMessage _ _) _ =
1058 -- this will not happen unless someone screws with Primitives.hs
1059 error "invalid local channel delivery"
1061-- [Issue #69]
1062ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
1063ncEffectKill from to reason = do
1064 node <- ask
1065 when (isLocal node (ProcessIdentifier to)) $
1066 throwException to $ ProcessKillException from reason
1068-- [Issue #69]
1069ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
1070ncEffectExit from to reason = do
1071 node <- ask
1072 when (isLocal node (ProcessIdentifier to)) $
1073 throwException to $ ProcessExitException from reason
1075-- [Issue #89]
1076ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
1077ncEffectGetInfo from pid =
1078 let lpid = processLocalId pid
1079 them = (ProcessIdentifier pid)
1080 in do
1081 node <- ask
1082 mProc <- liftIO $ withValidLocalState node
1083 $ return . (^. localProcessWithId lpid)
1084 case mProc of
1085 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1086 from (ProcessInfoNone DiedUnknownId)
1087 Just proc -> do
1088 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>
1089 gets (^. links)
1090 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)
1091 registered <- gets (^. registeredHere)
1092 size <- liftIO $ queueSize $ processQueue $ proc
1094 let reg = registeredNames registered
1095 dispatch (isLocal node (ProcessIdentifier from))
1096 from
1097 ProcessInfo {
1098 infoNode = (processNodeId pid)
1099 , infoRegisteredNames = reg
1100 , infoMessageQueueLength = size
1101 , infoMonitors = Set.toList itsMons
1102 , infoLinks = Set.toList itsLinks
1104 where dispatch :: (Serializable a)
1105 => Bool
1106 -> ProcessId
1107 -> a
1108 -> NC ()
1109 dispatch True dest pInfo = postAsMessage dest $ pInfo
1110 dispatch False dest pInfo =
1111 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1113 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1114 then (k:ks)
1115 else ks) []
1117ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1118ncEffectGetNodeStats from _nid = do
1119 node <- ask
1120 ncState <- StateT.get
1121 nodeState <- liftIO $ withValidLocalState node return
1122 let stats =
1123 NodeStats {
1124 nodeStatsNode = localNodeId node
1125 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1126 , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors
1127 , nodeStatsLinks = BiMultiMap.size $ ncState ^. links
1128 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1130 postAsMessage from stats
1132--------------------------------------------------------------------------------
1133-- Auxiliary --
1134--------------------------------------------------------------------------------
1136notifyDied :: ProcessId -- ^ Who to notify?
1137 -> Identifier -- ^ Who died?
1138 -> DiedReason -- ^ How did they die?
1139 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1140 -> NC ()
1141notifyDied dest src reason mRef = do
1142 node <- ask
1143 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1144 (True, Just ref, ProcessIdentifier pid) ->
1145 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1146 (True, Just ref, NodeIdentifier nid) ->
1147 postAsMessage dest $ NodeMonitorNotification ref nid reason
1148 (True, Just ref, SendPortIdentifier cid) ->
1149 postAsMessage dest $ PortMonitorNotification ref cid reason
1150 (True, Nothing, ProcessIdentifier pid) ->
1151 throwException dest $ ProcessLinkException pid reason
1152 (True, Nothing, NodeIdentifier pid) ->
1153 throwException dest $ NodeLinkException pid reason
1154 (True, Nothing, SendPortIdentifier pid) ->
1155 throwException dest $ PortLinkException pid reason
1156 (False, _, _) ->
1157 -- The change in sender comes from [Unified: Table 10]
1158 ncSendToNode (processNodeId dest) $ NCMsg
1159 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1160 , ctrlMsgSignal = Died src reason
1163-- | [Unified: Table 8]
1164destNid :: ProcessSignal -> Maybe NodeId
1165destNid (Link ident) = Just $ nodeOf ident
1166destNid (Unlink ident) = Just $ nodeOf ident
1167destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1168destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1169destNid (Spawn _ _) = Nothing
1170destNid (Register _ _ _ _) = Nothing
1171destNid (WhereIs _) = Nothing
1172destNid (NamedSend _ _) = Nothing
1173destNid (UnreliableSend _ _) = Nothing
1174-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1175-- then when a local process dies the monitoring/linking machinery will take
1176-- care of notifying remote nodes
1177destNid (Died _ _) = Nothing
1178destNid (Kill pid _) = Just $ processNodeId pid
1179destNid (Exit pid _) = Just $ processNodeId pid
1180destNid (GetInfo pid) = Just $ processNodeId pid
1181destNid (GetNodeStats nid) = Just nid
1182destNid (LocalSend pid _) = Just $ processNodeId pid
1183destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1184destNid (SigShutdown) = Nothing
1186-- | Check if a process is local to our own node
1187isLocal :: LocalNode -> Identifier -> Bool
1188isLocal nid ident = nodeOf ident == localNodeId nid
1190-- | Lookup a local closure
1191unClosure :: Typeable a => Closure a -> NC (Either String a)
1192unClosure closure = do
1193 rtable <- remoteTable <$> ask
1194 return (Static.unclosure rtable closure)
1196-- | Check if an identifier refers to a valid local object
1197isValidLocalIdentifier :: Identifier -> NC Bool
1198isValidLocalIdentifier ident = do
1199 node <- ask
1200 liftIO . withValidLocalState node $ \nSt ->
1201 case ident of
1202 NodeIdentifier nid ->
1203 return $ nid == localNodeId node
1204 ProcessIdentifier pid -> do
1205 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1206 return $ isJust mProc
1207 SendPortIdentifier cid -> do
1208 let pid = sendPortProcessId cid
1209 mProc = nSt ^. localProcessWithId (processLocalId pid)
1210 case mProc of
1211 Nothing -> return False
1212 Just proc -> withMVar (processState proc) $ \pSt -> do
1213 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1214 return $ isJust mCh
1216--------------------------------------------------------------------------------
1217-- Messages to local processes --
1218--------------------------------------------------------------------------------
1220postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1221postAsMessage pid = postMessage pid . createUnencodedMessage
1223postMessage :: ProcessId -> Message -> NC ()
1224postMessage pid msg = do
1225 node <- ask
1226 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1228throwException :: Exception e => ProcessId -> e -> NC ()
1229throwException pid e = do
1230 node <- ask
1231 -- throwTo blocks until the exception is received by the target thread.
1232 -- We cannot easily make it happen asynchronpusly because then 'unlink'
1233 -- semantics would break.
1234 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1236withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1237withLocalProc node pid p =
1238 -- By [Unified: table 6, rule missing_process] messages to dead processes
1239 -- can silently be dropped
1240 let lpid = processLocalId pid in do
1241 join $ withValidLocalState node $ \vst ->
1242 return $ forM_ (vst ^. localProcessWithId lpid) p
1244--------------------------------------------------------------------------------
1245-- Accessors --
1246--------------------------------------------------------------------------------
1248links :: Accessor NCState (BiMultiMap Identifier ProcessId ())
1249links = accessor _links (\ls st -> st { _links = ls })
1251monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)
1252monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1254registeredHere :: Accessor NCState (Map String ProcessId)
1255registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1257registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1258registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1260registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1261registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1263registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1264registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1266-- | @splitNotif ident@ splits a notifications map into those
1267-- notifications that should trigger when 'ident' fails and those links that
1268-- should not.
1270-- There is a hierarchy between identifiers: failure of a node implies failure
1271-- of all processes on that node, and failure of a process implies failure of
1272-- all typed channels to that process. In other words, if 'ident' refers to a
1273-- node, then the /should trigger/ set will include
1275-- * the notifications for the node specifically
1276-- * the notifications for processes on that node, and
1277-- * the notifications for typed channels to processes on that node.
1279-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1280-- include
1282-- * the notifications for that process specifically and
1283-- * the notifications for typed channels to that process.
1285-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1286splitNotif :: (Ord a, Ord v)
1287 => Identifier
1288 -> BiMultiMap Identifier a v
1289 -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)
1290splitNotif ident =
1291 BiMultiMap.partitionWithKeyBy1st (\k !_v -> ident `impliesDeathOf` k)
1293--------------------------------------------------------------------------------
1294-- Auxiliary --
1295--------------------------------------------------------------------------------
1297-- | Modify and evaluate the state
1298modify' :: MonadState s m => (s -> s) -> m ()
1299modify' f = StateT.get >>= \s -> StateT.put $! f s