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