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