distributed-process 0.5.5.1 → 0.6.0

raw patch · 11 files changed

+492/−220 lines, 11 filesdep ~network-transportdep ~network-transport-tcpPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: network-transport, network-transport-tcp

API changes (from Hackage documentation)

Files

• ChangeLog +28/−0
• distributed-process.cabal +3/−3
• src/Control/Distributed/Process.hs +28/−1
• src/Control/Distributed/Process/Debug.hs +7/−7
• src/Control/Distributed/Process/Internal/CQueue.hs +35/−13
• src/Control/Distributed/Process/Internal/Messaging.hs +34/−24
• src/Control/Distributed/Process/Internal/Primitives.hs +88/−20
• src/Control/Distributed/Process/Internal/Types.hs +73/−19
• src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs +1/−1
• src/Control/Distributed/Process/Node.hs +164/−132
• src/Control/Distributed/Process/UnsafePrimitives.hs +31/−0

@@ -1,3 +1,31 @@+2016-02-18 Facundo Domínguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 0.5.5  * Fix dependencies.

@@ -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,

@@ -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

@@ -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

@@ -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

@@ -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))

@@ -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                                                                   --

@@ -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'-

@@ -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

@@ -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                                                                  --

@@ -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 ()