diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,40 @@ +Copyright (c) 2014, Jeremy Huffman + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Jeremy Huffman nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +distributed-process-lifted incorporates code from distributed-process, +Copyright Copyright Well-Typed LLP, 2011-2012 + +distributed-process is distributed under the terms of the BSD3 license. + +distributed-process-lifted incorporates code from monad-control, +Copyright © 2010, Bas van Dijk, Anders Kaseorg + +monad-control is distributed under the terms of the BSD3 license. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/distributed-process-lifted.cabal b/distributed-process-lifted.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-lifted.cabal @@ -0,0 +1,76 @@ +-- Initial distributed-process-lifted.cabal generated by cabal init. For +-- further documentation, see http://haskell.org/cabal/users-guide/ + +name: distributed-process-lifted +version: 0.1.0.0 +synopsis: monad-control style typeclass and transformer instances for Process monad. + +description: This package provides typeclasses and functions for lifting functions and control operations (such as spawnLocal) from the @Process@ monad + into transformer stacks based on the Process monad. It uses + + and a new typeclass 'Control.Distributed.Process.Lifted.Class.MonadProcessBase' which plays the same role as + . + Instances are provided for all the types - so stacks based on any of these + (e.g. @ReaderT Config Process a@) can be used seamlessly. + . + The Control.Distributed.Process.Lifted module exports all the same symbols as found in + Control.Distributed.Process, but they are all generalized. + Where appropriate it re-exports the more general functions from lifted-base (e.g. catch) rather than the versions re-implemented for @Process@. +homepage: https://github.com/jeremyjh/distributed-process-lifted +license: BSD3 +license-file: LICENSE +author: Jeremy Huffman +maintainer: jeremy@jeremyhuffman.com +-- copyright: +category: Control, Cloud Haskell +build-type: Simple +-- extra-source-files: +cabal-version: >=1.10 + +source-repository head + type: git + location: git://github.com/jeremyjh/distributed-process-lifted.git + +library + exposed-modules: Control.Distributed.Process.Lifted + , Control.Distributed.Process.Lifted.Class + , Control.Distributed.Process.Node.Lifted + -- other-modules: + -- other-extensions: + build-depends: base < 5 + , distributed-process >= 0.5.1 && < 0.6 + , distributed-process-monad-control >= 0.5 && < 0.6 + , network-transport >= 0.2 && < 0.5 + , lifted-base >= 0.2 && < 0.3 + , transformers >= 0.3 && < 0.5 + , transformers-base >= 0.4.1 && <= 0.5.0 + , monad-control >= 0.3 && < 1.1 + , mtl >= 2.0 && < 2.3 + , deepseq >= 1.2 && < 1.5 + hs-source-dirs: src + default-language: Haskell2010 + ghc-options: -Wall + +test-suite testlifted + type: exitcode-stdio-1.0 + main-is: runTCP.hs + default-extensions: DeriveDataTypeable, ScopedTypeVariables, RecordWildCards, GeneralizedNewtypeDeriving + other-modules: Control.Distributed.Process.Lifted.Tests + , Network.Transport.Test + build-depends: base >= 4.4 && < 5, + mtl, + network, + transformers, + distributed-process, + distributed-process-lifted, + binary, + network-transport, + lifted-base, + network-transport-tcp, + test-framework >= 0.6 && < 0.9, + test-framework-hunit >= 0.3 && < 0.4, + rematch >= 0.2 && < 0.3, + HUnit + ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind + hs-source-dirs: test + default-language: Haskell2010 diff --git a/src/Control/Distributed/Process/Lifted.hs b/src/Control/Distributed/Process/Lifted.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Lifted.hs @@ -0,0 +1,391 @@ +{-# Language RankNTypes #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Control.Distributed.Process.Lifted + ( module Control.Distributed.Process.Lifted + , module Control.Distributed.Process + , module Control.Exception.Lifted + ) +where + +import Control.Distributed.Process.Lifted.Class + + +import Control.Distributed.Process + ( + Closure + , DidSpawn(..) + , DiedReason(..) + , Match + , Message + , MonitorRef + , NodeId(..) + , ProcessTerminationException(..) + , ProcessRegistrationException(..) + , ProcessLinkException(..) + , NodeLinkException(..) + , PortLinkException(..) + , ProcessMonitorNotification(..) + , NodeMonitorNotification(..) + , PortMonitorNotification(..) + , DiedReason(..) + , NodeStats(..) + , Process + , ProcessId + , ProcessInfo(..) + , ReceivePort + , RegisterReply(..) + , RemoteTable + , SendPort + , SendPortId + , SpawnRef + , Static + , WhereIsReply(..) + , liftIO + , unsafeWrapMessage + , wrapMessage + , closure + , infoLinks + , infoMessageQueueLength + , infoMonitors + , infoNode + , infoRegisteredNames + , isEncoded + , nodeAddress + , nodeStatsLinks + , nodeStatsMonitors + , nodeStatsNode + , nodeStatsProcesses + , nodeStatsRegisteredNames + , processNodeId + , sendPortId + , sendPortProcessId + , match + , matchAnyIf + , matchIf + , matchAny + , matchChan + , matchMessage + , matchMessageIf + , matchSTM + ) +import qualified Control.Distributed.Process as Base +import Control.Distributed.Process.MonadBaseControl () + +import Control.Distributed.Process.Serializable (Serializable) +import Control.Distributed.Process.Closure (SerializableDict) +import Control.Distributed.Process.Internal.Types + (ProcessExitException(..)) + +import Control.Exception.Lifted + (bracket, bracket_, catch, catches, Exception, finally, mask, + mask_, onException, try, Handler(..)) +import qualified Control.Exception.Lifted as EX +import Data.Typeable (Typeable) + +-- compose arity 2 functions +(.:) :: (c->d) -> (a->b->c) -> a->b->d +f .: i = \l r -> f $ i l r + +-- | Generalized version of 'Base.spawnLocal' +spawnLocal :: (MonadProcessBase m) => m () -> m ProcessId +spawnLocal = liftBaseDiscardP Base.spawnLocal + +-- | Generalized version of 'Base.getSelfPid' +getSelfPid :: (MonadProcess m) => m ProcessId +getSelfPid = liftP Base.getSelfPid + +-- | Generalized version of 'Base.expect' +expect :: (MonadProcess m) => forall a. Serializable a => m a +expect = liftP Base.expect + +-- | Generalized version of 'Base.expectTimeout' +expectTimeout :: (MonadProcess m) => forall a. Serializable a => Int -> m (Maybe a) +expectTimeout = liftP . Base.expectTimeout + +-- | Generalized version of 'Base.register' +register :: (MonadProcess m) => String -> ProcessId -> m () +register name = liftP . Base.register name + +-- | Generalized version of 'Base.whereis' +whereis :: (MonadProcess m) => String -> m (Maybe ProcessId) +whereis = liftP . Base.whereis + +-- | Generalized version of 'Base.catchesExit' +catchesExit :: forall m a. (MonadProcessBase m) => m a -> [ProcessId -> Message -> m (Maybe a)] -> m a +-- Had to re-implement due to requirement to preserve structure of Maybe a in +-- each possible handler. +catchesExit act handlers = catch act (`handleExit` handlers) + where + handleExit :: ProcessExitException + -> [ProcessId -> Message -> m (Maybe a)] + -> m a + handleExit ex [] = EX.throwIO ex + handleExit ex@(ProcessExitException from msg) (h:hs) = do + r <- h from msg + case r of + Nothing -> handleExit ex hs + Just p -> return p + +-- | Generalized version of 'Base.delegate' +delegate :: MonadProcess m => ProcessId -> (Message -> Bool) -> m () +delegate = liftP .: Base.delegate + +-- | Generalized version of 'Base.forward' +forward :: MonadProcess m => Message -> ProcessId -> m () +forward = liftP .: Base.forward + +-- | Generalized version of 'Base.getLocalNodeStats' +getLocalNodeStats :: MonadProcess m => m NodeStats +getLocalNodeStats = liftP Base.getLocalNodeStats + +-- | Generalized version of 'Base.getNodeStats' +getNodeStats :: MonadProcess m => NodeId -> m (Either DiedReason NodeStats) +getNodeStats = liftP . Base.getNodeStats + +-- | Generalized version of 'Base.getProcessInfo' +getProcessInfo :: MonadProcess m => ProcessId -> m (Maybe ProcessInfo) +getProcessInfo = liftP . Base.getProcessInfo + +-- | Generalized version of 'Base.getSelfNode' +getSelfNode :: MonadProcess m => m NodeId +getSelfNode = liftP Base.getSelfNode + +-- | Generalized version of 'Base.kill' +kill :: MonadProcess m => ProcessId -> String -> m () +kill = liftP .: Base.kill + +-- | Generalized version of 'Base.link' +link :: MonadProcess m => ProcessId -> m () +link = liftP . Base.link + +-- | Generalized version of 'Base.linkNode' +linkNode :: MonadProcess m => NodeId -> m () +linkNode = liftP . Base.linkNode + +-- | Generalized version of 'Base.linkPort' +linkPort :: MonadProcess m => SendPort a -> m () +linkPort = liftP . Base.linkPort + +-- | Generalized version of 'Base.monitor' +monitor :: MonadProcess m => ProcessId -> m MonitorRef +monitor = liftP . Base.monitor + +-- | Generalized version of 'Base.monitorNode' +monitorNode :: MonadProcess m => NodeId -> m MonitorRef +monitorNode = liftP . Base.monitorNode + +-- | Generalized version of 'Base.receiveTimeout' +receiveTimeout :: MonadProcess m => Int -> [Match b] -> m (Maybe b) +receiveTimeout = liftP .: Base.receiveTimeout + +-- | Generalized version of 'Base.receiveWait' +receiveWait :: MonadProcess m => [Match b] -> m b +receiveWait = liftP . Base.receiveWait + +-- | Generalized version of 'Base.reconnect' +reconnect :: MonadProcess m => ProcessId -> m () +reconnect = liftP . Base.reconnect + +-- | Generalized version of 'Base.reconnectPort' +reconnectPort :: MonadProcess m => SendPort a -> m () +reconnectPort = liftP . Base.reconnectPort + +-- | Generalized version of 'Base.registerRemoteAsync' +registerRemoteAsync :: MonadProcess m => NodeId -> String -> ProcessId -> m () +registerRemoteAsync n = liftP .: Base.registerRemoteAsync n + +-- | Generalized version of 'Base.relay' +relay :: MonadProcess m => ProcessId -> m () +relay = liftP . Base.relay + +-- | Generalized version of 'Base.reregister' +reregister :: MonadProcess m => String -> ProcessId -> m () +reregister = liftP .: Base.reregister + +-- | Generalized version of 'Base.reregisterRemoteAsync' +reregisterRemoteAsync :: MonadProcess m => NodeId -> String -> ProcessId -> m () +reregisterRemoteAsync n = liftP .: Base.reregisterRemoteAsync n + +-- | Generalized version of 'Base.say' +say :: MonadProcess m => String -> m () +say = liftP . Base.say + +-- | Generalized version of 'Base.spawn' +spawn :: MonadProcess m => NodeId -> Closure (Process ()) -> m ProcessId +spawn = liftP .: Base.spawn + +-- | Generalized version of 'Base.spawnAsync' +spawnAsync :: MonadProcess m => NodeId -> Closure (Process ()) -> m SpawnRef +spawnAsync = liftP .: Base.spawnAsync + +-- | Generalized version of 'Base.spawnLink' +spawnLink :: MonadProcess m => NodeId -> Closure (Process ()) -> m ProcessId +spawnLink = liftP .: Base.spawnLink + +-- | Generalized version of 'Base.spawnMonitor' +spawnMonitor :: MonadProcess m => NodeId -> Closure (Process ()) -> m (ProcessId, MonitorRef) +spawnMonitor = liftP .: Base.spawnMonitor + +-- | Generalized version of 'Base.spawnSupervised' +spawnSupervised :: MonadProcess m => NodeId -> Closure (Process ()) -> m (ProcessId, MonitorRef) +spawnSupervised = liftP .: Base.spawnSupervised + +-- | Generalized version of 'Base.terminate' +terminate :: MonadProcess m => m a +terminate = liftP Base.terminate + +-- | Generalized version of 'Base.unlink' +unlink :: MonadProcess m => ProcessId -> m () +unlink = liftP . Base.unlink + +-- | Generalized version of 'Base.unlinkNode' +unlinkNode :: MonadProcess m => NodeId -> m () +unlinkNode = liftP . Base.unlinkNode + +-- | Generalized version of 'Base.unlinkPort' +unlinkPort :: MonadProcess m => SendPort a -> m () +unlinkPort = liftP . Base.unlinkPort + +-- | Generalized version of 'Base.unmonitor' +unmonitor :: MonadProcess m => MonitorRef -> m () +unmonitor = liftP . Base.unmonitor + +-- | Generalized version of 'Base.unregister' +unregister :: MonadProcess m => String -> m () +unregister = liftP . Base.unregister + +-- | Generalized version of 'Base.unregisterRemoteAsync' +unregisterRemoteAsync :: MonadProcess m => NodeId -> String -> m () +unregisterRemoteAsync = liftP .: Base.unregisterRemoteAsync + +-- | Generalized version of 'Base.whereisRemoteAsync' +whereisRemoteAsync :: MonadProcess m => NodeId -> String -> m () +whereisRemoteAsync = liftP .: Base.whereisRemoteAsync + +-- | Generalized version of 'Base.withMonitor' +withMonitor :: MonadProcessBase m => ProcessId -> m a -> m a +withMonitor pid ma = controlP $ \runInP -> + Base.withMonitor pid (runInP ma) + +-- | Generalized version of 'Base.call' +call :: (MonadProcess m,Serializable a) + => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> m a +call s = liftP .: Base.call s + +-- | Generalized version of 'Base.catchExit' +catchExit :: (MonadProcessBase m,Show a,Serializable a) + => m b -> (ProcessId -> a -> m b) -> m b +catchExit ma handler = controlP $ \runInP -> + Base.catchExit (runInP ma) + (\pid msg -> runInP $ handler pid msg) + +-- | Generalized version of 'Base.die' +die :: (MonadProcess m, Serializable a) => a -> m b +die = liftP . Base.die + +-- | Generalized version of 'Base.exit' +exit :: (MonadProcess m, Serializable a) => ProcessId -> a -> m () +exit = liftP .: Base.exit + +-- | Generalized version of 'Base.handleMessage' +handleMessage :: (MonadProcess m,Serializable a) + => Message -> (a -> Process b) -> m (Maybe b) +handleMessage msg f = liftP $ Base.handleMessage msg f + +-- | Generalized version of 'Base.handleMessageIf' +handleMessageIf :: (MonadProcess m,Serializable a) + => Message -> (a -> Bool) -> (a -> Process b) -> m (Maybe b) +handleMessageIf msg p f = liftP $ Base.handleMessageIf msg p f + +-- | Generalized version of 'Base.handleMessageIf_' +handleMessageIf_ :: (MonadProcess m,Serializable a) + => Message -> (a -> Bool) -> (a -> Process ()) -> m () +handleMessageIf_ msg p f = liftP $ Base.handleMessageIf_ msg p f + +-- | Generalized version of 'Base.handleMessage_' +handleMessage_ :: (MonadProcess m, Serializable a) => Message -> (a -> Process ()) -> m () +handleMessage_ msg f = liftP $ Base.handleMessage_ msg f + +-- | Generalized version of 'Base.mergePortsBiased' +mergePortsBiased :: (MonadProcess m,Serializable a) + => [ReceivePort a] -> m (ReceivePort a) +mergePortsBiased = liftP . Base.mergePortsBiased + +-- | Generalized version of 'Base.mergePortsRR' +mergePortsRR :: (MonadProcess m, Serializable a) => [ReceivePort a] -> m (ReceivePort a) +mergePortsRR = liftP . Base.mergePortsRR + +-- | Generalized version of 'Base.monitorPort' +monitorPort :: (MonadProcess m, Serializable a) => SendPort a -> m MonitorRef +monitorPort = liftP . Base.monitorPort + +-- | Generalized version of 'Base.newChan' +newChan :: (MonadProcess m, Serializable a) => m (SendPort a, ReceivePort a) +newChan = liftP Base.newChan + +-- | Generalized version of 'Base.nsend' +nsend :: (MonadProcess m, Serializable a) => String -> a -> m () +nsend = liftP .: Base.nsend + +-- | Generalized version of 'Base.nsendRemote' +nsendRemote :: (MonadProcess m, Serializable a) => NodeId -> String -> a -> m () +nsendRemote n = liftP .: Base.nsendRemote n + +-- | Generalized version of 'Base.proxy' +proxy :: (MonadProcess m, Serializable a) => ProcessId -> (a -> Process Bool) -> m () +proxy = liftP .: Base.proxy + +-- | Generalized version of 'Base.receiveChan' +receiveChan :: (MonadProcess m, Serializable a) => ReceivePort a -> m a +receiveChan = liftP . Base.receiveChan + +-- | Generalized version of 'Base.receiveChanTimeout' +receiveChanTimeout :: (MonadProcess m, Serializable a) => Int -> ReceivePort a -> m (Maybe a) +receiveChanTimeout = liftP .: Base.receiveChanTimeout + +-- | Generalized version of 'Base.send' +send :: (MonadProcess m, Serializable a) => ProcessId -> a -> m () +send = liftP .: Base.send + +-- | Generalized version of 'Base.sendChan' +sendChan :: (MonadProcess m, Serializable a) => SendPort a -> a -> m () +sendChan = liftP .: Base.sendChan + +-- | Generalized version of 'Base.spawnChannel' +spawnChannel :: (MonadProcess m,Serializable a) + => Static (SerializableDict a) + -> NodeId + -> Closure (ReceivePort a -> Process ()) + -> m (SendPort a) +spawnChannel s = liftP .: Base.spawnChannel s + +-- | Generalized version of 'Base.spawnChannelLocal' +spawnChannelLocal :: (MonadProcess m,Serializable a) + => (ReceivePort a -> Process ()) -> m (SendPort a) +spawnChannelLocal = liftP . Base.spawnChannelLocal + +-- | Generalized version of 'Base.unClosure' +unClosure :: (MonadProcess m, Typeable a) => Closure a -> m a +unClosure = liftP . Base.unClosure + +-- | Generalized version of 'Base.unStatic' +unStatic :: (MonadProcess m, Typeable a) => Static a -> m a +unStatic = liftP . Base.unStatic + +-- | Generalized version of 'Base.unsafeNSend' +unsafeNSend :: (MonadProcess m, Serializable a) => String -> a -> m () +unsafeNSend = liftP .: Base.unsafeNSend + +-- | Generalized version of 'Base.unsafeSend' +unsafeSend :: (MonadProcess m, Serializable a) => ProcessId -> a -> m () +unsafeSend = liftP .: Base.unsafeSend + +-- | Generalized version of 'Base.unsafeSendChan' +unsafeSendChan :: (MonadProcess m, Serializable a) => SendPort a -> a -> m () +unsafeSendChan = liftP .: Base.unsafeSendChan + +-- | Generalized version of 'Base.unwrapMessage' +unwrapMessage :: (MonadProcess m, Serializable a) => Message -> m (Maybe a) +unwrapMessage = liftP . Base.unwrapMessage diff --git a/src/Control/Distributed/Process/Lifted/Class.hs b/src/Control/Distributed/Process/Lifted/Class.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Lifted/Class.hs @@ -0,0 +1,138 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE CPP #-} + +module Control.Distributed.Process.Lifted.Class where + +import Control.Distributed.Process (Process) +import Control.Distributed.Process.MonadBaseControl () + +import qualified Control.Monad.State.Strict as StateS +import Control.Monad.Trans (MonadIO, + lift) + +import Control.Monad.Trans.Control +import Control.Monad.Base (MonadBase(..)) + +import Data.Monoid ( Monoid ) + +import Control.Monad.Trans.Identity (IdentityT) +import Control.Monad.Trans.List (ListT) +import Control.Monad.Trans.Maybe (MaybeT) +import Control.Monad.Trans.Reader (ReaderT) +import Control.Monad.Trans.State (StateT) +import Control.Monad.Trans.Writer (WriterT) +import Control.Monad.Trans.RWS (RWST) +#if MIN_VERSION_transformers(0,4,0) +import Control.Monad.Trans.Except (ExceptT) +#endif +import qualified Control.Monad.Trans.RWS.Strict as Strict (RWST) +import qualified Control.Monad.Trans.State.Strict as Strict (StateT) +import qualified Control.Monad.Trans.Writer.Strict as Strict (WriterT) + +-- | A class into instances of which Process operations can be lifted; +-- similar to MonadIO or MonadBase. +class (Monad m, MonadIO m, MonadBase IO m, MonadBaseControl IO m) => MonadProcess m where + -- |lift a base 'Process' computation into the current monad + liftP :: Process a -> m a + +-- | A Clone of 'MonadBaseControl' specialized to the Process monad. This +-- uses the 'MonadTransControl' typeclass for transformer default instances, so the +-- core wrapping/unwrapping logic is not duplicated. This class +-- is needed because the MonadBaseControl instance for Process +-- has IO as the base. +class (MonadProcess m) => MonadProcessBase m where + type StMP m a :: * + liftBaseWithP :: (RunInBaseP m -> Process a) -> m a + restoreMP :: StMP m a -> m a + +-- | A clone of 'RunInBase' for MonadProcessBase. +type RunInBaseP m = forall a. m a -> Process (StMP m a) + +-- | A clone of 'ComposeSt' for MonadProcessBase. +type ComposeStP t m a = StMP m (StT t a) + +-- | A clone of 'RunInBaseDefault' for MonadProcessBase. +type RunInBaseDefaultP t m = forall a. t m a -> Process (ComposeStP t m a) + +-- | A clone of 'defaultLiftBaseWith' for MonadProcessBase. +-- This re-uses the MonadTransControl typeclass the same way as the +-- original; core wrapping/unwrapping logic for each transformer type is not duplicated. +defaultLiftBaseWithP :: (MonadTransControl t, MonadProcessBase m) + => (RunInBaseDefaultP t m -> Process a) -> t m a +defaultLiftBaseWithP f= liftWith $ \run -> + liftBaseWithP $ \runInBase -> + f $ runInBase . run + +-- | A clone of 'defaultRestoreMP' for MonadProcessBase. +-- This re-uses the MonadTransControl typeclass the same way as the +-- original; core wrapping/unwrapping logic for each transformer type is not duplicated. +defaultRestoreMP :: (MonadTransControl t, MonadProcessBase m) + => ComposeStP t m a -> t m a +defaultRestoreMP = restoreT . restoreMP + +-- | A clone of 'control' for MonadProcessBase. +controlP :: MonadProcessBase m => (RunInBaseP m -> Process (StMP m a)) -> m a +controlP f = liftBaseWithP f >>= restoreMP + +-- | A clone of 'liftBaseDiscard' for MonadProcessBase. +liftBaseDiscardP :: MonadProcessBase m => (Process () -> Process a) -> m () -> m a +liftBaseDiscardP f m = liftBaseWithP $ \runInBase -> f $ StateS.void $ runInBase m + +instance MonadProcess Process where + liftP = id + +instance MonadProcessBase Process where + type StMP Process a = a + liftBaseWithP f = f id + restoreMP = return + +#define LIFTP(T) \ +instance (MonadProcess m) => MonadProcess (T m) where liftP = lift . liftP \ + +LIFTP(IdentityT) +LIFTP(MaybeT) +LIFTP(ListT) +LIFTP(ReaderT r) +LIFTP(Strict.StateT s) +LIFTP( StateT s) +#if MIN_VERSION_transformers(0,4,0) +LIFTP(ExceptT e) +#endif + +#undef LIFTP +#define LIFTP(CTX, T) \ +instance (CTX, MonadProcess m) => MonadProcess (T m) where liftP = lift . liftP \ + +LIFTP(Monoid w, Strict.WriterT w) +LIFTP(Monoid w, WriterT w) +LIFTP(Monoid w, Strict.RWST r w s) +LIFTP(Monoid w, RWST r w s) + +#define BODY(T) { \ + type StMP (T m) a = ComposeStP (T) m a; \ + liftBaseWithP = defaultLiftBaseWithP; \ + restoreMP = defaultRestoreMP; \ + {-# INLINABLE liftBaseWithP #-}; \ + {-# INLINABLE restoreMP #-}} + +#define TRANS( T) \ + instance (MonadProcessBase m) => MonadProcessBase (T m) where BODY(T) +#define TRANS_CTX(CTX, T) \ + instance (CTX, MonadProcessBase m) => MonadProcessBase (T m) where BODY(T) + +TRANS(IdentityT) +TRANS(MaybeT) +TRANS(ListT) +TRANS(ReaderT r) +TRANS(Strict.StateT s) +TRANS( StateT s) +#if MIN_VERSION_transformers(0,4,0) +TRANS(ExceptT e) +#endif +TRANS_CTX(Monoid w, Strict.WriterT w) +TRANS_CTX(Monoid w, WriterT w) +TRANS_CTX(Monoid w, Strict.RWST r w s) +TRANS_CTX(Monoid w, RWST r w s) diff --git a/src/Control/Distributed/Process/Node/Lifted.hs b/src/Control/Distributed/Process/Node/Lifted.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Node/Lifted.hs @@ -0,0 +1,58 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Control.Distributed.Process.Node.Lifted + ( module Control.Distributed.Process.Node.Lifted + , Base.LocalNode + , Base.initRemoteTable + , Base.localNodeId + ) where + + +import Control.Monad.Base ( MonadBase, liftBase ) +import Control.Monad (void) +import Control.Distributed.Process (ProcessId, Process, RemoteTable) +import Control.Distributed.Process.Node (LocalNode) + +import Control.Exception (throw, SomeException) +import Control.Exception.Lifted (try) +import Network.Transport (Transport) +import qualified Control.Distributed.Process.Node as Base +import Control.Concurrent.MVar.Lifted + (newEmptyMVar, putMVar, takeMVar) +import Control.Distributed.Process.MonadBaseControl () +import Control.DeepSeq (NFData, deepseq) + + +-- | Generalized version of 'MVar.putMVar'. +closeLocalNode :: MonadBase IO m => LocalNode -> m () +closeLocalNode = liftBase . Base.closeLocalNode + +-- | Generalized version of 'Base.forkProcess'. +forkProcess :: MonadBase IO m => LocalNode -> Process () -> m ProcessId +forkProcess n = liftBase . Base.forkProcess n + +-- | Generalized version of 'Base.newLocalNode'. +newLocalNode :: MonadBase IO m => Transport -> RemoteTable -> m LocalNode +newLocalNode t = liftBase . Base.newLocalNode t + +-- | Generalized version of 'Base.runProcess' +runProcess :: MonadBase IO m => LocalNode -> Process () -> m () +runProcess n = liftBase . Base.runProcess n + +-- | A variant of 'runProcess' which returns a value. This works just +-- like 'runProcess' by forking a new process with a captured MVar, but it +-- will take the result of the computation. If the computation throws an +-- exception, it will be re-thrown by 'fromProcess' in the calling thread. +fromProcess :: forall a m. (NFData a, MonadBase IO m) + => LocalNode -> Process a -> m a +fromProcess node ma = + do resultMV <- newEmptyMVar + void . forkProcess node $ + do eresult <- try (do a <- ma; a `deepseq` return a) :: Process (Either SomeException a) + case eresult of + Right result -> putMVar resultMV result + Left exception -> putMVar resultMV (throw exception) + !result <- takeMVar resultMV + return result diff --git a/test/Control/Distributed/Process/Lifted/Tests.hs b/test/Control/Distributed/Process/Lifted/Tests.hs new file mode 100644 --- /dev/null +++ b/test/Control/Distributed/Process/Lifted/Tests.hs @@ -0,0 +1,1494 @@ +{-# LANGUAGE CPP #-} +module Control.Distributed.Process.Lifted.Tests (tests) where + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch) +#endif + +import Network.Transport.Test (TestTransport(..)) + +import Data.Binary (Binary(..)) +import Data.Typeable (Typeable) +import Data.Foldable (forM_) +import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId) +import Control.Concurrent.MVar.Lifted + ( MVar + , newEmptyMVar + , putMVar + , takeMVar + , readMVar + ) +import Control.Monad (replicateM_, replicateM, forever, void) +import qualified Control.Monad.Reader as R +import qualified Control.Monad.State.Strict as StS +import qualified Control.Monad.State.Lazy as StL +import qualified Control.Monad.RWS as RWST +import qualified Control.Monad.Trans.Except as EX +import Control.Exception.Lifted (SomeException, throwIO) +import qualified Control.Exception as Ex (catch) +import Control.Applicative ((<$>), (<*>), pure, (<|>)) +import qualified Network.Transport as NT (closeEndPoint) +import Control.Distributed.Process.Lifted +import Control.Distributed.Process.Internal.Types + ( + LocalNode(localEndPoint) + , ProcessExitException(..) + , nullProcessId + ) +import Control.Distributed.Process.Node.Lifted +import Control.Distributed.Process.Serializable (Serializable) + +import Test.HUnit (Assertion, assertFailure) +import Test.Framework (Test, testGroup) +import Test.Framework.Providers.HUnit (testCase) +import Control.Rematch hiding (match) +import Control.Rematch.Run (Match(..)) + +newtype Ping = Ping ProcessId + deriving (Typeable, Binary, Show) + +newtype Pong = Pong ProcessId + deriving (Typeable, Binary, Show) + +-------------------------------------------------------------------------------- +-- Supporting definitions -- +-------------------------------------------------------------------------------- + +expectThat :: a -> Matcher a -> Assertion +expectThat a matcher = case res of + MatchSuccess -> return () + (MatchFailure msg) -> assertFailure msg + where res = runMatch matcher a + +-- | Like fork, but throw exceptions in the child thread to the parent +forkTry :: IO () -> IO ThreadId +forkTry p = do + tid <- myThreadId + forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException)) + +-- | The ping server from the paper +ping :: Process () +ping = do + Pong partner <- expect + self <- getSelfPid + 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 + +data Add = Add ProcessId Double Double deriving (Typeable) +data Divide = Divide ProcessId Double Double deriving (Typeable) +data DivByZero = DivByZero deriving (Typeable) + +instance Binary Add where + put (Add pid x y) = put pid >> put x >> put y + get = Add <$> get <*> get <*> get + +instance Binary Divide where + put (Divide pid x y) = put pid >> put x >> put y + get = Divide <$> get <*> get <*> get + +instance Binary DivByZero where + put DivByZero = return () + get = return DivByZero + +-- The math server from the paper +math :: Process () +math = do + receiveWait + [ match (\(Add pid x y) -> send pid (x + y)) + , matchIf (\(Divide _ _ y) -> y /= 0) + (\(Divide pid x y) -> send pid (x / y)) + , match (\(Divide pid _ _) -> send pid DivByZero) + ] + math + +-- | Monitor or link to a remote node +monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link + -> ProcessId -- Process to monitor/link to + -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up + -> Process (Maybe MonitorRef) +monitorOrLink mOrL pid mSignal = do + result <- if mOrL then Just <$> monitor pid + else link pid >> return Nothing + -- Monitor is asynchronous, which usually does not matter but if we want a + -- *specific* signal then it does. Therefore we wait an arbitrary delay and + -- hope that this means the monitor has been set up + forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal () + return result + +monitorTestProcess :: ProcessId -- Process to monitor/link to + -> Bool -- 'True' for monitor, 'False' for link + -> Bool -- Should we unmonitor? + -> DiedReason -- Expected cause of death + -> Maybe (MVar ()) -- Signal for 'monitor set up' + -> MVar () -- Signal for successful termination + -> Process () +monitorTestProcess theirAddr mOrL un reason monitorSetup done = + catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup + case (un, mRef) of + (True, Nothing) -> do + unlink theirAddr + liftIO $ putMVar done () + (True, Just ref) -> do + unmonitor ref + liftIO $ putMVar done () + (False, ref) -> do + ProcessMonitorNotification ref' pid reason' <- expect + True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason' + liftIO $ putMVar done () + ) + (\(ProcessLinkException pid reason') -> do + True <- return $ pid == theirAddr && not mOrL && not un && reason == reason' + liftIO $ putMVar done () + ) + +-------------------------------------------------------------------------------- +-- The tests proper -- +-------------------------------------------------------------------------------- + +-- | Basic ping test +testPing :: TestTransport -> Assertion +testPing TestTransport{..} = do + serverAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- Server + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode ping + putMVar serverAddr addr + + -- Client + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + pingServer <- readMVar serverAddr + + let numPings = 10000 + + runProcess localNode $ do + pid <- getSelfPid + replicateM_ numPings $ do + send pingServer (Pong pid) + Ping _ <- expect + return () + + putMVar clientDone () + + takeMVar clientDone + +-- | Monitor a process on an unreachable node +testMonitorUnreachable :: TestTransport -> Bool -> Bool -> Assertion +testMonitorUnreachable TestTransport{..} mOrL un = do + deadProcess <- newEmptyMVar + done <- newEmptyMVar + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode . liftIO $ threadDelay 1000000 + closeLocalNode localNode + putMVar deadProcess addr + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + theirAddr <- readMVar deadProcess + runProcess localNode $ + monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done + + takeMVar done + +-- | Monitor a process which terminates normally +testMonitorNormalTermination :: TestTransport -> Bool -> Bool -> Assertion +testMonitorNormalTermination TestTransport{..} mOrL un = do + monitorSetup <- newEmptyMVar + monitoredProcess <- newEmptyMVar + done <- newEmptyMVar + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode $ + liftIO $ readMVar monitorSetup + putMVar monitoredProcess addr + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + theirAddr <- readMVar monitoredProcess + runProcess localNode $ + monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done + + takeMVar done + +-- | Monitor a process which terminates abnormally +testMonitorAbnormalTermination :: TestTransport -> Bool -> Bool -> Assertion +testMonitorAbnormalTermination TestTransport{..} mOrL un = do + monitorSetup <- newEmptyMVar + monitoredProcess <- newEmptyMVar + done <- newEmptyMVar + + let err = userError "Abnormal termination" + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode . liftIO $ do + readMVar monitorSetup + throwIO err + putMVar monitoredProcess addr + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + theirAddr <- readMVar monitoredProcess + runProcess localNode $ + monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done + + takeMVar done + +-- | Monitor a local process that is already dead +testMonitorLocalDeadProcess :: TestTransport -> Bool -> Bool -> Assertion +testMonitorLocalDeadProcess TestTransport{..} mOrL un = do + processDead <- newEmptyMVar + processAddr <- newEmptyMVar + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + forkIO $ do + addr <- forkProcess localNode . liftIO $ putMVar processDead () + putMVar processAddr addr + + forkIO $ do + theirAddr <- readMVar processAddr + readMVar processDead + runProcess localNode $ do + monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done + + takeMVar done + +-- | Monitor a remote process that is already dead +testMonitorRemoteDeadProcess :: TestTransport -> Bool -> Bool -> Assertion +testMonitorRemoteDeadProcess TestTransport{..} mOrL un = do + processDead <- newEmptyMVar + processAddr <- newEmptyMVar + done <- newEmptyMVar + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode . liftIO $ putMVar processDead () + putMVar processAddr addr + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + theirAddr <- readMVar processAddr + readMVar processDead + runProcess localNode $ do + monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done + + takeMVar done + +-- | Monitor a process that becomes disconnected +testMonitorDisconnect :: TestTransport -> Bool -> Bool -> Assertion +testMonitorDisconnect TestTransport{..} mOrL un = do + processAddr <- newEmptyMVar + monitorSetup <- newEmptyMVar + done <- newEmptyMVar + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode . liftIO $ threadDelay 1000000 + putMVar processAddr addr + readMVar monitorSetup + NT.closeEndPoint (localEndPoint localNode) + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + theirAddr <- readMVar processAddr + runProcess localNode $ do + monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done + + takeMVar done + +-- | Test the math server (i.e., receiveWait) +testMath :: TestTransport -> Assertion +testMath TestTransport{..} = do + serverAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- Server + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode math + putMVar serverAddr addr + + -- Client + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + mathServer <- readMVar serverAddr + + 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 + +-- | Send first message (i.e. connect) to an already terminated process +-- (without monitoring); then send another message to a second process on +-- the same remote node (we're checking that the remote node did not die) +testSendToTerminated :: TestTransport -> Assertion +testSendToTerminated TestTransport{..} = do + serverAddr1 <- newEmptyMVar + serverAddr2 <- newEmptyMVar + clientDone <- newEmptyMVar + + forkIO $ do + terminated <- newEmptyMVar + localNode <- newLocalNode testTransport initRemoteTable + addr1 <- forkProcess localNode $ liftIO $ putMVar terminated () + addr2 <- forkProcess localNode $ ping + readMVar terminated + putMVar serverAddr1 addr1 + putMVar serverAddr2 addr2 + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + server1 <- readMVar serverAddr1 + server2 <- readMVar serverAddr2 + runProcess localNode $ do + pid <- getSelfPid + send server1 "Hi" + send server2 (Pong pid) + Ping pid' <- expect + True <- return $ pid' == server2 + liftIO $ putMVar clientDone () + + takeMVar clientDone + +-- | Test (non-zero) timeout +testTimeout :: TestTransport -> Assertion +testTimeout TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + runProcess localNode $ do + Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())] + liftIO $ putMVar done () + + takeMVar done + +-- | Test zero timeout +testTimeout0 :: TestTransport -> Assertion +testTimeout0 TestTransport{..} = do + serverAddr <- newEmptyMVar + clientDone <- newEmptyMVar + messagesSent <- newEmptyMVar + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + addr <- forkProcess localNode $ do + liftIO $ readMVar messagesSent >> threadDelay 1000000 + -- Variation on the venerable ping server which uses a zero timeout + -- Since we wait for all messages to be sent before doing this receive, + -- we should nevertheless find the right message immediately + Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)] + self <- getSelfPid + send partner (Ping self) + putMVar serverAddr addr + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + server <- readMVar serverAddr + runProcess localNode $ do + pid <- getSelfPid + -- Send a bunch of messages. A large number of messages that the server + -- is not interested in, and then a single message that it wants + replicateM_ 10000 $ send server "Irrelevant message" + send server (Pong pid) + liftIO $ putMVar messagesSent () + Ping _ <- expect + liftIO $ putMVar clientDone () + + takeMVar clientDone + +-- | Test typed channels +testTypedChannels :: TestTransport -> Assertion +testTypedChannels TestTransport{..} = do + serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int))) + clientDone <- newEmptyMVar + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + forkProcess localNode $ do + (serverSendPort, rport) <- newChan + liftIO $ putMVar serverChannel serverSendPort + (clientSendPort, i) <- receiveChan rport + sendChan clientSendPort (even i) + return () + + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + serverSendPort <- readMVar serverChannel + runProcess localNode $ do + (clientSendPort, rport) <- newChan + sendChan serverSendPort (clientSendPort, 5) + False <- receiveChan rport + liftIO $ putMVar clientDone () + + takeMVar clientDone + +-- | Test merging receive ports +testMergeChannels :: TestTransport -> Assertion +testMergeChannels TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + testFlat localNode True "aaabbbccc" + testFlat localNode False "abcabcabc" + testNested localNode True True "aaabbbcccdddeeefffggghhhiii" + testNested localNode True False "adgadgadgbehbehbehcficficfi" + testNested localNode False True "abcabcabcdefdefdefghighighi" + testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi" + testBlocked localNode True + testBlocked localNode False + where + -- Single layer of merging + testFlat :: LocalNode -> Bool -> String -> IO () + testFlat localNode biased expected = do + done <- newEmptyMVar + forkProcess localNode $ do + rs <- mapM charChannel "abc" + m <- mergePorts biased rs + xs <- replicateM 9 $ receiveChan m + True <- return $ xs == expected + liftIO $ putMVar done () + takeMVar done + + -- Two layers of merging + testNested :: LocalNode -> Bool -> Bool -> String -> IO () + testNested localNode biasedInner biasedOuter expected = do + done <- newEmptyMVar + forkProcess localNode $ do + rss <- mapM (mapM charChannel) ["abc", "def", "ghi"] + ms <- mapM (mergePorts biasedInner) rss + m <- mergePorts biasedOuter ms + xs <- replicateM (9 * 3) $ receiveChan m + True <- return $ xs == expected + liftIO $ putMVar done () + takeMVar done + + -- Test that if no messages are (immediately) available, the scheduler makes no difference + testBlocked :: LocalNode -> Bool -> IO () + testBlocked localNode biased = do + vs <- replicateM 3 newEmptyMVar + done <- newEmptyMVar + + forkProcess localNode $ do + [sa, sb, sc] <- liftIO $ mapM readMVar vs + mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan) + [ -- a, b, c + (sa, 'a') + , (sb, 'b') + , (sc, 'c') + -- a, c, b + , (sa, 'a') + , (sc, 'c') + , (sb, 'b') + -- b, a, c + , (sb, 'b') + , (sa, 'a') + , (sc, 'c') + -- b, c, a + , (sb, 'b') + , (sc, 'c') + , (sa, 'a') + -- c, a, b + , (sc, 'c') + , (sa, 'a') + , (sb, 'b') + -- c, b, a + , (sc, 'c') + , (sb, 'b') + , (sa, 'a') + ] + + forkProcess localNode $ do + (ss, rs) <- unzip <$> replicateM 3 newChan + liftIO $ mapM_ (uncurry putMVar) $ zip vs ss + m <- mergePorts biased rs + xs <- replicateM (6 * 3) $ receiveChan m + True <- return $ xs == "abcacbbacbcacabcba" + liftIO $ putMVar done () + + takeMVar done + + mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a) + mergePorts True = mergePortsBiased + mergePorts False = mergePortsRR + + charChannel :: Char -> Process (ReceivePort Char) + charChannel c = do + (sport, rport) <- newChan + replicateM_ 3 $ sendChan sport c + liftIO $ threadDelay 10000 -- Make sure messages have been sent + return rport + +testTerminate :: TestTransport -> Assertion +testTerminate TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + pid <- forkProcess localNode $ do + liftIO $ threadDelay 100000 + terminate + + runProcess localNode $ do + ref <- monitor pid + ProcessMonitorNotification ref' pid' (DiedException ex) <- expect + True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException + liftIO $ putMVar done () + + takeMVar done + +testMonitorNode :: TestTransport -> Assertion +testMonitorNode TestTransport{..} = do + [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + closeLocalNode node1 + + runProcess node2 $ do + ref <- monitorNode (localNodeId node1) + NodeMonitorNotification ref' nid DiedDisconnect <- expect + True <- return $ ref == ref' && nid == localNodeId node1 + liftIO $ putMVar done () + + takeMVar done + +testMonitorLiveNode :: TestTransport -> Assertion +testMonitorLiveNode TestTransport{..} = do + [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable + ready <- newEmptyMVar + done <- newEmptyMVar + + forkProcess node2 $ do + ref <- monitorNode (localNodeId node1) + liftIO $ putMVar ready () + NodeMonitorNotification ref' nid _ <- expect + True <- return $ ref == ref' && nid == localNodeId node1 + liftIO $ putMVar done () + + takeMVar ready + closeLocalNode node1 + + takeMVar done + +testMonitorChannel :: TestTransport -> Assertion +testMonitorChannel TestTransport{..} = do + [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable + gotNotification <- newEmptyMVar + + pid <- forkProcess node1 $ do + sport <- expect :: Process (SendPort ()) + ref <- monitorPort 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 + (sport, _) <- newChan :: Process (SendPort (), ReceivePort ()) + send pid sport + liftIO $ threadDelay 100000 + + takeMVar gotNotification + +testRegistry :: TestTransport -> Assertion +testRegistry TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + pingServer <- forkProcess node ping + + runProcess node $ do + register "ping" pingServer + Just pid <- whereis "ping" + True <- return $ pingServer == pid + us <- getSelfPid + nsend "ping" (Pong us) + Ping pid' <- expect + True <- return $ pingServer == pid' + liftIO $ putMVar done () + + takeMVar done + +testRemoteRegistry :: TestTransport -> Assertion +testRemoteRegistry TestTransport{..} = do + node1 <- newLocalNode testTransport initRemoteTable + node2 <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + pingServer <- forkProcess node1 ping + + runProcess node2 $ do + let nid1 = localNodeId node1 + registerRemoteAsync nid1 "ping" pingServer + receiveWait [ + matchIf (\(RegisterReply label' _) -> "ping" == label') + (\(RegisterReply _ _) -> return ()) ] + + Just pid <- whereisRemote nid1 "ping" + True <- return $ pingServer == pid + us <- getSelfPid + nsendRemote nid1 "ping" (Pong us) + Ping pid' <- expect + True <- return $ pingServer == pid' + liftIO $ putMVar done () + + takeMVar done + +testSpawnLocal :: TestTransport -> Assertion +testSpawnLocal TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + runProcess node $ do + us <- getSelfPid + + pid <- spawnLocal $ do + sport <- expect + sendChan sport (1234 :: Int) + + sport <- spawnChannelLocal $ \rport -> do + (1234 :: Int) <- receiveChan rport + send us () + + send pid sport + () <- expect + liftIO $ putMVar done () + + takeMVar done + +testSpawnLocalET :: TestTransport -> Assertion +testSpawnLocalET TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + runProcess node $ do + Left "ono" <- EX.runExceptT $ do + us <- getSelfPid + + pid <- spawnLocal $ do + sport <- expect + sendChan sport (1234 :: Int) + + sport <- spawnChannelLocal $ \rport -> do + (1234 :: Int) <- receiveChan rport + send us () + + EX.throwE "ono" + + send pid sport + () <- expect + liftIO $ putMVar done () + return () + +testSpawnLocalStS :: TestTransport -> Assertion +testSpawnLocalStS TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + runProcess node $ do + flip StS.evalStateT (1234 :: Int) $ do + StS.put 5678 + us <- getSelfPid + + pid <- spawnLocal $ do + sport <- expect + StL.get >>= sendChan sport + + sport <- spawnChannelLocal $ \rport -> do + (5678 :: Int) <- receiveChan rport + send us () + + send pid sport + () <- expect + liftIO $ putMVar done () + +testSpawnLocalStL :: TestTransport -> Assertion +testSpawnLocalStL TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + runProcess node $ do + flip StL.evalStateT (1234 :: Int) $ do + StL.put 5678 + us <- getSelfPid + + pid <- spawnLocal $ do + sport <- expect + StL.get >>= sendChan sport + + sport <- spawnChannelLocal $ \rport -> do + (5678 :: Int) <- receiveChan rport + send us () + + send pid sport + () <- expect + liftIO $ putMVar done () + +testSpawnLocalRT :: TestTransport -> Assertion +testSpawnLocalRT TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + runProcess node $ do + flip R.runReaderT (1234 :: Int) $ do + us <- getSelfPid + + pid <- spawnLocal $ do + sport <- expect + R.ask >>= sendChan sport + + sport <- spawnChannelLocal $ \rport -> do + (1234 :: Int) <- receiveChan rport + send us () + + send pid sport + () <- expect + liftIO $ putMVar done () + + takeMVar done + +testReconnect :: TestTransport -> Assertion +testReconnect TestTransport{..} = do + [node1, node2] <- replicateM 2 $ newLocalNode testTransport 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 $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) + + -- 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 + registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous + receiveWait [ + matchIf (\(RegisterReply label' _) -> "a" == label') + (\(RegisterReply _ _) -> return ()) ] + + Just _ <- whereisRemote nid1 "a" + + + -- Simulate network failure + liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) + + -- This will happen due to implicit reconnect + registerRemoteAsync nid1 "b" us + receiveWait [ + matchIf (\(RegisterReply label' _) -> "b" == label') + (\(RegisterReply _ _) -> return ()) ] + + -- Should happen + registerRemoteAsync nid1 "c" us + receiveWait [ + matchIf (\(RegisterReply label' _) -> "c" == label') + (\(RegisterReply _ _) -> return ()) ] + + -- Check + Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected + Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect + Just _ <- whereisRemote nid1 "c" + + liftIO $ putMVar registerTestOk () + + takeMVar registerTestOk + +-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server +-- in between +testMatchAny :: TestTransport -> Assertion +testMatchAny TestTransport{..} = do + proxyAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- Math server + forkIO $ do + localNode <- newLocalNode testTransport 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 testTransport 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 + +-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server +-- in between, however we block 'Divide' requests .... +testMatchAnyHandle :: TestTransport -> Assertion +testMatchAnyHandle TestTransport{..} = do + proxyAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- Math server + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + mathServer <- forkProcess localNode math + proxyServer <- forkProcess localNode $ forever $ do + receiveWait [ + matchAny (maybeForward mathServer) + ] + putMVar proxyAddr proxyServer + + -- Client + forkIO $ do + localNode <- newLocalNode testTransport 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) + Nothing <- (expectTimeout 100000) :: Process (Maybe Double) + liftIO $ putMVar clientDone () + + takeMVar clientDone + where maybeForward :: ProcessId -> Message -> Process (Maybe ()) + maybeForward s msg = + handleMessage msg (\m@(Add _ _ _) -> send s m) + +testMatchAnyNoHandle :: TestTransport -> Assertion +testMatchAnyNoHandle TestTransport{..} = do + addr <- newEmptyMVar + clientDone <- newEmptyMVar + serverDone <- newEmptyMVar + + -- Math server + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + server <- forkProcess localNode $ forever $ do + receiveWait [ + matchAnyIf + -- the condition has type `Add -> Bool` + (\(Add _ _ _) -> True) + -- the match `AbstractMessage -> Process ()` will succeed! + (\m -> do + -- `String -> Process ()` does *not* match the input types however + r <- (handleMessage m (\(_ :: String) -> die "NONSENSE" )) + case r of + Nothing -> return () + Just _ -> die "NONSENSE") + ] + -- we *must* have removed the message from our mailbox though!!! + Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ] + liftIO $ putMVar serverDone () + putMVar addr server + + -- Client + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + server <- readMVar addr + + runProcess localNode $ do + pid <- getSelfPid + send server (Add pid 1 2) + -- we only care about the client having sent a message, so we're done + liftIO $ putMVar clientDone () + + takeMVar clientDone + takeMVar serverDone + +-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests +-- unless the text body @/= "bar"@ - this case should time out rather than +-- removing the message from the process mailbox. +testMatchAnyIf :: TestTransport -> Assertion +testMatchAnyIf TestTransport{..} = do + echoAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- echo server + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + echoServer <- forkProcess localNode $ forever $ do + receiveWait [ + matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar") + tryHandleMessage + ] + putMVar echoAddr echoServer + + -- Client + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + server <- readMVar echoAddr + + runProcess localNode $ do + pid <- getSelfPid + send server (pid, "foo") + "foo" <- expect + send server (pid, "baz") + "baz" <- expect + send server (pid, "bar") + Nothing <- (expectTimeout 100000) :: Process (Maybe Double) + liftIO $ putMVar clientDone () + + takeMVar clientDone + where tryHandleMessage :: Message -> Process (Maybe ()) + tryHandleMessage msg = + handleMessage msg (\(pid :: ProcessId, (m :: String)) + -> do { send pid m; return () }) + +testMatchMessageWithUnwrap :: TestTransport -> Assertion +testMatchMessageWithUnwrap TestTransport{..} = do + echoAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- echo server + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + echoServer <- forkProcess localNode $ forever $ do + msg <- receiveWait [ + matchMessage (\(m :: Message) -> do + return m) + ] + unwrapped <- unwrapMessage msg :: Process (Maybe (ProcessId, Message)) + case unwrapped of + (Just (p, msg')) -> forward msg' p + Nothing -> die "unable to unwrap the message" + putMVar echoAddr echoServer + + -- Client + forkIO $ do + localNode <- newLocalNode testTransport initRemoteTable + server <- readMVar echoAddr + + runProcess localNode $ do + pid <- getSelfPid + send server (pid, wrapMessage ("foo" :: String)) + "foo" <- expect + send server (pid, wrapMessage ("baz" :: String)) + "baz" <- expect + liftIO $ putMVar clientDone () + + takeMVar clientDone + +-- Test 'receiveChanTimeout' +testReceiveChanTimeout :: TestTransport -> Assertion +testReceiveChanTimeout TestTransport{..} = do + done <- newEmptyMVar + sendPort <- newEmptyMVar + + forkTry $ do + localNode <- newLocalNode testTransport initRemoteTable + runProcess localNode $ do + -- Create a typed channel + (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool) + liftIO $ putMVar sendPort sp + + -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second + Nothing <- receiveChanTimeout 1000000 rp + + -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds + Just True <- receiveChanTimeout 1000000 rp + + -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing + Nothing <- receiveChanTimeout 0 rp + + -- Again, but now there is a message available + liftIO $ threadDelay 1000000 + Just False <- receiveChanTimeout 0 rp + + liftIO $ putMVar done () + + forkTry $ do + localNode <- newLocalNode testTransport initRemoteTable + runProcess localNode $ do + sp <- liftIO $ readMVar sendPort + + liftIO $ threadDelay 1500000 + sendChan sp True + + liftIO $ threadDelay 500000 + sendChan sp False + + takeMVar done + +-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan +testReceiveChanFeatures :: TestTransport -> Assertion +testReceiveChanFeatures TestTransport{..} = do + done <- newEmptyMVar + + forkTry $ do + localNode <- newLocalNode testTransport initRemoteTable + runProcess localNode $ do + (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int) + (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool) + + -- Test Functor instance + + sendChan spInt 2 + sendChan spBool False + + rp1 <- mergePortsBiased [even <$> rpInt, rpBool] + + True <- receiveChan rp1 + False <- receiveChan rp1 + + -- Test Applicative instance + + sendChan spInt 3 + sendChan spInt 4 + + let rp2 = pure (+) <*> rpInt <*> rpInt + + 7 <- receiveChan rp2 + + -- Test Alternative instance + + sendChan spInt 3 + sendChan spBool True + + let rp3 = (even <$> rpInt) <|> rpBool + + False <- receiveChan rp3 + True <- receiveChan rp3 + + -- Test Monad instance + + sendChan spBool True + sendChan spBool False + sendChan spInt 5 + + let rp4 :: ReceivePort Int + rp4 = do b <- rpBool + if b + then rpInt + else return 7 + + 5 <- receiveChan rp4 + 7 <- receiveChan rp4 + + liftIO $ putMVar done () + + takeMVar done + +testKillLocal :: TestTransport -> Assertion +testKillLocal TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + pid <- forkProcess localNode $ do + liftIO $ threadDelay 1000000 + + runProcess localNode $ do + ref <- monitor pid + us <- getSelfPid + kill pid "TestKill" + ProcessMonitorNotification ref' pid' (DiedException ex) <- expect + True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill" + liftIO $ putMVar done () + + takeMVar done + +testKillRemote :: TestTransport -> Assertion +testKillRemote TestTransport{..} = do + node1 <- newLocalNode testTransport initRemoteTable + node2 <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + pid <- forkProcess node1 $ do + liftIO $ threadDelay 1000000 + + runProcess node2 $ do + ref <- monitor pid + us <- getSelfPid + kill pid "TestKill" + ProcessMonitorNotification ref' pid' (DiedException reason) <- expect + True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill" + liftIO $ putMVar done () + + takeMVar done + +testCatchesExit :: TestTransport -> Assertion +testCatchesExit TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + _ <- forkProcess localNode $ do + (die ("foobar", 123 :: Int)) + `catchesExit` [ + (\_ m -> handleMessage m (\(_ :: String) -> return ())) + , (\_ m -> handleMessage m (\(_ :: Maybe Int) -> return ())) + , (\_ m -> handleMessage m (\(_ :: String, _ :: Int) + -> (liftIO $ putMVar done ()) >> return ())) + ] + + takeMVar done + +testCatchesExitSTL :: TestTransport -> Assertion +testCatchesExitSTL TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + _ <- forkProcess localNode $ do + flip StL.evalStateT ("foobar", 123 :: Int) $ do + StL.get >>= die + `catchesExit` [ + (\_ m -> handleMessage m (\(_ :: String) -> return ())) + , (\_ m -> handleMessage m (\(_ :: Maybe Int) -> return ())) + , (\_ m -> handleMessage m (\(_ :: String, _ :: Int) + -> (liftIO $ putMVar done ()) >> return ())) + ] + + takeMVar done + +testHandleMessageIf :: TestTransport -> Assertion +testHandleMessageIf TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + _ <- forkProcess localNode $ do + self <- getSelfPid + send self (5 :: Integer, 10 :: Integer) + msg <- receiveWait [ matchMessage return ] + Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops") + handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10) + (\input -> liftIO $ putMVar done input) + return () + + result <- takeMVar done + expectThat result $ equalTo (5, 10) + +testCatches :: TestTransport -> Assertion +testCatches TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + _ <- forkProcess localNode $ do + node <- getSelfNode + (liftIO $ throwIO (ProcessLinkException (nullProcessId node) DiedNormal)) + `catches` [ + Handler (\(ProcessLinkException _ _) -> liftIO $ putMVar done ()) + ] + + takeMVar done + +testMaskRestoreScope :: TestTransport -> Assertion +testMaskRestoreScope TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + parentPid <- newEmptyMVar :: IO (MVar ProcessId) + spawnedPid <- newEmptyMVar :: IO (MVar ProcessId) + + void $ runProcess localNode $ mask $ \unmask -> do + getSelfPid >>= liftIO . putMVar parentPid + void $ spawnLocal $ unmask (getSelfPid >>= liftIO . putMVar spawnedPid) + + parent <- liftIO $ takeMVar parentPid + child <- liftIO $ takeMVar spawnedPid + expectThat parent $ isNot $ equalTo child + +testDieRWST :: TestTransport -> Assertion +testDieRWST TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + _ <- forkProcess localNode $ do + ((),"hear this") <- flip (`RWST.evalRWST` (123 :: Int)) "bazqux" $ do + RWST.put "foobar" + s <- RWST.get + n <- RWST.ask + RWST.tell "hear this" + (die (s, n)) + `catchExit` \_from reason -> do + True <- return $ reason == ("foobar", 123 :: Int) + liftIO $ putMVar done () + return () + + takeMVar done + +testDie :: TestTransport -> Assertion +testDie TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + _ <- forkProcess localNode $ do + (die ("foobar", 123 :: Int)) + `catchExit` \_from reason -> do + -- TODO: should verify that 'from' has the right value + True <- return $ reason == ("foobar", 123 :: Int) + liftIO $ putMVar done () + + takeMVar done + +testPrettyExit :: TestTransport -> Assertion +testPrettyExit TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + _ <- forkProcess localNode $ do + (die "timeout") + `catch` \ex@(ProcessExitException from _) -> + let expected = "exit-from=" ++ (show from) + in do + True <- return $ (show ex) == expected + liftIO $ putMVar done () + + takeMVar done + +testExitLocal :: TestTransport -> Assertion +testExitLocal TestTransport{..} = do + localNode <- newLocalNode testTransport initRemoteTable + supervisedDone <- newEmptyMVar + supervisorDone <- newEmptyMVar + + pid <- forkProcess localNode $ do + (liftIO $ threadDelay 100000) + `catchExit` \_from reason -> do + -- TODO: should verify that 'from' has the right value + True <- return $ reason == "TestExit" + liftIO $ putMVar supervisedDone () + + runProcess localNode $ do + ref <- monitor pid + exit pid "TestExit" + -- This time the client catches the exception, so it dies normally + ProcessMonitorNotification ref' pid' DiedNormal <- expect + True <- return $ ref == ref' && pid == pid' + liftIO $ putMVar supervisorDone () + + takeMVar supervisedDone + takeMVar supervisorDone + +testExitRemote :: TestTransport -> Assertion +testExitRemote TestTransport{..} = do + node1 <- newLocalNode testTransport initRemoteTable + node2 <- newLocalNode testTransport initRemoteTable + supervisedDone <- newEmptyMVar + supervisorDone <- newEmptyMVar + + pid <- forkProcess node1 $ do + (liftIO $ threadDelay 100000) + `catchExit` \_from reason -> do + -- TODO: should verify that 'from' has the right value + True <- return $ reason == "TestExit" + liftIO $ putMVar supervisedDone () + + runProcess node2 $ do + ref <- monitor pid + exit pid "TestExit" + ProcessMonitorNotification ref' pid' DiedNormal <- expect + True <- return $ ref == ref' && pid == pid' + liftIO $ putMVar supervisorDone () + + takeMVar supervisedDone + takeMVar supervisorDone + +testUnsafeSend :: TestTransport -> Assertion +testUnsafeSend TestTransport{..} = do + serverAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + localNode <- newLocalNode testTransport initRemoteTable + void $ forkProcess localNode $ do + self <- getSelfPid + liftIO $ putMVar serverAddr self + clientAddr <- expect + unsafeSend clientAddr () + + void $ forkProcess localNode $ do + serverPid <- liftIO $ takeMVar serverAddr + getSelfPid >>= unsafeSend serverPid + () <- expect + liftIO $ putMVar clientDone () + + takeMVar clientDone + +testUnsafeNSend :: TestTransport -> Assertion +testUnsafeNSend TestTransport{..} = do + clientDone <- newEmptyMVar + + localNode <- newLocalNode testTransport initRemoteTable + + pid <- forkProcess localNode $ do + () <- expect + liftIO $ putMVar clientDone () + + void $ runProcess localNode $ do + register "foobar" pid + unsafeNSend "foobar" () + + takeMVar clientDone + +testUnsafeSendChan :: TestTransport -> Assertion +testUnsafeSendChan TestTransport{..} = do + serverAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + localNode <- newLocalNode testTransport initRemoteTable + void $ forkProcess localNode $ do + self <- getSelfPid + liftIO $ putMVar serverAddr self + sp <- expect + unsafeSendChan sp () + + void $ forkProcess localNode $ do + serverPid <- liftIO $ takeMVar serverAddr + (sp, rp) <- newChan + unsafeSend serverPid sp + () <- receiveChan rp + liftIO $ putMVar clientDone () + + takeMVar clientDone + +testFromProcess :: TestTransport -> Assertion +testFromProcess TestTransport{..} = do + + localNode <- newLocalNode testTransport initRemoteTable + + "this" <- fromProcess localNode $ return "this" + return () + +testFromProcessException :: TestTransport -> Assertion +testFromProcessException TestTransport{..} = do + done <- newEmptyMVar + + localNode <- newLocalNode testTransport initRemoteTable + + "this" <- fromProcess localNode $ throwIO (userError "fails") + `catch` \e -> + do ignore e + putMVar done () + return "this" + + takeMVar done + +ignore :: Monad m => SomeException -> m () +ignore = return . const () + +tests :: TestTransport -> IO [Test] +tests testtrans = return [ + testGroup "Transformer variations" [ + testCase "SpawnLocalReaderT" (testSpawnLocalRT testtrans) + , testCase "SpawnLocalStateTLazy" (testSpawnLocalStL testtrans) + , testCase "SpawnLocalStateTStrict" (testSpawnLocalStS testtrans) + , testCase "SpawnLocalExceptT" (testSpawnLocalET testtrans) + , testCase "CatchesExitStateTLazy" (testCatchesExitSTL testtrans) + , testCase "TestDieRWST" (testDieRWST testtrans) + , testCase "fromProcessReturnsValue" (testFromProcess testtrans) + , testCase "fromProcessRelaysException" (testFromProcessException testtrans) + ] +#ifndef ONLY_TEST_TRANSFORMERS + , testGroup "Regression Basic Features" [ + testCase "Ping" (testPing testtrans) + , testCase "Math" (testMath testtrans) + , testCase "Timeout" (testTimeout testtrans) + , testCase "Timeout0" (testTimeout0 testtrans) + , testCase "SendToTerminated" (testSendToTerminated testtrans) + , testCase "TypedChannnels" (testTypedChannels testtrans) + , testCase "MergeChannels" (testMergeChannels testtrans) + , testCase "Terminate" (testTerminate testtrans) + , testCase "Registry" (testRegistry testtrans) + , testCase "RemoteRegistry" (testRemoteRegistry testtrans) + , testCase "SpawnLocal" (testSpawnLocal testtrans) + , testCase "HandleMessageIf" (testHandleMessageIf testtrans) + , testCase "MatchAny" (testMatchAny testtrans) + , testCase "MatchAnyHandle" (testMatchAnyHandle testtrans) + , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle testtrans) + , testCase "MatchAnyIf" (testMatchAnyIf testtrans) + , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans) + , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans) + , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans) + , testCase "KillLocal" (testKillLocal testtrans) + , testCase "KillRemote" (testKillRemote testtrans) + , testCase "Die" (testDie testtrans) + , testCase "PrettyExit" (testPrettyExit testtrans) + , testCase "CatchesExit" (testCatchesExit testtrans) + , testCase "Catches" (testCatches testtrans) + , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans) + , testCase "ExitLocal" (testExitLocal testtrans) + , testCase "ExitRemote" (testExitRemote testtrans) + -- Unsafe Primitives + , testCase "TestUnsafeSend" (testUnsafeSend testtrans) + , testCase "TestUnsafeNSend" (testUnsafeNSend testtrans) + , testCase "TestUnsafeSendChan" (testUnsafeSendChan testtrans) + ] + , testGroup "Regression Monitoring and Linking" [ + -- Monitoring processes + -- + -- The "missing" combinations in the list below don't make much sense, as + -- we cannot guarantee that the monitor reply or link exception will not + -- happen before the unmonitor or unlink + testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False) + , testCase "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False) + , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True False) + , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess testtrans True False) + , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans True False) + , testCase "MonitorDisconnect" (testMonitorDisconnect testtrans True False) + , testCase "LinkUnreachable" (testMonitorUnreachable testtrans False False) + , testCase "LinkNormalTermination" (testMonitorNormalTermination testtrans False False) + , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False False) + , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess testtrans False False) + , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans False False) + , testCase "LinkDisconnect" (testMonitorDisconnect testtrans False False) + , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination testtrans True True) + , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True True) + , testCase "UnmonitorDisconnect" (testMonitorDisconnect testtrans True True) + , testCase "UnlinkNormalTermination" (testMonitorNormalTermination testtrans False True) + , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False True) + , testCase "UnlinkDisconnect" (testMonitorDisconnect testtrans False True) + -- Monitoring nodes and channels + , testCase "MonitorNode" (testMonitorNode testtrans) + , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans) + , testCase "MonitorChannel" (testMonitorChannel testtrans) + -- Reconnect + , testCase "Reconnect" (testReconnect testtrans) + ] +#endif + ] diff --git a/test/Network/Transport/Test.hs b/test/Network/Transport/Test.hs new file mode 100644 --- /dev/null +++ b/test/Network/Transport/Test.hs @@ -0,0 +1,12 @@ +module Network.Transport.Test where + +import qualified Network.Transport as NT +import Data.Typeable (Typeable) + +-- | Extra operations required of transports for the purposes of testing. +data TestTransport = TestTransport + { -- | The transport to use for testing. + testTransport :: NT.Transport + -- | IO action to perform to simulate losing a connection. + , testBreakConnection :: NT.EndPointAddress -> NT.EndPointAddress -> IO () + } deriving (Typeable) diff --git a/test/runTCP.hs b/test/runTCP.hs new file mode 100644 --- /dev/null +++ b/test/runTCP.hs @@ -0,0 +1,40 @@ +-- Run tests using the TCP transport. + +module Main where + +import Control.Distributed.Process.Lifted.Tests (tests) + +import Network.Transport.Test (TestTransport(..)) +import Network.Socket (sClose) +import Network.Transport.TCP + ( createTransportExposeInternals + , TransportInternals(socketBetween) + , defaultTCPParameters + ) +import Test.Framework (defaultMainWithArgs) + +import Control.Concurrent (threadDelay) +import System.Environment (getArgs) + +main :: IO () +main = do + Right (transport, internals) <- + createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters + ts <- tests TestTransport + { testTransport = transport + , testBreakConnection = \addr1 addr2 -> do + sock <- socketBetween internals addr1 addr2 + sClose sock + threadDelay 10000 + } + args <- getArgs + -- Tests are time sensitive. Running the tests concurrently can slow them + -- down enough that threads using threadDelay would wake up later than + -- expected, thus changing the order in which messages were expected. + -- Therefore we run the tests sequentially by passing "-j 1" to + -- test-framework. This does not solve the issue but makes it less likely. + -- + -- The problem was first detected with + -- 'Control.Distributed.Process.Tests.CH.testMergeChannels' + -- in particular. + defaultMainWithArgs ts ("-j" : "1" : args)