distributed-process-lifted (empty) → 0.1.0.0
raw patch · 9 files changed
+2251/−0 lines, 9 filesdep +HUnitdep +basedep +binarysetup-changed
Dependencies added: HUnit, base, binary, deepseq, distributed-process, distributed-process-lifted, distributed-process-monad-control, lifted-base, monad-control, mtl, network, network-transport, network-transport-tcp, rematch, test-framework, test-framework-hunit, transformers, transformers-base
Files
- LICENSE +40/−0
- Setup.hs +2/−0
- distributed-process-lifted.cabal +76/−0
- src/Control/Distributed/Process/Lifted.hs +391/−0
- src/Control/Distributed/Process/Lifted/Class.hs +138/−0
- src/Control/Distributed/Process/Node/Lifted.hs +58/−0
- test/Control/Distributed/Process/Lifted/Tests.hs +1494/−0
- test/Network/Transport/Test.hs +12/−0
- test/runTCP.hs +40/−0
+ LICENSE view
@@ -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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ distributed-process-lifted.cabal view
@@ -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+ <http://hackage.haskell.org/package/monad-control-1.0.0.1/docs/Control-Monad-Trans-Control.html#t:MonadTransControl MonadTransControl> + and a new typeclass 'Control.Distributed.Process.Lifted.Class.MonadProcessBase' which plays the same role as + <http://hackage.haskell.org/package/monad-control-1.0.0.1/docs/Control-Monad-Trans-Control.html#t:MonadBaseControl MonadBaseControl>. + Instances are provided for all the <http://hackage.haskell.org/package/transformers transformers> 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
+ src/Control/Distributed/Process/Lifted.hs view
@@ -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
+ src/Control/Distributed/Process/Lifted/Class.hs view
@@ -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)
+ src/Control/Distributed/Process/Node/Lifted.hs view
@@ -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
+ test/Control/Distributed/Process/Lifted/Tests.hs view
@@ -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+ ]
+ test/Network/Transport/Test.hs view
@@ -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)
+ test/runTCP.hs view
@@ -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)