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