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