@@ -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,51 +28,73 @@ 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_)
43−import Data.Maybe (isJust, isNothing, catMaybes)
60+import 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 (catch)
78+import qualified Control.Exception as Exception
60−import Control.Concurrent (forkIO)
79+ ( Handler(..)
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
70import Control.Concurrent.Chan (newChan, writeChan, readChan)
96import Control.Concurrent.Chan (newChan, writeChan, readChan)
97+import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
71import Control.Concurrent.STM
98import Control.Concurrent.STM
72 ( atomically
99 ( atomically
@@ -76,6 +103,7 @@ import Control.Distributed.Process.Internal.CQueue
76 , enqueue
103 , enqueue
77 , newCQueue
104 , newCQueue
78 , mkWeakCQueue
105 , mkWeakCQueue
80import qualified Network.Transport as NT
108import qualified Network.Transport as NT
81 ( Transport
109 ( Transport
@@ -87,8 +115,8 @@ import qualified Network.Transport as NT
87 , TransportError(..)
115 , TransportError(..)
88 , address
116 , address
89 , closeEndPoint
117 , closeEndPoint
91 , Connection
118 , Connection
93 , EndPointAddress
121 , EndPointAddress
94 , Reliability(ReliableOrdered)
122 , Reliability(ReliableOrdered)
@@ -105,8 +133,11 @@ import Control.Distributed.Process.Internal.Types
105 , LocalProcessId(..)
133 , LocalProcessId(..)
106 , ProcessId(..)
134 , ProcessId(..)
107 , LocalNode(..)
135 , LocalNode(..)
108− , Tracer(InactiveTracer)
136+ , MxEventBus(..)
109 , LocalNodeState(..)
137 , LocalNodeState(..)
138+ , ValidLocalNodeState(..)
139+ , withValidLocalState
140+ , modifyValidLocalState
110 , LocalProcess(..)
141 , LocalProcess(..)
111 , LocalProcessState(..)
142 , LocalProcessState(..)
112 , Process(..)
143 , Process(..)
@@ -116,9 +147,11 @@ import Control.Distributed.Process.Internal.Types
116 , localPidCounter
147 , localPidCounter
117 , localPidUnique
148 , localPidUnique
118 , localProcessWithId
149 , localProcessWithId
119 , localConnections
151 , localConnections
120 , forever'
152 , forever'
121 , MonitorRef(..)
153 , MonitorRef(..)
154+ , NodeClosedException(..)
122 , ProcessMonitorNotification(..)
155 , ProcessMonitorNotification(..)
123 , NodeMonitorNotification(..)
156 , NodeMonitorNotification(..)
124 , PortMonitorNotification(..)
157 , PortMonitorNotification(..)
@@ -138,54 +171,81 @@ import Control.Distributed.Process.Internal.Types
139 , ProcessInfo(..)
172 , ProcessInfo(..)
140 , ProcessInfoNone(..)
173 , ProcessInfoNone(..)
141 , SendPortId(..)
175 , SendPortId(..)
142 , typedChannelWithId
176 , typedChannelWithId
143 , RegisterReply(..)
177 , RegisterReply(..)
144 , WhereIsReply(..)
178 , WhereIsReply(..)
146 , payloadToMessage
179 , payloadToMessage
147 , createUnencodedMessage
180 , createUnencodedMessage
181+ , unsafeCreateUnencodedMessage
148 , runLocalProcess
182 , runLocalProcess
149 , firstNonReservedProcessId
183 , firstNonReservedProcessId
150− , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
184+ , ImplicitReconnect(WithImplicitReconnect)
186+import Control.Distributed.Process.Management.Internal.Agent
189+import Control.Distributed.Process.Management.Internal.Types
192+import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
152−import Control.Distributed.Process.Internal.Trace
195+import Control.Distributed.Process.Management.Internal.Trace.Tracer
198+import Control.Distributed.Process.Management.Internal.Trace.Types
153 ( TraceArg(..)
199 ( TraceArg(..)
154− , traceFormat
200+ , traceEvent
155− , startTracing
201+ , traceLogFmt
156− , stopTracer
202+ , enableTrace
158import Control.Distributed.Process.Serializable (Serializable)
204import Control.Distributed.Process.Serializable (Serializable)
159import Control.Distributed.Process.Internal.Messaging
205import Control.Distributed.Process.Internal.Messaging
160 ( sendBinary
206 ( sendBinary
163 , closeImplicitReconnections
207 , closeImplicitReconnections
164 , impliesDeathOf
208 , impliesDeathOf
166import Control.Distributed.Process.Internal.Primitives
210import Control.Distributed.Process.Internal.Primitives
167 ( register
211 ( register
169 , receiveWait
212 , receiveWait
171 , sendChan
214 , sendChan
173−import Control.Distributed.Process.Internal.Types (SendPort)
221+import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
174import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
222import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
175import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
223import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
176import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
224import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
177 ( mapMaybe
225 ( mapMaybe
178 , mapDefault
226 , mapDefault
228+import Control.Monad.Catch (try)
229+import GHC.IO (IO(..), unsafeUnmask)
230+import GHC.Base ( maskAsyncExceptions# )
181import 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
183--------------------------------------------------------------------------------
243--------------------------------------------------------------------------------
184-- Initialization --
244-- Initialization --
185--------------------------------------------------------------------------------
245--------------------------------------------------------------------------------
187initRemoteTable :: RemoteTable
247initRemoteTable :: RemoteTable
188−initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable
248+initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
190-- | Initialize a new local node.
250-- | Initialize a new local node.
191newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
251newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
@@ -202,7 +262,7 @@ newLocalNode transport rtable = do
202createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
262createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
203createBareLocalNode endPoint rtable = do
263createBareLocalNode endPoint rtable = do
204 unq <- randomIO
264 unq <- randomIO
205− state <- newMVar LocalNodeState
265+ state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
206 { _localProcesses = Map.empty
266 { _localProcesses = Map.empty
207 , _localPidCounter = firstNonReservedProcessId
267 , _localPidCounter = firstNonReservedProcessId
208 , _localPidUnique = unq
268 , _localPidUnique = unq
@@ -213,22 +273,19 @@ createBareLocalNode endPoint rtable = do
213 , localEndPoint = endPoint
273 , localEndPoint = endPoint
214 , localState = state
274 , localState = state
215 , localCtrlChan = ctrlChan
275 , localCtrlChan = ctrlChan
216− , localTracer = InactiveTracer
276+ , localEventBus = MxEventBusInitialising
217 , remoteTable = rtable
277 , remoteTable = rtable
219− tracedNode <- startTracing node
279+ tracedNode <- startMxAgent node
221− ncGo <- newEmptyMVar
281+ -- Once the NC terminates, the endpoint isn't much use,
222− ncTid <- forkIO $ do
282+ void $ forkIO $ Exception.finally (runNodeController tracedNode)
223− tid <- takeMVar ncGo
283+ (NT.closeEndPoint (localEndPoint node))
224− (runNodeController tracedNode
225− `Exception.catch` \(e :: SomeException) -> throwTo tid e)
227− evTid <- forkIO $ do
285+ -- whilst a closed/failing endpoint will terminate the NC
228− (handleIncomingMessages tracedNode >> (stopNC node))
286+ void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
229− `Exception.catch` \(e :: SomeException) -> throwTo ncTid e
287+ (stopNC node)
232 return tracedNode
289 return tracedNode
234 stopNC node =
291 stopNC node =
@@ -237,48 +294,100 @@ createBareLocalNode endPoint rtable = do
237 , ctrlMsgSignal = SigShutdown
294 , ctrlMsgSignal = SigShutdown
297+startMxAgent :: LocalNode -> IO LocalNode
298+startMxAgent node = do
299+ -- see note [tracer/forkProcess races]
300+ let fork = forkProcess node
301+ mv <- MVar.newEmptyMVar
302+ pid <- fork $ mxAgentController fork mv
303+ (tracer', wqRef, mxNew') <- MVar.takeMVar mv
304+ return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
306+startDefaultTracer :: LocalNode -> IO ()
307+startDefaultTracer node' = do
308+ let t = localEventBus node'
310+ MxEventBus _ (Tracer pid _) _ _ -> do
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
240-- | Start and register the service processes on a node
319-- | Start and register the service processes on a node
241−-- (for now, this is only the logger)
242startServiceProcesses :: LocalNode -> IO ()
320startServiceProcesses :: LocalNode -> IO ()
243startServiceProcesses node = do
321startServiceProcesses node = do
322+ -- tracing /spawns/ relies on the tracer being enabled, but we start
323+ -- the default tracer first, even though it might @nsend@ to the logger
324+ -- before /that/ process has started - this is a totally harmless race
325+ -- however, so we deliberably ignore it
326+ startDefaultTracer node
244 logger <- forkProcess node loop
327 logger <- forkProcess node loop
245− runProcess node $ register "logger" logger
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.
333+ register "trace.logger" logger
247− loop = do
335+ loop = do
248− receiveWait
336+ receiveWait
249− [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
337+ [ match $ \(SayMessage time pid string) -> do
250− liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
338+ let time' = formatTime defaultTimeLocale "%c" time
251− loop
339+ liftIO . hPutStrLn stderr $ time' ++ " " ++ show pid ++ ": " ++ string
252− , match $ \((time, string) :: (String, String)) -> do
340+ loop
253− -- this is a 'trace' message from the local node tracer
341+ , match $ \((time, string) :: (String, String)) -> do
254− liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
342+ -- this is a 'trace' message from the local node tracer
255− loop
343+ liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
256− , match $ \(ch :: SendPort ()) -> -- a shutdown request
344+ loop
257− sendChan ch ()
345+ , match $ \(ch :: SendPort ()) -> -- a shutdown request
258− ]
346+ sendChan ch ()
260−-- | Force-close a local node
348+
261−--
349+-- | Force-close a local node, killing all processes on that node.
262−-- TODO: for now we just close the associated endpoint
263closeLocalNode :: LocalNode -> IO ()
350closeLocalNode :: LocalNode -> IO ()
264−closeLocalNode node =
351+closeLocalNode node = do
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').
265 NT.closeEndPoint (localEndPoint node)
368 NT.closeEndPoint (localEndPoint node)
267-- | 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
268runProcess :: LocalNode -> Process () -> IO ()
371runProcess :: LocalNode -> Process () -> IO ()
269runProcess node proc = do
372runProcess node proc = do
270 done <- newEmptyMVar
373 done <- newEmptyMVar
271− void $ forkProcess node (proc `finally` liftIO (putMVar done ()))
374+ -- TODO; When forkProcess inherits the masking state, protect the forked
272− takeMVar done
375+ -- thread against async exceptions that could occur before 'try' is evaluated.
376+ void $ forkProcess node $ try proc >>= liftIO . putMVar done
377+ takeMVar done >>= either (throwIO :: SomeException -> IO a) return
274-- | Spawn a new process on a local node
379-- | Spawn a new process on a local node
275forkProcess :: LocalNode -> Process () -> IO ProcessId
380forkProcess :: LocalNode -> Process () -> IO ProcessId
276−forkProcess node proc = modifyMVar (localState node) startProcess
381+forkProcess node proc = do
382+ ms <- getMaskingState
383+ modifyMVarMasked (localState node) (startProcess ms)
278− startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
385+ startProcess :: MaskingState
279− startProcess st = do
386+ -> LocalNodeState
280− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
387+ -> IO (LocalNodeState, ProcessId)
281− , lpidUnique = st ^. localPidUnique
388+ startProcess ms (LocalNodeValid vst) = do
389+ let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
390+ , lpidUnique = vst ^. localPidUnique
283 let pid = ProcessId { processNodeId = localNodeId node
392 let pid = ProcessId { processNodeId = localNodeId node
284 , processLocalId = lpid
393 , processLocalId = lpid
@@ -298,43 +407,81 @@ forkProcess node proc = modifyMVar (localState node) startProcess
298 , processThread = tid
407 , processThread = tid
299 , processNode = node
408 , processNode = node
301− tid' <- forkIO $ do
410+ -- Rewrite this code when this is fixed:
302− reason <- Exception.catch
411+ -- https://ghc.haskell.org/trac/ghc/ticket/10149
303− (runLocalProcess lproc proc >> return DiedNormal)
412+ let unmask = case ms of
304− (return . DiedException . (show :: SomeException -> String))
413+ Unmasked -> unblock
414+ MaskedInterruptible -> block
415+ MaskedUninterruptible -> id
416+ tid' <- uninterruptibleMask_ $ forkIO $ do
417+ reason <- Exception.catches
418+ (unmask $ runLocalProcess lproc proc >> return DiedNormal)
419+ [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
420+ mMsg <- unwrapMessage msg :: IO (Maybe String)
422+ Nothing -> return $ DiedException $ show ex
423+ Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
424+ , (Exception.Handler
425+ (return . DiedException . (show :: SomeException -> String)))]
306 -- [Unified: Table 4, rules termination and exiting]
427 -- [Unified: Table 4, rules termination and exiting]
307− 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
308 writeChan (localCtrlChan node) NCMsg
434 writeChan (localCtrlChan node) NCMsg
309 { ctrlMsgSender = ProcessIdentifier pid
435 { ctrlMsgSender = ProcessIdentifier pid
310 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
436 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
312 return (tid', lproc)
438 return (tid', lproc)
440+ -- see note [tracer/forkProcess races]
441+ trace node (MxSpawned pid)
314 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!?
316 newUnique <- randomIO
447 newUnique <- randomIO
317− return ( (localProcessWithId lpid ^= Just lproc)
448+ return ( LocalNodeValid
449+ $ (localProcessWithId lpid ^= Just lproc)
318 . (localPidCounter ^= firstNonReservedProcessId)
450 . (localPidCounter ^= firstNonReservedProcessId)
319 . (localPidUnique ^= newUnique)
451 . (localPidUnique ^= newUnique)
324− return ( (localProcessWithId lpid ^= Just lproc)
456+ return ( LocalNodeValid
457+ $ (localProcessWithId lpid ^= Just lproc)
325 . (localPidCounter ^: (+ 1))
458 . (localPidCounter ^: (+ 1))
462+ startProcess _ LocalNodeClosed =
463+ throwIO $ NodeClosedException $ localNodeId node
330− cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState
465+ cleanupProcess :: ProcessId
331− cleanupProcess pid st = do
466+ -> ValidLocalNodeState
467+ -> IO (ValidLocalNodeState, [NT.Connection])
468+ cleanupProcess pid vst = do
332 let pid' = ProcessIdentifier pid
469 let pid' = ProcessIdentifier pid
333− 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)
334− mapM_ (NT.close . fst) (Map.elems affected)
471+ return ( (localProcessWithId (processLocalId pid) ^= Nothing)
335− return $ (localProcessWithId (processLocalId pid) ^= Nothing)
472+ . (localConnections ^= unaffected)
336− . (localConnections ^= unaffected)
473+ $ vst
337− $ st
474+ , map fst $ Map.elems affected
477+-- note [tracer/forkProcess races]
479+-- Our startTracing function uses forkProcess to start the trace controller
480+-- process, and of course forkProcess attempts to call traceEvent once the
481+-- process has started. This is harmless, as the localEventBus is not updated
482+-- until /after/ the initial forkProcess completes, so the first call to
483+-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
339--------------------------------------------------------------------------------
486--------------------------------------------------------------------------------
340-- Handle incoming messages --
487-- Handle incoming messages --
@@ -344,8 +491,8 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
345data IncomingTarget =
492data IncomingTarget =
347− | ToProc (Weak (CQueue Message))
494+ | ToProc ProcessId (Weak (CQueue Message))
348− | ToChan TypedChannel
495+ | ToChan SendPortId TypedChannel
351data ConnectionState = ConnectionState {
498data ConnectionState = ConnectionState {
@@ -372,6 +519,7 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
373handleIncomingMessages :: LocalNode -> IO ()
520handleIncomingMessages :: LocalNode -> IO ()
374handleIncomingMessages node = go initConnectionState
521handleIncomingMessages node = go initConnectionState
522+ `Exception.catch` \(NodeClosedException _) -> return ()
376 go :: ConnectionState -> IO ()
524 go :: ConnectionState -> IO ()
377 go !st = do
525 go !st = do
@@ -379,28 +527,37 @@ handleIncomingMessages node = go initConnectionState
379 case event of
527 case event of
380 NT.ConnectionOpened cid rel theirAddr ->
528 NT.ConnectionOpened cid rel theirAddr ->
381 if rel == NT.ReliableOrdered
529 if rel == NT.ReliableOrdered
382− then go ( (incomingAt cid ^= Just (theirAddr, Uninit))
530+ then
531+ trace node (MxConnected cid theirAddr)
533+ (incomingAt cid ^= Just (theirAddr, Uninit))
383 . (incomingFrom theirAddr ^: Set.insert cid)
534 . (incomingFrom theirAddr ^: Set.insert cid)
386− else invalidRequest cid st
537+ else invalidRequest cid st $
538+ "attempt to connect with unsupported reliability " ++ show rel
387 NT.Received cid payload ->
539 NT.Received cid payload ->
388 case st ^. incomingAt cid of
540 case st ^. incomingAt cid of
389− Just (_, ToProc weakQueue) -> do
541+ Just (_, ToProc pid weakQueue) -> do
390 mQueue <- deRefWeak weakQueue
542 mQueue <- deRefWeak weakQueue
391 forM_ mQueue $ \queue -> do
543 forM_ mQueue $ \queue -> do
392 -- TODO: if we find that the queue is Nothing, should we remove
544 -- TODO: if we find that the queue is Nothing, should we remove
393 -- it from the NC state? (and same for channels, below)
545 -- it from the NC state? (and same for channels, below)
394 let msg = payloadToMessage payload
546 let msg = payloadToMessage payload
395 enqueue queue msg -- 'enqueue' is strict
547 enqueue queue msg -- 'enqueue' is strict
548+ trace node (MxReceived pid msg)
397− Just (_, ToChan (TypedChannel chan')) -> do
550+ Just (_, ToChan chId (TypedChannel chan')) -> do
398 mChan <- deRefWeak chan'
551 mChan <- deRefWeak chan'
399 -- 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
400 -- the channel and we simply ignore the incoming message
553 -- the channel and we simply ignore the incoming message
401− forM_ mChan $ \chan -> atomically $
554+ forM_ mChan $ \chan -> do
402− -- We make sure the message is fully decoded when it is enqueued
555+ msg' <- atomically $ do
403− writeTQueue chan $! decode (BSL.fromChunks payload)
556+ msg <- return $! decode (BSL.fromChunks payload)
557+ -- We make sure the message is fully decoded when it is enqueued
558+ writeTQueue chan msg
560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'
405 Just (_, ToNode) -> do
562 Just (_, ToNode) -> do
406 let ctrlMsg = decode . BSL.fromChunks $ payload
563 let ctrlMsg = decode . BSL.fromChunks $ payload
@@ -410,37 +567,49 @@ handleIncomingMessages node = go initConnectionState
410 case decode (BSL.fromChunks payload) of
567 case decode (BSL.fromChunks payload) of
411 ProcessIdentifier pid -> do
568 ProcessIdentifier pid -> do
412 let lpid = processLocalId pid
569 let lpid = processLocalId pid
413− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
570+ mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
414 case mProc of
571 case mProc of
415 Just proc ->
572 Just proc ->
416− go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)
573+ go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
417 Nothing ->
574 Nothing ->
418− invalidRequest cid st
575+ -- incoming attempt to connect to unknown process - might
577+ go (incomingAt cid ^= Nothing $ st)
419 SendPortIdentifier chId -> do
578 SendPortIdentifier chId -> do
420 let lcid = sendPortLocalId chId
579 let lcid = sendPortLocalId chId
421 lpid = processLocalId (sendPortProcessId chId)
580 lpid = processLocalId (sendPortProcessId chId)
422− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
581+ mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
423 case mProc of
582 case mProc of
424 Just proc -> do
583 Just proc -> do
425 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
426 case mChannel of
585 case mChannel of
427 Just channel ->
586 Just channel ->
428− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)
429 Nothing ->
588 Nothing ->
430− invalidRequest cid st
589+ invalidRequest cid st $
590+ "incoming attempt to connect to unknown channel of"
591+ ++ " process " ++ show (sendPortProcessId chId)
431 Nothing ->
592 Nothing ->
432− invalidRequest cid st
593+ -- incoming attempt to connect to channel of unknown
594+ -- process - might be dead already
595+ go (incomingAt cid ^= Nothing $ st)
433 NodeIdentifier nid ->
596 NodeIdentifier nid ->
434 if nid == localNodeId node
597 if nid == localNodeId node
435 then go (incomingAt cid ^= Just (src, ToNode) $ st)
598 then go (incomingAt cid ^= Just (src, ToNode) $ st)
436− 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 "
437 Nothing ->
604 Nothing ->
438 invalidRequest cid st
605 invalidRequest cid st
606+ "message received from an unknown connection"
439 NT.ConnectionClosed cid ->
607 NT.ConnectionClosed cid ->
440 case st ^. incomingAt cid of
608 case st ^. incomingAt cid of
441 Nothing ->
609 Nothing ->
442− invalidRequest cid st
610+ invalidRequest cid st "closed unknown connection"
443− Just (src, _) ->
611+ Just (src, _) -> do
612+ trace node (MxDisconnected cid src)
444 go ( (incomingAt cid ^= Nothing)
613 go ( (incomingAt cid ^= Nothing)
445 . (incomingFrom src ^: Set.delete cid)
614 . (incomingFrom src ^: Set.delete cid)
@@ -459,32 +628,30 @@ handleIncomingMessages node = go initConnectionState
461 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
630 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
462− fatal $ "Cloud Haskell fatal error: end point failed: " ++ str
631+ fail $ "Cloud Haskell fatal error: end point failed: " ++ str
463 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
632 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
464− fatal $ "Cloud Haskell fatal error: transport failed: " ++ str
633+ fail $ "Cloud Haskell fatal error: transport failed: " ++ str
465 NT.EndPointClosed ->
634 NT.EndPointClosed ->
466− stopTracer (localTracer node) >> return ()
635+ return ()
467 NT.ReceivedMulticast _ _ ->
636 NT.ReceivedMulticast _ _ ->
468 -- If we received a multicast message, something went horribly wrong
637 -- If we received a multicast message, something went horribly wrong
469 -- and we just give up
638 -- and we just give up
470− fatal "Cloud Haskell fatal error: received unexpected multicast"
639+ fail "Cloud Haskell fatal error: received unexpected multicast"
472− fatal :: String -> IO ()
473− fatal msg = stopTracer (localTracer node) >> fail msg
475− invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
641+ invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
476− invalidRequest cid st = do
642+ invalidRequest cid st msg = do
477 -- 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
478 -- 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
479 -- should close incoming connections (this requires a Transport layer
645 -- should close incoming connections (this requires a Transport layer
480 -- extension).
646 -- extension).
481− traceEventFmtIO node "" [(TraceStr "[network] invalid request: "),
647+ traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"
482− (Trace cid)]
648+ ++ " (" ++ msg ++ "): "
483 go ( incomingAt cid ^= Nothing
651 go ( incomingAt cid ^= Nothing
487− state = localState node
488 endpoint = localEndPoint node
655 endpoint = localEndPoint node
489 ctrlChan = localCtrlChan node
656 ctrlChan = localCtrlChan node
@@ -493,8 +660,9 @@ handleIncomingMessages node = go initConnectionState
493--------------------------------------------------------------------------------
660--------------------------------------------------------------------------------
495runNodeController :: LocalNode -> IO ()
662runNodeController :: LocalNode -> IO ()
496−runNodeController =
663+runNodeController node =
497− runReaderT (evalStateT (unNC nodeController) initNCState)
664+ runReaderT (evalStateT (unNC nodeController) initNCState) node
665+ `Exception.catch` \(NodeClosedException _) -> return ()
499--------------------------------------------------------------------------------
667--------------------------------------------------------------------------------
500-- Internal data types --
668-- Internal data types --
@@ -502,20 +670,26 @@ runNodeController =
503data NCState = NCState
671data NCState = NCState
504 { -- Mapping from remote processes to linked local processes
672 { -- Mapping from remote processes to linked local processes
505− _links :: !(Map Identifier (Set ProcessId))
673+ _links :: !(BiMultiMap Identifier ProcessId ())
506 -- Mapping from remote processes to monitoring local processes
674 -- Mapping from remote processes to monitoring local processes
507− , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
675+ , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef)
508 -- Process registry: names and where they live, mapped to the PIDs
676 -- Process registry: names and where they live, mapped to the PIDs
509 , _registeredHere :: !(Map String ProcessId)
677 , _registeredHere :: !(Map String ProcessId)
510 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
678 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
513newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
681newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
514− deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
682+ deriving ( Applicative
686+ , MonadState NCState
687+ , MonadReader LocalNode
516initNCState :: NCState
690initNCState :: NCState
517−initNCState = NCState { _links = Map.empty
691+initNCState = NCState { _links = BiMultiMap.empty
518− , _monitors = Map.empty
692+ , _monitors = BiMultiMap.empty
519 , _registeredHere = Map.empty
693 , _registeredHere = Map.empty
520 , _registeredOnNodes = Map.empty
694 , _registeredOnNodes = Map.empty
@@ -531,34 +705,60 @@ instance Show ProcessKillException where
531 show (ProcessKillException pid reason) =
705 show (ProcessKillException pid reason) =
532 "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
534--------------------------------------------------------------------------------
735--------------------------------------------------------------------------------
535-- Tracing/Debugging --
736-- Tracing/Debugging --
536--------------------------------------------------------------------------------
737--------------------------------------------------------------------------------
538−-- [Issue #104]
739+-- [Issue #104 / DP-13]
540traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
741traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
541traceNotifyDied node ident reason =
742traceNotifyDied node ident reason =
542− case reason of
743+ -- TODO: sendPortDied notifications
543− DiedNormal -> return ()
744+ liftIO $ withLocalTracer node $ \t ->
544− _ -> traceNcEventFmt node " "
745+ case ident of
545− [(TraceStr "[node-controller]"),
746+ (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
546− (Trace ident),
747+ (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
547− (Trace reason)]
748+ _ -> return ()
549−traceNcEventFmt :: LocalNode -> String -> [TraceArg] -> NC ()
550−traceNcEventFmt node fmt args =
551− liftIO $ traceEventFmtIO node fmt args
553traceEventFmtIO :: LocalNode
750traceEventFmtIO :: LocalNode
554 -> String
751 -> String
555 -> [TraceArg]
752 -> [TraceArg]
557traceEventFmtIO node fmt args =
754traceEventFmtIO node fmt args =
558− withLocalTracer node $ \t -> traceFormat t fmt args
755+ withLocalTracer node $ \t -> traceLogFmt t fmt args
560−withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
757+trace :: LocalNode -> MxEvent -> IO ()
561−withLocalTracer node act = act (localTracer node)
758+trace node ev = withLocalTracer node $ \t -> traceEvent t ev
760+withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
761+withLocalTracer node act = act (localEventBus node)
563--------------------------------------------------------------------------------
763--------------------------------------------------------------------------------
564-- Core functionality --
764-- Core functionality --
@@ -574,11 +774,7 @@ nodeController = do
574 -- [Unified: Table 7, rule nc_forward]
774 -- [Unified: Table 7, rule nc_forward]
575 case destNid (ctrlMsgSignal msg) of
775 case destNid (ctrlMsgSignal msg) of
576 Just nid' | nid' /= localNodeId node ->
776 Just nid' | nid' /= localNodeId node ->
577− liftIO $ sendBinary node
777+ ncSendToNode nid' msg
579− (NodeIdentifier nid')
580− WithImplicitReconnect
583 return ()
779 return ()
@@ -599,10 +795,12 @@ nodeController = do
599 ncEffectRegister from label atnode pid force
795 ncEffectRegister from label atnode pid force
600 NCMsg (ProcessIdentifier from) (WhereIs label) ->
796 NCMsg (ProcessIdentifier from) (WhereIs label) ->
601 ncEffectWhereIs from label
797 ncEffectWhereIs from label
602− NCMsg from (NamedSend label msg') ->
798+ NCMsg _ (NamedSend label msg') ->
603− ncEffectNamedSend from label msg'
799+ ncEffectNamedSend label msg'
800+ NCMsg _ (UnreliableSend lpid msg') ->
801+ ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
604 NCMsg _ (LocalSend to msg') ->
802 NCMsg _ (LocalSend to msg') ->
605− ncEffectLocalSend to msg'
803+ ncEffectLocalSend node to msg'
606 NCMsg _ (LocalPortSend to msg') ->
804 NCMsg _ (LocalPortSend to msg') ->
607 ncEffectLocalPortSend to msg'
805 ncEffectLocalPortSend to msg'
608 NCMsg (ProcessIdentifier from) (Kill to reason) ->
806 NCMsg (ProcessIdentifier from) (Kill to reason) ->
@@ -612,7 +810,11 @@ nodeController = do
612 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
810 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
613 ncEffectGetInfo from pid
811 ncEffectGetInfo from pid
614 NCMsg _ SigShutdown ->
812 NCMsg _ SigShutdown ->
615− liftIO $ throwIO ThreadKilled -- seems to make more sense than fail/error
813+ liftIO $ do
814+ NT.closeEndPoint (localEndPoint node)
815+ `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
816+ NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
817+ ncEffectGetNodeStats from nid
616 unexpected ->
818 unexpected ->
617 error $ "nodeController: unexpected message " ++ show unexpected
819 error $ "nodeController: unexpected message " ++ show unexpected
@@ -630,22 +832,18 @@ ncEffectMonitor from them mRef = do
630 case (shouldLink, isLocal node (ProcessIdentifier from)) of
832 case (shouldLink, isLocal node (ProcessIdentifier from)) of
631 (True, _) -> -- [Unified: first rule]
833 (True, _) -> -- [Unified: first rule]
632 case mRef of
834 case mRef of
633− Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
835+ Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref
634− Nothing -> modify' $ linksFor them ^: Set.insert from
836+ Nothing -> modify' $ links ^: BiMultiMap.insert them from ()
635 (False, True) -> -- [Unified: second rule]
837 (False, True) -> -- [Unified: second rule]
636 notifyDied from them DiedUnknownId mRef
838 notifyDied from them DiedUnknownId mRef
637 (False, False) -> -- [Unified: third rule]
839 (False, False) -> -- [Unified: third rule]
638 -- TODO: this is the right sender according to the Unified semantics,
840 -- TODO: this is the right sender according to the Unified semantics,
639 -- but perhaps having 'them' as the sender would make more sense
841 -- but perhaps having 'them' as the sender would make more sense
640 -- (see also: notifyDied)
842 -- (see also: notifyDied)
641− liftIO $ sendBinary node
843+ ncSendToNode (processNodeId from) $ NCMsg
642− (NodeIdentifier $ localNodeId node)
844+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
643− (NodeIdentifier $ processNodeId from)
845+ , ctrlMsgSignal = Died them DiedUnknownId
644− WithImplicitReconnect
846+ }
646− { ctrlMsgSender = NodeIdentifier (localNodeId node)
647− , ctrlMsgSignal = Died them DiedUnknownId
650-- [Unified: Table 11]
848-- [Unified: Table 11]
651ncEffectUnlink :: ProcessId -> Identifier -> NC ()
849ncEffectUnlink :: ProcessId -> Identifier -> NC ()
@@ -659,7 +857,7 @@ ncEffectUnlink from them = do
659 postAsMessage from $ DidUnlinkNode nid
857 postAsMessage from $ DidUnlinkNode nid
660 SendPortIdentifier cid ->
858 SendPortIdentifier cid ->
661 postAsMessage from $ DidUnlinkPort cid
859 postAsMessage from $ DidUnlinkPort cid
662− modify' $ linksFor them ^: Set.delete from
860+ modify' $ links ^: BiMultiMap.delete them from ()
664-- [Unified: Table 11]
862-- [Unified: Table 11]
665ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
863ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
@@ -667,7 +865,7 @@ ncEffectUnmonitor from ref = do
667 node <- ask
865 node <- ask
668 when (isLocal node (ProcessIdentifier from)) $
866 when (isLocal node (ProcessIdentifier from)) $
669 postAsMessage from $ DidUnmonitor ref
867 postAsMessage from $ DidUnmonitor ref
670− modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
868+ modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref
672-- [Unified: Table 12]
870-- [Unified: Table 12]
673ncEffectDied :: Identifier -> DiedReason -> NC ()
871ncEffectDied :: Identifier -> DiedReason -> NC ()
@@ -682,7 +880,7 @@ ncEffectDied ident reason = do
682 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
880 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
684 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
882 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
685− forM_ uss $ \us ->
883+ forM_ uss $ \(us, _) ->
686 when (localOnly <= isLocal node (ProcessIdentifier us)) $
884 when (localOnly <= isLocal node (ProcessIdentifier us)) $
687 notifyDied us them reason Nothing
885 notifyDied us them reason Nothing
@@ -691,9 +889,35 @@ ncEffectDied ident reason = do
691 when (localOnly <= isLocal node (ProcessIdentifier us)) $
889 when (localOnly <= isLocal node (ProcessIdentifier us)) $
692 notifyDied us them reason (Just ref)
890 notifyDied us them reason (Just ref)
694− modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
892+ -- Notify remote nodes that the process died so it can be removed from monitor
696− modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
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
914+ modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
916+ -- we now consider all labels for this identifier unregistered
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
698 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
922 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
699 mapM (\(pid,nidlist) ->
923 mapM (\(pid,nidlist) ->
@@ -701,20 +925,15 @@ ncEffectDied ident reason = do
702 do forM_ nidlist $ \(nid,_) ->
926 do forM_ nidlist $ \(nid,_) ->
703 when (not $ isLocal node (NodeIdentifier nid))
927 when (not $ isLocal node (NodeIdentifier nid))
704− (forwardNameDeath node nid)
928+ (forwardDeath node nid)
705 return Nothing
929 return Nothing
706 False -> return $ Just (pid,nidlist) )
930 False -> return $ Just (pid,nidlist) )
707 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
931 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
709− forwardNameDeath node nid =
933+ forwardDeath node nid = ncSendToNode nid
710− liftIO $ sendBinary node
934+ NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
711− (NodeIdentifier $ localNodeId node)
935+ , ctrlMsgSignal = Died ident reason
712− (NodeIdentifier $ nid)
936+ }
713− WithImplicitReconnect
715− { ctrlMsgSender = NodeIdentifier (localNodeId node)
716− , ctrlMsgSignal = Died ident reason
719-- [Unified: Table 13]
938-- [Unified: Table 13]
720ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
939ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
@@ -722,16 +941,13 @@ ncEffectSpawn pid cProc ref = do
722 mProc <- unClosure cProc
941 mProc <- unClosure cProc
723 -- 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
724 -- This allows the remote node to find out what's happening
943 -- This allows the remote node to find out what's happening
725 let proc = case mProc of
945 let proc = case mProc of
726 Left err -> fail $ "Error: Could not resolve closure: " ++ err
946 Left err -> fail $ "Error: Could not resolve closure: " ++ err
727 Right p -> p
947 Right p -> p
728 node <- ask
948 node <- ask
729 pid' <- liftIO $ forkProcess node proc
949 pid' <- liftIO $ forkProcess node proc
730− liftIO $ sendMessage node
950+ ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
731− (NodeIdentifier (localNodeId node))
732− (ProcessIdentifier pid)
733− WithImplicitReconnect
736-- Unified semantics does not explicitly describe how to implement 'register',
952-- Unified semantics does not explicitly describe how to implement 'register',
737-- but mentions it's "very similar to nsend" (Table 14)
953-- but mentions it's "very similar to nsend" (Table 14)
@@ -749,14 +965,19 @@ ncEffectRegister from label atnode mPid reregistration = do
749 return $ (isNothing currentVal /= reregistration) &&
965 return $ (isNothing currentVal /= reregistration) &&
750 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
966 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
751 if isLocal node (NodeIdentifier atnode)
967 if isLocal node (NodeIdentifier atnode)
752− then do when (isOk) $
968+ then do when isOk $
753 do modify' $ registeredHereFor label ^= mPid
969 do modify' $ registeredHereFor label ^= mPid
754 updateRemote node currentVal mPid
970 updateRemote node currentVal mPid
755− liftIO $ sendMessage node
971+ case mPid of
756− (NodeIdentifier (localNodeId node))
972+ (Just p) -> do
757− (ProcessIdentifier from)
973+ if reregistration
758− WithImplicitReconnect
974+ then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
759− (RegisterReply label isOk)
975+ else return ()
976+ liftIO $ trace node (MxRegistered p label)
977+ Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
978+ newVal <- gets (^. registeredHereFor label)
979+ ncSendToProcess from $ unsafeCreateUnencodedMessage $
980+ RegisterReply label isOk newVal
760 else let operation =
981 else let operation =
761 case reregistration of
982 case reregistration of
762 True -> flip decList
983 True -> flip decList
@@ -774,6 +995,7 @@ ncEffectRegister from label atnode mPid reregistration = do
774 updateRemote _ _ _ = return ()
995 updateRemote _ _ _ = return ()
775 maybeify f Nothing = unmaybeify $ f []
996 maybeify f Nothing = unmaybeify $ f []
776 maybeify f (Just x) = unmaybeify $ f x
997 maybeify f (Just x) = unmaybeify $ f x
777 unmaybeify [] = Nothing
999 unmaybeify [] = Nothing
778 unmaybeify x = Just x
1000 unmaybeify x = Just x
779 incList [] tag = [(tag,1)]
1001 incList [] tag = [(tag,1)]
@@ -785,51 +1007,44 @@ ncEffectRegister from label atnode mPid reregistration = do
785 decList (x:xs) tag = x:decList xs tag
1007 decList (x:xs) tag = x:decList xs tag
786 forward node to reg =
1008 forward node to reg =
787 when (not $ isLocal node (NodeIdentifier to)) $
1009 when (not $ isLocal node (NodeIdentifier to)) $
788− liftIO $ sendBinary node
1010+ ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
789− (ProcessIdentifier from)
1011+ , ctrlMsgSignal = reg
790− (NodeIdentifier to)
1012+ }
791− WithImplicitReconnect
793− { ctrlMsgSender = ProcessIdentifier from
794− , ctrlMsgSignal = reg
798-- Unified semantics does not explicitly describe 'whereis'
1015-- Unified semantics does not explicitly describe 'whereis'
799ncEffectWhereIs :: ProcessId -> String -> NC ()
1016ncEffectWhereIs :: ProcessId -> String -> NC ()
800ncEffectWhereIs from label = do
1017ncEffectWhereIs from label = do
802 mPid <- gets (^. registeredHereFor label)
1018 mPid <- gets (^. registeredHereFor label)
803− liftIO $ sendMessage node
1019+ ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
804− (NodeIdentifier (localNodeId node))
805− (ProcessIdentifier from)
806− WithImplicitReconnect
807− (WhereIsReply label mPid)
809-- [Unified: Table 14]
1021-- [Unified: Table 14]
810−ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
1022+ncEffectNamedSend :: String -> Message -> NC ()
811−ncEffectNamedSend from label msg = do
1023+ncEffectNamedSend label msg = do
813 mPid <- gets (^. registeredHereFor label)
1024 mPid <- gets (^. registeredHereFor label)
814 -- 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)
815− forM_ mPid $ \pid ->
1026+ forM_ mPid $ \to ->
816− liftIO $ sendPayload node
1027+ -- If this is a trace message we don't trace it to avoid entering a loop
817− from
1028+ -- where trace messages produce more trace messages.
818− (ProcessIdentifier pid)
1029+ ncSendToProcessAndTrace (label /= "trace.logger") to msg
820− (messageToPayload msg)
822-- [Issue #DP-20]
1031-- [Issue #DP-20]
823−ncEffectLocalSend :: ProcessId -> Message -> NC ()
1032+ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
824−ncEffectLocalSend = postMessage
1033+ncEffectLocalSend = ncEffectLocalSendAndTrace True
1035+ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
1036+ncEffectLocalSendAndTrace shouldTrace node to msg =
1037+ liftIO $ withLocalProc node to $ \p -> do
1038+ enqueue (processQueue p) msg
1039+ when shouldTrace $ trace node (MxReceived to msg)
826-- [Issue #DP-20]
1041-- [Issue #DP-20]
827ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
1042ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
828−ncEffectLocalPortSend from msg =
1043+ncEffectLocalPortSend from msg = do
829 let pid = sendPortProcessId from
1045 let pid = sendPortProcessId from
830 cid = sendPortLocalId from
1046 cid = sendPortLocalId from
831− in do
1047+ liftIO $ withLocalProc node pid $ \proc -> do
832− withLocalProc pid $ \proc -> do
833 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
1048 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
834 case mChan of
1049 case mChan of
835 -- in the unlikely event we know nothing about this channel id,
1050 -- in the unlikely event we know nothing about this channel id,
@@ -839,11 +1054,12 @@ ncEffectLocalPortSend from msg =
839 -- 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
840 -- the channel and we simply ignore the incoming message - this
1055 -- the channel and we simply ignore the incoming message - this
841 ch <- deRefWeak chan'
1056 ch <- deRefWeak chan'
842− forM_ ch $ \chan -> deliverChan msg chan
1057+ forM_ ch $ \chan -> deliverChan node from msg chan
843− where deliverChan :: forall a . Message -> TQueue a -> IO ()
1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()
844− deliverChan (UnencodedMessage _ raw) chan' =
1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do
845 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
846− deliverChan (EncodedMessage _ _) _ =
1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)
1062+ deliverChan _ _ (EncodedMessage _ _) _ =
847 -- this will not happen unless someone screws with Primitives.hs
1063 -- this will not happen unless someone screws with Primitives.hs
848 error "invalid local channel delivery"
1064 error "invalid local channel delivery"
@@ -868,46 +1084,56 @@ ncEffectGetInfo from pid =
868 them = (ProcessIdentifier pid)
1084 them = (ProcessIdentifier pid)
870 node <- ask
1086 node <- ask
871− mProc <- liftIO $
1087+ mProc <- liftIO $ withValidLocalState node
872− withMVar (localState node) $ return . (^. localProcessWithId lpid)
1088+ $ return . (^. localProcessWithId lpid)
873 case mProc of
1089 case mProc of
874 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1090 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
875− from node (ProcessInfoNone DiedUnknownId)
1091+ from (ProcessInfoNone DiedUnknownId)
876− Just _ -> do
1092+ Just proc -> do
877− itsLinks <- gets (^. linksFor them)
1093+ itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>
878− itsMons <- gets (^. monitorsFor them)
1094+ gets (^. links)
1095+ itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)
879 registered <- gets (^. registeredHere)
1096 registered <- gets (^. registeredHere)
1097+ size <- liftIO $ queueSize $ processQueue $ proc
881 let reg = registeredNames registered
1099 let reg = registeredNames registered
882 dispatch (isLocal node (ProcessIdentifier from))
1100 dispatch (isLocal node (ProcessIdentifier from))
885 ProcessInfo {
1102 ProcessInfo {
886 infoNode = (processNodeId pid)
1103 infoNode = (processNodeId pid)
887 , infoRegisteredNames = reg
1104 , infoRegisteredNames = reg
888− -- we cannot populate this field
1105+ , infoMessageQueueLength = size
889− , infoMessageQueueLength = Nothing
1106+ , infoMonitors = Set.toList itsMons
890− , infoMonitors = Set.toList itsMons
1107+ , infoLinks = Set.toList itsLinks
891− , infoLinks = Set.toList itsLinks
893− where dispatch :: (Serializable a, Show a)
1109+ where dispatch :: (Serializable a)
895 -> ProcessId
1111 -> ProcessId
898 -> NC ()
1113 -> NC ()
899− dispatch True dest _ pInfo = postAsMessage dest $ pInfo
1114+ dispatch True dest pInfo = postAsMessage dest $ pInfo
900− dispatch False dest node pInfo = do
1115+ dispatch False dest pInfo =
901− liftIO $ sendMessage node
1116+ ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
902− (NodeIdentifier (localNodeId node))
903− (ProcessIdentifier dest)
904− WithImplicitReconnect
907 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1118 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
908 then (k:ks)
1119 then (k:ks)
909 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
911--------------------------------------------------------------------------------
1137--------------------------------------------------------------------------------
912-- Auxiliary --
1138-- Auxiliary --
913--------------------------------------------------------------------------------
1139--------------------------------------------------------------------------------
@@ -934,14 +1160,10 @@ notifyDied dest src reason mRef = do
934 throwException dest $ PortLinkException pid reason
1160 throwException dest $ PortLinkException pid reason
935 (False, _, _) ->
1161 (False, _, _) ->
936 -- The change in sender comes from [Unified: Table 10]
1162 -- The change in sender comes from [Unified: Table 10]
937− liftIO $ sendBinary node
1163+ ncSendToNode (processNodeId dest) $ NCMsg
938− (NodeIdentifier $ localNodeId node)
1164+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
939− (NodeIdentifier $ processNodeId dest)
1165+ , ctrlMsgSignal = Died src reason
940− WithImplicitReconnect
1166+ }
942− { ctrlMsgSender = NodeIdentifier (localNodeId node)
943− , ctrlMsgSignal = Died src reason
946-- | [Unified: Table 8]
1168-- | [Unified: Table 8]
947destNid :: ProcessSignal -> Maybe NodeId
1169destNid :: ProcessSignal -> Maybe NodeId
@@ -953,6 +1175,7 @@ destNid (Spawn _ _) = Nothing
953destNid (Register _ _ _ _) = Nothing
1175destNid (Register _ _ _ _) = Nothing
954destNid (WhereIs _) = Nothing
1176destNid (WhereIs _) = Nothing
955destNid (NamedSend _ _) = Nothing
1177destNid (NamedSend _ _) = Nothing
1178+destNid (UnreliableSend _ _) = Nothing
956-- 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,
957-- 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
958-- care of notifying remote nodes
1181-- care of notifying remote nodes
@@ -960,6 +1183,7 @@ destNid (Died _ _) = Nothing
960destNid (Kill pid _) = Just $ processNodeId pid
1183destNid (Kill pid _) = Just $ processNodeId pid
961destNid (Exit pid _) = Just $ processNodeId pid
1184destNid (Exit pid _) = Just $ processNodeId pid
962destNid (GetInfo pid) = Just $ processNodeId pid
1185destNid (GetInfo pid) = Just $ processNodeId pid
1186+destNid (GetNodeStats nid) = Just nid
963destNid (LocalSend pid _) = Just $ processNodeId pid
1187destNid (LocalSend pid _) = Just $ processNodeId pid
964destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1188destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
965destNid (SigShutdown) = Nothing
1189destNid (SigShutdown) = Nothing
@@ -978,7 +1202,7 @@ unClosure closure = do
978isValidLocalIdentifier :: Identifier -> NC Bool
1202isValidLocalIdentifier :: Identifier -> NC Bool
979isValidLocalIdentifier ident = do
1203isValidLocalIdentifier ident = do
980 node <- ask
1204 node <- ask
981− liftIO . withMVar (localState node) $ \nSt ->
1205+ liftIO . withValidLocalState node $ \nSt ->
982 case ident of
1206 case ident of
983 NodeIdentifier nid ->
1207 NodeIdentifier nid ->
984 return $ nid == localNodeId node
1208 return $ nid == localNodeId node
@@ -1003,30 +1227,33 @@ postAsMessage pid = postMessage pid . createUnencodedMessage
1004postMessage :: ProcessId -> Message -> NC ()
1228postMessage :: ProcessId -> Message -> NC ()
1005postMessage pid msg = do
1229postMessage pid msg = do
1006− withLocalProc pid $ \p -> enqueue (processQueue p) msg
1230+ node <- ask
1231+ liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1008throwException :: Exception e => ProcessId -> e -> NC ()
1233throwException :: Exception e => ProcessId -> e -> NC ()
1009−throwException pid e = withLocalProc pid $ \p ->
1234+throwException pid e = do
1010− throwTo (processThread p) e
1012−withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
1013−withLocalProc pid p = do
1014 node <- ask
1235 node <- ask
1015− liftIO $ do
1236+ -- throwTo blocks until the exception is received by the target thread.
1016− -- By [Unified: table 6, rule missing_process] messages to dead processes
1237+ -- We cannot easily make it happen asynchronpusly because then 'unlink'
1017− -- can silently be dropped
1238+ -- semantics would break.
1018− let lpid = processLocalId pid
1239+ liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1019− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1240+
1020− forM_ mProc p
1241+withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1242+withLocalProc node pid p =
1243+ -- By [Unified: table 6, rule missing_process] messages to dead processes
1244+ -- can silently be dropped
1245+ let lpid = processLocalId pid in do
1246+ join $ withValidLocalState node $ \vst ->
1247+ return $ forM_ (vst ^. localProcessWithId lpid) p
1022--------------------------------------------------------------------------------
1249--------------------------------------------------------------------------------
1023-- Accessors --
1250-- Accessors --
1024--------------------------------------------------------------------------------
1251--------------------------------------------------------------------------------
1026−links :: Accessor NCState (Map Identifier (Set ProcessId))
1253+links :: Accessor NCState (BiMultiMap Identifier ProcessId ())
1027links = accessor _links (\ls st -> st { _links = ls })
1254links = accessor _links (\ls st -> st { _links = ls })
1029−monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1256+monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)
1030monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1257monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1032registeredHere :: Accessor NCState (Map String ProcessId)
1259registeredHere :: Accessor NCState (Map String ProcessId)
@@ -1035,12 +1262,6 @@ registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry }
1035registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1262registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1036registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1263registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1038−linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1039−linksFor ident = links >>> DAC.mapDefault Set.empty ident
1041−monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1042−monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1044registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1265registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1045registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1266registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
@@ -1067,10 +1288,12 @@ registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1067-- * the notifications for typed channels to that process.
1288-- * the notifications for typed channels to that process.
1069-- 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.
1070−splitNotif :: Identifier
1291+splitNotif :: (Ord a, Ord v)
1071− -> Map Identifier a
1292+ => Identifier
1072− -> (Map Identifier a, Map Identifier a)
1293+ -> BiMultiMap Identifier a v
1073−splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1294+ -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)
1296+ BiMultiMap.partitionWithKeyBy1st (\k !_v -> ident `impliesDeathOf` k)
1075--------------------------------------------------------------------------------
1298--------------------------------------------------------------------------------
1076-- Auxiliary --
1299-- Auxiliary --