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