packages feed

distributed-process-lifted 0.1.0.0 → 0.2.0.0

raw patch · 5 files changed

+116/−32 lines, 5 filesdep ~transformers

Dependency ranges changed: transformers

Files

distributed-process-lifted.cabal view
@@ -2,7 +2,7 @@ -- further documentation, see http://haskell.org/cabal/users-guide/  name:                distributed-process-lifted-version:             0.1.0.0+version:             0.2.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 @@ -33,6 +33,7 @@  library   exposed-modules:     Control.Distributed.Process.Lifted+                     , Control.Distributed.Process.Lifted.Extras                      , Control.Distributed.Process.Lifted.Class                      , Control.Distributed.Process.Node.Lifted   -- other-modules:       @@ -42,7 +43,7 @@                      , 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 >= 0.4 && < 0.5                      , transformers-base >= 0.4.1 && <= 0.5.0                       , monad-control >= 0.3 && < 1.1                      , mtl >= 2.0 && < 2.3
src/Control/Distributed/Process/Lifted/Class.hs view
@@ -25,9 +25,7 @@ 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)@@ -98,9 +96,7 @@ 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) \@@ -129,9 +125,7 @@ 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)
+ src/Control/Distributed/Process/Lifted/Extras.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveDataTypeable #-}+++-- | Utility functions for working with Processes outside of the+-- 'Process' monad.+module Control.Distributed.Process.Lifted.Extras+    ( fromProcess+    , ProcessProxy+    , proxyPid+    , spawnProxy+    , spawnProxyIO+    , inProxy+    , fromProxy+    )+where++import Control.Monad (void, join, forever)+import Control.Monad.Base (MonadBase(..))+import Control.Exception (throw, SomeException)+import Data.Typeable (Typeable)++import Control.Concurrent.Chan.Lifted+       (Chan, newChan, writeChan, readChan)+import Control.Concurrent.MVar.Lifted+       (newEmptyMVar, putMVar, takeMVar)++import Control.Distributed.Process.Lifted hiding (newChan)+import Control.Distributed.Process.Node.Lifted (LocalNode, forkProcess)+++-- | A variant of 'Control.Distributed.Process.Node.runProcess' which returns a value. This works just+-- like 'Control.Distributed.Process.Node.runProcess' by forking a new process with a captured 'Control.Concurrent.MVar.MVar', but it+-- will return 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. (MonadBase IO m)+            => LocalNode -> Process a -> m a+fromProcess node ma =+ do resultMV <- newEmptyMVar+    void . forkProcess node $+     do eresult <- try (do !a <- ma; 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++-- | Represents a handle to a process runner that communicates+-- through a 'Control.Concurrent.Chan.Chan'.+-- Create with 'spawnProxy' or 'spawnProxyIO'.+-- Use this to call process actions (using 'fromProxy' or 'inProxy') from any IO+-- that will be executed in a single process that will have a+-- persistent pid and mailbox across invocations.+-- Sharing a single proxy between threads may yield poor performance and is not advised.+data ProcessProxy = ProcessProxy {+     proxyPid  :: !ProcessId,+     proxyChan :: !(Chan (Process ()))+} deriving (Typeable)++instance Show ProcessProxy where+    show = show . proxyPid++-- | Spawn a new process and return a 'ProcessProxy' handle for it.+spawnProxy :: Process ProcessProxy+spawnProxy =+ do action <- newChan+    pid <- spawnLocal . forever $+        join (readChan action)+    return (ProcessProxy pid action)++-- | Same as spawnProxy but can be used from any IO+--+-- spawnProxyIO node = fromProcess node spawnProxy+spawnProxyIO :: forall m. (MonadBase IO m)+             => LocalNode -> m ProcessProxy+spawnProxyIO node = fromProcess node spawnProxy++-- | Use a 'ProcessProxy' created with 'spawnProxy' to run a+-- Process computation in the existing Process asynchronously.+inProxy :: forall m. (MonadBase IO m)+          => ProcessProxy -> Process () -> m ()+inProxy = writeChan . proxyChan++-- | Use a 'ProcessProxy' created with 'spawnProxy' to run a+-- Process computation in the existing Process and return the result+-- in any IO.+fromProxy :: forall a m. (MonadBase IO m)+          => ProcessProxy -> Process a -> m a+fromProxy (ProcessProxy _ prox) ma =+ do result <- newEmptyMVar+    writeChan prox (ma >>= putMVar result)+    takeMVar result
src/Control/Distributed/Process/Node/Lifted.hs view
@@ -1,6 +1,4 @@ {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ScopedTypeVariables #-}  module Control.Distributed.Process.Node.Lifted    ( module Control.Distributed.Process.Node.Lifted@@ -11,18 +9,12 @@   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'.@@ -40,19 +32,3 @@ -- | 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
@@ -29,6 +29,7 @@ import Control.Applicative ((<$>), (<*>), pure, (<|>)) import qualified Network.Transport as NT (closeEndPoint) import Control.Distributed.Process.Lifted+import Control.Distributed.Process.Lifted.Extras import Control.Distributed.Process.Internal.Types   (     LocalNode(localEndPoint)@@ -765,6 +766,23 @@    takeMVar done +testProcessProxy :: TestTransport -> Assertion+testProcessProxy TestTransport{..} = do+  node <- newLocalNode testTransport initRemoteTable++  other <- forkProcess node $ do+    pid <- expect+    send pid "hello"++  prox <- spawnProxyIO node+  inProxy prox $ getSelfPid >>= send other+  "hello" <- fromProxy prox expect+  inProxy prox (die "all done")+  threadDelay 10000++  return ()++ testReconnect :: TestTransport -> Assertion testReconnect TestTransport{..} = do   [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable@@ -1419,6 +1437,7 @@       , testCase "SpawnLocalStateTLazy"   (testSpawnLocalStL   testtrans)       , testCase "SpawnLocalStateTStrict"   (testSpawnLocalStS   testtrans)       , testCase "SpawnLocalExceptT"   (testSpawnLocalET   testtrans)+      , testCase "TestProcessProxy"   (testProcessProxy testtrans)       , testCase "CatchesExitStateTLazy"   (testCatchesExitSTL testtrans)       , testCase "TestDieRWST"   (testDieRWST testtrans)       , testCase "fromProcessReturnsValue"   (testFromProcess testtrans)