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