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