distributed-process-async (empty) → 0.1.0
raw patch · 6 files changed
+837/−0 lines, 6 filesdep +HUnitdep +ansi-terminaldep +basesetup-changed
Dependencies added: HUnit, ansi-terminal, base, binary, containers, data-accessor, deepseq, derive, distributed-process, distributed-process-async, distributed-process-extras, distributed-process-tests, fingertree, ghc-prim, hashable, mtl, network, network-transport, network-transport-tcp, rematch, stm, template-haskell, test-framework, test-framework-hunit, time, transformers, uniplate, unordered-containers
Files
- LICENCE +30/−0
- Setup.lhs +3/−0
- distributed-process-async.cabal +85/−0
- src/Control/Distributed/Process/Async.hs +371/−0
- src/Control/Distributed/Process/Async/Internal/Types.hs +80/−0
- tests/TestAsync.hs +268/−0
+ LICENCE view
@@ -0,0 +1,30 @@+Copyright Tim Watson, 2012-2013.++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 the author 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.
+ Setup.lhs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain
+ distributed-process-async.cabal view
@@ -0,0 +1,85 @@+name: distributed-process-async+version: 0.1.0+cabal-version: >=1.8+build-type: Simple+license: BSD3+license-file: LICENCE+stability: experimental+Copyright: Tim Watson 2012 - 2014+Author: Tim Watson+Maintainer: watson.timothy@gmail.com+Stability: experimental+Homepage: http://github.com/haskell-distributed/distributed-process-async+Bug-Reports: http://github.com/haskell-distributed/distributed-process-async/issues+synopsis: Cloud Haskell Async API +description: This package provides a higher-level interface over Processes, in which an Async a is a+ concurrent, possibly distributed Process that will eventually deliver a value of type a.+ The package provides ways to create Async computations, wait for their results, and cancel them.+category: Control+tested-with: GHC == 7.4.2 GHC == 7.6.2+data-dir: ""++source-repository head+ type: git+ location: https://github.com/haskell-distributed/distributed-process-async++flag perf+ description: Build with profiling enabled+ default: False++library+ build-depends:+ base >= 4.4 && < 5,+ data-accessor >= 0.2.2.3,+ distributed-process >= 0.5.2 && < 0.6,+ distributed-process-extras >= 0.1.0 && < 0.2,+ binary >= 0.6.3.0 && < 0.8,+ deepseq >= 1.3.0.1 && < 1.4,+ mtl,+ containers >= 0.4 && < 0.6,+ hashable >= 1.2.0.5 && < 1.3,+ unordered-containers >= 0.2.3.0 && < 0.3,+ fingertree < 0.2,+ stm >= 2.4 && < 2.5,+ time > 1.4 && < 1.5,+ transformers+ if impl(ghc <= 7.5) + Build-Depends: template-haskell == 2.7.0.0,+ derive == 2.5.5,+ uniplate == 1.6.12,+ ghc-prim+ extensions: CPP+ InstanceSigs+ hs-source-dirs: src+ ghc-options: -Wall+ exposed-modules:+ Control.Distributed.Process.Async+ other-modules:+ Control.Distributed.Process.Async.Internal.Types++test-suite AsyncTests+ type: exitcode-stdio-1.0+ x-uses-tf: true+ build-depends:+ base >= 4.4 && < 5,+ ansi-terminal >= 0.5 && < 0.6,+ distributed-process >= 0.5.2 && < 0.6,+ distributed-process-extras >= 0.1.0 && < 0.2,+ distributed-process-async,+ distributed-process-tests >= 0.4.1 && < 0.5,+ network >= 2.5 && < 2.7,+ network-transport >= 0.4 && < 0.5,+ network-transport-tcp >= 0.4 && < 0.5,+ binary >= 0.6.3.0 && < 0.8,+ deepseq >= 1.3.0.1 && < 1.4,+ HUnit >= 1.2 && < 2,+ stm >= 2.3 && < 2.5,+ test-framework >= 0.6 && < 0.9,+ test-framework-hunit,+ rematch >= 0.2.0.0,+ transformers+ hs-source-dirs:+ tests+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ extensions: CPP+ main-is: TestAsync.hs
+ src/Control/Distributed/Process/Async.hs view
@@ -0,0 +1,371 @@+-----------------------------------------------------------------------------+-- |+-- Module : Control.Distributed.Process.Async+-- Copyright : (c) Tim Watson 2012+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Tim Watson <watson.timothy@gmail.com>+-- Stability : experimental+-- Portability : non-portable (requires concurrency)+--+-- This API provides a means for spawning asynchronous operations, waiting+-- for their results, cancelling them and various other utilities.+-- Asynchronous operations can be executed on remote nodes.+--+-- [Asynchronous Operations]+--+-- There is an implicit contract for async workers; Workers must exit+-- normally (i.e., should not call the 'exit', 'die' or 'terminate'+-- Cloud Haskell primitives), otherwise the 'AsyncResult' will end up being+-- @AsyncFailed DiedException@ instead of containing the result.+--+-- Portions of this file are derived from the @Control.Concurrent.Async@+-- module, from the @async@ package written by Simon Marlow.+-----------------------------------------------------------------------------++module Control.Distributed.Process.Async+ ( -- * Exported types+ AsyncRef+ , AsyncTask(..)+ , Async+ , AsyncResult(..)+ -- * Spawning asynchronous operations+ , async+ , asyncLinked+ , task+ , remoteTask+ , monitorAsync+ -- * Cancelling asynchronous operations+ , cancel+ , cancelWait+ , cancelWith+ , cancelKill+ -- * Querying for results+ , poll+ , check+ , wait+ , waitAny+ -- * Waiting with timeouts+ , waitAnyTimeout+ , waitTimeout+ , waitCancelTimeout+ , waitCheckTimeout+ -- * STM versions+ , pollSTM+ , waitTimeoutSTM+ , waitAnyCancel+ , waitEither+ , waitEither_+ , waitBoth+ ) where++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif++import Control.Applicative+import Control.Concurrent.STM hiding (check)+import Control.Distributed.Process+import Control.Distributed.Process.Serializable+import Control.Distributed.Process.Async.Internal.Types+import Control.Distributed.Process.Extras+ ( CancelWait(..)+ , Channel+ , Resolvable(..)+ )+import Control.Distributed.Process.Extras.Time+ ( asTimeout+ , TimeInterval+ )+import Control.Monad+import Data.Maybe+ ( fromMaybe+ )++import System.Timeout (timeout)++-- | Wraps a regular @Process a@ as an 'AsyncTask'.+task :: Process a -> AsyncTask a+task = AsyncTask++-- | Wraps the components required and builds a remote 'AsyncTask'.+remoteTask :: Static (SerializableDict a)+ -> NodeId+ -> Closure (Process a)+ -> AsyncTask a+remoteTask = AsyncRemoteTask++-- | Given an 'Async' handle, monitor the worker process.+monitorAsync :: Async a -> Process MonitorRef+monitorAsync hAsync = do+ worker <- resolve hAsync+ case worker of+ Nothing -> die "Invalid Async Handle"+ Just p -> monitor p++-- | Spawns an asynchronous action and returns a handle to it,+-- which can be used to obtain its status and/or result or interact+-- with it (using the API exposed by this module).+--+async :: (Serializable a) => AsyncTask a -> Process (Async a)+async = asyncDo False++-- | This is a useful variant of 'async' that ensures an @Async@ task is+-- never left running unintentionally. We ensure that if the caller's process+-- exits, that the worker is killed.+--+-- There is currently a contract for async workers, that they should+-- exit normally (i.e., they should not call the @exit@ or @kill@ with their own+-- 'ProcessId' nor use the @terminate@ primitive to cease functining), otherwise+-- the 'AsyncResult' will end up being @AsyncFailed DiedException@ instead of+-- containing the desired result.+--+asyncLinked :: (Serializable a) => AsyncTask a -> Process (Async a)+asyncLinked = asyncDo True++-- private API+asyncDo :: (Serializable a) => Bool -> AsyncTask a -> Process (Async a)+asyncDo shouldLink (AsyncRemoteTask d n c) =+ let proc = call d n c in asyncDo shouldLink AsyncTask { asyncTask = proc }+asyncDo shouldLink (AsyncTask proc) = do+ root <- getSelfPid+ result <- liftIO $ newEmptyTMVarIO+ sigStart <- liftIO $ newEmptyTMVarIO+ (sp, rp) <- newChan++ -- listener/response proxy+ insulator <- spawnLocal $ do+ worker <- spawnLocal $ do+ liftIO $ atomically $ takeTMVar sigStart+ r <- proc+ void $ liftIO $ atomically $ putTMVar result (AsyncDone r)++ sendChan sp worker -- let the parent process know the worker pid++ wref <- monitor worker+ rref <- case shouldLink of+ True -> monitor root >>= return . Just+ False -> return Nothing+ finally (pollUntilExit worker result)+ (unmonitor wref >>+ return (maybe (return ()) unmonitor rref))++ workerPid <- receiveChan rp+ liftIO $ atomically $ putTMVar sigStart ()++ return Async {+ _asyncWorker = workerPid+ , _asyncMonitor = insulator+ , _asyncWait = (readTMVar result)+ }++ where+ pollUntilExit :: (Serializable a)+ => ProcessId+ -> TMVar (AsyncResult a)+ -> Process ()+ pollUntilExit wpid result' = do+ r <- receiveWait [+ match (\c@(CancelWait) -> kill wpid "cancel" >> return (Left c))+ , match (\(ProcessMonitorNotification _ pid' r) ->+ return (Right (pid', r)))+ ]+ case r of+ Left CancelWait+ -> liftIO $ atomically $ putTMVar result' AsyncCancelled+ Right (fpid, d)+ | fpid == wpid+ -> case d of+ DiedNormal -> return ()+ _ -> liftIO $ atomically $ putTMVar result' (AsyncFailed d)+ | otherwise -> kill wpid "linkFailed"++-- | Check whether an 'Async' has completed yet.+--+-- See "Control.Distributed.Process.Platform.Async".+poll :: (Serializable a) => Async a -> Process (AsyncResult a)+poll hAsync = do+ r <- liftIO $ atomically $ pollSTM hAsync+ return $ fromMaybe (AsyncPending) r++-- | Like 'poll' but returns 'Nothing' if @(poll hAsync) == AsyncPending@.+--+-- See "Control.Distributed.Process.Platform.Async".+check :: (Serializable a) => Async a -> Process (Maybe (AsyncResult a))+check hAsync = poll hAsync >>= \r -> case r of+ AsyncPending -> return Nothing+ ar -> return (Just ar)++-- | Wait for an asynchronous operation to complete or timeout.+--+-- See "Control.Distributed.Process.Platform.Async".+waitCheckTimeout :: (Serializable a) =>+ TimeInterval -> Async a -> Process (AsyncResult a)+waitCheckTimeout t hAsync =+ waitTimeout t hAsync >>= return . fromMaybe (AsyncPending)++-- | Wait for an asynchronous action to complete, and return its+-- value. The result (which can include failure and/or cancellation) is+-- encoded by the 'AsyncResult' type.+--+-- @wait = liftIO . atomically . waitSTM@+--+-- See "Control.Distributed.Process.Platform.Async".+{-# INLINE wait #-}+wait :: Async a -> Process (AsyncResult a)+wait = liftIO . atomically . waitSTM++-- | Wait for an asynchronous operation to complete or timeout.+--+-- See "Control.Distributed.Process.Platform.Async".+waitTimeout :: (Serializable a) =>+ TimeInterval -> Async a -> Process (Maybe (AsyncResult a))+waitTimeout t hAsync = do+ -- This is not the most efficient thing to do, but it's the most erlang-ish.+ (sp, rp) <- newChan :: (Serializable a) => Process (Channel (AsyncResult a))+ pid <- spawnLocal $ wait hAsync >>= sendChan sp+ receiveChanTimeout (asTimeout t) rp `finally` kill pid "timeout"++-- | Wait for an asynchronous operation to complete or timeout.+-- If it times out, then 'cancelWait' the async handle.+--+waitCancelTimeout :: (Serializable a)+ => TimeInterval+ -> Async a+ -> Process (AsyncResult a)+waitCancelTimeout t hAsync = do+ r <- waitTimeout t hAsync+ case r of+ Nothing -> cancelWait hAsync+ Just ar -> return ar++-- | As 'waitTimeout' but uses STM directly, which might be more efficient.+waitTimeoutSTM :: (Serializable a)+ => TimeInterval+ -> Async a+ -> Process (Maybe (AsyncResult a))+waitTimeoutSTM t hAsync =+ let t' = (asTimeout t)+ in liftIO $ timeout t' $ atomically $ waitSTM hAsync++-- | Wait for any of the supplied @Async@s to complete. If multiple+-- 'Async's complete, then the value returned corresponds to the first+-- completed 'Async' in the list.+--+-- NB: Unlike @AsyncChan@, 'Async' does not discard its 'AsyncResult' once+-- read, therefore the semantics of this function are different to the+-- former. Specifically, if @asyncs = [a1, a2, a3]@ and @(AsyncDone _) = a1@+-- then the remaining @a2, a3@ will never be returned by 'waitAny'.+--+waitAny :: (Serializable a)+ => [Async a]+ -> Process (Async a, AsyncResult a)+waitAny asyncs = do+ r <- liftIO $ waitAnySTM asyncs+ return r++-- | Like 'waitAny', but also cancels the other asynchronous+-- operations as soon as one has completed.+--+waitAnyCancel :: (Serializable a)+ => [Async a] -> Process (Async a, AsyncResult a)+waitAnyCancel asyncs =+ waitAny asyncs `finally` mapM_ cancel asyncs++-- | Wait for the first of two @Async@s to finish.+--+waitEither :: Async a+ -> Async b+ -> Process (Either (AsyncResult a) (AsyncResult b))+waitEither left right =+ liftIO $ atomically $+ (Left <$> waitSTM left)+ `orElse`+ (Right <$> waitSTM right)++-- | Like 'waitEither', but the result is ignored.+--+waitEither_ :: Async a -> Async b -> Process ()+waitEither_ left right =+ liftIO $ atomically $+ (void $ waitSTM left)+ `orElse`+ (void $ waitSTM right)++-- | Waits for both @Async@s to finish.+--+waitBoth :: Async a+ -> Async b+ -> Process ((AsyncResult a), (AsyncResult b))+waitBoth left right =+ liftIO $ atomically $ do+ a <- waitSTM left+ `orElse`+ (waitSTM right >> retry)+ b <- waitSTM right+ return (a,b)++-- | Like 'waitAny' but times out after the specified delay.+waitAnyTimeout :: (Serializable a)+ => TimeInterval+ -> [Async a]+ -> Process (Maybe (AsyncResult a))+waitAnyTimeout delay asyncs =+ let t' = asTimeout delay+ in liftIO $ timeout t' $ do+ r <- waitAnySTM asyncs+ return $ snd r++-- | Cancel an asynchronous operation.+--+-- See "Control.Distributed.Process.Platform.Async".+cancel :: Async a -> Process ()+cancel (Async _ g _) = send g CancelWait++-- | Cancel an asynchronous operation and wait for the cancellation to complete.+--+-- See "Control.Distributed.Process.Platform.Async".+cancelWait :: (Serializable a) => Async a -> Process (AsyncResult a)+cancelWait hAsync = cancel hAsync >> wait hAsync++-- | Cancel an asynchronous operation immediately.+--+-- See "Control.Distributed.Process.Platform.Async".+cancelWith :: (Serializable b) => b -> Async a -> Process ()+cancelWith reason hAsync = do+ worker <- resolve hAsync+ case worker of+ Nothing -> die "Invalid Async Handle"+ Just ref -> exit ref reason++-- | Like 'cancelWith' but sends a @kill@ instruction instead of an exit.+--+-- See 'Control.Distributed.Process.Platform.Async'.+cancelKill :: String -> Async a -> Process ()+cancelKill reason hAsync = do+ worker <- resolve hAsync+ case worker of+ Nothing -> die "Invalid Async Handle"+ Just ref -> kill ref reason++--------------------------------------------------------------------------------+-- STM Specific API --+--------------------------------------------------------------------------------++-- | STM version of 'waitAny'.+waitAnySTM :: [Async a] -> IO (Async a, AsyncResult a)+waitAnySTM asyncs =+ atomically $+ foldr orElse retry $+ map (\a -> do r <- waitSTM a; return (a, r)) asyncs++-- | A version of 'wait' that can be used inside an STM transaction.+--+waitSTM :: Async a -> STM (AsyncResult a)+waitSTM (Async _ _ w) = w++-- | A version of 'poll' that can be used inside an STM transaction.+--+{-# INLINE pollSTM #-}+pollSTM :: Async a -> STM (Maybe (AsyncResult a))+pollSTM (Async _ _ w) = (Just <$> w) `orElse` return Nothing
+ src/Control/Distributed/Process/Async/Internal/Types.hs view
@@ -0,0 +1,80 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE DeriveGeneric #-}++-- | shared, internal types for the Async package+module Control.Distributed.Process.Async.Internal.Types+ ( -- * Exported types+ Async(..)+ , AsyncRef+ , AsyncTask(..)+ , AsyncResult(..)+ ) where++import Control.Concurrent.STM+import Control.Distributed.Process+import Control.Distributed.Process.Extras (Resolvable(..))+import Control.Distributed.Process.Serializable+ ( Serializable+ , SerializableDict+ )+import Data.Binary+import Data.Typeable (Typeable)++import GHC.Generics++-- | A reference to an asynchronous action+type AsyncRef = ProcessId++-- | An handle for an asynchronous action spawned by 'async'.+-- Asynchronous operations are run in a separate process, and+-- operations are provided for waiting for asynchronous actions to+-- complete and obtaining their results (see e.g. 'wait').+--+-- Handles of this type cannot cross remote boundaries, nor are they+-- @Serializable@.+data Async a = Async {+ _asyncWorker :: AsyncRef+ , _asyncMonitor :: AsyncRef+ , _asyncWait :: STM (AsyncResult a)+ }++instance Eq (Async a) where+ Async a b _ == Async c d _ = a == c && b == d++instance Resolvable (Async a) where+ resolve :: Async a -> Process (Maybe ProcessId)+ resolve = return . Just . _asyncWorker++-- | A task to be performed asynchronously.+data AsyncTask a =+ AsyncTask {+ asyncTask :: Process a -- ^ the task to be performed+ }+ | AsyncRemoteTask {+ asyncTaskDict :: Static (SerializableDict a)+ -- ^ the serializable dict required to spawn a remote process+ , asyncTaskNode :: NodeId+ -- ^ the node on which to spawn the asynchronous task+ , asyncTaskProc :: Closure (Process a)+ -- ^ the task to be performed, wrapped in a closure environment+ }++-- | Represents the result of an asynchronous action, which can be in one of+-- several states at any given time.+data AsyncResult a =+ AsyncDone a -- ^ a completed action and its result+ | AsyncFailed DiedReason -- ^ a failed action and the failure reason+ | AsyncLinkFailed DiedReason -- ^ a link failure and the reason+ | AsyncCancelled -- ^ a cancelled action+ | AsyncPending -- ^ a pending action (that is still running)+ deriving (Typeable, Generic)++instance Serializable a => Binary (AsyncResult a) where++deriving instance Eq a => Eq (AsyncResult a)+deriving instance Show a => Show (AsyncResult a)+
+ tests/TestAsync.hs view
@@ -0,0 +1,268 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Main where++import Control.Applicative+import Control.Concurrent.MVar+import Control.Distributed.Process+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node+import Control.Distributed.Process.Serializable()+import Control.Distributed.Process.Async+import Control.Distributed.Process.Extras (Routable(..), Resolvable(..))+import Control.Distributed.Process.Extras.Test+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer+import Control.Distributed.Process.Tests.Internal.Utils (delayedAssertion)+import Data.Binary()+import Data.Typeable()+import Network.Transport.TCP+import qualified Network.Transport as NT++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif++import Test.Framework (Test, testGroup, defaultMain)+import Test.Framework.Providers.HUnit (testCase)+-- import TestUtils++testAsyncPoll :: TestResult (AsyncResult ()) -> Process ()+testAsyncPoll result = do+ hAsync <- async $ task $ do "go" <- expect; say "running" >> return ()+ ar <- poll hAsync+ case ar of+ AsyncPending ->+ sendTo hAsync "go" >> wait hAsync >>= stash result+ _ -> stash result ar >> return ()++testAsyncCancel :: TestResult (AsyncResult ()) -> Process ()+testAsyncCancel result = do+ hAsync <- async $ task $ runTestProcess $ say "running" >> return ()+ sleep $ milliSeconds 100++ p <- poll hAsync -- nasty kind of assertion: use assertEquals?+ case p of+ AsyncPending -> cancel hAsync >> wait hAsync >>= stash result+ _ -> say (show p) >> stash result p++testAsyncCancelWait :: TestResult (Maybe (AsyncResult ())) -> Process ()+testAsyncCancelWait result = do+ testPid <- getSelfPid+ p <- spawnLocal $ do+ hAsync <- async $ task $ runTestProcess $ sleep $ seconds 60+ sleep $ milliSeconds 100++ send testPid "running"++ AsyncPending <- poll hAsync+ cancelWait hAsync >>= send testPid++ "running" <- expect+ d <- expectTimeout (asTimeout $ seconds 5)+ case d of+ Nothing -> kill p "timed out" >> stash result Nothing+ Just ar -> stash result (Just ar)++testAsyncWaitTimeout :: TestResult (Maybe (AsyncResult ())) -> Process ()+testAsyncWaitTimeout result =+ let delay = seconds 1+ in do+ hAsync <- async $ task $ sleep $ seconds 20+ waitTimeout delay hAsync >>= stash result+ cancelWait hAsync >> return ()++testAsyncWaitTimeoutCompletes :: TestResult (Maybe (AsyncResult ()))+ -> Process ()+testAsyncWaitTimeoutCompletes result =+ let delay = seconds 1+ in do+ hAsync <- async $ task $ sleep $ seconds 20+ waitTimeout delay hAsync >>= stash result+ cancelWait hAsync >> return ()++testAsyncWaitTimeoutSTM :: TestResult (Maybe (AsyncResult ())) -> Process ()+testAsyncWaitTimeoutSTM result =+ let delay = seconds 1+ in do+ hAsync <- async $ task $ sleep $ seconds 20+ waitTimeoutSTM delay hAsync >>= stash result++testAsyncWaitTimeoutCompletesSTM :: TestResult (Maybe (AsyncResult Int))+ -> Process ()+testAsyncWaitTimeoutCompletesSTM result =+ let delay = seconds 1 in do++ hAsync <- async $ task $ do+ i <- expect+ return i++ r <- waitTimeoutSTM delay hAsync+ case r of+ Nothing -> sendTo hAsync (10 :: Int)+ >> wait hAsync >>= stash result . Just+ Just _ -> cancelWait hAsync >> stash result Nothing++testAsyncLinked :: TestResult Bool -> Process ()+testAsyncLinked result = do+ mv :: MVar (Async ()) <- liftIO $ newEmptyMVar+ pid <- spawnLocal $ do+ -- NB: async == asyncLinked for AsyncChan+ h <- asyncLinked $ task $ do+ "waiting" <- expect+ return ()+ stash mv h+ "sleeping" <- expect+ return ()++ hAsync <- liftIO $ takeMVar mv++ Just worker <- resolve hAsync+ mref <- monitor worker+ exit pid "stop"++ _ <- receiveTimeout (after 5 Seconds) [+ matchIf (\(ProcessMonitorNotification mref' _ _) -> mref == mref')+ (\_ -> return ())+ ]++ -- since the initial caller died and we used 'asyncLinked', the async should+ -- pick up on the exit signal and set the result accordingly. trying to match+ -- on 'DiedException String' is pointless though, as the *string* is highly+ -- context dependent.+ r <- waitTimeoutSTM (within 3 Seconds) hAsync+ case r of+ Nothing -> stash result True+ Just _ -> stash result False++testAsyncWaitAny :: TestResult [AsyncResult String] -> Process ()+testAsyncWaitAny result = do+ p1 <- async $ task $ expect >>= return+ p2 <- async $ task $ expect >>= return+ p3 <- async $ task $ expect >>= return+ sendTo p3 "c"+ r1 <- waitAny [p1, p2, p3]++ sendTo p1 "a"+ sendTo p2 "b"+ sleep $ seconds 1++ r2 <- waitAny [p2, p3]+ r3 <- waitAny [p1, p2, p3]++ stash result $ map snd [r1, r2, r3]++testAsyncWaitAnyTimeout :: TestResult (Maybe (AsyncResult String)) -> Process ()+testAsyncWaitAnyTimeout result = do+ p1 <- asyncLinked $ task $ expect >>= return+ p2 <- asyncLinked $ task $ expect >>= return+ p3 <- asyncLinked $ task $ expect >>= return+ waitAnyTimeout (seconds 1) [p1, p2, p3] >>= stash result++testAsyncCancelWith :: TestResult Bool -> Process ()+testAsyncCancelWith result = do+ p1 <- async $ task $ do { s :: String <- expect; return s }+ cancelWith "foo" p1+ AsyncFailed (DiedException _) <- wait p1+ stash result True++testAsyncWaitCancelTimeout :: TestResult (AsyncResult ()) -> Process ()+testAsyncWaitCancelTimeout result = do+ p1 <- async $ task $ sleep $ seconds 20+ waitCancelTimeout (seconds 1) p1 >>= stash result++remotableDecl [+ [d| fib :: (NodeId,Int) -> Process Integer ;+ fib (_,0) = return 0+ fib (_,1) = return 1+ fib (myNode,n) = do+ let tsk = remoteTask ($(functionTDict 'fib)) myNode ($(mkClosure 'fib) (myNode,n-2))+ future <- async tsk+ y <- fib (myNode,n-1)+ (AsyncDone z) <- wait future+ return $ y + z+ |]+ ]+testAsyncRecursive :: TestResult Integer -> Process ()+testAsyncRecursive result = do+ myNode <- processNodeId <$> getSelfPid+ fib (myNode,6) >>= stash result++tests :: LocalNode -> [Test]+tests localNode = [+ testGroup "Handling async results with STM" [+ testCase "testAsyncCancel"+ (delayedAssertion+ "expected async task to have been cancelled"+ localNode (AsyncCancelled) testAsyncCancel)+ , testCase "testAsyncPoll"+ (delayedAssertion+ "expected poll to return a valid AsyncResult"+ localNode (AsyncDone ()) testAsyncPoll)+ , testCase "testAsyncCancelWait"+ (delayedAssertion+ "expected cancelWait to complete some time"+ localNode (Just AsyncCancelled) testAsyncCancelWait)+ , testCase "testAsyncWaitTimeout"+ (delayedAssertion+ "expected waitTimeout to return Nothing when it times out"+ localNode (Nothing) testAsyncWaitTimeout)+ , testCase "testAsyncWaitTimeoutSTM"+ (delayedAssertion+ "expected waitTimeoutSTM to return Nothing when it times out"+ localNode (Nothing) testAsyncWaitTimeoutSTM)+ , testCase "testAsyncWaitTimeoutCompletes"+ (delayedAssertion+ "expected waitTimeout to return a value"+ localNode Nothing testAsyncWaitTimeoutCompletes)+ , testCase "testAsyncWaitTimeoutCompletesSTM"+ (delayedAssertion+ "expected waitTimeout to return a value"+ localNode (Just (AsyncDone 10)) testAsyncWaitTimeoutCompletesSTM)+ , testCase "testAsyncLinked"+ (delayedAssertion+ "expected linked process to die with originator"+ localNode True testAsyncLinked)+ , testCase "testAsyncWaitAny"+ (delayedAssertion+ "expected waitAny to pick the first result each time"+ localNode [AsyncDone "c",+ AsyncDone "b",+ AsyncDone "a"] testAsyncWaitAny)+ , testCase "testAsyncWaitAnyTimeout"+ (delayedAssertion+ "expected waitAnyTimeout to handle pending results properly"+ localNode Nothing testAsyncWaitAnyTimeout)+ , testCase "testAsyncCancelWith"+ (delayedAssertion+ "expected the worker to have been killed with the given signal"+ localNode True testAsyncCancelWith)+ , testCase "testAsyncRecursive"+ (delayedAssertion+ "expected Fibonacci 6 to be evaluated, and value of 8 returned"+ localNode 8 testAsyncRecursive)+ , testCase "testAsyncWaitCancelTimeout"+ (delayedAssertion+ "expected waitCancelTimeout to return a value"+ localNode AsyncCancelled testAsyncWaitCancelTimeout)+ ]+ ]++asyncStmTests :: NT.Transport -> IO [Test]+asyncStmTests transport = do+ localNode <- newLocalNode transport $ __remoteTableDecl initRemoteTable+ let testData = tests localNode+ return testData++-- | Given a @builder@ function, make and run a test suite on a single transport+testMain :: (NT.Transport -> IO [Test]) -> IO ()+testMain builder = do+ Right (transport, _) <- createTransportExposeInternals+ "127.0.0.1" "10501" defaultTCPParameters+ testData <- builder transport+ defaultMain testData++main :: IO ()+main = testMain $ asyncStmTests