diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,31 @@ +2016-02-18 Facundo Domínguez 0.6.0 + +* Have nsendRemote skip the transport for local communication. +* Unsafe primitives for usend and nsendRemote. +* Stop using the transport for local communication. +* Skip the transport for whereisRemoteAsync and registerRemoteAsync. +* Have nsendRemote skip the transport for local communication. +* Have runProcess forward exceptions. +* Reimport distributed-process-tests. d-p and d-p-test now can be kept in +sync. +* Add a stack.yaml file for building tests and d-p all at once conveniently. +* Implement unreliable forward (uforward). +* Add Functor instance for Match data type +* Have `spawnAsync` not use the transport in the local case. +* Fix monitor race in 'call'. +* Add compatibility with ghc-7.10: support new typeable, loosen deps, write +proper NFData instances, support new TH. +* Kill processes on a local node upon closeLocalNode. +* Fix getNodeStats function, see DP-97 +* Return size of the queue in ProcessInfo. +* Implement unreliable send (usend). +* Implement MonadFix instance for Process. +* Introduce callLocal primitive. +* Prevent message loss due to timeouts in CQueue. +* More informative ProcessRegistrationException. Now includes the identifier +of the process that owns the name, if any. +* Avoid message loop between threads when tracing received messages. + 2015-06-15 Facundo Domínguez 0.5.5 * Fix dependencies. diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,5 +1,5 @@ Name: distributed-process -Version: 0.5.5.1 +Version: 0.6.0 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -21,7 +21,7 @@ You will probably also want to install a Cloud Haskell backend such as distributed-process-simplelocalnet. -Tested-With: GHC==7.4.2 GHC==7.6.3 GHC==7.8.4 GHC==7.10.1 +Tested-With: GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 GHC==7.6.2 Category: Control extra-source-files: ChangeLog @@ -104,7 +104,7 @@ -- Tests are in distributed-process-test package, for convenience. benchmark distributed-process-throughput - Type: exitcode-stdio-1.0 + Type: exitcode-stdio-1.0 Build-Depends: base >= 4.4 && < 5, distributed-process, network-transport-tcp >= 0.3 && < 0.5, diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs --- a/src/Control/Distributed/Process.hs +++ b/src/Control/Distributed/Process.hs @@ -25,6 +25,7 @@ , liftIO -- Reexported for convenience -- * Basic messaging , send + , usend , expect , expectTimeout -- * Channels @@ -39,8 +40,10 @@ , mergePortsRR -- * Unsafe messaging variants , unsafeSend + , unsafeUSend , unsafeSendChan , unsafeNSend + , unsafeNSendRemote , unsafeWrapMessage -- * Advanced messaging , Match @@ -64,6 +67,7 @@ , handleMessage_ , handleMessageIf_ , forward + , uforward , delegate , relay , proxy @@ -149,6 +153,7 @@ -- * Local versions of 'spawn' , spawnLocal , spawnChannelLocal + , callLocal -- * Reconnecting , reconnect , reconnectPort @@ -161,7 +166,12 @@ import Control.Monad.IO.Class (liftIO) import Control.Applicative ((<$>)) import Control.Monad.Reader (ask) -import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) +import Control.Concurrent.MVar + ( MVar + , newEmptyMVar + , takeMVar + , putMVar + ) import Control.Distributed.Static ( Closure , closure @@ -195,6 +205,7 @@ import Control.Distributed.Process.Internal.Primitives ( -- Basic messaging send + , usend , expect -- Channels , newChan @@ -203,8 +214,10 @@ , mergePortsBiased , mergePortsRR , unsafeSend + , unsafeUSend , unsafeSendChan , unsafeNSend + , unsafeNSendRemote -- Advanced messaging , Match , receiveWait @@ -227,6 +240,7 @@ , handleMessage_ , handleMessageIf_ , forward + , uforward , delegate , relay , proxy @@ -302,6 +316,7 @@ , spawnSupervised , call ) +import Control.Exception (SomeException, throw) -- INTERNAL NOTES -- @@ -392,3 +407,15 @@ liftIO $ putMVar mvar sport proc rport takeMVar mvar + +-- | Local version of 'call'. Running a process in this way isolates it from +-- messages sent to the caller process, and also allows silently dropping late +-- or duplicate messages sent to the isolated process after it exits. +-- Silently dropping messages may not always be the best approach. +callLocal :: Process a -> Process a +callLocal proc = mask $ \release -> do + mv <- liftIO newEmptyMVar :: Process (MVar (Either SomeException a)) + child <- spawnLocal $ try (release proc) >>= liftIO . putMVar mv + rs <- liftIO (takeMVar mv) `onException` + (kill child "exception in parent process" >> liftIO (takeMVar mv)) + either throw return rs diff --git a/src/Control/Distributed/Process/Debug.hs b/src/Control/Distributed/Process/Debug.hs --- a/src/Control/Distributed/Process/Debug.hs +++ b/src/Control/Distributed/Process/Debug.hs @@ -83,13 +83,13 @@ -- environment variable accepts the following flags, which enable tracing specific -- event types: -- --- p = trace the spawning of new processes --- d = trace the death of processes --- n = trace registration of names (i.e., named processes) --- u = trace un-registration of names (i.e., named processes) --- s = trace the sending of messages to other processes --- r = trace the receipt of messages from other processes --- l = trace node up/down events +-- * @p@ = trace the spawning of new processes +-- * @d@ = trace the death of processes +-- * @n@ = trace registration of names (i.e., named processes) +-- * @u@ = trace un-registration of names (i.e., named processes) +-- * @s@ = trace the sending of messages to other processes +-- * @r@ = trace the receipt of messages from other processes +-- * @l@ = trace node up/down events -- -- Users of the /simplelocalnet/ Cloud Haskell backend should also note that -- because the trace file option only supports trace output from a single node diff --git a/src/Control/Distributed/Process/Internal/CQueue.hs b/src/Control/Distributed/Process/Internal/CQueue.hs --- a/src/Control/Distributed/Process/Internal/CQueue.hs +++ b/src/Control/Distributed/Process/Internal/CQueue.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE BangPatterns #-} {-# LANGUAGE MagicHash, UnboxedTuples, PatternGuards, ScopedTypeVariables, RankNTypes #-} -- | Concurrent queue for single reader, single writer @@ -10,6 +11,7 @@ , enqueueSTM , dequeue , mkWeakCQueue + , queueSize ) where import Prelude hiding (length, reverse) @@ -17,10 +19,14 @@ ( atomically , STM , TChan + , TVar + , modifyTVar' , tryReadTChan , newTChan + , newTVarIO , writeTChan , readTChan + , readTVarIO , orElse , retry ) @@ -38,6 +44,7 @@ , append ) import Data.Maybe (fromJust) +import Data.Traversable (traverse) import GHC.MVar (MVar(MVar)) import GHC.IO (IO(IO)) import GHC.Prim (mkWeak#) @@ -46,19 +53,22 @@ -- We use a TCHan rather than a Chan so that we have a non-blocking read data CQueue a = CQueue (StrictMVar (StrictList a)) -- Arrived (TChan a) -- Incoming + (TVar Int) -- Queue size newCQueue :: IO (CQueue a) -newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan +newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan <*> newTVarIO 0 -- | Enqueue an element -- -- Enqueue is strict. enqueue :: CQueue a -> a -> IO () -enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a +enqueue c !a = atomically (enqueueSTM c a) -- | Variant of enqueue for use in the STM monad. enqueueSTM :: CQueue a -> a -> STM () -enqueueSTM (CQueue _arrived incoming) !a = writeTChan incoming a +enqueueSTM (CQueue _arrived incoming size) !a = do + writeTChan incoming a + modifyTVar' size succ data BlockSpec = NonBlocking @@ -68,6 +78,7 @@ data MatchOn m a = MatchMsg (m -> Maybe a) | MatchChan (STM a) + deriving (Functor) type MatchChunks m a = [Either [m -> Maybe a] [STM a]] @@ -101,7 +112,7 @@ -> BlockSpec -- ^ Blocking behaviour -> [MatchOn m a] -- ^ List of matches -> IO (Maybe a) -- ^ 'Nothing' only on timeout -dequeue (CQueue arrived incoming) blockSpec matchons = +dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJust $ case blockSpec of Timeout n -> timeout n $ fmap fromJust run _other -> @@ -113,9 +124,16 @@ -- no onException needed _other -> run where + -- Decrement counter is smth is returned from the queue, + -- this is safe to use as method is called under a mask + -- and there is no 'unmasked' operation inside + decrementJust f = + traverse (either return (\x -> decrement >> return x)) =<< f + decrement = atomically $ modifyTVar' size pred + chunks = chunkMatches matchons - run = mask_ $ do + run = do arr <- takeMVar arrived let grabNew xs = do r <- atomically $ tryReadTChan incoming @@ -126,7 +144,7 @@ goCheck chunks arr' waitChans ports on_block = - foldr orElse on_block (map (fmap Just) ports) + foldr orElse on_block (map (fmap (Just . Left)) ports) -- -- First check the MatchChunks against the messages already in the @@ -135,7 +153,7 @@ -- goCheck :: MatchChunks m a -> StrictList m -- messages to check, in this order - -> IO (Maybe a) + -> IO (Maybe (Either a a)) goCheck [] old = goWait old @@ -152,7 +170,7 @@ -- of passing around restore and setting up exception handlers is -- high. So just don't use expensive matchIfs! case checkArrived matches old of - (old', Just r) -> returnOld old' (Just r) + (old', Just r) -> returnOld old' (Just (Right r)) (old', Nothing) -> goCheck rest old' -- use the result list, which is now left-biased @@ -181,6 +199,7 @@ -- Contents of 'arrived' from now on is (old ++ new), and -- messages that arrive are snocced onto new. -- + goWait :: StrictList m -> IO (Maybe (Either a a)) goWait old = do r <- waitIncoming `onException` putMVar arrived old case r of @@ -198,7 +217,7 @@ -- -- Right => message arrived on a channel first -- - Right a -> returnOld old (Just a) + Right a -> returnOld old (Just (Left a)) -- -- A message arrived in the process inbox; check the MatchChunks for @@ -207,7 +226,7 @@ goCheck1 :: MatchChunks m a -> m -- single message to check -> StrictList m -- old messages we have already checked - -> IO (Maybe a) + -> IO (Maybe (Either a a)) goCheck1 [] m old = goWait (Snoc old m) @@ -220,10 +239,10 @@ goCheck1 (Left matches : rest) m old = do case checkMatches matches m of Nothing -> goCheck1 rest m old - Just p -> returnOld old (Just p) + Just p -> returnOld old (Just (Right p)) -- a common pattern for putting back the arrived queue at the end - returnOld :: StrictList m -> Maybe a -> IO (Maybe a) + returnOld :: StrictList m -> Maybe (Either a a) -> IO (Maybe (Either a a)) returnOld old r = do putMVar arrived old; return r -- as a side-effect, this left-biases the list @@ -245,5 +264,8 @@ -- | Weak reference to a CQueue mkWeakCQueue :: CQueue a -> IO () -> IO (Weak (CQueue a)) -mkWeakCQueue m@(CQueue (StrictMVar (MVar m#)) _) f = IO $ \s -> +mkWeakCQueue m@(CQueue (StrictMVar (MVar m#)) _ _) f = IO $ \s -> case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #) + +queueSize :: CQueue a -> IO Int +queueSize (CQueue _ _ size) = readTVarIO size diff --git a/src/Control/Distributed/Process/Internal/Messaging.hs b/src/Control/Distributed/Process/Internal/Messaging.hs --- a/src/Control/Distributed/Process/Internal/Messaging.hs +++ b/src/Control/Distributed/Process/Internal/Messaging.hs @@ -13,10 +13,11 @@ import qualified Data.Map as Map (partitionWithKey, elems) import qualified Data.ByteString.Lazy as BSL (toChunks) import qualified Data.ByteString as BSS (ByteString) -import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_) import Control.Distributed.Process.Serializable () +import Control.Concurrent (forkIO) import Control.Concurrent.Chan (writeChan) +import Control.Exception (mask_) import Control.Monad (unless) import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask) @@ -30,6 +31,9 @@ ) import Control.Distributed.Process.Internal.Types ( LocalNode(localState, localEndPoint, localCtrlChan) + , withValidLocalState + , modifyValidLocalState + , modifyValidLocalState_ , Identifier , localConnections , localConnectionBetween @@ -50,6 +54,7 @@ , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier) ) import Control.Distributed.Process.Serializable (Serializable) +import Data.Foldable (forM_) -------------------------------------------------------------------------------- -- Message sending -- @@ -73,7 +78,7 @@ unless didSend $ do writeChan (localCtrlChan node) NCMsg { ctrlMsgSender = to - , ctrlMsgSignal = Died to DiedDisconnect + , ctrlMsgSignal = Died (NodeIdentifier $ nodeOf to) DiedDisconnect } sendBinary :: Binary a @@ -102,7 +107,7 @@ -> ImplicitReconnect -> IO (Maybe NT.Connection) setupConnBetween node from to implicitReconnect = do - mConn <- NT.connect endPoint + mConn <- NT.connect (localEndPoint node) (nodeAddress . nodeOf $ to) NT.ReliableOrdered NT.defaultConnectHints @@ -113,14 +118,11 @@ Left _ -> return Nothing Right () -> do - modifyMVar_ nodeState $ return . - (localConnectionBetween from to ^= Just (conn, implicitReconnect)) + modifyValidLocalState_ node $ + return . (localConnectionBetween from to ^= Just (conn, implicitReconnect)) return $ Just conn Left _ -> return Nothing - where - endPoint = localEndPoint node - nodeState = localState node connBetween :: LocalNode -> Identifier @@ -128,33 +130,37 @@ -> ImplicitReconnect -> IO (Maybe NT.Connection) connBetween node from to implicitReconnect = do - mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to) + mConn <- withValidLocalState node $ + return . (^. localConnectionBetween from to) case mConn of Just (conn, _) -> return $ Just conn Nothing -> setupConnBetween node from to implicitReconnect - where - nodeState = localState node disconnect :: LocalNode -> Identifier -> Identifier -> IO () -disconnect node from to = - modifyMVar_ (localState node) $ \st -> - case st ^. localConnectionBetween from to of +disconnect node from to = mask_ $ do + mio <- modifyValidLocalState node $ \vst -> + case vst ^. localConnectionBetween from to of Nothing -> - return st + return (vst, return ()) Just (conn, _) -> do - NT.close conn - return (localConnectionBetween from to ^= Nothing $ st) + return ( localConnectionBetween from to ^= Nothing $ vst + , NT.close conn + ) + forM_ mio forkIO closeImplicitReconnections :: LocalNode -> Identifier -> IO () -closeImplicitReconnections node to = - modifyMVar_ (localState node) $ \st -> do +closeImplicitReconnections node to = mask_ $ do + mconns <- modifyValidLocalState node $ \vst -> do let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to' shouldClose _ _ = False - let (affected, unaffected) = Map.partitionWithKey shouldClose (st ^. localConnections) - mapM_ (NT.close . fst) (Map.elems affected) - return (localConnections ^= unaffected $ st) + let (affected, unaffected) = + Map.partitionWithKey shouldClose (vst ^. localConnections) + return ( localConnections ^= unaffected $ vst + , map fst $ Map.elems affected + ) + forM_ mconns $ forkIO . mapM_ NT.close -- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node) -- implies the death of @b@ (for instance, a process on that node) @@ -177,7 +183,11 @@ False --- Send a control message +-- Send a control message. Evaluates the message to HNF before sending it. +-- +-- The message shouldn't produce more errors when further evaluated. If +-- evaluation threw errors the node controller or the receiver would crash when +-- inspecting it. sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node -> ProcessSignal -- ^ Message to send -> Process () @@ -188,7 +198,7 @@ } case mNid of Nothing -> do - liftIO $ writeChan (localCtrlChan (processNode proc)) msg + liftIO $ writeChan (localCtrlChan (processNode proc)) $! msg Just nid -> liftIO $ sendBinary (processNode proc) (ProcessIdentifier (processId proc)) diff --git a/src/Control/Distributed/Process/Internal/Primitives.hs b/src/Control/Distributed/Process/Internal/Primitives.hs --- a/src/Control/Distributed/Process/Internal/Primitives.hs +++ b/src/Control/Distributed/Process/Internal/Primitives.hs @@ -1,6 +1,7 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE BangPatterns #-} @@ -12,6 +13,7 @@ module Control.Distributed.Process.Internal.Primitives ( -- * Basic messaging send + , usend , expect -- * Channels , newChan @@ -21,8 +23,10 @@ , mergePortsRR -- * Unsafe messaging variants , unsafeSend + , unsafeUSend , unsafeSendChan , unsafeNSend + , unsafeNSendRemote -- * Advanced messaging , Match , receiveWait @@ -45,6 +49,7 @@ , handleMessage_ , handleMessageIf_ , forward + , uforward , delegate , relay , proxy @@ -258,6 +263,31 @@ unsafeSend :: Serializable a => ProcessId -> a -> Process () unsafeSend = Unsafe.send +-- | Send a message unreliably. +-- +-- Unlike 'send', this function is insensitive to 'reconnect'. It will +-- try to send the message regardless of the history of connection failures +-- between the nodes. +-- +-- Message passing with 'usend' is ordered for a given sender and receiver +-- if the messages arrive at all. +-- +usend :: Serializable a => ProcessId -> a -> Process () +usend them msg = do + here <- getSelfNode + let there = processNodeId them + if here == there + then sendLocal them msg + else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them) + (createMessage msg) + +-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize +-- the message when the destination process resides on the same local +-- node. Therefore, a local receiver would need to be prepared to cope with any +-- errors resulting from evaluation of the message. +unsafeUSend :: Serializable a => ProcessId -> a -> Process () +unsafeUSend = Unsafe.usend + -- | Wait for a message of a specific type expect :: forall a. Serializable a => Process a expect = receiveWait [match return] @@ -355,6 +385,7 @@ -- | Opaque type used in 'receiveWait' and 'receiveTimeout' newtype Match b = Match { unMatch :: MatchOn Message (Process b) } + deriving (Functor) -- | Test the matches in order against each message in the queue receiveWait :: [Match b] -> Process b @@ -452,6 +483,26 @@ liftIO $ traceEvent (localEventBus node) (MxSent them us msg) +-- | Forward a raw 'Message' to the given 'ProcessId'. +-- +-- Unlike 'forward', this function is insensitive to 'reconnect'. It will +-- try to send the message regardless of the history of connection failures +-- between the nodes. +uforward :: Message -> ProcessId -> Process () +uforward msg them = do + proc <- ask + let node = processNode proc + us = processId proc + nid = localNodeId node + destNode = (processNodeId them) in do + case destNode == nid of + True -> sendCtrlMsg Nothing (LocalSend them msg) + False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) + msg + -- We do not fire the trace event until after the sending is complete; + -- In the remote case, 'sendCtrlMsg' can block in the networking stack. + liftIO $ traceEvent (localEventBus node) + (MxSent them us msg) -- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are -- 'Serializable' - like the datum they contain - but also note, deserialising @@ -742,11 +793,11 @@ selfNode <- getSelfNode if nid == selfNode then Right `fmap` getLocalNodeStats -- optimisation - else getNodeStatsRemote + else getNodeStatsRemote selfNode where - getNodeStatsRemote :: Process (Either DiedReason NodeStats) - getNodeStatsRemote = do - sendCtrlMsg (Just nid) $ GetNodeStats nid + getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats) + getNodeStatsRemote selfNode = do + sendCtrlMsg (Just nid) $ GetNodeStats selfNode bracket (monitorNode nid) unmonitor $ \mRef -> receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats) , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) @@ -967,7 +1018,10 @@ spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef spawnAsync nid proc = do spawnRef <- getSpawnRef - sendCtrlMsg (Just nid) $ Spawn proc spawnRef + node <- getSelfNode + if nid == node + then sendCtrlMsg Nothing $ Spawn proc spawnRef + else sendCtrlMsg (Just nid) $ Spawn proc spawnRef return spawnRef -- | Monitor a node (asynchronous) @@ -1046,9 +1100,10 @@ registerImpl force label pid = do mynid <- getSelfNode sendCtrlMsg Nothing (Register label mynid (Just pid) force) - receiveWait [ matchIf (\(RegisterReply label' _) -> label == label') - (\(RegisterReply _ ok) -> handleRegistrationReply label ok) - ] + receiveWait + [ matchIf (\(RegisterReply label' _ _) -> label == label') + (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner) + ] -- | Register a process with a remote registry (asynchronous). -- @@ -1057,8 +1112,10 @@ -- -- See comments in 'whereisRemoteAsync' registerRemoteAsync :: NodeId -> String -> ProcessId -> Process () -registerRemoteAsync nid label pid = - sendCtrlMsg (Just nid) (Register label nid (Just pid) False) +registerRemoteAsync nid label pid = do + here <- getSelfNode + sendCtrlMsg (if nid == here then Nothing else Just nid) + (Register label nid (Just pid) False) reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process () reregisterRemoteAsync nid label pid = @@ -1071,16 +1128,17 @@ unregister label = do mynid <- getSelfNode sendCtrlMsg Nothing (Register label mynid Nothing False) - receiveWait [ matchIf (\(RegisterReply label' _) -> label == label') - (\(RegisterReply _ ok) -> handleRegistrationReply label ok) - ] + receiveWait + [ matchIf (\(RegisterReply label' _ _) -> label == label') + (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner) + ] -- | Deal with the result from an attempted registration or unregistration -- by throwing an exception if necessary -handleRegistrationReply :: String -> Bool -> Process () -handleRegistrationReply label ok = +handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process () +handleRegistrationReply label ok owner = when (not ok) $ - liftIO $ throwIO $ ProcessRegistrationException label + liftIO $ throwIO $ ProcessRegistrationException label owner -- | Remove a process from a remote registry (asynchronous). -- @@ -1109,8 +1167,9 @@ -- use 'monitorNode' and take appropriate action when you receive a -- 'NodeMonitorNotification'). whereisRemoteAsync :: NodeId -> String -> Process () -whereisRemoteAsync nid label = - sendCtrlMsg (Just nid) (WhereIs label) +whereisRemoteAsync nid label = do + here <- getSelfNode + sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label) -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process () @@ -1126,8 +1185,17 @@ -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () -nsendRemote nid label msg = - sendCtrlMsg (Just nid) (NamedSend label (createMessage msg)) +nsendRemote nid label msg = do + here <- getSelfNode + if here == nid then nsend label msg + else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg)) + +-- | Named send to a process in a remote registry (asynchronous) +-- This function makes /no/ attempt to serialize and (in the case when the +-- destination process resides on the same local node) therefore ensure that +-- the payload is fully evaluated before it is delivered. +unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process () +unsafeNSendRemote = Unsafe.nsendRemote -------------------------------------------------------------------------------- -- Closures -- diff --git a/src/Control/Distributed/Process/Internal/Types.hs b/src/Control/Distributed/Process/Internal/Types.hs --- a/src/Control/Distributed/Process/Internal/Types.hs +++ b/src/Control/Distributed/Process/Internal/Types.hs @@ -21,9 +21,14 @@ , nullProcessId -- * Local nodes and processes , LocalNode(..) + , LocalNodeState(..) + , ValidLocalNodeState(..) + , NodeClosedException(..) + , withValidLocalState + , modifyValidLocalState + , modifyValidLocalState_ , Tracer(..) , MxEventBus(..) - , LocalNodeState(..) , LocalProcess(..) , LocalProcessState(..) , Process(..) @@ -102,13 +107,14 @@ import Data.Accessor (Accessor, accessor) import Control.Category ((>>>)) import Control.DeepSeq (NFData(..)) -import Control.Exception (Exception) +import Control.Exception (Exception, throwIO) import Control.Concurrent (ThreadId) import Control.Concurrent.Chan (Chan) import Control.Concurrent.STM (STM) import Control.Concurrent.STM.TChan (TChan) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative (Applicative, Alternative, (<$>), (<*>)) +import Control.Monad.Fix (MonadFix) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT) import Control.Monad.IO.Class (MonadIO) import Control.Distributed.Process.Serializable @@ -121,7 +127,12 @@ , showFingerprint ) import Control.Distributed.Process.Internal.CQueue (CQueue) -import Control.Distributed.Process.Internal.StrictMVar (StrictMVar) +import Control.Distributed.Process.Internal.StrictMVar + ( StrictMVar + , withMVar + , modifyMVar + , modifyMVar_ + ) import Control.Distributed.Process.Internal.WeakTQueue (TQueue) import Control.Distributed.Static (RemoteTable, Closure) import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe) @@ -257,7 +268,11 @@ deriving (Eq, Show) -- | Local node state -data LocalNodeState = LocalNodeState +data LocalNodeState = + LocalNodeValid {-# UNPACK #-} !ValidLocalNodeState + | LocalNodeClosed + +data ValidLocalNodeState = ValidLocalNodeState { -- | Processes running on this node _localProcesses :: !(Map LocalProcessId LocalProcess) -- | Counter to assign PIDs @@ -270,6 +285,40 @@ (NT.Connection, ImplicitReconnect)) } +-- | Thrown by some primitives when they notice the node has been closed. +data NodeClosedException = NodeClosedException NodeId + deriving (Show, Typeable) + +instance Exception NodeClosedException + +-- | Wrapper around 'withMVar' that checks that the local node is still in +-- a valid state. +withValidLocalState :: LocalNode + -> (ValidLocalNodeState -> IO r) + -> IO r +withValidLocalState node f = withMVar (localState node) $ \st -> case st of + LocalNodeValid vst -> f vst + LocalNodeClosed -> throwIO $ NodeClosedException (localNodeId node) + +-- | Wrapper around 'modifyMVar' that checks that the local node is still in +-- a valid state. +modifyValidLocalState :: LocalNode + -> (ValidLocalNodeState -> IO (ValidLocalNodeState, a)) + -> IO (Maybe a) +modifyValidLocalState node f = modifyMVar (localState node) $ \st -> case st of + LocalNodeValid vst -> do (vst', a) <- f vst + return (LocalNodeValid vst', Just a) + LocalNodeClosed -> return (LocalNodeClosed, Nothing) + +-- | Wrapper around 'modifyMVar_' that checks that the local node is still in +-- a valid state. +modifyValidLocalState_ :: LocalNode + -> (ValidLocalNodeState -> IO ValidLocalNodeState) + -> IO () +modifyValidLocalState_ node f = modifyMVar_ (localState node) $ \st -> case st of + LocalNodeValid vst -> LocalNodeValid <$> f vst + LocalNodeClosed -> return LocalNodeClosed + -- | Processes running on our local node data LocalProcess = LocalProcess { processQueue :: !(CQueue Message) @@ -296,7 +345,7 @@ newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a } - deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative) + deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative, MonadFix) -------------------------------------------------------------------------------- -- Typed channels -- @@ -379,7 +428,8 @@ #else rnf (EncodedMessage _ e) = BSL.length e `seq` () #endif - rnf (UnencodedMessage _ a) = a `seq` () -- forced to WHNF only + rnf (UnencodedMessage _ a) = e `seq` () + where e = BSL.length (encode a) instance Show Message where show (EncodedMessage fp enc) = show enc ++ " :: " ++ showFingerprint fp [] @@ -481,9 +531,11 @@ -- | Exception thrown when a process attempts to register -- a process under an already-registered name or to --- unregister a name that hasn't been registered +-- unregister a name that hasn't been registered. Returns +-- the name and the identifier of the process that owns it, +-- if any. data ProcessRegistrationException = - ProcessRegistrationException !String + ProcessRegistrationException !String !(Maybe ProcessId) deriving (Typeable, Show) -- | Internal exception thrown indirectly by 'exit' @@ -548,7 +600,7 @@ deriving (Show, Typeable) -- | (Asynchronous) reply from 'register' and 'unregister' -data RegisterReply = RegisterReply String Bool +data RegisterReply = RegisterReply String Bool (Maybe ProcessId) deriving (Show, Typeable) data NodeStats = NodeStats { @@ -564,7 +616,7 @@ data ProcessInfo = ProcessInfo { infoNode :: NodeId , infoRegisteredNames :: [String] - , infoMessageQueueLength :: Maybe Int + , infoMessageQueueLength :: Int , infoMonitors :: [(ProcessId, MonitorRef)] , infoLinks :: [ProcessId] } deriving (Show, Eq, Typeable) @@ -594,6 +646,7 @@ | WhereIs !String | Register !String !NodeId !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister | NamedSend !String !Message + | UnreliableSend !LocalProcessId !Message | LocalSend !ProcessId !Message | LocalPortSend !SendPortId !Message | Kill !ProcessId !String @@ -649,6 +702,7 @@ put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason) put (LocalSend to' msg) = putWord8 11 >> put to' >> put (messageToPayload msg) put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg) + put (UnreliableSend lpid msg) = putWord8 13 >> put lpid >> put (messageToPayload msg) put (GetInfo about) = putWord8 30 >> put about put (SigShutdown) = putWord8 31 put (GetNodeStats nid) = putWord8 32 >> put nid @@ -668,6 +722,7 @@ 10 -> Exit <$> get <*> (payloadToMessage <$> get) 11 -> LocalSend <$> get <*> (payloadToMessage <$> get) 12 -> LocalPortSend <$> get <*> (payloadToMessage <$> get) + 13 -> UnreliableSend <$> get <*> (payloadToMessage <$> get) 30 -> GetInfo <$> get 31 -> return SigShutdown 32 -> GetNodeStats <$> get @@ -714,8 +769,8 @@ get = WhereIsReply <$> get <*> get instance Binary RegisterReply where - put (RegisterReply label ok) = put label >> put ok - get = RegisterReply <$> get <*> get + put (RegisterReply label ok owner) = put label >> put ok >> put owner + get = RegisterReply <$> get <*> get <*> get instance Binary ProcessInfo where get = ProcessInfo <$> get <*> get <*> get <*> get <*> get @@ -741,22 +796,22 @@ -- Accessors -- -------------------------------------------------------------------------------- -localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess) +localProcesses :: Accessor ValidLocalNodeState (Map LocalProcessId LocalProcess) localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs }) -localPidCounter :: Accessor LocalNodeState Int32 +localPidCounter :: Accessor ValidLocalNodeState Int32 localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr }) -localPidUnique :: Accessor LocalNodeState Int32 +localPidUnique :: Accessor ValidLocalNodeState Int32 localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq }) -localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect)) +localConnections :: Accessor ValidLocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect)) localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns }) -localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess) +localProcessWithId :: LocalProcessId -> Accessor ValidLocalNodeState (Maybe LocalProcess) localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid -localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect)) +localConnectionBetween :: Identifier -> Identifier -> Accessor ValidLocalNodeState (Maybe (NT.Connection, ImplicitReconnect)) localConnectionBetween from' to' = localConnections >>> DAC.mapMaybe (from', to') monitorCounter :: Accessor LocalProcessState Int32 @@ -782,4 +837,3 @@ {-# INLINE forever' #-} forever' :: Monad m => m a -> m b forever' a = let a' = a >> a' in a' - diff --git a/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs b/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs --- a/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs +++ b/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs @@ -150,7 +150,7 @@ emptyPid <- return $ (nullProcessId (localNodeId node)) traceMsg <- return $ NCMsg { ctrlMsgSender = ProcessIdentifier (emptyPid) - , ctrlMsgSignal = (NamedSend "logger" + , ctrlMsgSignal = (NamedSend "trace.logger" (createUnencodedMessage msg)) } liftIO $ writeChan (localCtrlChan node) traceMsg diff --git a/src/Control/Distributed/Process/Node.hs b/src/Control/Distributed/Process/Node.hs --- a/src/Control/Distributed/Process/Node.hs +++ b/src/Control/Distributed/Process/Node.hs @@ -53,7 +53,7 @@ import Data.Typeable (Typeable) import Control.Category ((>>>)) import Control.Applicative (Applicative, (<$>)) -import Control.Monad (void, when) +import Control.Monad (void, when, join) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets) import qualified Control.Monad.State.Strict as StateT (get, put) @@ -66,17 +66,22 @@ , throwTo , uninterruptibleMask_ ) -import qualified Control.Exception as Exception (Handler(..), catches, finally) -import Control.Concurrent (forkIO, forkIOWithUnmask, myThreadId) +import qualified Control.Exception as Exception + ( Handler(..) + , catch + , catches + , finally + ) +import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId) import Control.Distributed.Process.Internal.StrictMVar ( newMVar , withMVar , modifyMVarMasked , modifyMVar_ + , modifyMVar , newEmptyMVar , putMVar , takeMVar - , readMVar ) import Control.Concurrent.Chan (newChan, writeChan, readChan) import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar) @@ -88,6 +93,7 @@ , enqueue , newCQueue , mkWeakCQueue + , queueSize ) import qualified Network.Transport as NT ( Transport @@ -99,6 +105,7 @@ , TransportError(..) , address , closeEndPoint + , Connection , ConnectionId , close , EndPointAddress @@ -118,6 +125,9 @@ , LocalNode(..) , MxEventBus(..) , LocalNodeState(..) + , ValidLocalNodeState(..) + , withValidLocalState + , modifyValidLocalState , LocalProcess(..) , LocalProcessState(..) , Process(..) @@ -131,6 +141,7 @@ , localConnections , forever' , MonitorRef(..) + , NodeClosedException(..) , ProcessMonitorNotification(..) , NodeMonitorNotification(..) , PortMonitorNotification(..) @@ -158,6 +169,7 @@ , payloadToMessage , messageToPayload , createUnencodedMessage + , unsafeCreateUnencodedMessage , runLocalProcess , firstNonReservedProcessId , ImplicitReconnect(WithImplicitReconnect,NoImplicitReconnect) @@ -187,18 +199,16 @@ import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Messaging ( sendBinary - , sendMessage , sendPayload , closeImplicitReconnections , impliesDeathOf ) import Control.Distributed.Process.Internal.Primitives ( register - , finally , receiveWait , match , sendChan - , catch + , try , unwrapMessage ) import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..)) @@ -232,7 +242,7 @@ createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode createBareLocalNode endPoint rtable = do unq <- randomIO - state <- newMVar LocalNodeState + state <- newMVar $ LocalNodeValid $ ValidLocalNodeState { _localProcesses = Map.empty , _localPidCounter = firstNonReservedProcessId , _localPidUnique = unq @@ -297,7 +307,12 @@ tableCoordinatorPid <- fork $ Table.startTableCoordinator fork runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid logger <- forkProcess node loop - runProcess node $ register "logger" logger + runProcess node $ do + register "logger" logger + -- The trace.logger is used for tracing to the console to avoid feedback + -- loops during tracing if the user reregisters the "logger" with a custom + -- process which uses 'send' or other primitives which are traced. + register "trace.logger" logger where fork = forkProcess node @@ -314,23 +329,35 @@ sendChan ch () ] --- | Force-close a local node --- --- TODO: for now we just close the associated endpoint +-- | Force-close a local node, killing all processes on that node. closeLocalNode :: LocalNode -> IO () -closeLocalNode node = - -- TODO: close all our processes, surely!? +closeLocalNode node = do + -- Kill processes after refilling the mvar. Otherwise, there is potential for + -- deadlock as a dying process tries to get the mvar while masking exceptions + -- uninterruptibly. + join $ modifyMVar (localState node) $ \st -> case st of + LocalNodeValid vst -> do + return ( LocalNodeClosed + , forM_ (vst ^. localProcesses) $ \lproc -> + -- Semantics of 'throwTo' guarantee that target thread will get + -- delivered an exception. Therefore, target thread will be + -- killed eventually and that's as good as we can do. No need + -- to wait for thread to actually finish dying. + killThread (processThread lproc) + ) + LocalNodeClosed -> return (LocalNodeClosed, return ()) + -- This call will have the effect of shutting down the NC as well (see + -- 'createBareLocalNode'). NT.closeEndPoint (localEndPoint node) -- | Run a process on a local node and wait for it to finish runProcess :: LocalNode -> Process () -> IO () runProcess node proc = do done <- newEmptyMVar - tid <- myThreadId - void $ forkProcess node $ do - catch (proc `finally` liftIO (putMVar done ())) - (\(ex :: SomeException) -> liftIO $ throwTo tid ex) - takeMVar done + -- TODO; When forkProcess inherits the masking state, protect the forked + -- thread against async exceptions that could occur before 'try' is evaluated. + void $ forkProcess node $ try proc >>= liftIO . putMVar done + takeMVar done >>= either (throwIO :: SomeException -> IO a) return -- | Spawn a new process on a local node forkProcess :: LocalNode -> Process () -> IO ProcessId @@ -338,9 +365,9 @@ modifyMVarMasked (localState node) startProcess where startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId) - startProcess st = do - let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter - , lpidUnique = st ^. localPidUnique + startProcess (LocalNodeValid vst) = do + let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter + , lpidUnique = vst ^. localPidUnique } let pid = ProcessId { processNodeId = localNodeId node , processLocalId = lpid @@ -372,7 +399,12 @@ (return . DiedException . (show :: SomeException -> String)))] -- [Unified: Table 4, rules termination and exiting] - modifyMVar_ (localState node) (cleanupProcess pid) + mconns <- modifyValidLocalState node (cleanupProcess pid) + -- XXX: Revisit after agreeing on the bigger picture for the semantics + -- of transport operations. + -- https://github.com/haskell-distributed/distributed-process/issues/204 + forM_ mconns $ forkIO . mapM_ NT.close + writeChan (localCtrlChan node) NCMsg { ctrlMsgSender = ProcessIdentifier pid , ctrlMsgSignal = Died (ProcessIdentifier pid) reason @@ -387,27 +419,34 @@ -- TODO: this doesn't look right at all - how do we know -- that newUnique represents a process id that is available!? newUnique <- randomIO - return ( (localProcessWithId lpid ^= Just lproc) + return ( LocalNodeValid + $ (localProcessWithId lpid ^= Just lproc) . (localPidCounter ^= firstNonReservedProcessId) . (localPidUnique ^= newUnique) - $ st + $ vst , pid ) else - return ( (localProcessWithId lpid ^= Just lproc) + return ( LocalNodeValid + $ (localProcessWithId lpid ^= Just lproc) . (localPidCounter ^: (+ 1)) - $ st + $ vst , pid ) + startProcess LocalNodeClosed = + throwIO $ NodeClosedException $ localNodeId node - cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState - cleanupProcess pid st = do + cleanupProcess :: ProcessId + -> ValidLocalNodeState + -> IO (ValidLocalNodeState, [NT.Connection]) + cleanupProcess pid vst = do let pid' = ProcessIdentifier pid - let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections) - mapM_ (NT.close . fst) (Map.elems affected) - return $ (localProcessWithId (processLocalId pid) ^= Nothing) - . (localConnections ^= unaffected) - $ st + let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections) + return ( (localProcessWithId (processLocalId pid) ^= Nothing) + . (localConnections ^= unaffected) + $ vst + , map fst $ Map.elems affected + ) -- note [tracer/forkProcess races] -- @@ -454,6 +493,7 @@ handleIncomingMessages :: LocalNode -> IO () handleIncomingMessages node = go initConnectionState + `Exception.catch` \(NodeClosedException _) -> return () where go :: ConnectionState -> IO () go !st = do @@ -496,7 +536,7 @@ case decode (BSL.fromChunks payload) of ProcessIdentifier pid -> do let lpid = processLocalId pid - mProc <- withMVar state $ return . (^. localProcessWithId lpid) + mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid) case mProc of Just proc -> go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st) @@ -505,7 +545,7 @@ SendPortIdentifier chId -> do let lcid = sendPortLocalId chId lpid = processLocalId (sendPortProcessId chId) - mProc <- withMVar state $ return . (^. localProcessWithId lpid) + mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid) case mProc of Just proc -> do mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid) @@ -568,7 +608,6 @@ $ st ) - state = localState node endpoint = localEndPoint node ctrlChan = localCtrlChan node @@ -577,8 +616,9 @@ -------------------------------------------------------------------------------- runNodeController :: LocalNode -> IO () -runNodeController = - runReaderT (evalStateT (unNC nodeController) initNCState) +runNodeController node = + runReaderT (evalStateT (unNC nodeController) initNCState) node + `Exception.catch` \(NodeClosedException _) -> return () -------------------------------------------------------------------------------- -- Internal data types -- @@ -621,6 +661,33 @@ show (ProcessKillException pid reason) = "killed-by=" ++ show pid ++ ",reason=" ++ reason +ncSendToProcess :: ProcessId -> Message -> NC () +ncSendToProcess = ncSendToProcessAndTrace True + +ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC () +ncSendToProcessAndTrace shouldTrace pid msg = do + node <- ask + if processNodeId pid == localNodeId node + then ncEffectLocalSendAndTrace shouldTrace node pid msg + else liftIO $ sendBinary node + (NodeIdentifier $ localNodeId node) + (NodeIdentifier $ processNodeId pid) + WithImplicitReconnect + NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node + , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg + } + +ncSendToNode :: NodeId -> NCMsg -> NC () +ncSendToNode to msg = do + node <- ask + liftIO $ if to == localNodeId node + then writeChan (localCtrlChan node) $! msg + else sendBinary node + (NodeIdentifier $ localNodeId node) + (NodeIdentifier to) + WithImplicitReconnect + msg + -------------------------------------------------------------------------------- -- Tracing/Debugging -- -------------------------------------------------------------------------------- @@ -663,11 +730,7 @@ -- [Unified: Table 7, rule nc_forward] case destNid (ctrlMsgSignal msg) of Just nid' | nid' /= localNodeId node -> - liftIO $ sendBinary node - (ctrlMsgSender msg) - (NodeIdentifier nid') - WithImplicitReconnect - msg + ncSendToNode nid' msg _ -> return () @@ -688,8 +751,10 @@ ncEffectRegister from label atnode pid force NCMsg (ProcessIdentifier from) (WhereIs label) -> ncEffectWhereIs from label - NCMsg (ProcessIdentifier from) (NamedSend label msg') -> - ncEffectNamedSend from label msg' + NCMsg _ (NamedSend label msg') -> + ncEffectNamedSend label msg' + NCMsg _ (UnreliableSend lpid msg') -> + ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg' NCMsg _ (LocalSend to msg') -> ncEffectLocalSend node to msg' NCMsg _ (LocalPortSend to msg') -> @@ -703,8 +768,7 @@ NCMsg _ SigShutdown -> liftIO $ do NT.closeEndPoint (localEndPoint node) - `Exception.finally` throwIO ThreadKilled - -- ThreadKilled seems to make more sense than fail/error here + `Exception.finally` throwIO (NodeClosedException $ localNodeId node) NCMsg (ProcessIdentifier from) (GetNodeStats nid) -> ncEffectGetNodeStats from nid unexpected -> @@ -732,14 +796,10 @@ -- TODO: this is the right sender according to the Unified semantics, -- but perhaps having 'them' as the sender would make more sense -- (see also: notifyDied) - liftIO $ sendBinary node - (NodeIdentifier $ localNodeId node) - (NodeIdentifier $ processNodeId from) - WithImplicitReconnect - NCMsg - { ctrlMsgSender = NodeIdentifier (localNodeId node) - , ctrlMsgSignal = Died them DiedUnknownId - } + ncSendToNode (processNodeId from) $ NCMsg + { ctrlMsgSender = NodeIdentifier (localNodeId node) + , ctrlMsgSignal = Died them DiedUnknownId + } -- [Unified: Table 11] ncEffectUnlink :: ProcessId -> Identifier -> NC () @@ -800,15 +860,10 @@ False -> return $ Just (pid,nidlist) ) modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining)) where - forwardNameDeath node nid = - liftIO $ sendBinary node - (NodeIdentifier $ localNodeId node) - (NodeIdentifier $ nid) - WithImplicitReconnect - NCMsg - { ctrlMsgSender = NodeIdentifier (localNodeId node) - , ctrlMsgSignal = Died ident reason - } + forwardNameDeath node nid = ncSendToNode nid + NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node) + , ctrlMsgSignal = Died ident reason + } -- [Unified: Table 13] ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC () @@ -822,11 +877,7 @@ Right p -> p node <- ask pid' <- liftIO $ forkProcess node proc - liftIO $ sendMessage node - (NodeIdentifier (localNodeId node)) - (ProcessIdentifier pid) - WithImplicitReconnect - (DidSpawn ref pid') + ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid' -- Unified semantics does not explicitly describe how to implement 'register', -- but mentions it's "very similar to nsend" (Table 14) @@ -844,17 +895,15 @@ return $ (isNothing currentVal /= reregistration) && (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal ) if isLocal node (NodeIdentifier atnode) - then do when (isOk) $ + then do when isOk $ do modify' $ registeredHereFor label ^= mPid updateRemote node currentVal mPid case mPid of (Just p) -> liftIO $ trace node (MxRegistered p label) Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label) - liftIO $ sendMessage node - (NodeIdentifier (localNodeId node)) - (ProcessIdentifier from) - WithImplicitReconnect - (RegisterReply label isOk) + newVal <- gets (^. registeredHereFor label) + ncSendToProcess from $ unsafeCreateUnencodedMessage $ + RegisterReply label isOk newVal else let operation = case reregistration of True -> flip decList @@ -884,46 +933,36 @@ decList (x:xs) tag = x:decList xs tag forward node to reg = when (not $ isLocal node (NodeIdentifier to)) $ - liftIO $ sendBinary node - (ProcessIdentifier from) - (NodeIdentifier to) - WithImplicitReconnect - NCMsg - { ctrlMsgSender = ProcessIdentifier from - , ctrlMsgSignal = reg - } + ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from + , ctrlMsgSignal = reg + } -- Unified semantics does not explicitly describe 'whereis' ncEffectWhereIs :: ProcessId -> String -> NC () ncEffectWhereIs from label = do - node <- ask mPid <- gets (^. registeredHereFor label) - liftIO $ sendMessage node - (NodeIdentifier (localNodeId node)) - (ProcessIdentifier from) - WithImplicitReconnect - (WhereIsReply label mPid) + ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid -- [Unified: Table 14] -ncEffectNamedSend :: ProcessId -> String -> Message -> NC () -ncEffectNamedSend from label msg = do +ncEffectNamedSend :: String -> Message -> NC () +ncEffectNamedSend label msg = do mPid <- gets (^. registeredHereFor label) - node <- ask -- If mPid is Nothing, we just ignore the named send (as per Table 14) - forM_ mPid $ \pid -> - liftIO $ sendPayload node - (ProcessIdentifier from) - (ProcessIdentifier pid) - NoImplicitReconnect - (messageToPayload msg) + forM_ mPid $ \to -> + -- If this is a trace message we don't trace it to avoid entering a loop + -- where trace messages produce more trace messages. + ncSendToProcessAndTrace (label /= "trace.logger") to msg -- [Issue #DP-20] ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC () -ncEffectLocalSend node to msg = +ncEffectLocalSend = ncEffectLocalSendAndTrace True + +ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC () +ncEffectLocalSendAndTrace shouldTrace node to msg = liftIO $ withLocalProc node to $ \p -> do enqueue (processQueue p) msg - trace node (MxReceived to msg) + when shouldTrace $ trace node (MxReceived to msg) -- [Issue #DP-20] ncEffectLocalPortSend :: SendPortId -> Message -> NC () @@ -970,41 +1009,35 @@ them = (ProcessIdentifier pid) in do node <- ask - mProc <- liftIO $ - withMVar (localState node) $ return . (^. localProcessWithId lpid) + mProc <- liftIO $ withValidLocalState node + $ return . (^. localProcessWithId lpid) case mProc of Nothing -> dispatch (isLocal node (ProcessIdentifier from)) - from node (ProcessInfoNone DiedUnknownId) - Just _ -> do + from (ProcessInfoNone DiedUnknownId) + Just proc -> do itsLinks <- gets (^. linksFor them) itsMons <- gets (^. monitorsFor them) registered <- gets (^. registeredHere) + size <- liftIO $ queueSize $ processQueue $ proc let reg = registeredNames registered dispatch (isLocal node (ProcessIdentifier from)) from - node ProcessInfo { infoNode = (processNodeId pid) , infoRegisteredNames = reg - -- we cannot populate this field - , infoMessageQueueLength = Nothing + , infoMessageQueueLength = size , infoMonitors = Set.toList itsMons , infoLinks = Set.toList itsLinks } where dispatch :: (Serializable a, Show a) => Bool -> ProcessId - -> LocalNode -> a -> NC () - dispatch True dest _ pInfo = postAsMessage dest $ pInfo - dispatch False dest node pInfo = do - liftIO $ sendMessage node - (NodeIdentifier (localNodeId node)) - (ProcessIdentifier dest) - WithImplicitReconnect - pInfo + dispatch True dest pInfo = postAsMessage dest $ pInfo + dispatch False dest pInfo = + ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo registeredNames = Map.foldlWithKey (\ks k v -> if v == pid then (k:ks) @@ -1014,17 +1047,17 @@ ncEffectGetNodeStats from _nid = do node <- ask ncState <- StateT.get - nodeState <- liftIO $ readMVar (localState node) - let localProcesses' = nodeState ^. localProcesses - stats = + nodeState <- liftIO $ withValidLocalState node return + let stats = NodeStats { nodeStatsNode = localNodeId node , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere , nodeStatsMonitors = Map.size $ ncState ^. monitors , nodeStatsLinks = Map.size $ ncState ^. links - , nodeStatsProcesses = Map.size localProcesses' + , nodeStatsProcesses = Map.size (nodeState ^. localProcesses) } postAsMessage from stats + -------------------------------------------------------------------------------- -- Auxiliary -- -------------------------------------------------------------------------------- @@ -1051,14 +1084,10 @@ throwException dest $ PortLinkException pid reason (False, _, _) -> -- The change in sender comes from [Unified: Table 10] - liftIO $ sendBinary node - (NodeIdentifier $ localNodeId node) - (NodeIdentifier $ processNodeId dest) - WithImplicitReconnect - NCMsg - { ctrlMsgSender = NodeIdentifier (localNodeId node) - , ctrlMsgSignal = Died src reason - } + ncSendToNode (processNodeId dest) $ NCMsg + { ctrlMsgSender = NodeIdentifier (localNodeId node) + , ctrlMsgSignal = Died src reason + } -- | [Unified: Table 8] destNid :: ProcessSignal -> Maybe NodeId @@ -1070,6 +1099,7 @@ destNid (Register _ _ _ _) = Nothing destNid (WhereIs _) = Nothing destNid (NamedSend _ _) = Nothing +destNid (UnreliableSend _ _) = Nothing -- We don't need to forward 'Died' signals; if monitoring/linking is setup, -- then when a local process dies the monitoring/linking machinery will take -- care of notifying remote nodes @@ -1096,7 +1126,7 @@ isValidLocalIdentifier :: Identifier -> NC Bool isValidLocalIdentifier ident = do node <- ask - liftIO . withMVar (localState node) $ \nSt -> + liftIO . withValidLocalState node $ \nSt -> case ident of NodeIdentifier nid -> return $ nid == localNodeId node @@ -1127,15 +1157,17 @@ throwException :: Exception e => ProcessId -> e -> NC () throwException pid e = do node <- ask - liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e + -- throwTo blocks until the exception is received by the target thread. + -- We fork a helper thread to avoid blocking the node controller. + liftIO $ withLocalProc node pid $ \p -> void $ forkIO $ throwTo (processThread p) e withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO () withLocalProc node pid p = -- By [Unified: table 6, rule missing_process] messages to dead processes -- can silently be dropped let lpid = processLocalId pid in do - mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid) - forM_ mProc p + withValidLocalState node $ \vst -> + forM_ (vst ^. localProcessWithId lpid) p -------------------------------------------------------------------------------- -- Accessors -- diff --git a/src/Control/Distributed/Process/UnsafePrimitives.hs b/src/Control/Distributed/Process/UnsafePrimitives.hs --- a/src/Control/Distributed/Process/UnsafePrimitives.hs +++ b/src/Control/Distributed/Process/UnsafePrimitives.hs @@ -45,6 +45,8 @@ send , sendChan , nsend + , nsendRemote + , usend , wrapMessage ) where @@ -56,6 +58,7 @@ import Control.Distributed.Process.Internal.Types ( ProcessId(..) + , NodeId(..) , LocalNode(..) , LocalProcess(..) , Process(..) @@ -65,6 +68,7 @@ , ImplicitReconnect(..) , SendPortId(..) , Message + , createMessage , sendPortProcessId , unsafeCreateUnencodedMessage ) @@ -78,6 +82,14 @@ nsend label msg = sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg)) +-- | Named send to a process in a remote registry (asynchronous) +nsendRemote :: Serializable a => NodeId -> String -> a -> Process () +nsendRemote nid label msg = do + proc <- ask + if localNodeId (processNode proc) == nid + then nsend label msg + else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg)) + -- | Send a message send :: Serializable a => ProcessId -> a -> Process () send them msg = do @@ -95,6 +107,25 @@ unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process () unsafeSendLocal pid msg' = sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg') + +-- | Send a message unreliably. +-- +-- Unlike 'send', this function is insensitive to 'reconnect'. It will +-- try to send the message regardless of the history of connection failures +-- between the nodes. +-- +-- Message passing with 'usend' is ordered for a given sender and receiver +-- if the messages arrive at all. +-- +usend :: Serializable a => ProcessId -> a -> Process () +usend them msg = do + proc <- ask + let there = processNodeId them + if localNodeId (processNode proc) == there + then sendCtrlMsg Nothing $ + LocalSend them (unsafeCreateUnencodedMessage msg) + else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them) + (createMessage msg) -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process ()