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