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