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