diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,6 +1,8 @@ -[Unrealeased] +2018-06-12 David Simmons-Duffin 0.7.5 -* +* Bumped dependencies to build with ghc-9.8 +* Turn Serializable into a type synonym +* Remove tests for ghc-8.8.* and earlier 2018-06-12 Facundo Domínguez 0.7.4 diff --git a/benchmarks/Channels.hs b/benchmarks/Channels.hs --- a/benchmarks/Channels.hs +++ b/benchmarks/Channels.hs @@ -4,7 +4,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import Data.Binary (encode, decode) import qualified Data.ByteString.Lazy as BSL @@ -36,7 +36,8 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport - host port (\sn -> (host, port)) defaultTCPParameters - node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role + trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters + case trans of + Right transport -> do node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role + Left other -> error $ show other diff --git a/benchmarks/Latency.hs b/benchmarks/Latency.hs --- a/benchmarks/Latency.hs +++ b/benchmarks/Latency.hs @@ -3,7 +3,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import Data.Binary (encode, decode) import qualified Data.ByteString.Lazy as BSL @@ -31,7 +31,9 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport - host port (\sn -> (host, sn)) defaultTCPParameters - node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role + trans <- createTransport + (defaultTCPAddr host port) defaultTCPParameters + case trans of + Right transport -> do node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role + Left other -> error $ show other diff --git a/benchmarks/ProcessRing.hs b/benchmarks/ProcessRing.hs --- a/benchmarks/ProcessRing.hs +++ b/benchmarks/ProcessRing.hs @@ -12,7 +12,7 @@ import Control.Distributed.Process hiding (catch) import Control.Distributed.Process.Node import Control.Exception (catch, SomeException) -import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import System.Environment import System.Console.GetOpt @@ -111,7 +111,7 @@ (opt, _) <- parseArgv argv putStrLn $ "options: " ++ (show opt) Right transport <- createTransport - "127.0.0.1" "8090" (\sn -> ("127.0.0.1", sn)) defaultTCPParameters + (defaultTCPAddr "127.0.0.1" "8090" ) defaultTCPParameters node <- newLocalNode transport initRemoteTable catch (void $ runProcess node $ initialProcess opt) (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e)) diff --git a/benchmarks/Spawns.hs b/benchmarks/Spawns.hs --- a/benchmarks/Spawns.hs +++ b/benchmarks/Spawns.hs @@ -7,7 +7,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import Data.Binary (encode, decode) import qualified Data.ByteString.Lazy as BSL @@ -42,7 +42,8 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport - host port (\sn -> (host, sn)) defaultTCPParameters - node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role + trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters + case trans of + Right transport -> do node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role + Left other -> error $ show other diff --git a/benchmarks/Throughput.hs b/benchmarks/Throughput.hs --- a/benchmarks/Throughput.hs +++ b/benchmarks/Throughput.hs @@ -6,7 +6,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Transport.TCP (createTransport, defaultTCPParameters, defaultTCPAddr) import Data.Binary import qualified Data.ByteString.Lazy as BSL import Data.Typeable @@ -67,7 +67,8 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport - host port (\sn -> (host, sn)) defaultTCPParameters - node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role + trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters + case trans of + Right transport -> do node <- newLocalNode transport initRemoteTable + runProcess node $ initialProcess role + Left other -> error $ show other diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,12 +1,12 @@ Name: distributed-process -Version: 0.7.4 -Cabal-Version: >=1.8 +Version: 0.7.5 +Cabal-Version: >=1.10 Build-Type: Simple License: BSD3 License-File: LICENSE Copyright: Well-Typed LLP, Tweag I/O Limited Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: Facundo Domínguez +Maintainer: Tim Watson Stability: experimental Homepage: http://haskell-distributed.github.com/ Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues @@ -21,7 +21,7 @@ You will probably also want to install a Cloud Haskell backend such as distributed-process-simplelocalnet. -Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 GHC==8.0.1 +Tested-With: GHC==7.10.3 GHC==8.0.2 GHC==8.2.2 GHC==8.4.4 Category: Control extra-source-files: ChangeLog @@ -41,21 +41,21 @@ default: False Library - Build-Depends: base >= 4.6 && < 5, + Build-Depends: base >= 4.9 && < 5, binary >= 0.6.3 && < 0.10, - hashable >= 1.2.0.5 && < 1.3, + hashable >= 1.2.0.5 && <= 1.4.3.0, network-transport >= 0.4.1.0 && < 0.6, - stm >= 2.4 && < 2.5, - transformers >= 0.2 && < 0.6, + stm >= 2.4 && < 2.6, + transformers >= 0.2 && < 0.7, mtl >= 2.0 && < 2.4, data-accessor >= 0.2 && < 0.3, - bytestring >= 0.9 && < 0.11, - random >= 1.0 && < 1.2, + bytestring >= 0.9 && < 0.13, + random >= 1.0 && < 1.3, distributed-static >= 0.2 && < 0.4, rank1dynamic >= 0.1 && < 0.5, syb >= 0.3 && < 0.8, exceptions >= 0.5, - containers >= 0.5 && < 0.6, + containers >= 0.5 && < 0.7, deepseq >= 1.3.0.1 && < 1.6 Exposed-modules: Control.Distributed.Process, Control.Distributed.Process.Closure, @@ -83,6 +83,7 @@ Control.Distributed.Process.Management.Internal.Trace.Remote, Control.Distributed.Process.Management.Internal.Trace.Types, Control.Distributed.Process.Management.Internal.Trace.Tracer + default-language: Haskell2010 ghc-options: -Wall HS-Source-Dirs: src other-extensions: BangPatterns @@ -118,51 +119,56 @@ -- Tests are in distributed-process-test package, for convenience. benchmark distributed-process-throughput - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.6 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.7, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.10 - Main-Is: benchmarks/Throughput.hs - ghc-options: -Wall + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.9 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.81, + bytestring >= 0.9 && < 0.13, + binary >= 0.6.3 && < 0.10 + Main-Is: benchmarks/Throughput.hs + default-language: Haskell2010 + ghc-options: -Wall benchmark distributed-process-latency - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.6 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.7, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.10 - Main-Is: benchmarks/Latency.hs - ghc-options: -Wall + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.9 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.81, + bytestring >= 0.9 && < 0.13, + binary >= 0.6.3 && < 0.10 + Main-Is: benchmarks/Latency.hs + default-language: Haskell2010 + ghc-options: -Wall benchmark distributed-process-channels - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.6 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.7, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.10 - Main-Is: benchmarks/Channels.hs - ghc-options: -Wall + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.9 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.81, + bytestring >= 0.9 && < 0.13, + binary >= 0.6.3 && < 0.10 + Main-Is: benchmarks/Channels.hs + default-language: Haskell2010 + ghc-options: -Wall benchmark distributed-process-spawns - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.6 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.7, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.10 - Main-Is: benchmarks/Spawns.hs - ghc-options: -Wall + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.9 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.81, + bytestring >= 0.9 && < 0.13, + binary >= 0.6.3 && < 0.10 + Main-Is: benchmarks/Spawns.hs + default-language: Haskell2010 + ghc-options: -Wall benchmark distributed-process-ring - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.6 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.7, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.10 - Main-Is: benchmarks/ProcessRing.hs - ghc-options: -Wall -threaded -O2 -rtsopts + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.9 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.81, + bytestring >= 0.9 && < 0.13, + binary >= 0.6.3 && < 0.10 + Main-Is: benchmarks/ProcessRing.hs + default-language: Haskell2010 + ghc-options: -Wall -threaded -O2 -rtsopts 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 @@ -31,7 +31,6 @@ , orElse , retry ) -import Control.Applicative ((<$>), (<*>)) import Control.Exception (mask_, onException) import System.Timeout (timeout) import Control.Distributed.Process.Internal.StrictMVar @@ -45,7 +44,6 @@ , append ) import Data.Maybe (fromJust) -import Data.Traversable (traverse) import GHC.MVar (MVar(MVar)) import GHC.IO (IO(IO), unIO) import GHC.Exts (mkWeak#) diff --git a/src/Control/Distributed/Process/Internal/Closure/Explicit.hs b/src/Control/Distributed/Process/Internal/Closure/Explicit.hs --- a/src/Control/Distributed/Process/Internal/Closure/Explicit.hs +++ b/src/Control/Distributed/Process/Internal/Closure/Explicit.hs @@ -7,6 +7,7 @@ , KindSignatures , GADTs , EmptyDataDecls + , TypeOperators , DeriveDataTypeable #-} module Control.Distributed.Process.Internal.Closure.Explicit ( @@ -29,6 +30,7 @@ import Data.Rank1Typeable import Data.Binary(encode,put,get,Binary) import qualified Data.ByteString.Lazy as B +import Data.Kind (Type) -- | A RemoteRegister is a trasformer on a RemoteTable to register additional static values. type RemoteRegister = RemoteTable -> RemoteTable @@ -118,7 +120,7 @@ -- This generic uncurry courtesy Andrea Vezzosi data HTrue data HFalse -data Fun :: * -> * -> * -> * where +data Fun :: Type -> Type -> Type -> Type where Done :: Fun EndOfTuple r r Moar :: Fun xs f r -> Fun (x,xs) (x -> f) r 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 @@ -12,7 +12,6 @@ ) where import Prelude hiding (succ, any) -import Control.Applicative ((<$>)) import Language.Haskell.TH ( -- Q monad and operations Q @@ -28,6 +27,9 @@ , Exp , Type(AppT, ForallT, VarT, ArrowT) , Info(VarI) +#if MIN_VERSION_template_haskell(2,17,0) + , Specificity +#endif , TyVarBndr(PlainTV, KindedTV) , Pred #if MIN_VERSION_template_haskell(2,10,0) @@ -203,7 +205,11 @@ , concat [register, registerSDict, registerTDict] ) where +#if MIN_VERSION_template_haskell(2,17,0) + makeStatic :: [TyVarBndr Specificity] -> Type -> Q ([Dec], [Q Exp]) +#else makeStatic :: [TyVarBndr] -> Type -> Q ([Dec], [Q Exp]) +#endif makeStatic typVars typ = do static <- generateStatic origName typVars typ let dyn = case typVars of @@ -222,7 +228,12 @@ ) -- | Turn a polymorphic type into a monomorphic type using ANY and co +#if MIN_VERSION_template_haskell(2,17,0) +monomorphize :: [TyVarBndr Specificity] -> Type -> Q Type +#else monomorphize :: [TyVarBndr] -> Type -> Q Type +#endif + monomorphize tvs = let subst = zip (map tyVarBndrName tvs) anys in everywhereM (mkM (applySubst subst)) @@ -247,7 +258,11 @@ applySubst s t = gmapM (mkM (applySubst s)) t -- | Generate a static value +#if MIN_VERSION_template_haskell(2,17,0) +generateStatic :: Name -> [TyVarBndr Specificity] -> Type -> Q [Dec] +#else generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec] +#endif generateStatic n xs typ = do staticTyp <- [t| Static |] sequence @@ -259,7 +274,11 @@ , sfnD (staticName n) [| staticLabel $(showFQN n) |] ] where +#if MIN_VERSION_template_haskell(2,17,0) + typeable :: TyVarBndr Specificity -> Q Pred +#else typeable :: TyVarBndr -> Q Pred +#endif typeable tv = #if MIN_VERSION_template_haskell(2,10,0) conT (mkName "Typeable") `appT` varT (tyVarBndrName tv) @@ -315,9 +334,16 @@ sfnD n e = funD n [clause [] (normalB e) []] -- | The name of a type variable binding occurrence +#if MIN_VERSION_template_haskell(2,17,0) +tyVarBndrName :: TyVarBndr Specificity -> Name +tyVarBndrName (PlainTV n _) = n +tyVarBndrName (KindedTV n _ _) = n +#else tyVarBndrName :: TyVarBndr -> Name tyVarBndrName (PlainTV n) = n tyVarBndrName (KindedTV n _) = n +#endif + -- | Fully qualified name; that is, the name and the _current_ module -- 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 @@ -76,6 +76,7 @@ , unlink , monitor , unmonitor + , unmonitorAsync , withMonitor , withMonitor_ -- * Logging @@ -252,18 +253,16 @@ let us = processId proc node = processNode proc nodeId = localNodeId node - destNode = (processNodeId them) in do - case destNode == nodeId of - True -> sendLocal them msg - False -> liftIO $ sendMessage (processNode proc) - (ProcessIdentifier (processId proc)) - (ProcessIdentifier them) - NoImplicitReconnect - msg - -- We do not fire the trace event until after the sending is complete; - -- In the remote case, 'sendMessage' can block in the networking stack. + destNode = (processNodeId them) liftIO $ traceEvent (localEventBus node) (MxSent them us (createUnencodedMessage msg)) + if destNode == nodeId + then sendLocal them msg + else liftIO $ sendMessage (processNode proc) + (ProcessIdentifier (processId proc)) + (ProcessIdentifier them) + NoImplicitReconnect + msg -- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize -- and (in the case when the destination process resides on the same local @@ -283,9 +282,12 @@ -- usend :: Serializable a => ProcessId -> a -> Process () usend them msg = do - here <- getSelfNode + proc <- ask let there = processNodeId them - if here == there + let (us, node) = (processId proc, processNode proc) + let msg' = wrapMessage msg + liftIO $ traceEvent (localEventBus node) (MxSent them us msg') + if localNodeId (processNode proc) == there then sendLocal them msg else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them) (createMessage msg) @@ -305,7 +307,13 @@ -- Channels -- -------------------------------------------------------------------------------- --- | Create a new typed channel +-- | Create a new typed channel, bound to the calling Process +-- +-- Note that the channel is bound to the lifecycle of the process that evaluates +-- this function, such that when it dies/exits, the channel will no longer +-- function, but will remain accessible. Thus reading from the ReceivePort will +-- fail silently thereafter, blocking indefinitely (unless a timeout is used). +-- newChan :: Serializable a => Process (SendPort a, ReceivePort a) newChan = do proc <- ask @@ -333,13 +341,16 @@ sendChan :: Serializable a => SendPort a -> a -> Process () sendChan (SendPort cid) msg = do proc <- ask - let node = localNodeId (processNode proc) - destNode = processNodeId (sendPortProcessId cid) in do - case destNode == node of + let node = processNode proc + pid = processId proc + us = localNodeId node + them = processNodeId (sendPortProcessId cid) + liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg) + case them == us of True -> sendChanLocal cid msg False -> do - liftIO $ sendBinary (processNode proc) - (ProcessIdentifier (processId proc)) + liftIO $ sendBinary node + (ProcessIdentifier pid) (SendPortIdentifier cid) NoImplicitReconnect msg @@ -400,8 +411,11 @@ receiveWait :: [Match b] -> Process b receiveWait ms = do queue <- processQueue <$> ask - Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms) - proc + mProc <- liftIO $ dequeue queue Blocking (map unMatch ms) + case mProc of + Just proc' -> proc' + Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` " + ++ "in the absence of a timeout value." -- | Like 'receiveWait' but with a timeout. -- @@ -479,18 +493,15 @@ let node = processNode proc us = processId proc nid = localNodeId node - destNode = (processNodeId them) in do - case destNode == nid of - True -> sendCtrlMsg Nothing (LocalSend them msg) - False -> liftIO $ sendPayload (processNode proc) - (ProcessIdentifier (processId proc)) - (ProcessIdentifier them) - NoImplicitReconnect - (messageToPayload msg) - -- We do not fire the trace event until after the sending is complete; - -- In the remote case, 'sendMessage' can block in the networking stack. - liftIO $ traceEvent (localEventBus node) - (MxSent them us msg) + destNode = (processNodeId them) + liftIO $ traceEvent (localEventBus node) (MxSent them us msg) + if destNode == nid + then sendCtrlMsg Nothing (LocalSend them msg) + else liftIO $ sendPayload (processNode proc) + (ProcessIdentifier (processId proc)) + (ProcessIdentifier them) + NoImplicitReconnect + (messageToPayload msg) -- | Forward a raw 'Message' to the given 'ProcessId'. -- @@ -503,15 +514,11 @@ let node = processNode proc us = processId proc nid = localNodeId node - destNode = (processNodeId them) in do - case destNode == nid of - True -> sendCtrlMsg Nothing (LocalSend them msg) - False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) - msg - -- We do not fire the trace event until after the sending is complete; - -- In the remote case, 'sendCtrlMsg' can block in the networking stack. - liftIO $ traceEvent (localEventBus node) - (MxSent them us msg) + destNode = (processNodeId them) + liftIO $ traceEvent (localEventBus node) (MxSent them us msg) + if destNode == nid + then sendCtrlMsg Nothing (LocalSend them msg) + else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg -- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are -- 'Serializable' - like the datum they contain - but also note, deserialising @@ -1220,8 +1227,12 @@ -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process () -nsend label msg = - sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg)) +nsend label msg = do + proc <- ask + let msg' = createUnencodedMessage msg + liftIO $ traceEvent (localEventBus (processNode proc)) + (MxSentToName label (processId proc) msg') + sendCtrlMsg Nothing (NamedSend label msg') -- | Named send to a process in the local registry (asynchronous). -- This function makes /no/ attempt to serialize and (in the case when the @@ -1233,9 +1244,15 @@ -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () nsendRemote nid label msg = do - here <- getSelfNode - if here == nid then nsend label msg - else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg)) + proc <- ask + let us = processId proc + let node = processNode proc + if localNodeId node == nid + then nsend label msg + else let lbl = label ++ "@" ++ show nid in do + liftIO $ traceEvent (localEventBus node) + (MxSentToName lbl us (wrapMessage msg)) + sendCtrlMsg (Just nid) (NamedSend label (createMessage msg)) -- | Named send to a process in a remote registry (asynchronous) -- This function makes /no/ attempt to serialize and (in the case when the 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 @@ -14,13 +14,8 @@ , mkWeakMVar ) where -import Control.Applicative ((<$>)) import Control.Monad ((>=>)) -#if MIN_VERSION_base(4,6,0) -import Control.Exception (evaluate) -#else import Control.Exception (evaluate, mask_, onException) -#endif import qualified Control.Concurrent.MVar as MVar ( MVar , newEmptyMVar @@ -31,9 +26,6 @@ , withMVar , modifyMVar_ , modifyMVar -#if MIN_VERSION_base(4,6,0) - , modifyMVarMasked -#endif ) import GHC.MVar (MVar(MVar)) import GHC.IO (IO(IO), unIO) @@ -71,18 +63,11 @@ modifyMVarMasked :: StrictMVar a -> (a -> IO (a, b)) -> IO b modifyMVarMasked (StrictMVar v) f = -#if MIN_VERSION_base(4,6,0) - MVar.modifyMVarMasked v (f >=> evaluateFst) -#else mask_ $ do a <- MVar.takeMVar v (a',b) <- (f a >>= evaluate) `onException` MVar.putMVar v a MVar.putMVar v a' return b -#endif - where - evaluateFst :: (a, b) -> IO (a, b) - evaluateFst (x, y) = evaluate x >> return (x, y) mkWeakMVar :: StrictMVar a -> IO () -> IO (Weak (StrictMVar a)) mkWeakMVar q@(StrictMVar (MVar m#)) f = IO $ \s -> 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 @@ -118,6 +118,9 @@ import Control.Monad.Catch (MonadThrow(..), MonadCatch(..), MonadMask(..)) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative +#if !MIN_VERSION_base(4,13,0) && MIN_VERSION_base(4,9,0) +import Control.Monad.Fail (MonadFail) +#endif import Control.Monad.Fix (MonadFix) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT) import Control.Monad.IO.Class (MonadIO(..)) @@ -353,6 +356,9 @@ deriving ( Applicative , Functor , Monad +#if MIN_VERSION_base(4,9,0) + , MonadFail +#endif , MonadFix , MonadIO , MonadReader LocalProcess diff --git a/src/Control/Distributed/Process/Management.hs b/src/Control/Distributed/Process/Management.hs --- a/src/Control/Distributed/Process/Management.hs +++ b/src/Control/Distributed/Process/Management.hs @@ -76,6 +76,40 @@ -- -- * Whether messages will be taken from the mailbox first, or the event bus. -- +-- Since the event bus uses STM broadcast channels to communicate with agents, +-- no message written to the bus successfully can be lost. +-- +-- Agents can also receive messages via their mailboxes - these are subject to +-- the same guarantees as all inter-process message sending. +-- +-- Messages dispatched on an STM broadcast channel (i.e., management event bus) +-- are guaranteed to be delivered with the same FIFO ordering guarantees that +-- exist between two communicating processes, such that communication from the +-- node controller's threads (i.e., MxEvent's) will never be re-ordered, but +-- messages dispatched to the event bus by other processes (including, but not +-- limited to agents) are only guaranteed to be ordered between one sender and +-- one receiver. +-- +-- No guarantee exists for the ordering in which messages sent to an agent's +-- mailbox will be delivered, vs messages dispatched via the event bus. +-- +-- Because of the above, there are no ordering guarantees for messages sent +-- between agents, or for processes to agents, except for those that apply to +-- messages sent between regular processes, since agents are +-- implemented as such. +-- +-- The event bus is serial and single threaded. Anything that is published by +-- the node controller will be seen in FIFO order. There are no ordering +-- guarantees pertaining to entries published to the event bus by other +-- processes or agents. +-- +-- It should not be possible to see, for example, an @MxReceived@ before the +-- corresponding @MxSent@ event, since the places where we issue the @MxSent@ +-- write directly to the event bus (using STM) in the calling (green) thread, +-- before dispatching instructions to the node controller to perform the +-- necessary routing to deliver the message to a process (or registered name, +-- or typed channel) locally or remotely. +-- -- [Management Data API] -- -- Both management agents and clients of the API have access to a variety of @@ -143,8 +177,6 @@ -- -- > monitorNames = getSelfPid >>= nsend "name-monitor" -- > monitorNames2 = getSelfPid >>= mxNotify --- --- For some real-world examples, see the distributed-process-platform package. -- -- [Performance, Stablity and Scalability] -- diff --git a/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs b/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs --- a/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs +++ b/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs @@ -33,6 +33,7 @@ ( whereis , newChan , receiveChan + , die ) import Control.Distributed.Process.Management.Internal.Trace.Types ( TraceArg(..) @@ -168,6 +169,11 @@ withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp currentTracer <- receiveChan rp case currentTracer of - Nothing -> do { (Just p') <- whereis "tracer.initial"; act p' } + Nothing -> do mTP <- whereis "tracer.initial" + -- NB: this should NOT ever happen, but forcing pattern matches + -- is not considered cool in later versions of MonadFail + case mTP of + Just p' -> act p' + Nothing -> die $ "System Invariant Violation: Tracer Process " + ++ "Name Not Found (whereis tracer.initial)" (Just p) -> act p - diff --git a/src/Control/Distributed/Process/Management/Internal/Types.hs b/src/Control/Distributed/Process/Management/Internal/Types.hs --- a/src/Control/Distributed/Process/Management/Internal/Types.hs +++ b/src/Control/Distributed/Process/Management/Internal/Types.hs @@ -14,13 +14,13 @@ , Addressable(..) ) where -import Control.Applicative (Applicative) import Control.Concurrent.STM ( TChan ) import Control.Distributed.Process.Internal.Types ( Process , ProcessId + , SendPortId , Message , DiedReason , NodeId @@ -30,6 +30,7 @@ ( MonadState , StateT ) +import Control.Monad.Fix (MonadFix) import Data.Binary import Data.Typeable (Typeable) import GHC.Generics @@ -56,8 +57,14 @@ -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected) | MxSent ProcessId ProcessId Message -- ^ fired whenever a message is sent from a local process + | MxSentToName String ProcessId Message + -- ^ fired whenever a named send occurs + | MxSentToPort ProcessId SendPortId Message + -- ^ fired whenever a sendChan occurs | MxReceived ProcessId Message -- ^ fired whenever a message is received by a local process + | MxReceivedPort SendPortId Message + -- ^ fired whenever a message is received via a typed channel | MxConnected ConnectionId EndPointAddress -- ^ fired when a network-transport connection is first established | MxDisconnected ConnectionId EndPointAddress @@ -112,6 +119,7 @@ } deriving ( Functor , Monad , MonadIO + , MonadFix , ST.MonadState (MxAgentState s) , Typeable , Applicative 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 @@ -30,7 +30,7 @@ ( empty , toList , fromList - , filter + , partition , partitionWithKey , elems , size @@ -183,6 +183,9 @@ import Control.Distributed.Process.Management.Internal.Agent ( mxAgentController ) +import Control.Distributed.Process.Management.Internal.Types + ( MxEvent(..) + ) import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace ( remoteTable ) @@ -195,9 +198,6 @@ , traceLogFmt , enableTrace ) -import Control.Distributed.Process.Management.Internal.Types - ( MxEvent(..) - ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Messaging ( sendBinary @@ -486,7 +486,7 @@ data IncomingTarget = Uninit | ToProc ProcessId (Weak (CQueue Message)) - | ToChan TypedChannel + | ToChan SendPortId TypedChannel | ToNode data ConnectionState = ConnectionState { @@ -541,13 +541,17 @@ enqueue queue msg -- 'enqueue' is strict trace node (MxReceived pid msg) go st - Just (_, ToChan (TypedChannel chan')) -> do + Just (_, ToChan chId (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) + forM_ mChan $ \chan -> do + msg' <- atomically $ do + msg <- return $! decode (BSL.fromChunks payload) + -- We make sure the message is fully decoded when it is enqueued + writeTQueue chan msg + return msg + trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg' go st Just (_, ToNode) -> do let ctrlMsg = decode . BSL.fromChunks $ payload @@ -574,7 +578,7 @@ mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid) case mChannel of Just channel -> - go (incomingAt cid ^= Just (src, ToChan channel) $ st) + go (incomingAt cid ^= Just (src, ToChan chId channel) $ st) Nothing -> invalidRequest cid st $ "incoming attempt to connect to unknown channel of" @@ -903,7 +907,11 @@ modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons') - modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid) + -- we now consider all labels for this identifier unregistered + let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid + (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere) + mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames) + modify' $ registeredHere ^= keepNames remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>= mapM (\(pid,nidlist) -> @@ -955,7 +963,11 @@ do modify' $ registeredHereFor label ^= mPid updateRemote node currentVal mPid case mPid of - (Just p) -> liftIO $ trace node (MxRegistered p label) + (Just p) -> do + if reregistration + then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label) + else return () + liftIO $ trace node (MxRegistered p label) Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label) newVal <- gets (^. registeredHereFor label) ncSendToProcess from $ unsafeCreateUnencodedMessage $ @@ -1036,11 +1048,12 @@ -- If ch is Nothing, the process has given up the read end of -- the channel and we simply ignore the incoming message - this ch <- deRefWeak chan' - forM_ ch $ \chan -> deliverChan msg chan - where deliverChan :: forall a . Message -> TQueue a -> IO () - deliverChan (UnencodedMessage _ raw) chan' = + forM_ ch $ \chan -> deliverChan node from msg chan + where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO () + deliverChan n p (UnencodedMessage _ raw) chan' = do atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a) - deliverChan (EncodedMessage _ _) _ = + trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw) + deliverChan _ _ (EncodedMessage _ _) _ = -- this will not happen unless someone screws with Primitives.hs error "invalid local channel delivery" @@ -1070,7 +1083,7 @@ case mProc of Nothing -> dispatch (isLocal node (ProcessIdentifier from)) from (ProcessInfoNone DiedUnknownId) - Just proc -> do + Just proc -> do itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$> gets (^. links) itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors) @@ -1084,8 +1097,8 @@ infoNode = (processNodeId pid) , infoRegisteredNames = reg , infoMessageQueueLength = size - , infoMonitors = Set.toList itsMons - , infoLinks = Set.toList itsLinks + , infoMonitors = Set.toList itsMons + , infoLinks = Set.toList itsLinks } where dispatch :: (Serializable a) => Bool diff --git a/src/Control/Distributed/Process/Serializable.hs b/src/Control/Distributed/Process/Serializable.hs --- a/src/Control/Distributed/Process/Serializable.hs +++ b/src/Control/Distributed/Process/Serializable.hs @@ -1,4 +1,5 @@ {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -40,8 +41,7 @@ deriving (Typeable) -- | Objects that can be sent across the network -class (Binary a, Typeable a) => Serializable a -instance (Binary a, Typeable a) => Serializable a +type Serializable a = (Binary a, Typeable a) -- | Encode type representation as a bytestring encodeFingerprint :: Fingerprint -> ByteString diff --git a/src/Control/Distributed/Process/UnsafePrimitives.hs b/src/Control/Distributed/Process/UnsafePrimitives.hs --- a/src/Control/Distributed/Process/UnsafePrimitives.hs +++ b/src/Control/Distributed/Process/UnsafePrimitives.hs @@ -32,8 +32,25 @@ -- the /normal/ strategy). -- -- Use of the functions in this module can potentially change the runtime --- behaviour of your application. You have been warned! +-- behaviour of your application. In addition, messages passed between Cloud +-- Haskell processes are written to a tracing infrastructure on the local node, +-- to provide improved introspection and debugging facilities for complex actor +-- based systems. This module makes no attempt to force evaluation in these +-- cases either, thus evaluation problems in passed data structures could not +-- only crash your processes, but could also bring down critical internal +-- services on which the node relies to function correctly. -- +-- If you wish to repudiate such issues, you are advised to consider the use +-- of NFSerialisable in the distributed-process-extras package, which type +-- class brings NFData into scope along with Serializable, such that we can +-- force evaluation. Intended for use with modules such as this one, this +-- approach guarantees correct evaluatedness in terms of @NFData@. Please note +-- however, that we /cannot/ guarantee that an @NFData@ instance will behave the +-- same way as a @Binary@ one with regards evaluation, so it is still possible +-- to introduce unexpected behaviour by using /unsafe/ primitives in this way. +-- +-- You have been warned! +-- -- This module is exported so that you can replace the use of Cloud Haskell's -- /safe/ messaging primitives. If you want to use both variants, then you can -- take advantage of qualified imports, however "Control.Distributed.Process" @@ -55,7 +72,12 @@ , sendBinary , sendCtrlMsg ) - +import Control.Distributed.Process.Management.Internal.Types + ( MxEvent(..) + ) +import Control.Distributed.Process.Management.Internal.Trace.Types + ( traceEvent + ) import Control.Distributed.Process.Internal.Types ( ProcessId(..) , NodeId(..) @@ -79,34 +101,48 @@ -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process () -nsend label msg = - sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg)) +nsend label msg = do + proc <- ask + let us = processId proc + let msg' = wrapMessage msg + -- see [note: tracing] + liftIO $ traceEvent (localEventBus (processNode proc)) + (MxSentToName label us msg') + sendCtrlMsg Nothing (NamedSend label msg') -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () nsendRemote nid label msg = do proc <- ask - if localNodeId (processNode proc) == nid + let us = processId proc + let node = processNode proc + if localNodeId node == nid then nsend label msg - else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg)) + else + let lbl = label ++ "@" ++ show nid in do + -- see [note: tracing] NB: this is a remote call to another NC... + liftIO $ traceEvent (localEventBus node) + (MxSentToName lbl us (wrapMessage msg)) + sendCtrlMsg (Just nid) (NamedSend label (createMessage msg)) -- | Send a message send :: Serializable a => ProcessId -> a -> Process () send them msg = do proc <- ask let node = localNodeId (processNode proc) - destNode = (processNodeId them) in do - case destNode == node of - True -> unsafeSendLocal them msg - False -> liftIO $ sendMessage (processNode proc) - (ProcessIdentifier (processId proc)) - (ProcessIdentifier them) - NoImplicitReconnect - msg - where - unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process () - unsafeSendLocal pid msg' = - sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg') + destNode = (processNodeId them) + us = (processId proc) + msg' = wrapMessage msg in do + -- see [note: tracing] + liftIO $ traceEvent (localEventBus (processNode proc)) + (MxSent them us msg') + if destNode == node + then sendCtrlMsg Nothing $ LocalSend them msg' + else liftIO $ sendMessage (processNode proc) + (ProcessIdentifier (processId proc)) + (ProcessIdentifier them) + NoImplicitReconnect + msg -- | Send a message unreliably. -- @@ -121,30 +157,44 @@ usend them msg = do proc <- ask let there = processNodeId them + let (us, node) = (processId proc, processNode proc) + let msg' = wrapMessage msg + -- see [note: tracing] + liftIO $ traceEvent (localEventBus node) (MxSent them us msg') if localNodeId (processNode proc) == there - then sendCtrlMsg Nothing $ - LocalSend them (unsafeCreateUnencodedMessage msg) + then sendCtrlMsg Nothing $ LocalSend them msg' else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them) (createMessage msg) +-- [note: tracing] +-- Note that tracing writes to the local node's control channel, and this +-- module explicitly specifies to its clients that it does unsafe message +-- encoding. The same is true for the messages it puts onto the Management +-- event bus, however we do *not* want unevaluated thunks hitting the event +-- bus control thread. Hence the word /Unsafe/ in this module's name! +-- + -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process () sendChan (SendPort cid) msg = do proc <- ask - let node = localNodeId (processNode proc) - destNode = processNodeId (sendPortProcessId cid) in do - case destNode == node of - True -> unsafeSendChanLocal cid msg - False -> do - liftIO $ sendBinary (processNode proc) - (ProcessIdentifier (processId proc)) - (SendPortIdentifier cid) - NoImplicitReconnect - msg + let + node = processNode proc + pid = processId proc + us = localNodeId node + them = processNodeId (sendPortProcessId cid) + msg' = wrapMessage msg + liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg') + if them == us + then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!! + else liftIO $ sendBinary node + (ProcessIdentifier pid) + (SendPortIdentifier cid) + NoImplicitReconnect + msg where - unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process () - unsafeSendChanLocal spId msg' = - sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg') + unsafeSendChanLocal :: SendPortId -> Message -> Process () + unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m -- | Create an unencoded @Message@ for any @Serializable@ type. wrapMessage :: Serializable a => a -> Message