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