diff --git a/benchmarks/Channels.hs b/benchmarks/Channels.hs new file mode 100644 --- /dev/null +++ b/benchmarks/Channels.hs @@ -0,0 +1,41 @@ +-- | Like Latency, but creating lots of channels +import System.Environment +import Control.Monad +import Control.Applicative +import Control.Distributed.Process +import Control.Distributed.Process.Node +import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Data.Binary (encode, decode) +import qualified Data.ByteString.Lazy as BSL + +pingServer :: Process () +pingServer = forever $ do + them <- expect + sendChan them () + -- TODO: should this be automatic? + reconnectPort them + +pingClient :: Int -> ProcessId -> Process () +pingClient n them = do + replicateM_ n $ do + (sc, rc) <- newChan :: Process (SendPort (), ReceivePort ()) + send them sc + receiveChan rc + liftIO . putStrLn $ "Did " ++ show n ++ " pings" + +initialProcess :: String -> Process () +initialProcess "SERVER" = do + us <- getSelfPid + liftIO $ BSL.writeFile "pingServer.pid" (encode us) + pingServer +initialProcess "CLIENT" = do + n <- liftIO $ getLine + them <- liftIO $ decode <$> BSL.readFile "pingServer.pid" + pingClient (read n) them + +main :: IO () +main = do + [role, host, port] <- getArgs + Right transport <- createTransport host port defaultTCPParameters + node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role diff --git a/benchmarks/Latency.hs b/benchmarks/Latency.hs new file mode 100644 --- /dev/null +++ b/benchmarks/Latency.hs @@ -0,0 +1,36 @@ +import System.Environment +import Control.Monad +import Control.Applicative +import Control.Distributed.Process +import Control.Distributed.Process.Node +import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Data.Binary (encode, decode) +import qualified Data.ByteString.Lazy as BSL + +pingServer :: Process () +pingServer = forever $ do + them <- expect + send them () + +pingClient :: Int -> ProcessId -> Process () +pingClient n them = do + us <- getSelfPid + replicateM_ n $ send them us >> (expect :: Process ()) + liftIO . putStrLn $ "Did " ++ show n ++ " pings" + +initialProcess :: String -> Process () +initialProcess "SERVER" = do + us <- getSelfPid + liftIO $ BSL.writeFile "pingServer.pid" (encode us) + pingServer +initialProcess "CLIENT" = do + n <- liftIO $ getLine + them <- liftIO $ decode <$> BSL.readFile "pingServer.pid" + pingClient (read n) them + +main :: IO () +main = do + [role, host, port] <- getArgs + Right transport <- createTransport host port defaultTCPParameters + node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role diff --git a/benchmarks/Spawns.hs b/benchmarks/Spawns.hs new file mode 100644 --- /dev/null +++ b/benchmarks/Spawns.hs @@ -0,0 +1,45 @@ +-- | Like Throughput, but send every ping from a different process +-- (i.e., require a lightweight connection per ping) +import System.Environment +import Control.Monad +import Control.Applicative +import Control.Distributed.Process +import Control.Distributed.Process.Node +import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Data.Binary (encode, decode) +import qualified Data.ByteString.Lazy as BSL + +counter :: Process () +counter = go 0 + where + go :: Int -> Process () + go !n = do + b <- expect + case b of + Nothing -> go (n + 1) + Just them -> send them n >> go 0 + +count :: Int -> ProcessId -> Process () +count n them = do + us <- getSelfPid + replicateM_ n . spawnLocal $ send them (Nothing :: Maybe ProcessId) + send them (Just us) + n' <- expect + liftIO $ print (n == n') + +initialProcess :: String -> Process () +initialProcess "SERVER" = do + us <- getSelfPid + liftIO $ BSL.writeFile "counter.pid" (encode us) + counter +initialProcess "CLIENT" = do + n <- liftIO $ getLine + them <- liftIO $ decode <$> BSL.readFile "counter.pid" + count (read n) them + +main :: IO () +main = do + [role, host, port] <- getArgs + Right transport <- createTransport host port defaultTCPParameters + node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role diff --git a/benchmarks/Throughput.hs b/benchmarks/Throughput.hs new file mode 100644 --- /dev/null +++ b/benchmarks/Throughput.hs @@ -0,0 +1,70 @@ +{-# LANGUAGE DeriveDataTypeable #-} +import System.Environment +import Control.Monad +import Control.Applicative +import Control.Distributed.Process +import Control.Distributed.Process.Node +import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Data.Binary +import qualified Data.ByteString.Lazy as BSL +import Data.Typeable + +data SizedList a = SizedList { size :: Int , elems :: [a] } + deriving (Typeable) + +instance Binary a => Binary (SizedList a) where + put (SizedList sz xs) = put sz >> mapM_ put xs + get = do + sz <- get + xs <- getMany sz + return (SizedList sz xs) + +-- Copied from Data.Binary +getMany :: Binary a => Int -> Get [a] +getMany = go [] + where + go xs 0 = return $! reverse xs + go xs i = do x <- get + x `seq` go (x:xs) (i-1) +{-# INLINE getMany #-} + +nats :: Int -> SizedList Int +nats = \n -> SizedList n (aux n) + where + aux 0 = [] + aux n = n : aux (n - 1) + +counter :: Process () +counter = go 0 + where + go :: Int -> Process () + go !n = + receiveWait + [ match $ \xs -> go (n + size (xs :: SizedList Int)) + , match $ \them -> send them n >> go 0 + ] + +count :: (Int, Int) -> ProcessId -> Process () +count (packets, sz) them = do + us <- getSelfPid + replicateM_ packets $ send them (nats sz) + send them us + n' <- expect + liftIO $ print (packets * sz, n' == packets * sz) + +initialProcess :: String -> Process () +initialProcess "SERVER" = do + us <- getSelfPid + liftIO $ BSL.writeFile "counter.pid" (encode us) + counter +initialProcess "CLIENT" = do + n <- liftIO getLine + them <- liftIO $ decode <$> BSL.readFile "counter.pid" + count (read n) them + +main :: IO () +main = do + [role, host, port] <- getArgs + Right transport <- createTransport host port defaultTCPParameters + node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role 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.1 +Version: 0.4.0 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -33,10 +33,14 @@ description: Build with Template Haskell support default: True +flag benchmarks + description: Build benchmarks + default: False + Library Build-Depends: base >= 4.4 && < 5, binary >= 0.5 && < 0.6, - network-transport >= 0.2 && < 0.3, + network-transport >= 0.3 && < 0.4, stm >= 2.3 && < 2.5, transformers >= 0.2 && < 0.4, mtl >= 2.0 && < 2.2, @@ -58,8 +62,11 @@ Control.Distributed.Process.Internal.CQueue, Control.Distributed.Process.Internal.Types, Control.Distributed.Process.Internal.Closure.BuiltIn, - Control.Distributed.Process.Internal.Node - Control.Distributed.Process.Internal.StrictMVar + Control.Distributed.Process.Internal.Messaging, + Control.Distributed.Process.Internal.StrictList, + Control.Distributed.Process.Internal.StrictMVar, + Control.Distributed.Process.Internal.WeakTQueue + Control.Distributed.Process.Internal.StrictContainerAccessors Extensions: RankNTypes, ScopedTypeVariables, FlexibleInstances, @@ -84,9 +91,10 @@ random >= 1.0 && < 1.1, ansi-terminal >= 0.5 && < 0.6, distributed-process >= 0.3 && < 0.4, - network-transport >= 0.2 && < 0.3, - network-transport-tcp >= 0.2 && < 0.3, - binary >= 0.5 && < 0.6 + network-transport >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4, + binary >= 0.5 && < 0.6, + network >= 2.3 && < 2.5 Other-modules: TestAuxiliary Extensions: CPP, ScopedTypeVariables, @@ -103,11 +111,64 @@ ansi-terminal >= 0.5 && < 0.6, 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 + network-transport >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4, + bytestring >= 0.9 && < 0.11, + network >= 2.3 && < 2.5 Other-modules: TestAuxiliary Extensions: CPP, ScopedTypeVariables ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests + +Executable distributed-process-throughput + if flag(benchmarks) + Build-Depends: base >= 4.4 && < 5, + distributed-process >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4, + bytestring >= 0.9 && < 0.11, + binary >= 0.5 && < 0.6 + else + buildable: False + Main-Is: benchmarks/Throughput.hs + ghc-options: -Wall + Extensions: BangPatterns + +Executable distributed-process-latency + if flag(benchmarks) + Build-Depends: base >= 4.4 && < 5, + distributed-process >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4, + bytestring >= 0.9 && < 0.11, + binary >= 0.5 && < 0.6 + else + buildable: False + Main-Is: benchmarks/Latency.hs + ghc-options: -Wall + Extensions: BangPatterns + +Executable distributed-process-channels + if flag(benchmarks) + Build-Depends: base >= 4.4 && < 5, + distributed-process >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4, + bytestring >= 0.9 && < 0.11, + binary >= 0.5 && < 0.6 + else + buildable: False + Main-Is: benchmarks/Channels.hs + ghc-options: -Wall + Extensions: BangPatterns + +Executable distributed-process-spawns + if flag(benchmarks) + Build-Depends: base >= 4.4 && < 5, + distributed-process >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4, + bytestring >= 0.9 && < 0.11, + binary >= 0.5 && < 0.6 + else + buildable: False + Main-Is: benchmarks/Spawns.hs + ghc-options: -Wall + Extensions: BangPatterns 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 @@ -35,6 +35,8 @@ , match , matchIf , matchUnknown + , AbstractMessage(..) + , matchAny -- * Process management , spawn , call @@ -66,6 +68,7 @@ , Closure , closure , Static + , unStatic , unClosure , RemoteTable -- * Logging @@ -77,7 +80,6 @@ , nsend , registerRemote , unregisterRemote - , whereisRemote , whereisRemoteAsync , nsendRemote , WhereIsReply(..) @@ -101,7 +103,6 @@ , spawnChannelLocal -- * Reconnecting , reconnect - , reconnectNode , reconnectPort ) where @@ -113,11 +114,14 @@ import Control.Monad.IO.Class (liftIO) import Control.Applicative ((<$>)) import Control.Monad.Reader (ask) +import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) import Control.Distributed.Static ( Closure , closure , Static , RemoteTable + , closureCompose + , staticClosure ) import Control.Distributed.Process.Internal.Types ( NodeId(..) @@ -138,8 +142,9 @@ , SendPortId(..) , WhereIsReply(..) , LocalProcess(processNode) + , nullProcessId ) -import Control.Distributed.Process.Serializable (SerializableDict) +import Control.Distributed.Process.Serializable (Serializable, SerializableDict) import Control.Distributed.Process.Internal.Closure.BuiltIn ( sdictUnit , sdictSendPort @@ -154,7 +159,6 @@ , cpSend , cpNewChan ) -import Control.Distributed.Static (closureCompose, staticClosure) import Control.Distributed.Process.Internal.Primitives ( -- Basic messaging send @@ -172,6 +176,8 @@ , match , matchIf , matchUnknown + , AbstractMessage(..) + , matchAny -- Process management , terminate , ProcessTerminationException(..) @@ -197,10 +203,10 @@ , nsend , registerRemote , unregisterRemote - , whereisRemote , whereisRemoteAsync , nsendRemote -- Closures + , unStatic , unClosure -- Exception handling , catch @@ -214,10 +220,8 @@ , spawnAsync -- Reconnecting , reconnect - , reconnectNode , reconnectPort ) -import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Node (forkProcess) -- INTERNAL NOTES @@ -293,10 +297,6 @@ spawn :: NodeId -> Closure (Process ()) -> Process ProcessId spawn nid proc = do us <- getSelfPid - -- Since we throw an exception when the remote node dies, we could use - -- linkNode instead. However, we don't have a way of "scoped" linking so if - -- we call linkNode here, and unlinkNode after, then we might remove a link - -- that was already set up mRef <- monitorNode nid sRef <- spawnAsync nid $ cpLink us `seqCP` cpExpect sdictUnit @@ -310,7 +310,7 @@ ] unmonitor mRef case mPid of - Left err -> fail $ "spawn: remote node failed: " ++ show err + Left _err -> return (nullProcessId nid) Right pid -> send pid () >> return pid -- | Spawn a process and link to it @@ -411,6 +411,12 @@ -> Process (SendPort a) spawnChannelLocal proc = do node <- processNode <$> ask - (sport, rport) <- newChan - _ <- liftIO $ forkProcess node (proc rport) - return sport + liftIO $ do + mvar <- newEmptyMVar + forkProcess node $ do + -- It is important that we allocate the new channel in the new process, + -- because otherwise it will be associated with the wrong process ID + (sport, rport) <- newChan + liftIO $ putMVar mvar sport + proc rport + takeMVar mvar 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 @@ -179,6 +179,7 @@ #ifdef TemplateHaskellSupport -- * Template Haskell support for creating static values and closures , remotable + , remotableDecl , mkStatic , mkClosure , functionSDict @@ -210,6 +211,7 @@ #ifdef TemplateHaskellSupport import Control.Distributed.Process.Internal.Closure.TH ( remotable + , remotableDecl , mkStatic , functionSDict , functionTDict diff --git a/src/Control/Distributed/Process/Internal/CQueue.hs b/src/Control/Distributed/Process/Internal/CQueue.hs --- a/src/Control/Distributed/Process/Internal/CQueue.hs +++ b/src/Control/Distributed/Process/Internal/CQueue.hs @@ -1,13 +1,15 @@ -- | Concurrent queue for single reader, single writer +{-# LANGUAGE MagicHash, UnboxedTuples #-} module Control.Distributed.Process.Internal.CQueue ( CQueue , BlockSpec(..) , newCQueue , enqueue , dequeue + , mkWeakCQueue ) where -import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar) +import Prelude hiding (length, reverse) import Control.Concurrent.STM ( atomically , TChan @@ -19,16 +21,34 @@ import Control.Applicative ((<$>), (<*>)) import Control.Exception (mask, onException) import System.Timeout (timeout) +import Control.Distributed.Process.Internal.StrictMVar + ( StrictMVar(StrictMVar) + , newMVar + , takeMVar + , putMVar + ) +import Control.Distributed.Process.Internal.StrictList + ( StrictList(..) + , reverse + , reverse' + ) +import GHC.MVar (MVar(MVar)) +import GHC.IO (IO(IO)) +import GHC.Prim (mkWeak#) +import GHC.Weak (Weak(Weak)) -- We use a TCHan rather than a Chan so that we have a non-blocking read -data CQueue a = CQueue (MVar [a]) -- Arrived - (TChan a) -- Incoming +data CQueue a = CQueue (StrictMVar (StrictList a)) -- Arrived + (TChan a) -- Incoming newCQueue :: IO (CQueue a) -newCQueue = CQueue <$> newMVar [] <*> atomically newTChan +newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan +-- | Enqueue an element +-- +-- Enqueue is strict. enqueue :: CQueue a -> a -> IO () -enqueue (CQueue _arrived incoming) a = atomically $ writeTChan incoming a +enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a data BlockSpec = NonBlocking @@ -51,7 +71,7 @@ arr <- takeMVar arrived -- We first check the arrived messages. If we get interrupted during this -- search, we just put the MVar back (we haven't read from the Chan yet) - (arr', mb) <- onException (restore (checkArrived [] arr)) + (arr', mb) <- onException (restore (checkArrived Nil arr)) (putMVar arrived arr) case (mb, blockSpec) of (Just b, _) -> do @@ -65,15 +85,15 @@ timeout n $ checkBlocking arr' -- We reverse the accumulator on return only if we find a match - checkArrived :: [a] -> [a] -> IO ([a], Maybe b) - checkArrived acc [] = return (acc, Nothing) - checkArrived acc (x:xs) = + checkArrived :: StrictList a -> StrictList a -> IO (StrictList a, Maybe b) + checkArrived acc Nil = return (acc, Nothing) + checkArrived acc (Cons x xs) = case check x of - Just y -> return (reverse acc ++ xs, Just y) - Nothing -> checkArrived (x:acc) xs + Just y -> return (reverse' acc xs, Just y) + Nothing -> checkArrived (Cons x acc) xs -- If we call checkBlocking there may or may not be a timeout - checkBlocking :: [a] -> IO b + checkBlocking :: StrictList a -> IO b checkBlocking acc = do -- readTChan is a blocking call, and hence is interruptable. If it is -- interrupted, we put the value of the accumulator in 'arrived' @@ -82,18 +102,18 @@ x <- onException (atomically $ readTChan incoming) (putMVar arrived $ reverse acc) case check x of - Nothing -> checkBlocking (x:acc) + Nothing -> checkBlocking (Cons x acc) Just y -> putMVar arrived (reverse acc) >> return y -- checkNonBlocking is only called if there is no timeout - checkNonBlocking :: [a] -> IO (Maybe b) + checkNonBlocking :: StrictList a -> IO (Maybe b) checkNonBlocking acc = do -- tryReadTChan is *not* interruptible mx <- atomically $ tryReadTChan incoming case mx of Nothing -> putMVar arrived (reverse acc) >> return Nothing Just x -> case check x of - Nothing -> checkNonBlocking (x:acc) + Nothing -> checkNonBlocking (Cons x acc) Just y -> putMVar arrived (reverse acc) >> return (Just y) check :: a -> Maybe b @@ -103,3 +123,8 @@ checkMatches [] _ = Nothing checkMatches (m:ms) a = case m a of Nothing -> checkMatches ms a Just b -> Just b + +-- | Weak reference to a CQueue +mkWeakCQueue :: CQueue a -> IO () -> IO (Weak (CQueue a)) +mkWeakCQueue m@(CQueue (StrictMVar (MVar m#)) _) f = IO $ \s -> + case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #) 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 @@ -3,6 +3,7 @@ module Control.Distributed.Process.Internal.Closure.TH ( -- * User-level API remotable + , remotableDecl , mkStatic , functionSDict , functionTDict @@ -20,7 +21,7 @@ , mkName , nameBase -- Algebraic data types - , Dec + , Dec(SigD) , Exp , Type(AppT, ForallT, VarT, ArrowT) , Info(VarI) @@ -39,6 +40,7 @@ , funD , sigD ) +import Data.Maybe (catMaybes) import Data.Binary (encode) import Data.Generics (everywhereM, mkM, gmapM) import Data.Rank1Dynamic (toDynamic) @@ -69,10 +71,47 @@ -- of functions remotable :: [Name] -> Q [Dec] remotable ns = do - (closures, inserts) <- unzip <$> mapM generateDefs ns - rtable <- createMetaData (concat inserts) - return $ concat closures ++ rtable + types <- mapM getType ns + (closures, inserts) <- unzip <$> mapM generateDefs types + rtable <- createMetaData (mkName "__remoteTable") (concat inserts) + return $ concat closures ++ rtable +-- | Like 'remotable', but parameterized by the declaration of a function +-- instead of the function name. So where for 'remotable' you'd do +-- +-- > f :: T1 -> T2 +-- > f = ... +-- > +-- > remotable ['f] +-- +-- with 'remotableDecl' you would instead do +-- +-- > remotableDecl [ +-- > [d| f :: T1 -> T2 ; +-- > f = ... +-- > |] +-- > ] +-- +-- 'remotableDecl' creates the function specified as well as the various +-- dictionaries and static versions that 'remotable' also creates. +-- 'remotableDecl' is sometimes necessary when you want to refer to, say, +-- @$(mkClosure 'f)@ within the definition of @f@ itself. +-- +-- NOTE: 'remotableDecl' creates @__remoteTableDecl@ instead of @__remoteTable@ +-- so that you can use both 'remotable' and 'remotableDecl' within the same +-- module. +remotableDecl :: [Q [Dec]] -> Q [Dec] +remotableDecl qDecs = do + decs <- concat <$> sequence qDecs + let types = catMaybes (map typeOf decs) + (closures, inserts) <- unzip <$> mapM generateDefs types + rtable <- createMetaData (mkName "__remoteTableDecl") (concat inserts) + return $ decs ++ concat closures ++ rtable + where + typeOf :: Dec -> Maybe (Name, Type) + typeOf (SigD name typ) = Just (name, typ) + typeOf _ = Nothing + -- | Construct a static value. -- -- If @f : forall a1 .. an. T@ @@ -95,6 +134,10 @@ functionTDict :: Name -> Q Exp functionTDict = varE . tdictName +-- | If @f : T1 -> T2@ then @$(mkClosure 'f) :: T1 -> Closure T2@. +-- +-- TODO: The current version of mkClosure is too polymorphic +-- (@forall a. Binary a => a -> Closure T2). mkClosure :: Name -> Q Exp mkClosure n = [| closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n)) @@ -106,49 +149,44 @@ -------------------------------------------------------------------------------- -- | Generate the code to add the metadata to the CH runtime -createMetaData :: [Q Exp] -> Q [Dec] -createMetaData is = - [d| __remoteTable :: RemoteTable -> RemoteTable ; - __remoteTable = $(compose is) - |] +createMetaData :: Name -> [Q Exp] -> Q [Dec] +createMetaData name is = + sequence [ sigD name [t| RemoteTable -> RemoteTable |] + , sfnD name (compose is) + ] -generateDefs :: Name -> Q ([Dec], [Q Exp]) -generateDefs n = do +generateDefs :: (Name, Type) -> Q ([Dec], [Q Exp]) +generateDefs (origName, fullType) = do proc <- [t| Process |] - mType <- getType n - case mType of - Just (origName, typ) -> do - let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono) - _ -> ([], typ) + let (typVars, typ') = case fullType of ForallT vars [] mono -> (vars, mono) + _ -> ([], fullType) - -- The main "static" entry - (static, register) <- makeStatic origName typVars typ' - - -- If n :: T1 -> T2, static serializable dictionary for T1 - -- TODO: we should check if arg is an instance of Serializable, but we cannot - -- http://hackage.haskell.org/trac/ghc/ticket/7066 - (sdict, registerSDict) <- case (typVars, typ') of - ([], ArrowT `AppT` arg `AppT` _res) -> - makeDict (sdictName origName) arg - _ -> - return ([], []) - - -- If n :: T1 -> Process T2, static serializable dictionary for T2 - -- TODO: check if T2 is serializable (same as above) - (tdict, registerTDict) <- case (typVars, typ') of - ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc -> - makeDict (tdictName origName) res - _ -> - return ([], []) - - return ( concat [static, sdict, tdict] - , concat [register, registerSDict, registerTDict] - ) + -- The main "static" entry + (static, register) <- makeStatic typVars typ' + + -- If n :: T1 -> T2, static serializable dictionary for T1 + -- TODO: we should check if arg is an instance of Serializable, but we cannot + -- http://hackage.haskell.org/trac/ghc/ticket/7066 + (sdict, registerSDict) <- case (typVars, typ') of + ([], ArrowT `AppT` arg `AppT` _res) -> + makeDict (sdictName origName) arg _ -> - fail $ "remotable: " ++ show n ++ " not found" + return ([], []) + + -- If n :: T1 -> Process T2, static serializable dictionary for T2 + -- TODO: check if T2 is serializable (same as above) + (tdict, registerTDict) <- case (typVars, typ') of + ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc -> + makeDict (tdictName origName) res + _ -> + return ([], []) + + return ( concat [static, sdict, tdict] + , concat [register, registerSDict, registerTDict] + ) where - makeStatic :: Name -> [TyVarBndr] -> Type -> Q ([Dec], [Q Exp]) - makeStatic origName typVars typ = do + makeStatic :: [TyVarBndr] -> Type -> Q ([Dec], [Q Exp]) + makeStatic typVars typ = do static <- generateStatic origName typVars typ let dyn = case typVars of [] -> [| toDynamic $(varE origName) |] @@ -190,8 +228,6 @@ Just t -> t applySubst s t = gmapM (mkM (applySubst s)) t - - -- | Generate a static value generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec] generateStatic n xs typ = do @@ -240,12 +276,12 @@ stringE = litE . stringL -- | Look up the "original name" (module:name) and type of a top-level function -getType :: Name -> Q (Maybe (Name, Type)) +getType :: Name -> Q (Name, Type) getType name = do info <- reify name case info of - VarI origName typ _ _ -> return $ Just (origName, typ) - _ -> return Nothing + VarI origName typ _ _ -> return (origName, typ) + _ -> fail $ show name ++ " not found" -- | Variation on 'funD' which takes a single expression to define the function sfnD :: Name -> Q Exp -> Q Dec diff --git a/src/Control/Distributed/Process/Internal/Messaging.hs b/src/Control/Distributed/Process/Internal/Messaging.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Messaging.hs @@ -0,0 +1,169 @@ +module Control.Distributed.Process.Internal.Messaging + ( sendPayload + , sendBinary + , sendMessage + , disconnect + , closeImplicitReconnections + , impliesDeathOf + ) where + +import Data.Accessor ((^.), (^=)) +import Data.Binary (Binary, encode) +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.Concurrent.Chan (writeChan) +import Control.Monad (unless) +import qualified Network.Transport as NT + ( Connection + , send + , defaultConnectHints + , connect + , Reliability(ReliableOrdered) + , close + ) +import Control.Distributed.Process.Internal.Types + ( LocalNode(localState, localEndPoint, localCtrlChan) + , Identifier + , localConnections + , localConnectionBetween + , nodeAddress + , nodeOf + , messageToPayload + , createMessage + , NCMsg(..) + , ProcessSignal(Died) + , DiedReason(DiedDisconnect) + , ImplicitReconnect(WithImplicitReconnect) + , NodeId + , ProcessId(processNodeId) + , SendPortId(sendPortProcessId) + , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier) + ) +import Control.Distributed.Process.Serializable (Serializable) + +-------------------------------------------------------------------------------- +-- Message sending -- +-------------------------------------------------------------------------------- + +sendPayload :: LocalNode + -> Identifier + -> Identifier + -> ImplicitReconnect + -> [BSS.ByteString] + -> IO () +sendPayload node from to implicitReconnect payload = do + mConn <- connBetween node from to implicitReconnect + didSend <- case mConn of + Just conn -> do + didSend <- NT.send conn payload + case didSend of + Left _err -> return False + Right () -> return True + Nothing -> return False + unless didSend $ do + writeChan (localCtrlChan node) NCMsg + { ctrlMsgSender = to + , ctrlMsgSignal = Died to DiedDisconnect + } + +sendBinary :: Binary a + => LocalNode + -> Identifier + -> Identifier + -> ImplicitReconnect + -> a + -> IO () +sendBinary node from to implicitReconnect + = sendPayload node from to implicitReconnect . BSL.toChunks . encode + +sendMessage :: Serializable a + => LocalNode + -> Identifier + -> Identifier + -> ImplicitReconnect + -> a + -> IO () +sendMessage node from to implicitReconnect = + sendPayload node from to implicitReconnect . messageToPayload . createMessage + +setupConnBetween :: LocalNode + -> Identifier + -> Identifier + -> ImplicitReconnect + -> IO (Maybe NT.Connection) +setupConnBetween node from to implicitReconnect = do + mConn <- NT.connect endPoint + (nodeAddress . nodeOf $ to) + NT.ReliableOrdered + NT.defaultConnectHints + case mConn of + Right conn -> do + didSend <- NT.send conn (BSL.toChunks . encode $ to) + case didSend of + Left _ -> + return Nothing + Right () -> do + modifyMVar_ nodeState $ return . + (localConnectionBetween from to ^= Just (conn, implicitReconnect)) + return $ Just conn + Left _ -> + return Nothing + where + endPoint = localEndPoint node + nodeState = localState node + +connBetween :: LocalNode + -> Identifier + -> Identifier + -> ImplicitReconnect + -> IO (Maybe NT.Connection) +connBetween node from to implicitReconnect = do + mConn <- withMVar nodeState $ 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 + Nothing -> + return st + Just (conn, _) -> do + NT.close conn + return (localConnectionBetween from to ^= Nothing $ st) + +closeImplicitReconnections :: LocalNode -> Identifier -> IO () +closeImplicitReconnections node to = + modifyMVar_ (localState node) $ \st -> 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) + +-- | @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) +impliesDeathOf :: Identifier + -> Identifier + -> Bool +NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' = + nid' == nid +NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid = + processNodeId pid == nid +NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid = + processNodeId (sendPortProcessId cid) == nid +ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' = + pid' == pid +ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid = + sendPortProcessId cid == pid +SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' = + cid' == cid +_ `impliesDeathOf` _ = + False diff --git a/src/Control/Distributed/Process/Internal/Node.hs b/src/Control/Distributed/Process/Internal/Node.hs deleted file mode 100644 --- a/src/Control/Distributed/Process/Internal/Node.hs +++ /dev/null @@ -1,103 +0,0 @@ -module Control.Distributed.Process.Internal.Node - ( -- * Message sending - 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.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_) -import Control.Concurrent.Chan (writeChan) -import Control.Monad (unless) -import qualified Network.Transport as NT - ( Connection - , send - , defaultConnectHints - , connect - , Reliability(ReliableOrdered) - , close - ) -import Control.Distributed.Process.Internal.Types - ( LocalNode(localState, localEndPoint, localCtrlChan) - , Identifier - , localConnectionBetween - , nodeAddress - , nodeOf - , messageToPayload - , createMessage - , NCMsg(..) - , ProcessSignal(Died) - , DiedReason(DiedDisconnect) - ) -import Control.Distributed.Process.Serializable (Serializable) - --------------------------------------------------------------------------------- --- Message sending -- --------------------------------------------------------------------------------- - -sendPayload :: LocalNode -> Identifier -> Identifier -> [BSS.ByteString] -> IO () -sendPayload node from to payload = do - mConn <- connBetween node from to - didSend <- case mConn of - Just conn -> do - didSend <- NT.send conn payload - case didSend of - Left _err -> return False - Right () -> return True - Nothing -> return False - unless didSend $ - writeChan (localCtrlChan node) NCMsg - { ctrlMsgSender = to - , ctrlMsgSignal = Died to DiedDisconnect - } - -sendBinary :: Binary a => LocalNode -> Identifier -> Identifier -> a -> IO () -sendBinary node from to = sendPayload node from to . BSL.toChunks . encode - -sendMessage :: Serializable a => LocalNode -> Identifier -> Identifier -> a -> IO () -sendMessage node from to = sendPayload node from to . messageToPayload . createMessage - -setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection) -setupConnBetween node from to = do - mConn <- NT.connect endPoint - (nodeAddress . nodeOf $ to) - NT.ReliableOrdered - NT.defaultConnectHints - case mConn of - Right conn -> do - didSend <- NT.send conn (BSL.toChunks . encode $ to) - case didSend of - Left _ -> - return Nothing - Right () -> do - modifyMVar_ nodeState $ return . - (localConnectionBetween from to ^= Just conn) - return $ Just conn - Left _ -> - return Nothing - where - endPoint = localEndPoint node - nodeState = localState node - -connBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection) -connBetween node from to = do - mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to) - case mConn of - Just conn -> return $ Just conn - 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 @@ -19,6 +19,8 @@ , match , matchIf , matchUnknown + , AbstractMessage(..) + , matchAny -- * Process management , terminate , ProcessTerminationException(..) @@ -38,11 +40,11 @@ , nsend , registerRemote , unregisterRemote - , whereisRemote , whereisRemoteAsync , nsendRemote -- * Closures , unClosure + , unStatic -- * Exception handling , catch , mask @@ -61,7 +63,6 @@ , monitorPort -- * Reconnecting , reconnect - , reconnectNode , reconnectPort ) where @@ -78,14 +79,16 @@ import Control.Applicative ((<$>)) import Control.Exception (Exception, throwIO, SomeException) import qualified Control.Exception as Ex (catch, mask) -import Control.Distributed.Process.Internal.StrictMVar (modifyMVar) +import Control.Distributed.Process.Internal.StrictMVar + ( StrictMVar + , modifyMVar + , modifyMVar_ + ) import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM ( STM , atomically , orElse - , newTChan - , readTChan , newTVar , readTVar , writeTVar @@ -93,9 +96,9 @@ import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..)) import Control.Distributed.Process.Serializable (Serializable, fingerprint) import Data.Accessor ((^.), (^:), (^=)) -import Control.Distributed.Static (Closure) +import Control.Distributed.Static (Closure, Static) import Data.Rank1Typeable (Typeable) -import qualified Control.Distributed.Static as Static (unclosure) +import qualified Control.Distributed.Static as Static (unstatic, unclosure) import Control.Distributed.Process.Internal.Types ( NodeId(..) , ProcessId(..) @@ -123,8 +126,22 @@ , WhereIsReply(..) , createMessage , runLocalProcess + , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect) + , LocalProcessState + , LocalSendPortId + , messageToPayload ) -import Control.Distributed.Process.Internal.Node (sendMessage, sendBinary) +import Control.Distributed.Process.Internal.Messaging + ( sendMessage + , sendBinary + , sendPayload + , disconnect + ) +import Control.Distributed.Process.Internal.WeakTQueue + ( newTQueueIO + , readTQueue + , mkWeakTQueue + ) -------------------------------------------------------------------------------- -- Basic messaging -- @@ -139,6 +156,7 @@ liftIO $ sendMessage (processNode proc) (ProcessIdentifier (processId proc)) (ProcessIdentifier them) + NoImplicitReconnect msg -- | Wait for a message of a specific type @@ -152,21 +170,26 @@ -- | Create a new typed channel newChan :: Serializable a => Process (SendPort a, ReceivePort a) newChan = do - proc <- ask - liftIO . modifyMVar (processState proc) $ \st -> do - chan <- liftIO . atomically $ newTChan - let lcid = st ^. channelCounter - cid = SendPortId { sendPortProcessId = processId proc - , sendPortLocalId = lcid - } - sport = SendPort cid - rport = ReceivePortSingle chan - tch = TypedChannel chan - return ( (channelCounter ^: (+ 1)) - . (typedChannelWithId lcid ^= Just tch) - $ st - , (sport, rport) - ) + proc <- ask + liftIO . modifyMVar (processState proc) $ \st -> do + let lcid = st ^. channelCounter + let cid = SendPortId { sendPortProcessId = processId proc + , sendPortLocalId = lcid + } + let sport = SendPort cid + chan <- liftIO newTQueueIO + chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid + let rport = ReceivePortSingle chan + let tch = TypedChannel chan' + return ( (channelCounter ^: (+ 1)) + . (typedChannelWithId lcid ^= Just tch) + $ st + , (sport, rport) + ) + where + finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO () + finalizer processState lcid = modifyMVar_ processState $ + return . (typedChannelWithId lcid ^= Nothing) -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process () @@ -175,6 +198,7 @@ liftIO $ sendBinary (processNode proc) (ProcessIdentifier (processId proc)) (SendPortIdentifier cid) + NoImplicitReconnect msg -- | Wait for a message on a typed channel @@ -183,7 +207,7 @@ where receiveSTM :: ReceivePort a -> STM a receiveSTM (ReceivePortSingle c) = - readTChan c + readTQueue c receiveSTM (ReceivePortBiased ps) = foldr1 orElse (map receiveSTM ps) receiveSTM (ReceivePortRR psVar) = do @@ -243,12 +267,35 @@ -- | Match against any message of the right type that satisfies a predicate matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b matchIf c p = Match $ \msg -> - let decoded :: a - decoded = decode . messageEncoding $ msg in - if messageFingerprint msg == fingerprint (undefined :: a) && c decoded - then Just $ p decoded - else Nothing + case messageFingerprint msg == fingerprint (undefined :: a) of + True | c decoded -> Just (p decoded) + where + decoded :: a + -- Make sure the value is fully decoded so that we don't hang to + -- bytestrings when the process calling 'matchIf' doesn't process + -- the values immediately + !decoded = decode (messageEncoding msg) + _ -> Nothing +data AbstractMessage = AbstractMessage { + forward :: ProcessId -> Process () + } + +-- | Match against an arbitrary message +matchAny :: forall b. (AbstractMessage -> Process b) -> Match b +matchAny p = Match $ Just . p . abstract + where + abstract :: Message -> AbstractMessage + abstract msg = AbstractMessage { + forward = \them -> do + proc <- ask + liftIO $ sendPayload (processNode proc) + (ProcessIdentifier (processId proc)) + (ProcessIdentifier them) + NoImplicitReconnect + (messageToPayload msg) + } + -- | Remove any message from the queue matchUnknown :: Process b -> Match b matchUnknown = Match . const . Just @@ -291,7 +338,12 @@ monitor :: ProcessId -> Process MonitorRef monitor = monitor' . ProcessIdentifier --- | Remove a link (synchronous) +-- | Remove a link +-- +-- This is synchronous in the sense that once it returns you are guaranteed +-- that no exception will be raised if the remote process dies. However, it is +-- asynchronous in the sense that we do not wait for a response from the remote +-- node. unlink :: ProcessId -> Process () unlink pid = do unlinkAsync pid @@ -299,7 +351,9 @@ (\_ -> return ()) ] --- | Remove a node link (synchronous) +-- | Remove a node link +-- +-- This has the same synchronous/asynchronous nature as 'unlink'. unlinkNode :: NodeId -> Process () unlinkNode nid = do unlinkNodeAsync nid @@ -307,7 +361,9 @@ (\_ -> return ()) ] --- | Remove a channel (send port) link (synchronous) +-- | Remove a channel (send port) link +-- +-- This has the same synchronous/asynchronous nature as 'unlink'. unlinkPort :: SendPort a -> Process () unlinkPort sport = do unlinkPortAsync sport @@ -315,7 +371,9 @@ (\_ -> return ()) ] --- | Remove a monitor (synchronous) +-- | Remove a monitor +-- +-- This has the same synchronous/asynchronous nature as 'unlink'. unmonitor :: MonitorRef -> Process () unmonitor ref = do unmonitorAsync ref @@ -382,12 +440,12 @@ sendCtrlMsg (Just nid) $ Spawn proc spawnRef return spawnRef --- | Monitor a node +-- | Monitor a node (asynchronous) monitorNode :: NodeId -> Process MonitorRef monitorNode = monitor' . NodeIdentifier --- | Monitor a typed channel +-- | Monitor a typed channel (asynchronous) monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef monitorPort (SendPort cid) = monitor' (SendPortIdentifier cid) @@ -397,11 +455,11 @@ unmonitorAsync = sendCtrlMsg Nothing . Unmonitor --- | Link to a node +-- | Link to a node (asynchronous) linkNode :: NodeId -> Process () linkNode = link' . NodeIdentifier --- | Link to a channel (send port) +-- | Link to a channel (asynchronous) linkPort :: SendPort a -> Process () linkPort (SendPort cid) = link' (SendPortIdentifier cid) @@ -465,7 +523,7 @@ unregisterRemote nid label = sendCtrlMsg (Just nid) (Register label Nothing) --- | Query the local process registry (synchronous). +-- | Query the local process registry whereis :: String -> Process (Maybe ProcessId) whereis label = do sendCtrlMsg Nothing (WhereIs label) @@ -473,17 +531,15 @@ (\(WhereIsReply _ mPid) -> return mPid) ] --- | Query a remote process registry (synchronous) -whereisRemote :: NodeId -> String -> Process (Maybe ProcessId) -whereisRemote nid label = do - whereisRemoteAsync nid label - receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label') - (\(WhereIsReply _ mPid) -> return mPid) - ] - -- | Query a remote process registry (asynchronous) -- --- Reply will come in the form of a 'WhereIsReply' message +-- Reply will come in the form of a 'WhereIsReply' message. +-- +-- There is currently no synchronous version of 'whereisRemoteAsync': if +-- you implement one yourself, be sure to take into account that the remote +-- node might die or get disconnect before it can respond (i.e. you should +-- use 'monitorNode' and take appropriate action when you receive a +-- 'NodeMonitorNotification'). whereisRemoteAsync :: NodeId -> String -> Process () whereisRemoteAsync nid label = sendCtrlMsg (Just nid) (WhereIs label) @@ -502,8 +558,16 @@ -- Closures -- -------------------------------------------------------------------------------- --- | Deserialize a closure -unClosure :: forall a. Typeable a => Closure a -> Process a +-- | Resolve a static value +unStatic :: Typeable a => Static a -> Process a +unStatic static = do + rtable <- remoteTable . processNode <$> ask + case Static.unstatic rtable static of + Left err -> fail $ "Could not resolve static value: " ++ err + Right x -> return x + +-- | Resolve a closure +unClosure :: Typeable a => Closure a -> Process a unClosure closure = do rtable <- remoteTable . processNode <$> ask case Static.unclosure rtable closure of @@ -535,18 +599,17 @@ -- 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 them = do + us <- getSelfPid + node <- processNode <$> ask + liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them) -- | Reconnect to a sendport. See 'reconnect' for more information. reconnectPort :: SendPort a -> Process () -reconnectPort = - sendCtrlMsg Nothing . Reconnect . SendPortIdentifier . sendPortId +reconnectPort them = do + us <- getSelfPid + node <- processNode <$> ask + liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them)) -------------------------------------------------------------------------------- -- Auxiliary functions -- @@ -598,4 +661,5 @@ liftIO $ sendBinary (processNode proc) (ProcessIdentifier (processId proc)) (NodeIdentifier nid) + WithImplicitReconnect msg diff --git a/src/Control/Distributed/Process/Internal/StrictContainerAccessors.hs b/src/Control/Distributed/Process/Internal/StrictContainerAccessors.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/StrictContainerAccessors.hs @@ -0,0 +1,21 @@ +module Control.Distributed.Process.Internal.StrictContainerAccessors + ( mapMaybe + , mapDefault + ) where + +import Prelude hiding (map) +import Data.Accessor +import Data.Map (Map) +import qualified Data.Map as Map (lookup, insert, delete, findWithDefault) + +mapMaybe :: Ord key => key -> Accessor (Map key elem) (Maybe elem) +mapMaybe key = accessor + (Map.lookup key) + (\mVal map -> case mVal of + Nothing -> Map.delete key map + Just val -> val `seq` Map.insert key val map) + +mapDefault :: Ord key => elem -> key -> Accessor (Map key elem) elem +mapDefault def key = accessor + (Map.findWithDefault def key) + (\val map -> val `seq` Map.insert key val map) diff --git a/src/Control/Distributed/Process/Internal/StrictList.hs b/src/Control/Distributed/Process/Internal/StrictList.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/StrictList.hs @@ -0,0 +1,25 @@ +-- | Spine and element strict list +module Control.Distributed.Process.Internal.StrictList + ( StrictList(Cons, Nil) + , length + , reverse + , reverse' + ) where + +import Prelude hiding (length, reverse) + +-- | Strict list +data StrictList a = Cons !a !(StrictList a) | Nil + +length :: StrictList a -> Int +length Nil = 0 +length (Cons _ xs) = 1 + length xs + +-- | Reverse a strict list +reverse :: StrictList a -> StrictList a +reverse xs = reverse' xs Nil + +-- | @reverseStrict' xs ys@ is 'reverse xs ++ ys' if they were lists +reverse' :: StrictList a -> StrictList a -> StrictList a +reverse' Nil ys = ys +reverse' (Cons x xs) ys = reverse' xs (Cons x ys) diff --git a/src/Control/Distributed/Process/Internal/StrictMVar.hs b/src/Control/Distributed/Process/Internal/StrictMVar.hs --- a/src/Control/Distributed/Process/Internal/StrictMVar.hs +++ b/src/Control/Distributed/Process/Internal/StrictMVar.hs @@ -1,6 +1,7 @@ -- | Like Control.Concurrent.MVar.Strict but reduce to HNF, not NF +{-# LANGUAGE MagicHash, UnboxedTuples #-} module Control.Distributed.Process.Internal.StrictMVar - ( StrictMVar + ( StrictMVar(StrictMVar) , newEmptyMVar , newMVar , takeMVar @@ -8,6 +9,7 @@ , withMVar , modifyMVar_ , modifyMVar + , mkWeakMVar ) where import Control.Applicative ((<$>)) @@ -23,6 +25,10 @@ , modifyMVar_ , modifyMVar ) +import GHC.MVar (MVar(MVar)) +import GHC.IO (IO(IO)) +import GHC.Prim (mkWeak#) +import GHC.Weak (Weak(Weak)) newtype StrictMVar a = StrictMVar (MVar.MVar a) @@ -49,3 +55,7 @@ where evaluateFst :: (a, b) -> IO (a, b) evaluateFst (x, y) = evaluate x >> return (x, y) + +mkWeakMVar :: StrictMVar a -> IO () -> IO (Weak (StrictMVar a)) +mkWeakMVar m@(StrictMVar (MVar m#)) f = IO $ \s -> + case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #) 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 @@ -10,6 +10,8 @@ , ProcessId(..) , Identifier(..) , nodeOf + , firstNonReservedProcessId + , nullProcessId -- * Local nodes and processes , LocalNode(..) , LocalNodeState(..) @@ -17,6 +19,7 @@ , LocalProcessState(..) , Process(..) , runLocalProcess + , ImplicitReconnect(..) -- * Typed channels , LocalSendPortId , SendPortId(..) @@ -61,24 +64,25 @@ , typedChannelWithId ) where +import System.Mem.Weak (Weak) import Data.Map (Map) import Data.Int (Int32) import Data.Typeable (Typeable) import Data.Binary (Binary(put, get), putWord8, getWord8, encode) -import qualified Data.ByteString as BSS (ByteString, concat) +import qualified Data.ByteString as BSS (ByteString, concat, copy) import qualified Data.ByteString.Lazy as BSL ( ByteString , toChunks , splitAt , fromChunks ) +import qualified Data.ByteString.Lazy.Internal as BSL (ByteString(..)) import Data.Accessor (Accessor, accessor) -import qualified Data.Accessor.Container as DAC (mapMaybe) import Control.Category ((>>>)) import Control.Exception (Exception) import Control.Concurrent (ThreadId) import Control.Concurrent.Chan (Chan) -import Control.Concurrent.STM (TChan, TVar) +import Control.Concurrent.STM (TVar) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative (Applicative, (<$>), (<*>)) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT) @@ -94,10 +98,9 @@ ) import Control.Distributed.Process.Internal.CQueue (CQueue) import Control.Distributed.Process.Internal.StrictMVar (StrictMVar) +import Control.Distributed.Process.Internal.WeakTQueue (TQueue) import Control.Distributed.Static (RemoteTable, Closure) - --- import Control.Distributed.Process.Internal.Dynamic (Dynamic) --- import Control.Distributed.Process.Internal.TypeRep (compareTypeRep) -- and Binary instances +import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe) -------------------------------------------------------------------------------- -- Node and process identifiers -- @@ -105,7 +108,7 @@ -- | Node identifier newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress } - deriving (Eq, Ord, Binary) + deriving (Eq, Ord, Binary, Typeable) instance Show NodeId where show (NodeId addr) = "nid://" ++ show addr @@ -133,9 +136,9 @@ -- | Union of all kinds of identifiers data Identifier = - NodeIdentifier NodeId - | ProcessIdentifier ProcessId - | SendPortIdentifier SendPortId + NodeIdentifier !NodeId + | ProcessIdentifier !ProcessId + | SendPortIdentifier !SendPortId deriving (Eq, Ord) instance Show Identifier where @@ -149,39 +152,64 @@ nodeOf (SendPortIdentifier cid) = processNodeId (sendPortProcessId cid) -------------------------------------------------------------------------------- +-- Special PIDs -- +-------------------------------------------------------------------------------- + +firstNonReservedProcessId :: Int32 +firstNonReservedProcessId = 1 + +nullProcessId :: NodeId -> ProcessId +nullProcessId nid = + ProcessId { processNodeId = nid + , processLocalId = LocalProcessId { lpidUnique = 0 + , lpidCounter = 0 + } + } + +-------------------------------------------------------------------------------- -- Local nodes and processes -- -------------------------------------------------------------------------------- -- | Local nodes data LocalNode = LocalNode { -- | 'NodeId' of the node - localNodeId :: NodeId + localNodeId :: !NodeId -- | The network endpoint associated with this node - , localEndPoint :: NT.EndPoint + , localEndPoint :: !NT.EndPoint -- | Local node state - , localState :: StrictMVar LocalNodeState + , localState :: !(StrictMVar LocalNodeState) -- | Channel for the node controller - , localCtrlChan :: Chan NCMsg + , localCtrlChan :: !(Chan NCMsg) -- | Runtime lookup table for supporting closures -- TODO: this should be part of the CH state, not the local endpoint state - , remoteTable :: RemoteTable + , remoteTable :: !RemoteTable } +data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect + deriving (Eq, Show) + -- | Local node state data LocalNodeState = LocalNodeState - { _localProcesses :: !(Map LocalProcessId LocalProcess) + { -- | Processes running on this node + _localProcesses :: !(Map LocalProcessId LocalProcess) + -- | Counter to assign PIDs , _localPidCounter :: !Int32 + -- | The 'unique' value used to create PIDs (so that processes on + -- restarted nodes have new PIDs) , _localPidUnique :: !Int32 - , _localConnections :: !(Map (Identifier, Identifier) NT.Connection) + -- | Outgoing connections + , _localConnections :: !(Map (Identifier, Identifier) + (NT.Connection, ImplicitReconnect)) } -- | Processes running on our local node data LocalProcess = LocalProcess - { processQueue :: CQueue Message - , processId :: ProcessId - , processState :: StrictMVar LocalProcessState - , processThread :: ThreadId - , processNode :: LocalNode + { processQueue :: !(CQueue Message) + , processWeakQ :: !(Weak (CQueue Message)) + , processId :: !ProcessId + , processState :: !(StrictMVar LocalProcessState) + , processThread :: !ThreadId + , processNode :: !LocalNode } -- | Deconstructor for 'Process' (not exported to the public API) @@ -224,7 +252,7 @@ show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid) = "cid://" ++ show addr ++ ":" ++ show plid ++ ":" ++ show clid -data TypedChannel = forall a. Serializable a => TypedChannel (TChan a) +data TypedChannel = forall a. Serializable a => TypedChannel (Weak (TQueue a)) -- | The send send of a typed channel (serializable) newtype SendPort a = SendPort { @@ -236,7 +264,7 @@ -- | The receive end of a typed channel (not serializable) data ReceivePort a = -- | A single receive port - ReceivePortSingle (TChan a) + ReceivePortSingle (TQueue a) -- | A left-biased combination of receive ports | ReceivePortBiased [ReceivePort a] -- | A round-robin combination of receive ports @@ -249,8 +277,8 @@ -- | Messages consist of their typeRep fingerprint and their encoding data Message = Message - { messageFingerprint :: Fingerprint - , messageEncoding :: BSL.ByteString + { messageFingerprint :: !Fingerprint + , messageEncoding :: !BSL.ByteString } instance Show Message where @@ -266,12 +294,20 @@ -- | Deserialize a message payloadToMessage :: [BSS.ByteString] -> Message -payloadToMessage payload = Message fp msg +payloadToMessage payload = Message fp (copy msg) where + encFp :: BSL.ByteString + msg :: BSL.ByteString (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint) $ BSL.fromChunks payload + + fp :: Fingerprint fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp + copy :: BSL.ByteString -> BSL.ByteString + copy (BSL.Chunk bs BSL.Empty) = BSL.Chunk (BSS.copy bs) BSL.Empty + copy bsl = BSL.fromChunks . return . BSS.concat . BSL.toChunks $ bsl + -------------------------------------------------------------------------------- -- Node controller user-visible data types -- -------------------------------------------------------------------------------- @@ -287,32 +323,32 @@ -- | Message sent by process monitors data ProcessMonitorNotification = - ProcessMonitorNotification MonitorRef ProcessId DiedReason + ProcessMonitorNotification !MonitorRef !ProcessId !DiedReason deriving (Typeable, Show) -- | Message sent by node monitors data NodeMonitorNotification = - NodeMonitorNotification MonitorRef NodeId DiedReason + NodeMonitorNotification !MonitorRef !NodeId !DiedReason deriving (Typeable, Show) -- | Message sent by channel (port) monitors data PortMonitorNotification = - PortMonitorNotification MonitorRef SendPortId DiedReason + PortMonitorNotification !MonitorRef !SendPortId !DiedReason deriving (Typeable, Show) -- | Exceptions thrown when a linked process dies data ProcessLinkException = - ProcessLinkException ProcessId DiedReason + ProcessLinkException !ProcessId !DiedReason deriving (Typeable, Show) -- | Exception thrown when a linked node dies data NodeLinkException = - NodeLinkException NodeId DiedReason + NodeLinkException !NodeId !DiedReason deriving (Typeable, Show) -- | Exception thrown when a linked channel (port) dies data PortLinkException = - PortLinkException SendPortId DiedReason + PortLinkException !SendPortId !DiedReason deriving (Typeable, Show) instance Exception ProcessLinkException @@ -325,7 +361,7 @@ DiedNormal -- | The process exited with an exception -- (provided as 'String' because 'Exception' does not implement 'Binary') - | DiedException String + | DiedException !String -- | We got disconnected from the process node | DiedDisconnect -- | The process node died @@ -368,23 +404,22 @@ -- | Messages to the node controller data NCMsg = NCMsg - { ctrlMsgSender :: Identifier - , ctrlMsgSignal :: ProcessSignal + { ctrlMsgSender :: !Identifier + , ctrlMsgSignal :: !ProcessSignal } deriving Show -- | Signals to the node controller (see 'NCMsg') data ProcessSignal = - Link Identifier - | Unlink Identifier - | Monitor MonitorRef - | Unmonitor MonitorRef - | Died Identifier DiedReason - | Spawn (Closure (Process ())) SpawnRef - | WhereIs String - | Register String (Maybe ProcessId) -- Nothing to unregister - | NamedSend String Message - | Reconnect Identifier + Link !Identifier + | Unlink !Identifier + | Monitor !MonitorRef + | Unmonitor !MonitorRef + | Died Identifier !DiedReason + | Spawn !(Closure (Process ())) !SpawnRef + | WhereIs !String + | Register !String !(Maybe ProcessId) -- Use 'Nothing' to unregister + | NamedSend !String !Message deriving Show -------------------------------------------------------------------------------- @@ -429,7 +464,6 @@ 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 @@ -442,7 +476,6 @@ 6 -> WhereIs <$> get 7 -> Register <$> get <*> get 8 -> NamedSend <$> get <*> (payloadToMessage <$> get) - 9 -> Reconnect <$> get _ -> fail "ProcessSignal.get: invalid" instance Binary DiedReason where @@ -498,13 +531,13 @@ localPidUnique :: Accessor LocalNodeState Int32 localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq }) -localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) NT.Connection) +localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect)) localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns }) localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess) localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid -localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe NT.Connection) +localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect)) localConnectionBetween from to = localConnections >>> DAC.mapMaybe (from, to) monitorCounter :: Accessor LocalProcessState Int32 diff --git a/src/Control/Distributed/Process/Internal/WeakTQueue.hs b/src/Control/Distributed/Process/Internal/WeakTQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/WeakTQueue.hs @@ -0,0 +1,101 @@ +-- | Clone of Control.Concurrent.STM.TQueue with support for mkWeakTQueue +-- +-- Not all functionality from the original module is available: unGetTQueue, +-- peekTQueue and tryPeekTQueue are missing. In order to implement these we'd +-- need to be able to touch# the write end of the queue inside unGetTQueue, but +-- that means we need a version of touch# that works within the STM monad. +{-# LANGUAGE MagicHash, UnboxedTuples #-} +module Control.Distributed.Process.Internal.WeakTQueue ( + -- * Original functionality + TQueue, + newTQueue, + newTQueueIO, + readTQueue, + tryReadTQueue, + writeTQueue, + isEmptyTQueue, + -- * New functionality + mkWeakTQueue + ) where + +import Prelude hiding (read) +import GHC.Conc +import Data.Typeable (Typeable) +import GHC.IO (IO(IO)) +import GHC.Prim (mkWeak#) +import GHC.Weak (Weak(Weak)) + +-------------------------------------------------------------------------------- +-- Original functionality -- +-------------------------------------------------------------------------------- + +-- | 'TQueue' is an abstract type representing an unbounded FIFO channel. +data TQueue a = TQueue {-# UNPACK #-} !(TVar [a]) + {-# UNPACK #-} !(TVar [a]) + deriving Typeable + +instance Eq (TQueue a) where + TQueue a _ == TQueue b _ = a == b + +-- |Build and returns a new instance of 'TQueue' +newTQueue :: STM (TQueue a) +newTQueue = do + read <- newTVar [] + write <- newTVar [] + return (TQueue read write) + +-- |@IO@ version of 'newTQueue'. This is useful for creating top-level +-- 'TQueue's using 'System.IO.Unsafe.unsafePerformIO', because using +-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't +-- possible. +newTQueueIO :: IO (TQueue a) +newTQueueIO = do + read <- newTVarIO [] + write <- newTVarIO [] + return (TQueue read write) + +-- |Write a value to a 'TQueue'. +writeTQueue :: TQueue a -> a -> STM () +writeTQueue (TQueue _read write) a = do + listend <- readTVar write + writeTVar write (a:listend) + +-- |Read the next value from the 'TQueue'. +readTQueue :: TQueue a -> STM a +readTQueue (TQueue read write) = do + xs <- readTVar read + case xs of + (x:xs') -> do writeTVar read xs' + return x + [] -> do ys <- readTVar write + case ys of + [] -> retry + _ -> case reverse ys of + [] -> error "readTQueue" + (z:zs) -> do writeTVar write [] + writeTVar read zs + return z + +-- | A version of 'readTQueue' which does not retry. Instead it +-- returns @Nothing@ if no value is available. +tryReadTQueue :: TQueue a -> STM (Maybe a) +tryReadTQueue c = fmap Just (readTQueue c) `orElse` return Nothing + +-- |Returns 'True' if the supplied 'TQueue' is empty. +isEmptyTQueue :: TQueue a -> STM Bool +isEmptyTQueue (TQueue read write) = do + xs <- readTVar read + case xs of + (_:_) -> return False + [] -> do ys <- readTVar write + case ys of + [] -> return True + _ -> return False + +-------------------------------------------------------------------------------- +-- New functionality -- +-------------------------------------------------------------------------------- + +mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a)) +mkWeakTQueue q@(TQueue _read (TVar write#)) f = IO $ \s -> + case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #) 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 @@ -1,4 +1,7 @@ -- | Local nodes +-- +-- TODO: Calls to 'sendBinary' and co by the node controller may stall the +-- node controller. module Control.Distributed.Process.Node ( LocalNode , newLocalNode @@ -14,21 +17,19 @@ #endif import System.IO (fixIO, hPutStrLn, stderr) +import System.Mem.Weak (Weak, deRefWeak) import qualified Data.ByteString.Lazy as BSL (fromChunks) import Data.Binary (decode) import Data.Map (Map) import qualified Data.Map as Map ( empty - , lookup - , insert - , delete , toList , partitionWithKey - , filterWithKey , elems + , filterWithKey ) import Data.Set (Set) -import qualified Data.Set as Set (empty, insert, delete, member, filter) +import qualified Data.Set as Set (empty, insert, delete, member) import Data.Foldable (forM_) import Data.Maybe (isJust) import Data.Typeable (Typeable) @@ -52,8 +53,13 @@ , takeMVar ) import Control.Concurrent.Chan (newChan, writeChan, readChan) -import Control.Concurrent.STM (atomically, writeTChan) -import Control.Distributed.Process.Internal.CQueue (enqueue, newCQueue) +import Control.Concurrent.STM (atomically) +import Control.Distributed.Process.Internal.CQueue + ( CQueue + , enqueue + , newCQueue + , mkWeakCQueue + ) import qualified Network.Transport as NT ( Transport , EndPoint @@ -67,9 +73,10 @@ , ConnectionId , Connection , close + , EndPointAddress + , Reliability(ReliableOrdered) ) 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 @@ -116,16 +123,24 @@ , payloadToMessage , createMessage , runLocalProcess + , firstNonReservedProcessId + , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect) ) import Control.Distributed.Process.Serializable (Serializable) -import Control.Distributed.Process.Internal.Node +import Control.Distributed.Process.Internal.Messaging ( sendBinary , sendMessage , sendPayload - , reconnect + , closeImplicitReconnections + , impliesDeathOf ) import Control.Distributed.Process.Internal.Primitives (expect, register, finally) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable) +import Control.Distributed.Process.Internal.WeakTQueue (writeTQueue) +import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC + ( mapMaybe + , mapDefault + ) -------------------------------------------------------------------------------- -- Initialization -- @@ -151,7 +166,7 @@ unq <- randomIO state <- newMVar LocalNodeState { _localProcesses = Map.empty - , _localPidCounter = 0 + , _localPidCounter = firstNonReservedProcessId , _localPidUnique = unq , _localConnections = Map.empty } @@ -212,8 +227,10 @@ , _typedChannels = Map.empty } queue <- newCQueue + weakQueue <- mkWeakCQueue queue (return ()) (_, lproc) <- fixIO $ \ ~(tid, _) -> do let lproc = LocalProcess { processQueue = queue + , processWeakQ = weakQueue , processId = pid , processState = pst , processThread = tid @@ -235,7 +252,7 @@ then do newUnique <- randomIO return ( (localProcessWithId lpid ^= Just lproc) - . (localPidCounter ^= 0) + . (localPidCounter ^= firstNonReservedProcessId) . (localPidUnique ^= newUnique) $ st , pid @@ -251,63 +268,91 @@ 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) + mapM_ (NT.close . fst) (Map.elems affected) return $ (localProcessWithId (processLocalId pid) ^= Nothing) . (localConnections ^= unaffected) $ st +-------------------------------------------------------------------------------- +-- Handle incoming messages -- +-------------------------------------------------------------------------------- + +type IncomingConnection = (NT.EndPointAddress, IncomingTarget) + +data IncomingTarget = + Uninit + | ToProc (Weak (CQueue Message)) + | ToChan TypedChannel + | ToNode + +data ConnectionState = ConnectionState { + _incoming :: !(Map NT.ConnectionId IncomingConnection) + , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId)) + } + +initConnectionState :: ConnectionState +initConnectionState = ConnectionState { + _incoming = Map.empty + , _incomingFrom = Map.empty + } + +incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection) +incoming = accessor _incoming (\conns st -> st { _incoming = conns }) + +incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection) +incomingAt cid = incoming >>> DAC.mapMaybe cid + +incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId) +incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr + where + aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr }) + handleIncomingMessages :: LocalNode -> IO () -handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty +handleIncomingMessages node = go initConnectionState where - 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 :: ConnectionState -> IO () + go !st = 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 (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 `Set.member` uninitConns - ) of - (Just proc, _, _, _) -> do - let msg = payloadToMessage payload - enqueue (processQueue proc) msg - go uninitConns procs chans ctrls - (_, Just (TypedChannel chan), _, _) -> do - atomically $ writeTChan chan . decode . BSL.fromChunks $ payload - go uninitConns procs chans ctrls - (_, _, True, _) -> do + NT.ConnectionOpened cid rel theirAddr -> + if rel == NT.ReliableOrdered + then go ( (incomingAt cid ^= Just (theirAddr, Uninit)) + . (incomingFrom theirAddr ^: Set.insert cid) + $ st + ) + else invalidRequest cid st + NT.Received cid payload -> + case st ^. incomingAt cid of + Just (_, ToProc weakQueue) -> do + mQueue <- deRefWeak weakQueue + forM_ mQueue $ \queue -> do + -- TODO: if we find that the queue is Nothing, should we remove + -- it from the NC state? (and same for channels, below) + let msg = payloadToMessage payload + enqueue queue msg -- 'enqueue' is strict + go st + Just (_, ToChan (TypedChannel chan')) -> do + mChan <- deRefWeak chan' + -- If mChan is Nothing, the process has given up the read end of + -- the channel and we simply ignore the incoming message + forM_ mChan $ \chan -> atomically $ + -- We make sure the message is fully decoded when it is enqueued + writeTQueue chan $! decode (BSL.fromChunks payload) + go st + Just (_, ToNode) -> do let ctrlMsg = decode . BSL.fromChunks $ payload - writeChan ctrlChan ctrlMsg - go uninitConns procs chans ctrls - (_, _, _, True) -> + writeChan ctrlChan $! ctrlMsg + go st + Just (src, Uninit) -> case decode (BSL.fromChunks payload) of ProcessIdentifier pid -> do let lpid = processLocalId pid mProc <- withMVar state $ return . (^. localProcessWithId lpid) case mProc of Just proc -> - go (Set.delete cid uninitConns) - (Map.insert cid proc procs) - chans - ctrls + go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st) Nothing -> - -- Request for an unknown process. - -- - -- TODO: We should treat this as a fatal error on the - -- part of the remote node. That is, we should report the - -- remote node as having died, and we should close - -- incoming connections (this requires a Transport layer - -- extension). (**) - go (Set.delete cid uninitConns) procs chans ctrls + invalidRequest cid st SendPortIdentifier chId -> do let lcid = sendPortLocalId chId lpid = processLocalId (sendPortProcessId chId) @@ -316,50 +361,40 @@ Just proc -> do mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid) case mChannel of - Just channel -> - go (Set.delete cid uninitConns) - procs - (Map.insert cid channel chans) - ctrls + Just channel -> + go (incomingAt cid ^= Just (src, ToChan channel) $ st) Nothing -> - -- Unknown typed channel - -- TODO (**) above - go (Set.delete cid uninitConns) procs chans ctrls + invalidRequest cid st Nothing -> - -- Unknown process - -- TODO (**) above - go (Set.delete cid uninitConns) procs chans ctrls - NodeIdentifier _ -> - go (Set.delete cid uninitConns) - procs - chans - (Set.insert cid ctrls) - _ -> - -- Unexpected message - -- TODO (**) above - go uninitConns procs chans ctrls + invalidRequest cid st + NodeIdentifier nid -> + if nid == localNodeId node + then go (incomingAt cid ^= Just (src, ToNode) $ st) + else invalidRequest cid st + Nothing -> + invalidRequest cid st NT.ConnectionClosed cid -> - go (Set.delete cid uninitConns) - (Map.delete cid procs) - (Map.delete cid chans) - (Set.delete cid ctrls) - NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) cids) _) -> do + case st ^. incomingAt cid of + Nothing -> + invalidRequest cid st + Just (src, _) -> + go ( (incomingAt cid ^= Nothing) + . (incomingFrom src ^: Set.delete cid) + $ st + ) + NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do -- [Unified table 9, rule node_disconnect] let nid = NodeIdentifier $ NodeId theirAddr writeChan ctrlChan NCMsg { ctrlMsgSender = nid , ctrlMsgSignal = Died nid DiedDisconnect } - 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) - -- as a fatal error on the part of the remote node (like (**), above) - fail "handleIncomingMessages: TODO" + let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr)) + closeImplicitReconnections node nid + go ( (incomingFrom theirAddr ^= Set.empty) + . (incoming ^: Map.filterWithKey (const . notLost)) + $ st + ) NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) -> fail $ "Cloud Haskell fatal error: end point failed: " ++ str NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) -> @@ -370,7 +405,17 @@ -- If we received a multicast message, something went horribly wrong -- and we just give up fail "Cloud Haskell fatal error: received unexpected multicast" - + + invalidRequest :: NT.ConnectionId -> ConnectionState -> IO () + invalidRequest cid st = + -- TODO: We should treat this as a fatal error on the part of the remote + -- node. That is, we should report the remote node as having died, and we + -- should close incoming connections (this requires a Transport layer + -- extension). + go ( incomingAt cid ^= Nothing + $ st + ) + state = localState node endpoint = localEndPoint node ctrlChan = localCtrlChan node @@ -422,6 +467,7 @@ liftIO $ sendBinary node (ctrlMsgSender msg) (NodeIdentifier nid') + WithImplicitReconnect msg _ -> return () @@ -445,8 +491,6 @@ 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 @@ -475,6 +519,7 @@ liftIO $ sendBinary node (NodeIdentifier $ localNodeId node) (NodeIdentifier $ processNodeId from) + WithImplicitReconnect NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node) , ctrlMsgSignal = Died them DiedUnknownId @@ -537,6 +582,7 @@ liftIO $ sendMessage node (NodeIdentifier (localNodeId node)) (ProcessIdentifier pid) + WithImplicitReconnect (DidSpawn ref pid') -- Unified semantics does not explicitly describe how to implement 'register', @@ -556,6 +602,7 @@ liftIO $ sendMessage node (NodeIdentifier (localNodeId node)) (ProcessIdentifier from) + WithImplicitReconnect (WhereIsReply label mPid) -- [Unified: Table 14] @@ -568,14 +615,9 @@ liftIO $ sendPayload node from (ProcessIdentifier pid) + NoImplicitReconnect (messageToPayload msg) --- Reconnecting -ncEffectReconnect :: Identifier -> Identifier -> NC () -ncEffectReconnect from to = do - node <- ask - liftIO $ reconnect node from to - -------------------------------------------------------------------------------- -- Auxiliary -- -------------------------------------------------------------------------------- @@ -601,11 +643,11 @@ (True, Nothing, SendPortIdentifier pid) -> throwException dest $ PortLinkException pid reason (False, _, _) -> - -- TODO: why the change in sender? How does that affect 'reconnect' semantics? - -- (see [Unified: Table 10] + -- 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 @@ -621,7 +663,6 @@ 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 @@ -721,33 +762,15 @@ -- -- * the notifications for that process specifically and -- * the notifications for typed channels to that process. +-- +-- See https://github.com/haskell/containers/issues/14 for the bang on _v. splitNotif :: Identifier -> Map Identifier a -> (Map Identifier a, Map Identifier a) -splitNotif ident = Map.partitionWithKey (\k !v -> impliesDeathOf ident k) - --- | Does the death of one entity (node, project, channel) imply the death --- of another? -impliesDeathOf :: Identifier -- ^ Who died - -> Identifier -- ^ Who's being watched - -> Bool -- ^ Does this death implies the death of the watchee? -NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' = - nid' == nid -NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid = - processNodeId pid == nid -NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid = - processNodeId (sendPortProcessId cid) == nid -ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' = - pid' == pid -ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid = - sendPortProcessId cid == pid -SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' = - cid' == cid -_ `impliesDeathOf` _ = - False +splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k) -------------------------------------------------------------------------------- --- Strict evaluation of the state -- +-- Auxiliary -- -------------------------------------------------------------------------------- -- | Modify and evaluate the state diff --git a/tests/TestCH.hs b/tests/TestCH.hs --- a/tests/TestCH.hs +++ b/tests/TestCH.hs @@ -15,13 +15,21 @@ , takeMVar , readMVar ) -import Control.Monad (replicateM_, replicateM) +import Control.Monad (replicateM_, replicateM, forever) import Control.Exception (throwIO) import Control.Applicative ((<$>), (<*>)) import qualified Network.Transport as NT (Transport, closeEndPoint) -import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Socket (sClose) +import Network.Transport.TCP + ( createTransportExposeInternals + , TransportInternals(socketBetween) + , defaultTCPParameters + ) import Control.Distributed.Process -import Control.Distributed.Process.Internal.Types (LocalNode(localEndPoint)) +import Control.Distributed.Process.Internal.Types + ( NodeId(nodeAddress) + , LocalNode(localEndPoint) + ) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable (Serializable) import TestAuxiliary @@ -40,6 +48,13 @@ send partner (Ping self) ping +-- | Quick and dirty synchronous version of whereisRemoteAsync +whereisRemote :: NodeId -> String -> Process (Maybe ProcessId) +whereisRemote nid string = do + whereisRemoteAsync nid string + WhereIsReply _ mPid <- expect + return mPid + -- | Basic ping test testPing :: NT.Transport -> IO () testPing transport = do @@ -515,8 +530,10 @@ pid <- forkProcess node1 $ do sport <- expect :: Process (SendPort ()) ref <- monitorPort sport - PortMonitorNotification ref' port' DiedNormal <- expect - return $ ref' == ref && port' == sendPortId sport + PortMonitorNotification ref' port' reason <- expect + -- reason might be DiedUnknownId if the receive port is GCed before the + -- monitor is established (TODO: not sure that this is reasonable) + return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId) liftIO $ putMVar gotNotification () runProcess node2 $ do @@ -541,7 +558,7 @@ Ping pid' <- expect True <- return $ pingServer == pid' return () - + testRemoteRegistry :: NT.Transport -> IO () testRemoteRegistry transport = do node1 <- newLocalNode transport initRemoteTable @@ -578,9 +595,109 @@ send pid sport expect +testReconnect :: NT.Transport -> TransportInternals -> IO () +testReconnect transport transportInternals = do + [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable + let nid1 = localNodeId node1 + nid2 = localNodeId node2 + processA <- newEmptyMVar + [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar + + forkProcess node1 $ do + us <- getSelfPid + liftIO $ putMVar processA us + msg1 <- expect + msg2 <- expect + True <- return $ msg1 == "message 1" && msg2 == "message 3" + liftIO $ putMVar sendTestOk () + + forkProcess node2 $ do + {- + - Make sure there is no implicit reconnect on normal message sending + -} + + them <- liftIO $ readMVar processA + send them "message 1" >> liftIO (threadDelay 100000) + + -- Simulate network failure + liftIO $ do + sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2) + sClose sock + threadDelay 10000 + + -- Should not arrive + send them "message 2" + + -- Should arrive + reconnect them + send them "message 3" + + liftIO $ takeMVar sendTestOk + + {- + - Test that there *is* implicit reconnect on node controller messages + -} + + us <- getSelfPid + registerRemote nid1 "a" us >> liftIO (threadDelay 100000) -- registerRemote is asynchronous + + -- Simulate network failure + liftIO $ do + sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2) + sClose sock + threadDelay 10000 + + -- This will happen due to implicit reconnect + registerRemote nid1 "b" us + + -- Should happen + registerRemote nid1 "c" us + + -- Check + Just _ <- whereisRemote nid1 "a" + Just _ <- whereisRemote nid1 "b" + Just _ <- whereisRemote nid1 "c" + + liftIO $ putMVar registerTestOk () + + takeMVar registerTestOk + +-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server +-- in between +testMatchAny :: NT.Transport -> IO () +testMatchAny transport = do + proxyAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- Math server + forkIO $ do + localNode <- newLocalNode transport initRemoteTable + mathServer <- forkProcess localNode math + proxyServer <- forkProcess localNode $ forever $ do + msg <- receiveWait [ matchAny return ] + forward msg mathServer + putMVar proxyAddr proxyServer + + -- Client + forkIO $ do + localNode <- newLocalNode transport initRemoteTable + mathServer <- readMVar proxyAddr + + runProcess localNode $ do + pid <- getSelfPid + send mathServer (Add pid 1 2) + 3 <- expect :: Process Double + send mathServer (Divide pid 8 2) + 4 <- expect :: Process Double + send mathServer (Divide pid 8 0) + DivByZero <- expect + liftIO $ putMVar clientDone () + + takeMVar clientDone + main :: IO () main = do - Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters + Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters runTests [ ("Ping", testPing transport) , ("Math", testMath transport) @@ -593,6 +710,7 @@ , ("Registry", testRegistry transport) , ("RemoteRegistry", testRemoteRegistry transport) , ("SpawnLocal", testSpawnLocal transport) + , ("MatchAny", testMatchAny transport) -- Monitoring processes -- -- The "missing" combinations in the list below don't make much sense, as @@ -619,4 +737,6 @@ -- Monitoring nodes and channels , ("MonitorNode", testMonitorNode transport) , ("MonitorChannel", testMonitorChannel transport) + -- Reconnect + , ("Reconnect", testReconnect transport transportInternals) ] diff --git a/tests/TestClosure.hs b/tests/TestClosure.hs --- a/tests/TestClosure.hs +++ b/tests/TestClosure.hs @@ -6,13 +6,28 @@ import Control.Monad (join, replicateM, forever) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay) -import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar) +import Control.Concurrent.MVar + ( MVar + , newEmptyMVar + , readMVar + , takeMVar + , putMVar + , modifyMVar_ + , newMVar + ) import Control.Applicative ((<$>)) +import System.Random (randomIO) import Network.Transport (Transport) -import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Transport.TCP + ( createTransportExposeInternals + , defaultTCPParameters + , TransportInternals(socketBetween) + ) +import Network.Socket (sClose) import Control.Distributed.Process import Control.Distributed.Process.Closure import Control.Distributed.Process.Node +import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress)) import Control.Distributed.Static (staticLabel, staticClosure) import TestAuxiliary @@ -53,6 +68,9 @@ sport <- receiveChan rport sendChan sport () +signal :: ProcessId -> Process () +signal pid = send pid () + remotable [ 'factorial , 'addInt , 'putInt @@ -62,8 +80,30 @@ , 'typedPingServer , 'isPrime , 'quintuple + , 'signal ] +randomElement :: [a] -> IO a +randomElement xs = do + ix <- randomIO + return (xs !! (ix `mod` length xs)) + +remotableDecl [ + [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ; + dfib (_, reply, 0) = sendChan reply 0 + dfib (_, reply, 1) = sendChan reply 1 + dfib (nids, reply, n) = do + nid1 <- liftIO $ randomElement nids + nid2 <- liftIO $ randomElement nids + (sport, rport) <- newChan + spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2) + spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1) + n1 <- receiveChan rport + n2 <- receiveChan rport + sendChan reply $ n1 + n2 + |] + ] + -- Just try creating a static polymorphic value staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e) => Static (a -> b -> c -> d -> e -> (a, b, c, d, e)) @@ -337,10 +377,57 @@ liftIO $ putMVar done () takeMVar done +simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process () +simulateNetworkFailure transportInternals fr to = liftIO $ do + threadDelay 10000 + sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to) + sClose sock + threadDelay 10000 + +testFib :: Transport -> RemoteTable -> IO () +testFib transport rtable = do + nodes <- replicateM 4 $ newLocalNode transport rtable + done <- newEmptyMVar + + forkProcess (head nodes) $ do + (sport, rport) <- newChan + spawnLocal $ dfib (map localNodeId nodes, sport, 10) + 55 <- receiveChan rport :: Process Integer + liftIO $ putMVar done () + + takeMVar done + +testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO () +testSpawnReconnect transport rtable transportInternals = do + [node1, node2] <- replicateM 2 $ newLocalNode transport rtable + let nid1 = localNodeId node1 + nid2 = localNodeId node2 + done <- newEmptyMVar + iv <- newMVar (0 :: Int) + + incr <- forkProcess node1 $ forever $ do + () <- expect + liftIO $ modifyMVar_ iv (return . (+ 1)) + + forkProcess node2 $ do + _pid1 <- spawn nid1 ($(mkClosure 'signal) incr) + simulateNetworkFailure transportInternals nid2 nid1 + _pid2 <- spawn nid1 ($(mkClosure 'signal) incr) + _pid3 <- spawn nid1 ($(mkClosure 'signal) incr) + + liftIO $ threadDelay 100000 + + count <- liftIO $ takeMVar iv + True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn' + + liftIO $ putMVar done () + + takeMVar done + main :: IO () main = do - Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters - let rtable = __remoteTable initRemoteTable + Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters + let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable runTests [ ("Unclosure", testUnclosure transport rtable) , ("Bind", testBind transport rtable) @@ -356,4 +443,6 @@ , ("ClosureExpect", testClosureExpect transport rtable) , ("SpawnChannel", testSpawnChannel transport rtable) , ("TDict", testTDict transport rtable) + , ("Fib", testFib transport rtable) + , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals) ]