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