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