@@ -1,3 +1,12 @@
2+{-# LANGUAGE DeriveDataTypeable #-}
3+{-# LANGUAGE ScopedTypeVariables #-}
4+{-# LANGUAGE RankNTypes #-}
5+{-# LANGUAGE BangPatterns #-}
6+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7+{-# LANGUAGE RecordWildCards #-}
8+{-# LANGUAGE MagicHash #-}
1-- | Local nodes
10-- | Local nodes
3module Control.Distributed.Process.Node
12module Control.Distributed.Process.Node
@@ -12,10 +21,6 @@ module Control.Distributed.Process.Node
13-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
22-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
15−#if ! MIN_VERSION_base(4,6,0)
16−import Prelude hiding (catch)
19import System.IO (fixIO, hPutStrLn, stderr)
24import System.IO (fixIO, hPutStrLn, stderr)
20import System.Mem.Weak (Weak, deRefWeak)
25import System.Mem.Weak (Weak, deRefWeak)
21import qualified Data.ByteString.Lazy as BSL (fromChunks)
26import qualified Data.ByteString.Lazy as BSL (fromChunks)
@@ -23,46 +28,67 @@ import Data.Binary (decode)
23import Data.Map (Map)
28import Data.Map (Map)
24import qualified Data.Map as Map
29import qualified Data.Map as Map
27 , fromList
33 , fromList
29 , partitionWithKey
37 , partitionWithKey
31 , filterWithKey
40 , filterWithKey
32 , foldlWithKey
41 , foldlWithKey
43+import Data.Time.Format (formatTime)
44+#if MIN_VERSION_time(1,5,0)
45+import Data.Time.Format (defaultTimeLocale)
47+import System.Locale (defaultTimeLocale)
34import Data.Set (Set)
49import Data.Set (Set)
35import qualified Data.Set as Set
50import qualified Data.Set as Set
42import Data.Foldable (forM_)
59import Data.Foldable (forM_)
43import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
60import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
44import Data.Typeable (Typeable)
61import Data.Typeable (Typeable)
45import Control.Category ((>>>))
62import Control.Category ((>>>))
46−import Control.Applicative ((<$>))
63+import Control.Applicative
47−import Control.Monad (void, when)
64+import Control.Monad (void, when, join)
48import Control.Monad.IO.Class (MonadIO, liftIO)
65import Control.Monad.IO.Class (MonadIO, liftIO)
49import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
66import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
50import qualified Control.Monad.State.Strict as StateT (get, put)
67import qualified Control.Monad.State.Strict as StateT (get, put)
51import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
68import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
52import Control.Exception
69import Control.Exception
54− , AsyncException(ThreadKilled)
55 , SomeException
71 , SomeException
56 , Exception
72 , Exception
74+ , uninterruptibleMask_
59−import qualified Control.Exception as Exception (Handler(..), catch, catches, finally)
78+import qualified Control.Exception as Exception
60−import Control.Concurrent (forkIO, myThreadId)
84+import Control.Concurrent (forkIO, killThread)
85+import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)
86+import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap
61import Control.Distributed.Process.Internal.StrictMVar
87import Control.Distributed.Process.Internal.StrictMVar
63 , withMVar
89 , withMVar
64 , modifyMVar
91 , modifyMVar
66 , newEmptyMVar
92 , newEmptyMVar
68 , takeMVar
94 , takeMVar
@@ -77,6 +103,7 @@ import Control.Distributed.Process.Internal.CQueue
77 , enqueue
103 , enqueue
78 , newCQueue
104 , newCQueue
79 , mkWeakCQueue
105 , mkWeakCQueue
81import qualified Network.Transport as NT
108import qualified Network.Transport as NT
82 ( Transport
109 ( Transport
@@ -88,8 +115,8 @@ import qualified Network.Transport as NT
88 , TransportError(..)
115 , TransportError(..)
89 , address
116 , address
90 , closeEndPoint
117 , closeEndPoint
92 , Connection
118 , Connection
94 , EndPointAddress
121 , EndPointAddress
95 , Reliability(ReliableOrdered)
122 , Reliability(ReliableOrdered)
@@ -106,8 +133,11 @@ import Control.Distributed.Process.Internal.Types
106 , LocalProcessId(..)
133 , LocalProcessId(..)
107 , ProcessId(..)
134 , ProcessId(..)
108 , LocalNode(..)
135 , LocalNode(..)
109− , Tracer(..)
136+ , MxEventBus(..)
110 , LocalNodeState(..)
137 , LocalNodeState(..)
138+ , ValidLocalNodeState(..)
139+ , withValidLocalState
140+ , modifyValidLocalState
111 , LocalProcess(..)
141 , LocalProcess(..)
112 , LocalProcessState(..)
142 , LocalProcessState(..)
113 , Process(..)
143 , Process(..)
@@ -117,9 +147,11 @@ import Control.Distributed.Process.Internal.Types
117 , localPidCounter
147 , localPidCounter
118 , localPidUnique
148 , localPidUnique
119 , localProcessWithId
149 , localProcessWithId
120 , localConnections
151 , localConnections
121 , forever'
152 , forever'
122 , MonitorRef(..)
153 , MonitorRef(..)
154+ , NodeClosedException(..)
123 , ProcessMonitorNotification(..)
155 , ProcessMonitorNotification(..)
124 , NodeMonitorNotification(..)
156 , NodeMonitorNotification(..)
125 , PortMonitorNotification(..)
157 , PortMonitorNotification(..)
@@ -139,55 +171,74 @@ import Control.Distributed.Process.Internal.Types
140 , ProcessInfo(..)
172 , ProcessInfo(..)
141 , ProcessInfoNone(..)
173 , ProcessInfoNone(..)
142 , SendPortId(..)
175 , SendPortId(..)
143 , typedChannelWithId
176 , typedChannelWithId
144 , RegisterReply(..)
177 , RegisterReply(..)
145 , WhereIsReply(..)
178 , WhereIsReply(..)
146 , payloadToMessage
179 , payloadToMessage
147 , createUnencodedMessage
180 , createUnencodedMessage
181+ , unsafeCreateUnencodedMessage
148 , runLocalProcess
182 , runLocalProcess
149 , firstNonReservedProcessId
183 , firstNonReservedProcessId
150 , ImplicitReconnect(WithImplicitReconnect)
184 , ImplicitReconnect(WithImplicitReconnect)
186+import Control.Distributed.Process.Management.Internal.Agent
189+import Control.Distributed.Process.Management.Internal.Types
152−import qualified Control.Distributed.Process.Internal.Trace.Remote as Trace
192+import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
153 ( remoteTable
193 ( remoteTable
155−import Control.Distributed.Process.Internal.Trace.Types
195+import Control.Distributed.Process.Management.Internal.Trace.Tracer
198+import Control.Distributed.Process.Management.Internal.Trace.Types
156 ( TraceArg(..)
199 ( TraceArg(..)
158 , traceEvent
200 , traceEvent
159 , traceLogFmt
201 , traceLogFmt
160 , enableTrace
202 , enableTrace
162−import Control.Distributed.Process.Internal.Trace.Tracer
166import Control.Distributed.Process.Serializable (Serializable)
204import Control.Distributed.Process.Serializable (Serializable)
167import Control.Distributed.Process.Internal.Messaging
205import Control.Distributed.Process.Internal.Messaging
168 ( sendBinary
206 ( sendBinary
170 , closeImplicitReconnections
207 , closeImplicitReconnections
171 , impliesDeathOf
208 , impliesDeathOf
173import Control.Distributed.Process.Internal.Primitives
210import Control.Distributed.Process.Internal.Primitives
174 ( register
211 ( register
176 , receiveWait
212 , receiveWait
178 , sendChan
214 , sendChan
180 , unwrapMessage
215 , unwrapMessage
182−import Control.Distributed.Process.Internal.Types (SendPort)
221+import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
183import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
222import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
184import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
223import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
185import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
224import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
186 ( mapMaybe
225 ( mapMaybe
187 , mapDefault
226 , mapDefault
189−import System.Environment (getEnv)
228+import Control.Monad.Catch (try)
229+import GHC.IO (IO(..), unsafeUnmask)
230+import GHC.Base ( maskAsyncExceptions# )
190import Unsafe.Coerce
232import Unsafe.Coerce
235+-- Remove these definitions when the fix for
236+-- https://ghc.haskell.org/trac/ghc/ticket/10149
237+-- is included in all supported compilers:
238+block :: IO a -> IO a
239+block (IO io) = IO $ maskAsyncExceptions# io
240+unblock :: IO a -> IO a
241+unblock = unsafeUnmask
192--------------------------------------------------------------------------------
243--------------------------------------------------------------------------------
193-- Initialization --
244-- Initialization --
@@ -211,7 +262,7 @@ newLocalNode transport rtable = do
211createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
262createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
212createBareLocalNode endPoint rtable = do
263createBareLocalNode endPoint rtable = do
213 unq <- randomIO
264 unq <- randomIO
214− state <- newMVar LocalNodeState
265+ state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
215 { _localProcesses = Map.empty
266 { _localProcesses = Map.empty
216 , _localPidCounter = firstNonReservedProcessId
267 , _localPidCounter = firstNonReservedProcessId
217 , _localPidUnique = unq
268 , _localPidUnique = unq
@@ -222,10 +273,10 @@ createBareLocalNode endPoint rtable = do
222 , localEndPoint = endPoint
273 , localEndPoint = endPoint
223 , localState = state
274 , localState = state
224 , localCtrlChan = ctrlChan
275 , localCtrlChan = ctrlChan
225− , localTracer = InactiveTracer
276+ , localEventBus = MxEventBusInitialising
226 , remoteTable = rtable
277 , remoteTable = rtable
228− tracedNode <- startTracing node
279+ tracedNode <- startMxAgent node
230 -- Once the NC terminates, the endpoint isn't much use,
281 -- Once the NC terminates, the endpoint isn't much use,
231 void $ forkIO $ Exception.finally (runNodeController tracedNode)
282 void $ forkIO $ Exception.finally (runNodeController tracedNode)
@@ -243,38 +294,27 @@ createBareLocalNode endPoint rtable = do
243 , ctrlMsgSignal = SigShutdown
294 , ctrlMsgSignal = SigShutdown
246−startTracing :: LocalNode -> IO LocalNode
297+startMxAgent :: LocalNode -> IO LocalNode
247−startTracing node =
298+startMxAgent node = do
248− -- TODO: use tryGetEnv once we drop support for GHC 7.2.x
250− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)
251− (\(_ :: IOError) -> return node)
252− where procTracer :: LocalNode -> IO LocalNode
254− -- see note [tracer/forkProcess races]
299+ -- see note [tracer/forkProcess races]
300+ let fork = forkProcess node
255− mv <- MVar.newEmptyMVar
301+ mv <- MVar.newEmptyMVar
256− pid <- forkProcess n (traceController mv)
302+ pid <- fork $ mxAgentController fork mv
257− wqRef <- MVar.takeMVar mv
303+ (tracer', wqRef, mxNew') <- MVar.takeMVar mv
258− return n { localTracer = (ActiveTracer pid wqRef) }
304+ return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
260−startDefaultTracer :: LocalNode -> IO (Maybe ProcessId)
306+startDefaultTracer :: LocalNode -> IO ()
261−startDefaultTracer node =
307+startDefaultTracer node' = do
262− Exception.catch
308+ let t = localEventBus node'
263− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> go node)
264− (\(_ :: IOError) -> return Nothing)
265− where go :: LocalNode -> IO (Maybe ProcessId)
267− registerTraceController node'
268− pid <- forkProcess node' defaultTracer
269− enableTrace (localTracer node') pid
272− registerTraceController :: LocalNode -> IO ()
273− registerTraceController n =
274− let t = localTracer n in
275− case t of
309+ case t of
276− InactiveTracer -> return ()
310+ MxEventBus _ (Tracer pid _) _ _ -> do
277− (ActiveTracer pid _) -> runProcess n $ register "trace.controller" pid
311+ runProcess node' $ register "trace.controller" pid
312+ pid' <- forkProcess node' defaultTracer
313+ enableTrace (localEventBus node') pid'
314+ runProcess node' $ register "tracer.initial" pid'
317+-- TODO: we need a better mechanism for defining and registering services
279-- | Start and register the service processes on a node
319-- | Start and register the service processes on a node
280startServiceProcesses :: LocalNode -> IO ()
320startServiceProcesses :: LocalNode -> IO ()
@@ -283,53 +323,71 @@ startServiceProcesses node = do
283 -- the default tracer first, even though it might @nsend@ to the logger
323 -- the default tracer first, even though it might @nsend@ to the logger
284 -- before /that/ process has started - this is a totally harmless race
324 -- before /that/ process has started - this is a totally harmless race
285 -- however, so we deliberably ignore it
325 -- however, so we deliberably ignore it
286− mPid <- startDefaultTracer node
326+ startDefaultTracer node
288− Nothing -> return ()
289− Just pid -> runProcess node $ register "tracer.initial" pid
291 logger <- forkProcess node loop
327 logger <- forkProcess node loop
328+ runProcess node $ do
329+ register "logger" logger
330+ -- The trace.logger is used for tracing to the console to avoid feedback
331+ -- loops during tracing if the user reregisters the "logger" with a custom
332+ -- process which uses 'send' or other primitives which are traced.
292− runProcess node $ register "logger" logger
333+ register "trace.logger" logger
294− loop = do
335+ loop = do
295− receiveWait
336+ receiveWait
296− [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
337+ [ match $ \(SayMessage time pid string) -> do
338+ let time' = formatTime defaultTimeLocale "%c" time
297− liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
339+ liftIO . hPutStrLn stderr $ time' ++ " " ++ show pid ++ ": " ++ string
299− , match $ \((time, string) :: (String, String)) -> do
341+ , match $ \((time, string) :: (String, String)) -> do
300− -- this is a 'trace' message from the local node tracer
342+ -- this is a 'trace' message from the local node tracer
301− liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
343+ liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
303− , match $ \(ch :: SendPort ()) -> -- a shutdown request
345+ , match $ \(ch :: SendPort ()) -> -- a shutdown request
304− sendChan ch ()
346+ sendChan ch ()
307−-- | Force-close a local node
349+-- | Force-close a local node, killing all processes on that node.
309−-- TODO: for now we just close the associated endpoint
310closeLocalNode :: LocalNode -> IO ()
350closeLocalNode :: LocalNode -> IO ()
311−closeLocalNode node =
351+closeLocalNode node = do
312− -- TODO: close all our processes, surely!?
352+ -- Kill processes after refilling the mvar. Otherwise, there is potential for
353+ -- deadlock as a dying process tries to get the mvar while masking exceptions
355+ join $ modifyMVar (localState node) $ \st -> case st of
356+ LocalNodeValid vst -> do
357+ return ( LocalNodeClosed
358+ , forM_ (vst ^. localProcesses) $ \lproc ->
359+ -- Semantics of 'throwTo' guarantee that target thread will get
360+ -- delivered an exception. Therefore, target thread will be
361+ -- killed eventually and that's as good as we can do. No need
362+ -- to wait for thread to actually finish dying.
363+ killThread (processThread lproc)
365+ LocalNodeClosed -> return (LocalNodeClosed, return ())
366+ -- This call will have the effect of shutting down the NC as well (see
367+ -- 'createBareLocalNode').
313 NT.closeEndPoint (localEndPoint node)
368 NT.closeEndPoint (localEndPoint node)
315-- | Run a process on a local node and wait for it to finish
370-- | Run a process on a local node and wait for it to finish
316runProcess :: LocalNode -> Process () -> IO ()
371runProcess :: LocalNode -> Process () -> IO ()
317runProcess node proc = do
372runProcess node proc = do
318 done <- newEmptyMVar
373 done <- newEmptyMVar
319− tid <- myThreadId
374+ -- TODO; When forkProcess inherits the masking state, protect the forked
375+ -- thread against async exceptions that could occur before 'try' is evaluated.
320− void $ forkProcess node $ do
376+ void $ forkProcess node $ try proc >>= liftIO . putMVar done
321− catch (proc `finally` liftIO (putMVar done ()))
377+ takeMVar done >>= either (throwIO :: SomeException -> IO a) return
322− (\(ex :: SomeException) -> liftIO $ throwTo tid ex)
325-- | Spawn a new process on a local node
379-- | Spawn a new process on a local node
326forkProcess :: LocalNode -> Process () -> IO ProcessId
380forkProcess :: LocalNode -> Process () -> IO ProcessId
381+forkProcess node proc = do
382+ ms <- getMaskingState
327−forkProcess node proc = modifyMVar (localState node) startProcess
383+ modifyMVarMasked (localState node) (startProcess ms)
385+ startProcess :: MaskingState
329− startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
387+ -> IO (LocalNodeState, ProcessId)
330− startProcess st = do
388+ startProcess ms (LocalNodeValid vst) = do
331− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
389+ let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
332− , lpidUnique = st ^. localPidUnique
390+ , lpidUnique = vst ^. localPidUnique
334 let pid = ProcessId { processNodeId = localNodeId node
392 let pid = ProcessId { processNodeId = localNodeId node
335 , processLocalId = lpid
393 , processLocalId = lpid
@@ -349,9 +407,15 @@ forkProcess node proc = modifyMVar (localState node) startProcess
349 , processThread = tid
407 , processThread = tid
350 , processNode = node
408 , processNode = node
410+ -- Rewrite this code when this is fixed:
411+ -- https://ghc.haskell.org/trac/ghc/ticket/10149
412+ let unmask = case ms of
414+ MaskedInterruptible -> block
415+ MaskedUninterruptible -> id
352− tid' <- forkIO $ do
416+ tid' <- uninterruptibleMask_ $ forkIO $ do
353 reason <- Exception.catches
417 reason <- Exception.catches
354− (runLocalProcess lproc proc >> return DiedNormal)
418+ (unmask $ runLocalProcess lproc proc >> return DiedNormal)
355 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
419 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
356 mMsg <- unwrapMessage msg :: IO (Maybe String)
420 mMsg <- unwrapMessage msg :: IO (Maybe String)
357 case mMsg of
421 case mMsg of
@@ -361,47 +425,60 @@ forkProcess node proc = modifyMVar (localState node) startProcess
361 (return . DiedException . (show :: SomeException -> String)))]
425 (return . DiedException . (show :: SomeException -> String)))]
363 -- [Unified: Table 4, rules termination and exiting]
427 -- [Unified: Table 4, rules termination and exiting]
364− modifyMVar_ (localState node) (cleanupProcess pid)
428+ mconns <- modifyValidLocalState node (cleanupProcess pid)
429+ -- XXX: Revisit after agreeing on the bigger picture for the semantics
430+ -- of transport operations.
431+ -- https://github.com/haskell-distributed/distributed-process/issues/204
432+ forM_ mconns $ forkIO . mapM_ NT.close
365 writeChan (localCtrlChan node) NCMsg
434 writeChan (localCtrlChan node) NCMsg
366 { ctrlMsgSender = ProcessIdentifier pid
435 { ctrlMsgSender = ProcessIdentifier pid
367 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
436 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
369 return (tid', lproc)
438 return (tid', lproc)
372 -- see note [tracer/forkProcess races]
440 -- see note [tracer/forkProcess races]
373− trace node (TraceEvSpawned pid)
441+ trace node (MxSpawned pid)
375 if lpidCounter lpid == maxBound
443 if lpidCounter lpid == maxBound
445+ -- TODO: this doesn't look right at all - how do we know
446+ -- that newUnique represents a process id that is available!?
377 newUnique <- randomIO
447 newUnique <- randomIO
448+ return ( LocalNodeValid
378− return ( (localProcessWithId lpid ^= Just lproc)
449+ $ (localProcessWithId lpid ^= Just lproc)
379 . (localPidCounter ^= firstNonReservedProcessId)
450 . (localPidCounter ^= firstNonReservedProcessId)
380 . (localPidUnique ^= newUnique)
451 . (localPidUnique ^= newUnique)
456+ return ( LocalNodeValid
385− return ( (localProcessWithId lpid ^= Just lproc)
457+ $ (localProcessWithId lpid ^= Just lproc)
386 . (localPidCounter ^: (+ 1))
458 . (localPidCounter ^: (+ 1))
462+ startProcess _ LocalNodeClosed =
463+ throwIO $ NodeClosedException $ localNodeId node
391− cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState
465+ cleanupProcess :: ProcessId
466+ -> ValidLocalNodeState
467+ -> IO (ValidLocalNodeState, [NT.Connection])
392− cleanupProcess pid st = do
468+ cleanupProcess pid vst = do
393 let pid' = ProcessIdentifier pid
469 let pid' = ProcessIdentifier pid
394− let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)
470+ let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
395− mapM_ (NT.close . fst) (Map.elems affected)
396− return $ (localProcessWithId (processLocalId pid) ^= Nothing)
471+ return ( (localProcessWithId (processLocalId pid) ^= Nothing)
397− . (localConnections ^= unaffected)
472+ . (localConnections ^= unaffected)
474+ , map fst $ Map.elems affected
400-- note [tracer/forkProcess races]
477-- note [tracer/forkProcess races]
402-- Our startTracing function uses forkProcess to start the trace controller
479-- Our startTracing function uses forkProcess to start the trace controller
403-- process, and of course forkProcess attempts to call traceEvent once the
480-- process, and of course forkProcess attempts to call traceEvent once the
404−-- process has started. This is harmless, as the localTracer is not updated
481+-- process has started. This is harmless, as the localEventBus is not updated
405-- until /after/ the initial forkProcess completes, so the first call to
482-- until /after/ the initial forkProcess completes, so the first call to
406-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
483-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
@@ -415,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
415data IncomingTarget =
492data IncomingTarget =
417 | ToProc ProcessId (Weak (CQueue Message))
494 | ToProc ProcessId (Weak (CQueue Message))
418− | ToChan TypedChannel
495+ | ToChan SendPortId TypedChannel
421data ConnectionState = ConnectionState {
498data ConnectionState = ConnectionState {
@@ -442,6 +519,7 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
443handleIncomingMessages :: LocalNode -> IO ()
520handleIncomingMessages :: LocalNode -> IO ()
444handleIncomingMessages node = go initConnectionState
521handleIncomingMessages node = go initConnectionState
522+ `Exception.catch` \(NodeClosedException _) -> return ()
446 go :: ConnectionState -> IO ()
524 go :: ConnectionState -> IO ()
447 go !st = do
525 go !st = do
@@ -450,13 +528,14 @@ handleIncomingMessages node = go initConnectionState
450 NT.ConnectionOpened cid rel theirAddr ->
528 NT.ConnectionOpened cid rel theirAddr ->
451 if rel == NT.ReliableOrdered
529 if rel == NT.ReliableOrdered
453− trace node (TraceEvConnected cid)
531+ trace node (MxConnected cid theirAddr)
455 (incomingAt cid ^= Just (theirAddr, Uninit))
533 (incomingAt cid ^= Just (theirAddr, Uninit))
456 . (incomingFrom theirAddr ^: Set.insert cid)
534 . (incomingFrom theirAddr ^: Set.insert cid)
459− else invalidRequest cid st
537+ else invalidRequest cid st $
538+ "attempt to connect with unsupported reliability " ++ show rel
460 NT.Received cid payload ->
539 NT.Received cid payload ->
461 case st ^. incomingAt cid of
540 case st ^. incomingAt cid of
462 Just (_, ToProc pid weakQueue) -> do
541 Just (_, ToProc pid weakQueue) -> do
@@ -466,15 +545,19 @@ handleIncomingMessages node = go initConnectionState
466 -- it from the NC state? (and same for channels, below)
545 -- it from the NC state? (and same for channels, below)
467 let msg = payloadToMessage payload
546 let msg = payloadToMessage payload
468 enqueue queue msg -- 'enqueue' is strict
547 enqueue queue msg -- 'enqueue' is strict
469− trace node (TraceEvReceived pid msg)
548+ trace node (MxReceived pid msg)
471− Just (_, ToChan (TypedChannel chan')) -> do
550+ Just (_, ToChan chId (TypedChannel chan')) -> do
472 mChan <- deRefWeak chan'
551 mChan <- deRefWeak chan'
473 -- If mChan is Nothing, the process has given up the read end of
552 -- If mChan is Nothing, the process has given up the read end of
474 -- the channel and we simply ignore the incoming message
553 -- the channel and we simply ignore the incoming message
475− forM_ mChan $ \chan -> atomically $
554+ forM_ mChan $ \chan -> do
555+ msg' <- atomically $ do
556+ msg <- return $! decode (BSL.fromChunks payload)
476− -- We make sure the message is fully decoded when it is enqueued
557+ -- We make sure the message is fully decoded when it is enqueued
477− writeTQueue chan $! decode (BSL.fromChunks payload)
558+ writeTQueue chan msg
560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'
479 Just (_, ToNode) -> do
562 Just (_, ToNode) -> do
480 let ctrlMsg = decode . BSL.fromChunks $ payload
563 let ctrlMsg = decode . BSL.fromChunks $ payload
@@ -484,38 +567,49 @@ handleIncomingMessages node = go initConnectionState
484 case decode (BSL.fromChunks payload) of
567 case decode (BSL.fromChunks payload) of
485 ProcessIdentifier pid -> do
568 ProcessIdentifier pid -> do
486 let lpid = processLocalId pid
569 let lpid = processLocalId pid
487− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
570+ mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
488 case mProc of
571 case mProc of
489 Just proc ->
572 Just proc ->
490 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
573 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
491 Nothing ->
574 Nothing ->
492− invalidRequest cid st
575+ -- incoming attempt to connect to unknown process - might
577+ go (incomingAt cid ^= Nothing $ st)
493 SendPortIdentifier chId -> do
578 SendPortIdentifier chId -> do
494 let lcid = sendPortLocalId chId
579 let lcid = sendPortLocalId chId
495 lpid = processLocalId (sendPortProcessId chId)
580 lpid = processLocalId (sendPortProcessId chId)
496− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
581+ mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
497 case mProc of
582 case mProc of
498 Just proc -> do
583 Just proc -> do
499 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
500 case mChannel of
585 case mChannel of
501 Just channel ->
586 Just channel ->
502− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)
503 Nothing ->
588 Nothing ->
504− invalidRequest cid st
589+ invalidRequest cid st $
590+ "incoming attempt to connect to unknown channel of"
591+ ++ " process " ++ show (sendPortProcessId chId)
505 Nothing ->
592 Nothing ->
506− invalidRequest cid st
593+ -- incoming attempt to connect to channel of unknown
594+ -- process - might be dead already
595+ go (incomingAt cid ^= Nothing $ st)
507 NodeIdentifier nid ->
596 NodeIdentifier nid ->
508 if nid == localNodeId node
597 if nid == localNodeId node
509 then go (incomingAt cid ^= Just (src, ToNode) $ st)
598 then go (incomingAt cid ^= Just (src, ToNode) $ st)
510− else invalidRequest cid st
599+ else invalidRequest cid st $
600+ "incoming attempt to connect to a different node -"
601+ ++ " I'm " ++ show (localNodeId node)
602+ ++ " but the remote peer wants to connect to "
511 Nothing ->
604 Nothing ->
512 invalidRequest cid st
605 invalidRequest cid st
606+ "message received from an unknown connection"
513 NT.ConnectionClosed cid ->
607 NT.ConnectionClosed cid ->
514 case st ^. incomingAt cid of
608 case st ^. incomingAt cid of
515 Nothing ->
609 Nothing ->
516− invalidRequest cid st
610+ invalidRequest cid st "closed unknown connection"
517 Just (src, _) -> do
611 Just (src, _) -> do
518− trace node (TraceEvDisconnected cid)
612+ trace node (MxDisconnected cid src)
519 go ( (incomingAt cid ^= Nothing)
613 go ( (incomingAt cid ^= Nothing)
520 . (incomingFrom src ^: Set.delete cid)
614 . (incomingFrom src ^: Set.delete cid)
@@ -544,19 +638,20 @@ handleIncomingMessages node = go initConnectionState
544 -- and we just give up
638 -- and we just give up
545 fail "Cloud Haskell fatal error: received unexpected multicast"
639 fail "Cloud Haskell fatal error: received unexpected multicast"
547− invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
641+ invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
548− invalidRequest cid st = do
642+ invalidRequest cid st msg = do
549 -- TODO: We should treat this as a fatal error on the part of the remote
643 -- TODO: We should treat this as a fatal error on the part of the remote
550 -- node. That is, we should report the remote node as having died, and we
644 -- node. That is, we should report the remote node as having died, and we
551 -- should close incoming connections (this requires a Transport layer
645 -- should close incoming connections (this requires a Transport layer
552 -- extension).
646 -- extension).
553− traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
647+ traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"
648+ ++ " (" ++ msg ++ "): "
554− (Trace cid)]
649+ , (Trace cid)
555 go ( incomingAt cid ^= Nothing
651 go ( incomingAt cid ^= Nothing
559− state = localState node
560 endpoint = localEndPoint node
655 endpoint = localEndPoint node
561 ctrlChan = localCtrlChan node
656 ctrlChan = localCtrlChan node
@@ -565,8 +660,9 @@ handleIncomingMessages node = go initConnectionState
565--------------------------------------------------------------------------------
660--------------------------------------------------------------------------------
567runNodeController :: LocalNode -> IO ()
662runNodeController :: LocalNode -> IO ()
568−runNodeController =
663+runNodeController node =
569− runReaderT (evalStateT (unNC nodeController) initNCState)
664+ runReaderT (evalStateT (unNC nodeController) initNCState) node
665+ `Exception.catch` \(NodeClosedException _) -> return ()
571--------------------------------------------------------------------------------
667--------------------------------------------------------------------------------
572-- Internal data types --
668-- Internal data types --
@@ -574,20 +670,26 @@ runNodeController =
575data NCState = NCState
671data NCState = NCState
576 { -- Mapping from remote processes to linked local processes
672 { -- Mapping from remote processes to linked local processes
577− _links :: !(Map Identifier (Set ProcessId))
673+ _links :: !(BiMultiMap Identifier ProcessId ())
578 -- Mapping from remote processes to monitoring local processes
674 -- Mapping from remote processes to monitoring local processes
579− , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
675+ , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef)
580 -- Process registry: names and where they live, mapped to the PIDs
676 -- Process registry: names and where they live, mapped to the PIDs
581 , _registeredHere :: !(Map String ProcessId)
677 , _registeredHere :: !(Map String ProcessId)
582 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
678 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
585newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
681newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
682+ deriving ( Applicative
686+ , MonadState NCState
586− deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
687+ , MonadReader LocalNode
588initNCState :: NCState
690initNCState :: NCState
589−initNCState = NCState { _links = Map.empty
691+initNCState = NCState { _links = BiMultiMap.empty
590− , _monitors = Map.empty
692+ , _monitors = BiMultiMap.empty
591 , _registeredHere = Map.empty
693 , _registeredHere = Map.empty
592 , _registeredOnNodes = Map.empty
694 , _registeredOnNodes = Map.empty
@@ -603,19 +705,46 @@ instance Show ProcessKillException where
603 show (ProcessKillException pid reason) =
705 show (ProcessKillException pid reason) =
604 "killed-by=" ++ show pid ++ ",reason=" ++ reason
706 "killed-by=" ++ show pid ++ ",reason=" ++ reason
708+ncSendToProcess :: ProcessId -> Message -> NC ()
709+ncSendToProcess = ncSendToProcessAndTrace True
711+ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
712+ncSendToProcessAndTrace shouldTrace pid msg = do
714+ if processNodeId pid == localNodeId node
715+ then ncEffectLocalSendAndTrace shouldTrace node pid msg
716+ else liftIO $ sendBinary node
717+ (NodeIdentifier $ localNodeId node)
718+ (NodeIdentifier $ processNodeId pid)
719+ WithImplicitReconnect
720+ NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
721+ , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
724+ncSendToNode :: NodeId -> NCMsg -> NC ()
725+ncSendToNode to msg = do
727+ liftIO $ if to == localNodeId node
728+ then writeChan (localCtrlChan node) $! msg
729+ else sendBinary node
730+ (NodeIdentifier $ localNodeId node)
732+ WithImplicitReconnect
606--------------------------------------------------------------------------------
735--------------------------------------------------------------------------------
607-- Tracing/Debugging --
736-- Tracing/Debugging --
608--------------------------------------------------------------------------------
737--------------------------------------------------------------------------------
610−-- [Issue #104]
739+-- [Issue #104 / DP-13]
612traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
741traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
613traceNotifyDied node ident reason =
742traceNotifyDied node ident reason =
614 -- TODO: sendPortDied notifications
743 -- TODO: sendPortDied notifications
615 liftIO $ withLocalTracer node $ \t ->
744 liftIO $ withLocalTracer node $ \t ->
616 case ident of
745 case ident of
617− (NodeIdentifier nid) -> traceEvent t (TraceEvNodeDied nid reason)
746+ (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
618− (ProcessIdentifier pid) -> traceEvent t (TraceEvDied pid reason)
747+ (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
619 _ -> return ()
748 _ -> return ()
621traceEventFmtIO :: LocalNode
750traceEventFmtIO :: LocalNode
@@ -625,11 +754,11 @@ traceEventFmtIO :: LocalNode
625traceEventFmtIO node fmt args =
754traceEventFmtIO node fmt args =
626 withLocalTracer node $ \t -> traceLogFmt t fmt args
755 withLocalTracer node $ \t -> traceLogFmt t fmt args
628−trace :: LocalNode -> TraceEvent -> IO ()
757+trace :: LocalNode -> MxEvent -> IO ()
629trace node ev = withLocalTracer node $ \t -> traceEvent t ev
758trace node ev = withLocalTracer node $ \t -> traceEvent t ev
631−withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
760+withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
632−withLocalTracer node act = act (localTracer node)
761+withLocalTracer node act = act (localEventBus node)
634--------------------------------------------------------------------------------
763--------------------------------------------------------------------------------
635-- Core functionality --
764-- Core functionality --
@@ -645,11 +774,7 @@ nodeController = do
645 -- [Unified: Table 7, rule nc_forward]
774 -- [Unified: Table 7, rule nc_forward]
646 case destNid (ctrlMsgSignal msg) of
775 case destNid (ctrlMsgSignal msg) of
647 Just nid' | nid' /= localNodeId node ->
776 Just nid' | nid' /= localNodeId node ->
648− liftIO $ sendBinary node
777+ ncSendToNode nid' msg
650− (NodeIdentifier nid')
651− WithImplicitReconnect
654 return ()
779 return ()
@@ -672,6 +797,8 @@ nodeController = do
672 ncEffectWhereIs from label
797 ncEffectWhereIs from label
673 NCMsg _ (NamedSend label msg') ->
798 NCMsg _ (NamedSend label msg') ->
674 ncEffectNamedSend label msg'
799 ncEffectNamedSend label msg'
800+ NCMsg _ (UnreliableSend lpid msg') ->
801+ ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
675 NCMsg _ (LocalSend to msg') ->
802 NCMsg _ (LocalSend to msg') ->
676 ncEffectLocalSend node to msg'
803 ncEffectLocalSend node to msg'
677 NCMsg _ (LocalPortSend to msg') ->
804 NCMsg _ (LocalPortSend to msg') ->
@@ -685,8 +812,9 @@ nodeController = do
685 NCMsg _ SigShutdown ->
812 NCMsg _ SigShutdown ->
686 liftIO $ do
813 liftIO $ do
687 NT.closeEndPoint (localEndPoint node)
814 NT.closeEndPoint (localEndPoint node)
688− `Exception.finally` throwIO ThreadKilled
815+ `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
689− -- ThreadKilled seems to make more sense than fail/error here
816+ NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
817+ ncEffectGetNodeStats from nid
690 unexpected ->
818 unexpected ->
691 error $ "nodeController: unexpected message " ++ show unexpected
819 error $ "nodeController: unexpected message " ++ show unexpected
@@ -704,22 +832,18 @@ ncEffectMonitor from them mRef = do
704 case (shouldLink, isLocal node (ProcessIdentifier from)) of
832 case (shouldLink, isLocal node (ProcessIdentifier from)) of
705 (True, _) -> -- [Unified: first rule]
833 (True, _) -> -- [Unified: first rule]
706 case mRef of
834 case mRef of
707− Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
835+ Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref
708− Nothing -> modify' $ linksFor them ^: Set.insert from
836+ Nothing -> modify' $ links ^: BiMultiMap.insert them from ()
709 (False, True) -> -- [Unified: second rule]
837 (False, True) -> -- [Unified: second rule]
710 notifyDied from them DiedUnknownId mRef
838 notifyDied from them DiedUnknownId mRef
711 (False, False) -> -- [Unified: third rule]
839 (False, False) -> -- [Unified: third rule]
712 -- TODO: this is the right sender according to the Unified semantics,
840 -- TODO: this is the right sender according to the Unified semantics,
713 -- but perhaps having 'them' as the sender would make more sense
841 -- but perhaps having 'them' as the sender would make more sense
714 -- (see also: notifyDied)
842 -- (see also: notifyDied)
715− liftIO $ sendBinary node
716− (NodeIdentifier $ localNodeId node)
717− (NodeIdentifier $ processNodeId from)
843+ ncSendToNode (processNodeId from) $ NCMsg
718− WithImplicitReconnect
720− { ctrlMsgSender = NodeIdentifier (localNodeId node)
844+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
721− , ctrlMsgSignal = Died them DiedUnknownId
845+ , ctrlMsgSignal = Died them DiedUnknownId
724-- [Unified: Table 11]
848-- [Unified: Table 11]
725ncEffectUnlink :: ProcessId -> Identifier -> NC ()
849ncEffectUnlink :: ProcessId -> Identifier -> NC ()
@@ -733,7 +857,7 @@ ncEffectUnlink from them = do
733 postAsMessage from $ DidUnlinkNode nid
857 postAsMessage from $ DidUnlinkNode nid
734 SendPortIdentifier cid ->
858 SendPortIdentifier cid ->
735 postAsMessage from $ DidUnlinkPort cid
859 postAsMessage from $ DidUnlinkPort cid
736− modify' $ linksFor them ^: Set.delete from
860+ modify' $ links ^: BiMultiMap.delete them from ()
738-- [Unified: Table 11]
862-- [Unified: Table 11]
739ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
863ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
@@ -741,7 +865,7 @@ ncEffectUnmonitor from ref = do
741 node <- ask
865 node <- ask
742 when (isLocal node (ProcessIdentifier from)) $
866 when (isLocal node (ProcessIdentifier from)) $
743 postAsMessage from $ DidUnmonitor ref
867 postAsMessage from $ DidUnmonitor ref
744− modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
868+ modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref
746-- [Unified: Table 12]
870-- [Unified: Table 12]
747ncEffectDied :: Identifier -> DiedReason -> NC ()
871ncEffectDied :: Identifier -> DiedReason -> NC ()
@@ -756,7 +880,7 @@ ncEffectDied ident reason = do
756 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
880 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
758 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
882 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
759− forM_ uss $ \us ->
883+ forM_ uss $ \(us, _) ->
760 when (localOnly <= isLocal node (ProcessIdentifier us)) $
884 when (localOnly <= isLocal node (ProcessIdentifier us)) $
761 notifyDied us them reason Nothing
885 notifyDied us them reason Nothing
@@ -765,9 +889,35 @@ ncEffectDied ident reason = do
765 when (localOnly <= isLocal node (ProcessIdentifier us)) $
889 when (localOnly <= isLocal node (ProcessIdentifier us)) $
766 notifyDied us them reason (Just ref)
890 notifyDied us them reason (Just ref)
892+ -- Notify remote nodes that the process died so it can be removed from monitor
894+ mapM_ (forwardDeath node) $
895+ [ nid | ProcessIdentifier pid <- [ident]
896+ , i <- Set.toList $ Set.union
897+ (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedLinks)
898+ (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedMons)
899+ , let nid = nodeOf i
900+ , nid /= localNodeId node
903+ -- Delete monitors in the local node.
904+ let deleteDeads :: (Ord a, Ord v)
905+ => BiMultiMap a ProcessId v -> BiMultiMap a ProcessId v
906+ deleteDeads = case ident of
907+ -- deleteAllBy2nd is faster than partitionWithKeyBy2nd
908+ ProcessIdentifier pid -> BiMultiMap.deleteAllBy2nd pid
909+ _ -> snd . BiMultiMap.partitionWithKeyBy2nd
910+ (\pid _ -> ident `impliesDeathOf` ProcessIdentifier pid)
911+ unaffectedLinks' = deleteDeads unaffectedLinks
912+ unaffectedMons' = deleteDeads unaffectedMons
768− modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
914+ modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
916+ -- we now consider all labels for this identifier unregistered
770− modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
917+ let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid
918+ (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)
919+ mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)
920+ modify' $ registeredHere ^= keepNames
772 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
922 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
773 mapM (\(pid,nidlist) ->
923 mapM (\(pid,nidlist) ->
@@ -775,20 +925,15 @@ ncEffectDied ident reason = do
776 do forM_ nidlist $ \(nid,_) ->
926 do forM_ nidlist $ \(nid,_) ->
777 when (not $ isLocal node (NodeIdentifier nid))
927 when (not $ isLocal node (NodeIdentifier nid))
778− (forwardNameDeath node nid)
928+ (forwardDeath node nid)
779 return Nothing
929 return Nothing
780 False -> return $ Just (pid,nidlist) )
930 False -> return $ Just (pid,nidlist) )
781 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
931 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
783− forwardNameDeath node nid =
933+ forwardDeath node nid = ncSendToNode nid
784− liftIO $ sendBinary node
785− (NodeIdentifier $ localNodeId node)
786− (NodeIdentifier $ nid)
787− WithImplicitReconnect
789− { ctrlMsgSender = NodeIdentifier (localNodeId node)
934+ NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
790− , ctrlMsgSignal = Died ident reason
935+ , ctrlMsgSignal = Died ident reason
793-- [Unified: Table 13]
938-- [Unified: Table 13]
794ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
939ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
@@ -796,17 +941,13 @@ ncEffectSpawn pid cProc ref = do
796 mProc <- unClosure cProc
941 mProc <- unClosure cProc
797 -- If the closure does not exist, we spawn a process that throws an exception
942 -- If the closure does not exist, we spawn a process that throws an exception
798 -- This allows the remote node to find out what's happening
943 -- This allows the remote node to find out what's happening
799− -- TODO:
944+ -- TODO:
800 let proc = case mProc of
945 let proc = case mProc of
801 Left err -> fail $ "Error: Could not resolve closure: " ++ err
946 Left err -> fail $ "Error: Could not resolve closure: " ++ err
802 Right p -> p
947 Right p -> p
803 node <- ask
948 node <- ask
804 pid' <- liftIO $ forkProcess node proc
949 pid' <- liftIO $ forkProcess node proc
805− liftIO $ sendMessage node
950+ ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
806− (NodeIdentifier (localNodeId node))
807− (ProcessIdentifier pid)
808− WithImplicitReconnect
811-- Unified semantics does not explicitly describe how to implement 'register',
952-- Unified semantics does not explicitly describe how to implement 'register',
812-- but mentions it's "very similar to nsend" (Table 14)
953-- but mentions it's "very similar to nsend" (Table 14)
@@ -824,17 +965,19 @@ ncEffectRegister from label atnode mPid reregistration = do
824 return $ (isNothing currentVal /= reregistration) &&
965 return $ (isNothing currentVal /= reregistration) &&
825 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
966 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
826 if isLocal node (NodeIdentifier atnode)
967 if isLocal node (NodeIdentifier atnode)
827− then do when (isOk) $
968+ then do when isOk $
828 do modify' $ registeredHereFor label ^= mPid
969 do modify' $ registeredHereFor label ^= mPid
829 updateRemote node currentVal mPid
970 updateRemote node currentVal mPid
830 case mPid of
971 case mPid of
831− (Just p) -> liftIO $ trace node (TraceEvRegistered p label)
972+ (Just p) -> do
974+ then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
976+ liftIO $ trace node (MxRegistered p label)
832− Nothing -> liftIO $ trace node (TraceEvUnRegistered (fromJust currentVal) label)
977+ Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
833− liftIO $ sendMessage node
978+ newVal <- gets (^. registeredHereFor label)
834− (NodeIdentifier (localNodeId node))
979+ ncSendToProcess from $ unsafeCreateUnencodedMessage $
835− (ProcessIdentifier from)
836− WithImplicitReconnect
837− (RegisterReply label isOk)
980+ RegisterReply label isOk newVal
838 else let operation =
981 else let operation =
839 case reregistration of
982 case reregistration of
840 True -> flip decList
983 True -> flip decList
@@ -864,40 +1007,36 @@ ncEffectRegister from label atnode mPid reregistration = do
864 decList (x:xs) tag = x:decList xs tag
1007 decList (x:xs) tag = x:decList xs tag
865 forward node to reg =
1008 forward node to reg =
866 when (not $ isLocal node (NodeIdentifier to)) $
1009 when (not $ isLocal node (NodeIdentifier to)) $
867− liftIO $ sendBinary node
868− (ProcessIdentifier from)
870− WithImplicitReconnect
872− { ctrlMsgSender = ProcessIdentifier from
1010+ ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
873− , ctrlMsgSignal = reg
1011+ , ctrlMsgSignal = reg
877-- Unified semantics does not explicitly describe 'whereis'
1015-- Unified semantics does not explicitly describe 'whereis'
878ncEffectWhereIs :: ProcessId -> String -> NC ()
1016ncEffectWhereIs :: ProcessId -> String -> NC ()
879ncEffectWhereIs from label = do
1017ncEffectWhereIs from label = do
881 mPid <- gets (^. registeredHereFor label)
1018 mPid <- gets (^. registeredHereFor label)
882− liftIO $ sendMessage node
883− (NodeIdentifier (localNodeId node))
884− (ProcessIdentifier from)
885− WithImplicitReconnect
886− (WhereIsReply label mPid)
1019+ ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
888-- [Unified: Table 14]
1021-- [Unified: Table 14]
889ncEffectNamedSend :: String -> Message -> NC ()
1022ncEffectNamedSend :: String -> Message -> NC ()
890ncEffectNamedSend label msg = do
1023ncEffectNamedSend label msg = do
891 mPid <- gets (^. registeredHereFor label)
1024 mPid <- gets (^. registeredHereFor label)
892 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
1025 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
893− forM_ mPid $ \pid -> postMessage pid msg
1026+ forM_ mPid $ \to ->
1027+ -- If this is a trace message we don't trace it to avoid entering a loop
1028+ -- where trace messages produce more trace messages.
1029+ ncSendToProcessAndTrace (label /= "trace.logger") to msg
895-- [Issue #DP-20]
1031-- [Issue #DP-20]
896ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
1032ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
897−ncEffectLocalSend node to msg =
1033+ncEffectLocalSend = ncEffectLocalSendAndTrace True
1035+ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
1036+ncEffectLocalSendAndTrace shouldTrace node to msg =
898 liftIO $ withLocalProc node to $ \p -> do
1037 liftIO $ withLocalProc node to $ \p -> do
899 enqueue (processQueue p) msg
1038 enqueue (processQueue p) msg
900− trace node (TraceEvReceived to msg)
1039+ when shouldTrace $ trace node (MxReceived to msg)
902-- [Issue #DP-20]
1041-- [Issue #DP-20]
903ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
1042ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
@@ -915,11 +1054,12 @@ ncEffectLocalPortSend from msg = do
915 -- If ch is Nothing, the process has given up the read end of
1054 -- If ch is Nothing, the process has given up the read end of
916 -- the channel and we simply ignore the incoming message - this
1055 -- the channel and we simply ignore the incoming message - this
917 ch <- deRefWeak chan'
1056 ch <- deRefWeak chan'
918− forM_ ch $ \chan -> deliverChan msg chan
1057+ forM_ ch $ \chan -> deliverChan node from msg chan
919− where deliverChan :: forall a . Message -> TQueue a -> IO ()
1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()
920− deliverChan (UnencodedMessage _ raw) chan' =
1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do
921 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)
922− deliverChan (EncodedMessage _ _) _ =
1062+ deliverChan _ _ (EncodedMessage _ _) _ =
923 -- this will not happen unless someone screws with Primitives.hs
1063 -- this will not happen unless someone screws with Primitives.hs
924 error "invalid local channel delivery"
1064 error "invalid local channel delivery"
@@ -944,46 +1084,56 @@ ncEffectGetInfo from pid =
944 them = (ProcessIdentifier pid)
1084 them = (ProcessIdentifier pid)
946 node <- ask
1086 node <- ask
947− mProc <- liftIO $
1087+ mProc <- liftIO $ withValidLocalState node
948− withMVar (localState node) $ return . (^. localProcessWithId lpid)
1088+ $ return . (^. localProcessWithId lpid)
949 case mProc of
1089 case mProc of
950 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1090 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
951− from node (ProcessInfoNone DiedUnknownId)
1091+ from (ProcessInfoNone DiedUnknownId)
952− Just _ -> do
1092+ Just proc -> do
953− itsLinks <- gets (^. linksFor them)
1093+ itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>
954− itsMons <- gets (^. monitorsFor them)
1094+ gets (^. links)
1095+ itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)
955 registered <- gets (^. registeredHere)
1096 registered <- gets (^. registeredHere)
1097+ size <- liftIO $ queueSize $ processQueue $ proc
957 let reg = registeredNames registered
1099 let reg = registeredNames registered
958 dispatch (isLocal node (ProcessIdentifier from))
1100 dispatch (isLocal node (ProcessIdentifier from))
961 ProcessInfo {
1102 ProcessInfo {
962 infoNode = (processNodeId pid)
1103 infoNode = (processNodeId pid)
963 , infoRegisteredNames = reg
1104 , infoRegisteredNames = reg
964− -- we cannot populate this field
965− , infoMessageQueueLength = Nothing
1105+ , infoMessageQueueLength = size
966− , infoMonitors = Set.toList itsMons
1106+ , infoMonitors = Set.toList itsMons
967− , infoLinks = Set.toList itsLinks
1107+ , infoLinks = Set.toList itsLinks
969− where dispatch :: (Serializable a, Show a)
1109+ where dispatch :: (Serializable a)
971 -> ProcessId
1111 -> ProcessId
974 -> NC ()
1113 -> NC ()
975− dispatch True dest _ pInfo = postAsMessage dest $ pInfo
1114+ dispatch True dest pInfo = postAsMessage dest $ pInfo
976− dispatch False dest node pInfo = do
1115+ dispatch False dest pInfo =
977− liftIO $ sendMessage node
1116+ ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
978− (NodeIdentifier (localNodeId node))
979− (ProcessIdentifier dest)
980− WithImplicitReconnect
983 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1118 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
984 then (k:ks)
1119 then (k:ks)
985 else ks) []
1120 else ks) []
1122+ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1123+ncEffectGetNodeStats from _nid = do
1125+ ncState <- StateT.get
1126+ nodeState <- liftIO $ withValidLocalState node return
1129+ nodeStatsNode = localNodeId node
1130+ , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1131+ , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors
1132+ , nodeStatsLinks = BiMultiMap.size $ ncState ^. links
1133+ , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1135+ postAsMessage from stats
987--------------------------------------------------------------------------------
1137--------------------------------------------------------------------------------
988-- Auxiliary --
1138-- Auxiliary --
989--------------------------------------------------------------------------------
1139--------------------------------------------------------------------------------
@@ -1010,14 +1160,10 @@ notifyDied dest src reason mRef = do
1010 throwException dest $ PortLinkException pid reason
1160 throwException dest $ PortLinkException pid reason
1011 (False, _, _) ->
1161 (False, _, _) ->
1012 -- The change in sender comes from [Unified: Table 10]
1162 -- The change in sender comes from [Unified: Table 10]
1013− liftIO $ sendBinary node
1014− (NodeIdentifier $ localNodeId node)
1015− (NodeIdentifier $ processNodeId dest)
1163+ ncSendToNode (processNodeId dest) $ NCMsg
1016− WithImplicitReconnect
1018− { ctrlMsgSender = NodeIdentifier (localNodeId node)
1164+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
1019− , ctrlMsgSignal = Died src reason
1165+ , ctrlMsgSignal = Died src reason
1022-- | [Unified: Table 8]
1168-- | [Unified: Table 8]
1023destNid :: ProcessSignal -> Maybe NodeId
1169destNid :: ProcessSignal -> Maybe NodeId
@@ -1029,6 +1175,7 @@ destNid (Spawn _ _) = Nothing
1029destNid (Register _ _ _ _) = Nothing
1175destNid (Register _ _ _ _) = Nothing
1030destNid (WhereIs _) = Nothing
1176destNid (WhereIs _) = Nothing
1031destNid (NamedSend _ _) = Nothing
1177destNid (NamedSend _ _) = Nothing
1178+destNid (UnreliableSend _ _) = Nothing
1032-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1179-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1033-- then when a local process dies the monitoring/linking machinery will take
1180-- then when a local process dies the monitoring/linking machinery will take
1034-- care of notifying remote nodes
1181-- care of notifying remote nodes
@@ -1036,6 +1183,7 @@ destNid (Died _ _) = Nothing
1036destNid (Kill pid _) = Just $ processNodeId pid
1183destNid (Kill pid _) = Just $ processNodeId pid
1037destNid (Exit pid _) = Just $ processNodeId pid
1184destNid (Exit pid _) = Just $ processNodeId pid
1038destNid (GetInfo pid) = Just $ processNodeId pid
1185destNid (GetInfo pid) = Just $ processNodeId pid
1186+destNid (GetNodeStats nid) = Just nid
1039destNid (LocalSend pid _) = Just $ processNodeId pid
1187destNid (LocalSend pid _) = Just $ processNodeId pid
1040destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1188destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1041destNid (SigShutdown) = Nothing
1189destNid (SigShutdown) = Nothing
@@ -1054,7 +1202,7 @@ unClosure closure = do
1054isValidLocalIdentifier :: Identifier -> NC Bool
1202isValidLocalIdentifier :: Identifier -> NC Bool
1055isValidLocalIdentifier ident = do
1203isValidLocalIdentifier ident = do
1056 node <- ask
1204 node <- ask
1057− liftIO . withMVar (localState node) $ \nSt ->
1205+ liftIO . withValidLocalState node $ \nSt ->
1058 case ident of
1206 case ident of
1059 NodeIdentifier nid ->
1207 NodeIdentifier nid ->
1060 return $ nid == localNodeId node
1208 return $ nid == localNodeId node
@@ -1085,6 +1233,9 @@ postMessage pid msg = do
1085throwException :: Exception e => ProcessId -> e -> NC ()
1233throwException :: Exception e => ProcessId -> e -> NC ()
1086throwException pid e = do
1234throwException pid e = do
1087 node <- ask
1235 node <- ask
1236+ -- throwTo blocks until the exception is received by the target thread.
1237+ -- We cannot easily make it happen asynchronpusly because then 'unlink'
1238+ -- semantics would break.
1088 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1239 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1090withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1241withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
@@ -1092,17 +1243,17 @@ withLocalProc node pid p =
1092 -- By [Unified: table 6, rule missing_process] messages to dead processes
1243 -- By [Unified: table 6, rule missing_process] messages to dead processes
1093 -- can silently be dropped
1244 -- can silently be dropped
1094 let lpid = processLocalId pid in do
1245 let lpid = processLocalId pid in do
1246+ join $ withValidLocalState node $ \vst ->
1095− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1247+ return $ forM_ (vst ^. localProcessWithId lpid) p
1098--------------------------------------------------------------------------------
1249--------------------------------------------------------------------------------
1099-- Accessors --
1250-- Accessors --
1100--------------------------------------------------------------------------------
1251--------------------------------------------------------------------------------
1102−links :: Accessor NCState (Map Identifier (Set ProcessId))
1253+links :: Accessor NCState (BiMultiMap Identifier ProcessId ())
1103links = accessor _links (\ls st -> st { _links = ls })
1254links = accessor _links (\ls st -> st { _links = ls })
1105−monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1256+monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)
1106monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1257monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1108registeredHere :: Accessor NCState (Map String ProcessId)
1259registeredHere :: Accessor NCState (Map String ProcessId)
@@ -1111,12 +1262,6 @@ registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry }
1111registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1262registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1112registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1263registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1114−linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1115−linksFor ident = links >>> DAC.mapDefault Set.empty ident
1117−monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1118−monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1120registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1265registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1121registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1266registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
@@ -1143,10 +1288,12 @@ registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1143-- * the notifications for typed channels to that process.
1288-- * the notifications for typed channels to that process.
1145-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1290-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1291+splitNotif :: (Ord a, Ord v)
1146−splitNotif :: Identifier
1292+ => Identifier
1147− -> Map Identifier a
1293+ -> BiMultiMap Identifier a v
1148− -> (Map Identifier a, Map Identifier a)
1294+ -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)
1149−splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1295+splitNotif ident =
1296+ BiMultiMap.partitionWithKeyBy1st (\k !_v -> ident `impliesDeathOf` k)
1151--------------------------------------------------------------------------------
1298--------------------------------------------------------------------------------
1152-- Auxiliary --
1299-- Auxiliary --