diff --git a/LICENCE b/LICENCE new file mode 100644 --- /dev/null +++ b/LICENCE @@ -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. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,3 @@ +#!/usr/bin/env runhaskell +> import Distribution.Simple +> main = defaultMain diff --git a/distributed-process-async.cabal b/distributed-process-async.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-async.cabal @@ -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 diff --git a/src/Control/Distributed/Process/Async.hs b/src/Control/Distributed/Process/Async.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Async.hs @@ -0,0 +1,371 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.Async +-- Copyright : (c) Tim Watson 2012 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- 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 diff --git a/src/Control/Distributed/Process/Async/Internal/Types.hs b/src/Control/Distributed/Process/Async/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Async/Internal/Types.hs @@ -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) + diff --git a/tests/TestAsync.hs b/tests/TestAsync.hs new file mode 100644 --- /dev/null +++ b/tests/TestAsync.hs @@ -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