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.3.0 +Version: 0.3.1 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -29,6 +29,10 @@ Location: https://github.com/haskell-distributed/distributed-process SubDir: distributed-process +flag th + description: Build with Template Haskell support + default: True + Library Build-Depends: base >= 4.4 && < 5, binary >= 0.5 && < 0.6, @@ -41,10 +45,9 @@ containers >= 0.4 && < 0.6, old-locale >= 1.0 && < 1.1, time >= 1.2 && < 1.5, - template-haskell >= 2.6 && < 2.8, random >= 1.0 && < 1.1, - ghc-prim >= 0.2 && < 0.3, - distributed-static >= 0.1 && < 0.2, + ghc-prim >= 0.2 && < 0.4, + distributed-static >= 0.2 && < 0.3, rank1dynamic >= 0.1 && < 0.2, syb >= 0.3 && < 0.4 Exposed-modules: Control.Distributed.Process, @@ -54,9 +57,9 @@ Control.Distributed.Process.Internal.Primitives, Control.Distributed.Process.Internal.CQueue, Control.Distributed.Process.Internal.Types, - Control.Distributed.Process.Internal.Closure.TH, Control.Distributed.Process.Internal.Closure.BuiltIn, Control.Distributed.Process.Internal.Node + Control.Distributed.Process.Internal.StrictMVar Extensions: RankNTypes, ScopedTypeVariables, FlexibleInstances, @@ -65,75 +68,46 @@ GADTs, GeneralizedNewtypeDeriving, DeriveDataTypeable, - CPP + CPP, + BangPatterns ghc-options: -Wall HS-Source-Dirs: src + if flag(th) + Build-Depends: template-haskell >= 2.6 && < 2.9 + Exposed-modules: Control.Distributed.Process.Internal.Closure.TH + CPP-Options: -DTemplateHaskellSupport Test-Suite TestCH Type: exitcode-stdio-1.0 Main-Is: TestCH.hs Build-Depends: base >= 4.4 && < 5, - binary >= 0.5 && < 0.6, - network-transport >= 0.2 && < 0.3, - network-transport-tcp >= 0.2 && < 0.3, - stm >= 2.3 && < 2.5, - transformers >= 0.2 && < 0.4, - mtl >= 2.0 && < 2.2, - data-accessor >= 0.2 && < 0.3, - bytestring >= 0.9 && < 0.11, - containers >= 0.4 && < 0.6, - old-locale >= 1.0 && < 1.1, - time >= 1.2 && < 1.5, - template-haskell >= 2.6 && < 2.8, random >= 1.0 && < 1.1, - ghc-prim >= 0.2 && < 0.3, ansi-terminal >= 0.5 && < 0.6, - distributed-static >= 0.1 && < 0.2, - rank1dynamic >= 0.1 && < 0.2 - Extensions: RankNTypes, + distributed-process >= 0.3 && < 0.4, + network-transport >= 0.2 && < 0.3, + network-transport-tcp >= 0.2 && < 0.3, + binary >= 0.5 && < 0.6 + Other-modules: TestAuxiliary + Extensions: CPP, ScopedTypeVariables, - FlexibleInstances, - UndecidableInstances, - ExistentialQuantification, - GADTs, - GeneralizedNewtypeDeriving, DeriveDataTypeable, - CPP + GeneralizedNewtypeDeriving ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind - HS-Source-Dirs: tests src + HS-Source-Dirs: tests Test-Suite TestClosure Type: exitcode-stdio-1.0 Main-Is: TestClosure.hs Build-Depends: base >= 4.4 && < 5, - binary >= 0.5 && < 0.6, - network-transport >= 0.2 && < 0.3, - network-transport-tcp >= 0.2 && < 0.3, - stm >= 2.3 && < 2.5, - transformers >= 0.2 && < 0.4, - mtl >= 2.0 && < 2.2, - data-accessor >= 0.2 && < 0.3, - bytestring >= 0.9 && < 0.11, - containers >= 0.4 && < 0.6, - old-locale >= 1.0 && < 1.1, - time >= 1.2 && < 1.5, - template-haskell >= 2.6 && < 2.8, random >= 1.0 && < 1.1, - ghc-prim >= 0.2 && < 0.3, ansi-terminal >= 0.5 && < 0.6, - distributed-static >= 0.1 && < 0.2, - rank1dynamic >= 0.1 && < 0.2, - syb >= 0.3 && < 0.4 + distributed-static >= 0.2 && < 0.3, + distributed-process >= 0.3 && < 0.4, + network-transport >= 0.2 && < 0.3, + network-transport-tcp >= 0.2 && < 0.3, + bytestring >= 0.9 && < 0.11 Other-modules: TestAuxiliary - Extensions: RankNTypes, - ScopedTypeVariables, - FlexibleInstances, - UndecidableInstances, - ExistentialQuantification, - GADTs, - GeneralizedNewtypeDeriving, - DeriveDataTypeable, - TemplateHaskell, - CPP + Extensions: CPP, + ScopedTypeVariables ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind - HS-Source-Dirs: tests src + HS-Source-Dirs: tests 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 @@ -63,7 +63,8 @@ , PortMonitorNotification(..) , DiedReason(..) -- * Closures - , Closure(..) + , Closure + , closure , Static , unClosure , RemoteTable @@ -98,6 +99,10 @@ -- * Local versions of 'spawn' , spawnLocal , spawnChannelLocal + -- * Reconnecting + , reconnect + , reconnectNode + , reconnectPort ) where #if ! MIN_VERSION_base(4,6,0) @@ -109,7 +114,8 @@ import Control.Applicative ((<$>)) import Control.Monad.Reader (ask) import Control.Distributed.Static - ( Closure(..) + ( Closure + , closure , Static , RemoteTable ) @@ -206,6 +212,10 @@ -- Auxiliary API , expectTimeout , spawnAsync + -- Reconnecting + , reconnect + , reconnectNode + , reconnectPort ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Node (forkProcess) @@ -294,14 +304,14 @@ `seqCP` proc mPid <- receiveWait [ matchIf (\(DidSpawn ref _) -> ref == sRef) - (\(DidSpawn _ pid) -> return $ Just pid) + (\(DidSpawn _ pid) -> return $ Right pid) , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) - (\_ -> return Nothing) + (\(NodeMonitorNotification _ _ err) -> return $ Left err) ] unmonitor mRef case mPid of - Nothing -> fail "spawn: remote node failed" - Just pid -> send pid () >> return pid + Left err -> fail $ "spawn: remote node failed: " ++ show err + Right pid -> send pid () >> return pid -- | Spawn a process and link to it -- @@ -333,7 +343,7 @@ call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a call dict nid proc = do us <- getSelfPid - (_, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us) + (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us) -- We are guaranteed to receive the reply before the monitor notification -- (if a reply is sent at all) -- NOTE: This might not be true if we switch to unreliable delivery. @@ -343,8 +353,17 @@ (\(ProcessMonitorNotification _ _ reason) -> return (Left reason)) ] case mResult of - Right a -> unmonitor mRef >> return a - Left err -> fail $ "call: remote process died: " ++ show err + Right a -> do + -- Wait for the monitor message so that we the mailbox doesn't grow + receiveWait + [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef) + (\(ProcessMonitorNotification {}) -> return ()) + ] + -- Clean up connection to pid + reconnect pid + return a + Left err -> + fail $ "call: remote process died: " ++ show err -- | Spawn a child process, have the child link to the parent and the parent -- monitor the child diff --git a/src/Control/Distributed/Process/Closure.hs b/src/Control/Distributed/Process/Closure.hs --- a/src/Control/Distributed/Process/Closure.hs +++ b/src/Control/Distributed/Process/Closure.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE CPP #-} -- | /Towards Haskell in the Cloud/ (Epstein et al., Haskell Symposium 2011) -- proposes a new type construct called 'static' that characterizes values that -- are known statically. Cloud Haskell uses the @@ -156,14 +157,8 @@ -- instance into scope, but unless proper 'static' support is added to -- ghc we need both the type class argument and the explicit dictionary. module Control.Distributed.Process.Closure - ( -- * Template Haskell support for creating static values and closures - remotable - , mkStatic - , mkClosure - , functionSDict - , functionTDict - -- * Serialization dictionaries (and their static versions) - , SerializableDict(..) + ( -- * Serialization dictionaries (and their static versions) + SerializableDict(..) , staticDecode , sdictUnit , sdictProcessId @@ -181,16 +176,17 @@ , cpSend , cpExpect , cpNewChan - ) where - -import Control.Distributed.Process.Serializable (SerializableDict(..)) -import Control.Distributed.Process.Internal.Closure.TH - ( remotable +#ifdef TemplateHaskellSupport + -- * Template Haskell support for creating static values and closures + , remotable , mkStatic + , mkClosure , functionSDict , functionTDict - , mkClosure - ) +#endif + ) where + +import Control.Distributed.Process.Serializable (SerializableDict(..)) import Control.Distributed.Process.Internal.Closure.BuiltIn ( -- Static dictionaries and associated operations staticDecode @@ -211,3 +207,12 @@ , cpExpect , cpNewChan ) +#ifdef TemplateHaskellSupport +import Control.Distributed.Process.Internal.Closure.TH + ( remotable + , mkStatic + , functionSDict + , functionTDict + , mkClosure + ) +#endif diff --git a/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs --- a/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +++ b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs @@ -33,7 +33,8 @@ , Static , staticLabel , staticApply - , Closure(Closure) + , Closure + , closure , closureApplyStatic , closureApply , staticCompose @@ -159,7 +160,7 @@ -- | 'CP' version of 'Control.Monad.return' returnCP :: forall a. Serializable a => Static (SerializableDict a) -> a -> Closure (Process a) -returnCP dict x = Closure decoder (encode x) +returnCP dict x = closure decoder (encode x) where decoder :: Static (ByteString -> Process a) decoder = returnProcessStatic @@ -193,14 +194,14 @@ -- | 'CP' version of 'link' cpLink :: ProcessId -> Closure (Process ()) -cpLink = Closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode +cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode where linkStatic :: Static (ProcessId -> Process ()) linkStatic = staticLabel "$link" -- | 'CP' version of 'unlink' cpUnlink :: ProcessId -> Closure (Process ()) -cpUnlink = Closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode +cpUnlink = closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode where unlinkStatic :: Static (ProcessId -> Process ()) unlinkStatic = staticLabel "$unlink" @@ -208,7 +209,7 @@ -- | 'CP' version of 'send' cpSend :: forall a. Typeable a => Static (SerializableDict a) -> ProcessId -> CP a () -cpSend dict pid = Closure decoder (encode pid) +cpSend dict pid = closure decoder (encode pid) where decoder :: Static (ByteString -> a -> Process ()) decoder = (sendDictStatic `staticApply` dict) diff --git a/src/Control/Distributed/Process/Internal/Closure/TH.hs b/src/Control/Distributed/Process/Internal/Closure/TH.hs --- a/src/Control/Distributed/Process/Internal/Closure/TH.hs +++ b/src/Control/Distributed/Process/Internal/Closure/TH.hs @@ -52,7 +52,7 @@ , registerStatic , Static , staticLabel - , Closure(Closure) + , closure , staticCompose ) import Control.Distributed.Process.Internal.Types (Process) @@ -97,7 +97,7 @@ mkClosure :: Name -> Q Exp mkClosure n = - [| Closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n)) + [| closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n)) . encode |] diff --git a/src/Control/Distributed/Process/Internal/Node.hs b/src/Control/Distributed/Process/Internal/Node.hs --- a/src/Control/Distributed/Process/Internal/Node.hs +++ b/src/Control/Distributed/Process/Internal/Node.hs @@ -3,13 +3,14 @@ sendPayload , sendBinary , sendMessage + , reconnect ) where import Data.Accessor ((^.), (^=)) import Data.Binary (Binary, encode) import qualified Data.ByteString.Lazy as BSL (toChunks) import qualified Data.ByteString as BSS (ByteString) -import Control.Concurrent.MVar (withMVar, modifyMVar_) +import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_) import Control.Concurrent.Chan (writeChan) import Control.Monad (unless) import qualified Network.Transport as NT @@ -18,6 +19,7 @@ , defaultConnectHints , connect , Reliability(ReliableOrdered) + , close ) import Control.Distributed.Process.Internal.Types ( LocalNode(localState, localEndPoint, localCtrlChan) @@ -44,8 +46,8 @@ Just conn -> do didSend <- NT.send conn payload case didSend of - Left _ -> return False - Right _ -> return True + Left _err -> return False + Right () -> return True Nothing -> return False unless didSend $ writeChan (localCtrlChan node) NCMsg @@ -89,3 +91,13 @@ Nothing -> setupConnBetween node from to where nodeState = localState node + +reconnect :: LocalNode -> Identifier -> Identifier -> IO () +reconnect node from to = + modifyMVar_ (localState node) $ \st -> + case st ^. localConnectionBetween from to of + Nothing -> + return st + Just conn -> do + NT.close conn + return (localConnectionBetween from to ^= Nothing $ st) 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 @@ -59,6 +59,10 @@ , unlinkPort , monitorNode , monitorPort + -- * Reconnecting + , reconnect + , reconnectNode + , reconnectPort ) where #if ! MIN_VERSION_base(4,6,0) @@ -74,7 +78,7 @@ import Control.Applicative ((<$>)) import Control.Exception (Exception, throwIO, SomeException) import qualified Control.Exception as Ex (catch, mask) -import Control.Concurrent.MVar (modifyMVar) +import Control.Distributed.Process.Internal.StrictMVar (modifyMVar) import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM ( STM @@ -346,7 +350,7 @@ -- | Lift 'Control.Exception.bracket' bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c -bracket before after thing = do +bracket before after thing = mask $ \restore -> do a <- before r <- restore (thing a) `onException` after a @@ -505,6 +509,44 @@ case Static.unclosure rtable closure of Left err -> fail $ "Could not resolve closure: " ++ err Right x -> return x + +-------------------------------------------------------------------------------- +-- Reconnecting -- +-------------------------------------------------------------------------------- + +-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered +-- message passing. However, when network connections get disrupted this +-- illusion cannot always be maintained. Once a network connection breaks (even +-- temporarily) no further communication on that connection will be possible. +-- For example, if process A sends a message to process B, and A is then +-- notified (by monitor notification) that it got disconnected from B, A will +-- not be able to send any further messages to B, /unless/ A explicitly +-- indicates that it is acceptable to attempt to reconnect to B using the +-- Cloud Haskell 'reconnect' primitive. +-- +-- Importantly, when A calls 'reconnect' it acknowledges that some messages to +-- B might have been lost. For instance, if A sends messages m1 and m2 to B, +-- then receives a monitor notification that its connection to B has been lost, +-- calls 'reconnect' and then sends m3, it is possible that B will receive m1 +-- and m3 but not m2. +-- +-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay +-- to attempt to reconnect on the next send/. In particular, if no further +-- communication attempts are made to B then A can use reconnect to clean up +-- its connection to B. +reconnect :: ProcessId -> Process () +reconnect = + sendCtrlMsg Nothing . Reconnect . ProcessIdentifier + +-- | Reconnect to a node. See 'reconnect' for more information. +reconnectNode :: NodeId -> Process () +reconnectNode = + sendCtrlMsg Nothing . Reconnect . NodeIdentifier + +-- | Reconnect to a sendport. See 'reconnect' for more information. +reconnectPort :: SendPort a -> Process () +reconnectPort = + sendCtrlMsg Nothing . Reconnect . SendPortIdentifier . sendPortId -------------------------------------------------------------------------------- -- Auxiliary functions -- diff --git a/src/Control/Distributed/Process/Internal/StrictMVar.hs b/src/Control/Distributed/Process/Internal/StrictMVar.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/StrictMVar.hs @@ -0,0 +1,51 @@ +-- | Like Control.Concurrent.MVar.Strict but reduce to HNF, not NF +module Control.Distributed.Process.Internal.StrictMVar + ( StrictMVar + , newEmptyMVar + , newMVar + , takeMVar + , putMVar + , withMVar + , modifyMVar_ + , modifyMVar + ) where + +import Control.Applicative ((<$>)) +import Control.Monad ((>=>)) +import Control.Exception (evaluate) +import qualified Control.Concurrent.MVar as MVar + ( MVar + , newEmptyMVar + , newMVar + , takeMVar + , putMVar + , withMVar + , modifyMVar_ + , modifyMVar + ) + +newtype StrictMVar a = StrictMVar (MVar.MVar a) + +newEmptyMVar :: IO (StrictMVar a) +newEmptyMVar = StrictMVar <$> MVar.newEmptyMVar + +newMVar :: a -> IO (StrictMVar a) +newMVar x = evaluate x >> StrictMVar <$> MVar.newMVar x + +takeMVar :: StrictMVar a -> IO a +takeMVar (StrictMVar v) = MVar.takeMVar v + +putMVar :: StrictMVar a -> a -> IO () +putMVar (StrictMVar v) x = evaluate x >> MVar.putMVar v x + +withMVar :: StrictMVar a -> (a -> IO b) -> IO b +withMVar (StrictMVar v) = MVar.withMVar v + +modifyMVar_ :: StrictMVar a -> (a -> IO a) -> IO () +modifyMVar_ (StrictMVar v) f = MVar.modifyMVar_ v (f >=> evaluate) + +modifyMVar :: StrictMVar a -> (a -> IO (a, b)) -> IO b +modifyMVar (StrictMVar v) f = MVar.modifyMVar v (f >=> evaluateFst) + where + evaluateFst :: (a, b) -> IO (a, b) + evaluateFst (x, y) = evaluate x >> return (x, y) 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 @@ -77,7 +77,6 @@ import Control.Category ((>>>)) import Control.Exception (Exception) import Control.Concurrent (ThreadId) -import Control.Concurrent.MVar (MVar) import Control.Concurrent.Chan (Chan) import Control.Concurrent.STM (TChan, TVar) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) @@ -94,6 +93,7 @@ , showFingerprint ) import Control.Distributed.Process.Internal.CQueue (CQueue) +import Control.Distributed.Process.Internal.StrictMVar (StrictMVar) import Control.Distributed.Static (RemoteTable, Closure) -- import Control.Distributed.Process.Internal.Dynamic (Dynamic) @@ -113,17 +113,17 @@ -- | A local process ID consists of a seed which distinguishes processes from -- different instances of the same local node and a counter data LocalProcessId = LocalProcessId - { lpidUnique :: Int32 - , lpidCounter :: Int32 + { lpidUnique :: {-# UNPACK #-} !Int32 + , lpidCounter :: {-# UNPACK #-} !Int32 } deriving (Eq, Ord, Typeable, Show) -- | Process identifier data ProcessId = ProcessId { -- | The ID of the node the process is running on - processNodeId :: NodeId + processNodeId :: !NodeId -- | Node-local identifier for the process - , processLocalId :: LocalProcessId + , processLocalId :: {-# UNPACK #-} !LocalProcessId } deriving (Eq, Ord, Typeable) @@ -159,7 +159,7 @@ -- | The network endpoint associated with this node , localEndPoint :: NT.EndPoint -- | Local node state - , localState :: MVar LocalNodeState + , localState :: StrictMVar LocalNodeState -- | Channel for the node controller , localCtrlChan :: Chan NCMsg -- | Runtime lookup table for supporting closures @@ -169,17 +169,17 @@ -- | Local node state data LocalNodeState = LocalNodeState - { _localProcesses :: Map LocalProcessId LocalProcess - , _localPidCounter :: Int32 - , _localPidUnique :: Int32 - , _localConnections :: Map (Identifier, Identifier) NT.Connection + { _localProcesses :: !(Map LocalProcessId LocalProcess) + , _localPidCounter :: !Int32 + , _localPidUnique :: !Int32 + , _localConnections :: !(Map (Identifier, Identifier) NT.Connection) } -- | Processes running on our local node data LocalProcess = LocalProcess { processQueue :: CQueue Message , processId :: ProcessId - , processState :: MVar LocalProcessState + , processState :: StrictMVar LocalProcessState , processThread :: ThreadId , processNode :: LocalNode } @@ -190,10 +190,10 @@ -- | Local process state data LocalProcessState = LocalProcessState - { _monitorCounter :: Int32 - , _spawnCounter :: Int32 - , _channelCounter :: Int32 - , _typedChannels :: Map LocalSendPortId TypedChannel + { _monitorCounter :: !Int32 + , _spawnCounter :: !Int32 + , _channelCounter :: !Int32 + , _typedChannels :: !(Map LocalSendPortId TypedChannel) } -- | The Cloud Haskell 'Process' type @@ -214,9 +214,9 @@ -- to create a SendPort. data SendPortId = SendPortId { -- | The ID of the process that will receive messages sent on this port - sendPortProcessId :: ProcessId + sendPortProcessId :: {-# UNPACK #-} !ProcessId -- | Process-local ID of the channel - , sendPortLocalId :: LocalSendPortId + , sendPortLocalId :: {-# UNPACK #-} !LocalSendPortId } deriving (Eq, Ord) @@ -279,9 +279,9 @@ -- | MonitorRef is opaque for regular Cloud Haskell processes data MonitorRef = MonitorRef { -- | ID of the entity to be monitored - monitorRefIdent :: Identifier + monitorRefIdent :: !Identifier -- | Unique to distinguish multiple monitor requests by the same process - , monitorRefCounter :: Int32 + , monitorRefCounter :: !Int32 } deriving (Eq, Ord, Show) @@ -384,6 +384,7 @@ | WhereIs String | Register String (Maybe ProcessId) -- Nothing to unregister | NamedSend String Message + | Reconnect Identifier deriving Show -------------------------------------------------------------------------------- @@ -428,6 +429,7 @@ put (WhereIs label) = putWord8 6 >> put label put (Register label pid) = putWord8 7 >> put label >> put pid put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg) + put (Reconnect dest) = putWord8 9 >> put dest get = do header <- getWord8 case header of @@ -440,6 +442,7 @@ 6 -> WhereIs <$> get 7 -> Register <$> get <*> get 8 -> NamedSend <$> get <*> (payloadToMessage <$> get) + 9 -> Reconnect <$> get _ -> fail "ProcessSignal.get: invalid" instance Binary DiedReason where 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 @@ -25,23 +25,24 @@ , toList , partitionWithKey , filterWithKey + , elems ) -import qualified Data.List as List (delete, (\\)) import Data.Set (Set) -import qualified Data.Set as Set (empty, insert, delete, member, (\\), fromList) +import qualified Data.Set as Set (empty, insert, delete, member, filter) import Data.Foldable (forM_) import Data.Maybe (isJust) import Data.Typeable (Typeable) import Control.Category ((>>>)) import Control.Applicative ((<$>)) -import Control.Monad (void, when, forever) +import Control.Monad (void, when) import Control.Monad.IO.Class (MonadIO, liftIO) -import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify) +import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets) +import qualified Control.Monad.State.Strict as StateT (get, put) import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask) import Control.Exception (throwIO, SomeException, Exception, throwTo) import qualified Control.Exception as Exception (catch) import Control.Concurrent (forkIO) -import Control.Concurrent.MVar +import Control.Distributed.Process.Internal.StrictMVar ( newMVar , withMVar , modifyMVar @@ -64,12 +65,17 @@ , address , closeEndPoint , ConnectionId + , Connection + , close ) import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe) import System.Random (randomIO) import Control.Distributed.Static (RemoteTable, Closure) -import qualified Control.Distributed.Static as Static (unclosure) +import qualified Control.Distributed.Static as Static + ( unclosure + , initRemoteTable + ) import Control.Distributed.Process.Internal.Types ( NodeId(..) , LocalProcessId(..) @@ -85,6 +91,7 @@ , localPidCounter , localPidUnique , localProcessWithId + , localConnections , MonitorRef(..) , ProcessMonitorNotification(..) , NodeMonitorNotification(..) @@ -115,10 +122,10 @@ ( sendBinary , sendMessage , sendPayload + , reconnect ) import Control.Distributed.Process.Internal.Primitives (expect, register, finally) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable) -import qualified Control.Distributed.Static as Static (initRemoteTable) -------------------------------------------------------------------------------- -- Initialization -- @@ -159,11 +166,16 @@ void . forkIO $ handleIncomingMessages node return node +-- Like 'Control.Monad.forever' but sans space leak +{-# INLINE forever' #-} +forever' :: Monad m => m a -> m b +forever' a = let a' = a >> a' in a' + -- | Start and register the service processes on a node -- (for now, this is only the logger) startServiceProcesses :: LocalNode -> IO () startServiceProcesses node = do - logger <- forkProcess node . forever $ do + logger <- forkProcess node . forever' $ do (time, pid, string) <- expect :: Process (String, ProcessId, String) liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string runProcess node $ register "logger" logger @@ -184,75 +196,86 @@ -- | Spawn a new process on a local node forkProcess :: LocalNode -> Process () -> IO ProcessId -forkProcess node proc = modifyMVar (localState node) $ \st -> do - let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter - , lpidUnique = st ^. localPidUnique - } - let pid = ProcessId { processNodeId = localNodeId node - , processLocalId = lpid - } - pst <- newMVar LocalProcessState { _monitorCounter = 0 - , _spawnCounter = 0 - , _channelCounter = 0 - , _typedChannels = Map.empty - } - queue <- newCQueue - (_, lproc) <- fixIO $ \ ~(tid, _) -> do - let lproc = LocalProcess { processQueue = queue - , processId = pid - , processState = pst - , processThread = tid - , processNode = node - } - tid' <- forkIO $ do - reason <- Exception.catch - (runLocalProcess lproc proc >> return DiedNormal) - (return . DiedException . (show :: SomeException -> String)) - -- [Unified: Table 4, rules termination and exiting] - modifyMVar_ (localState node) $ - return . (localProcessWithId lpid ^= Nothing) - writeChan (localCtrlChan node) NCMsg - { ctrlMsgSender = ProcessIdentifier pid - , ctrlMsgSignal = Died (ProcessIdentifier pid) reason - } - return (tid', lproc) +forkProcess node proc = modifyMVar (localState node) startProcess + where + startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId) + startProcess st = do + let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter + , lpidUnique = st ^. localPidUnique + } + let pid = ProcessId { processNodeId = localNodeId node + , processLocalId = lpid + } + pst <- newMVar LocalProcessState { _monitorCounter = 0 + , _spawnCounter = 0 + , _channelCounter = 0 + , _typedChannels = Map.empty + } + queue <- newCQueue + (_, lproc) <- fixIO $ \ ~(tid, _) -> do + let lproc = LocalProcess { processQueue = queue + , processId = pid + , processState = pst + , processThread = tid + , processNode = node + } + tid' <- forkIO $ do + reason <- Exception.catch + (runLocalProcess lproc proc >> return DiedNormal) + (return . DiedException . (show :: SomeException -> String)) + -- [Unified: Table 4, rules termination and exiting] + modifyMVar_ (localState node) (cleanupProcess pid) + writeChan (localCtrlChan node) NCMsg + { ctrlMsgSender = ProcessIdentifier pid + , ctrlMsgSignal = Died (ProcessIdentifier pid) reason + } + return (tid', lproc) - if lpidCounter lpid == maxBound - then do - newUnique <- randomIO - return ( (localProcessWithId lpid ^= Just lproc) - . (localPidCounter ^= 0) - . (localPidUnique ^= newUnique) - $ st - , pid - ) - else - return ( (localProcessWithId lpid ^= Just lproc) - . (localPidCounter ^: (+ 1)) + if lpidCounter lpid == maxBound + then do + newUnique <- randomIO + return ( (localProcessWithId lpid ^= Just lproc) + . (localPidCounter ^= 0) + . (localPidUnique ^= newUnique) + $ st + , pid + ) + else + return ( (localProcessWithId lpid ^= Just lproc) + . (localPidCounter ^: (+ 1)) + $ st + , pid + ) + + cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState + cleanupProcess pid st = do + let pid' = ProcessIdentifier pid + let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections) + mapM_ NT.close (Map.elems affected) + return $ (localProcessWithId (processLocalId pid) ^= Nothing) + . (localConnections ^= unaffected) $ st - , pid - ) handleIncomingMessages :: LocalNode -> IO () -handleIncomingMessages node = go [] Map.empty Map.empty Set.empty +handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty where - go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know + go :: Set NT.ConnectionId -- ^ Connections whose purpose we don't yet know -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes -> Map NT.ConnectionId TypedChannel -- ^ Connections to typed channels -> Set NT.ConnectionId -- ^ Connections to our controller -> IO () - go uninitConns procs chans ctrls = do + go !uninitConns !procs !chans !ctrls = do event <- NT.receive endpoint case event of NT.ConnectionOpened cid _rel _theirAddr -> -- TODO: Check if _rel is ReliableOrdered, and if not, treat as -- (**) below. - go (cid : uninitConns) procs chans ctrls + go (Set.insert cid uninitConns) procs chans ctrls NT.Received cid payload -> case ( Map.lookup cid procs , Map.lookup cid chans , cid `Set.member` ctrls - , cid `elem` uninitConns + , cid `Set.member` uninitConns ) of (Just proc, _, _, _) -> do let msg = payloadToMessage payload @@ -272,7 +295,7 @@ mProc <- withMVar state $ return . (^. localProcessWithId lpid) case mProc of Just proc -> - go (List.delete cid uninitConns) + go (Set.delete cid uninitConns) (Map.insert cid proc procs) chans ctrls @@ -284,7 +307,7 @@ -- remote node as having died, and we should close -- incoming connections (this requires a Transport layer -- extension). (**) - go (List.delete cid uninitConns) procs chans ctrls + go (Set.delete cid uninitConns) procs chans ctrls SendPortIdentifier chId -> do let lcid = sendPortLocalId chId lpid = processLocalId (sendPortProcessId chId) @@ -294,20 +317,20 @@ mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid) case mChannel of Just channel -> - go (List.delete cid uninitConns) + go (Set.delete cid uninitConns) procs (Map.insert cid channel chans) ctrls Nothing -> -- Unknown typed channel -- TODO (**) above - go (List.delete cid uninitConns) procs chans ctrls + go (Set.delete cid uninitConns) procs chans ctrls Nothing -> -- Unknown process -- TODO (**) above - go (List.delete cid uninitConns) procs chans ctrls + go (Set.delete cid uninitConns) procs chans ctrls NodeIdentifier _ -> - go (List.delete cid uninitConns) + go (Set.delete cid uninitConns) procs chans (Set.insert cid ctrls) @@ -316,7 +339,7 @@ -- TODO (**) above go uninitConns procs chans ctrls NT.ConnectionClosed cid -> - go (List.delete cid uninitConns) + go (Set.delete cid uninitConns) (Map.delete cid procs) (Map.delete cid chans) (Set.delete cid ctrls) @@ -327,10 +350,11 @@ { ctrlMsgSender = nid , ctrlMsgSignal = Died nid DiedDisconnect } - go (uninitConns List.\\ cids) - (Map.filterWithKey (\k _ -> k `notElem` cids) procs) - (Map.filterWithKey (\k _ -> k `notElem` cids) chans) - (ctrls Set.\\ Set.fromList cids) + let notRemoved k = k `notElem` cids + go (Set.filter notRemoved uninitConns) + (Map.filterWithKey (const . notRemoved) procs) + (Map.filterWithKey (const . notRemoved) chans) + (Set.filter notRemoved ctrls) NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost Nothing _) _) -> -- TODO: We should treat an asymetrical connection loss (incoming -- connection broken, but outgoing connection still potentially ok) @@ -365,11 +389,11 @@ data NCState = NCState { -- Mapping from remote processes to linked local processes - _links :: Map Identifier (Set ProcessId) + _links :: !(Map Identifier (Set ProcessId)) -- Mapping from remote processes to monitoring local processes - , _monitors :: Map Identifier (Set (ProcessId, MonitorRef)) + , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef))) -- Process registry - , _registry :: Map String ProcessId + , _registry :: !(Map String ProcessId) } newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a } @@ -389,7 +413,7 @@ nodeController :: NC () nodeController = do node <- ask - forever $ do + forever' $ do msg <- liftIO $ readChan (localCtrlChan node) -- [Unified: Table 7, rule nc_forward] @@ -421,6 +445,8 @@ ncEffectWhereIs from label NCMsg from (NamedSend label msg') -> ncEffectNamedSend from label msg' + NCMsg from (Reconnect to) -> + ncEffectReconnect from to unexpected -> error $ "nodeController: unexpected message " ++ show unexpected @@ -438,8 +464,8 @@ case (shouldLink, isLocal node (ProcessIdentifier from)) of (True, _) -> -- [Unified: first rule] case mRef of - Just ref -> modify $ monitorsFor them ^: Set.insert (from, ref) - Nothing -> modify $ linksFor them ^: Set.insert from + Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref) + Nothing -> modify' $ linksFor them ^: Set.insert from (False, True) -> -- [Unified: second rule] notifyDied from them DiedUnknownId mRef (False, False) -> -- [Unified: third rule] @@ -466,7 +492,7 @@ postAsMessage from $ DidUnlinkNode nid SendPortIdentifier cid -> postAsMessage from $ DidUnlinkPort cid - modify $ linksFor them ^: Set.delete from + modify' $ linksFor them ^: Set.delete from -- [Unified: Table 11] ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC () @@ -474,7 +500,7 @@ node <- ask when (isLocal node (ProcessIdentifier from)) $ postAsMessage from $ DidUnmonitor ref - modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref) + modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref) -- [Unified: Table 12] ncEffectDied :: Identifier -> DiedReason -> NC () @@ -495,7 +521,7 @@ when (localOnly <= isLocal node (ProcessIdentifier us)) $ notifyDied us them reason (Just ref) - modify $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons) + modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons) -- [Unified: Table 13] ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC () @@ -517,7 +543,7 @@ -- but mentions it's "very similar to nsend" (Table 14) ncEffectRegister :: String -> Maybe ProcessId -> NC () ncEffectRegister label mPid = - modify $ registryFor label ^= mPid + modify' $ registryFor label ^= mPid -- An acknowledgement is not necessary. If we want a synchronous register, -- it suffices to send a whereis requiry immediately after the register -- (that may not suffice if we do decide for unreliable messaging instead) @@ -544,6 +570,12 @@ (ProcessIdentifier pid) (messageToPayload msg) +-- Reconnecting +ncEffectReconnect :: Identifier -> Identifier -> NC () +ncEffectReconnect from to = do + node <- ask + liftIO $ reconnect node from to + -------------------------------------------------------------------------------- -- Auxiliary -- -------------------------------------------------------------------------------- @@ -589,6 +621,7 @@ destNid (Register _ _) = Nothing destNid (WhereIs _) = Nothing destNid (NamedSend _ _) = Nothing +destNid (Reconnect _) = 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 @@ -691,7 +724,7 @@ splitNotif :: Identifier -> Map Identifier a -> (Map Identifier a, Map Identifier a) -splitNotif ident = Map.partitionWithKey (const . impliesDeathOf ident) +splitNotif ident = Map.partitionWithKey (\k !v -> impliesDeathOf ident k) -- | Does the death of one entity (node, project, channel) imply the death -- of another? @@ -712,3 +745,11 @@ cid' == cid _ `impliesDeathOf` _ = False + +-------------------------------------------------------------------------------- +-- Strict evaluation of the state -- +-------------------------------------------------------------------------------- + +-- | Modify and evaluate the state +modify' :: MonadState s m => (s -> s) -> m () +modify' f = StateT.get >>= \s -> StateT.put $! f s diff --git a/tests/TestClosure.hs b/tests/TestClosure.hs --- a/tests/TestClosure.hs +++ b/tests/TestClosure.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE TemplateHaskell #-} module Main where import Data.ByteString.Lazy (empty) @@ -290,7 +291,7 @@ node <- newLocalNode transport rtable done <- newEmptyMVar forkProcess node $ do - (pid, ref) <- spawnMonitor (localNodeId node) (Closure (staticLabel "ThisDoesNotExist") empty) + (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty) ProcessMonitorNotification ref' pid' _reason <- expect -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure True <- return $ ref' == ref && pid == pid'