distributed-process-task (empty) → 0.1.0
raw patch · 7 files changed
+750/−0 lines, 7 filesdep +HUnitdep +QuickCheckdep +ansi-terminalsetup-changed
Dependencies added: HUnit, QuickCheck, ansi-terminal, base, binary, bytestring, containers, data-accessor, deepseq, derive, distributed-process, distributed-process-async, distributed-process-client-server, distributed-process-extras, distributed-process-task, distributed-static, fingertree, ghc-prim, hashable, mtl, network, network-transport, network-transport-tcp, rematch, stm, template-haskell, test-framework, test-framework-hunit, test-framework-quickcheck2, time, transformers, uniplate, unordered-containers
Files
- LICENCE +30/−0
- Setup.lhs +3/−0
- distributed-process-task.cabal +100/−0
- src/Control/Distributed/Process/Task.hs +30/−0
- src/Control/Distributed/Process/Task/Queue/BlockingQueue.hs +216/−0
- tests/TestTaskQueues.hs +137/−0
- tests/TestUtils.hs +234/−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-task.cabal view
@@ -0,0 +1,100 @@+name: distributed-process-task+version: 0.1.0+cabal-version: >=1.8+build-type: Simple+license: BSD3+license-file: LICENCE+stability: experimental+Copyright: Tim Watson 2012 - 2013+Author: Tim Watson+Maintainer: watson.timothy@gmail.com+Stability: experimental+Homepage: http://github.com/haskell-distributed/distributed-process-task+Bug-Reports: http://github.com/haskell-distributed/distributed-process-task/issues+synopsis: Task Framework for The Cloud Haskell Application Platform+description: The Task Framework intends to provide tools for task management, work scheduling and distributed task coordination.+ These capabilities build on the Async Framework as well as other tools and libraries.+ The framework is currently a work in progress. The current release includes a simple bounded blocking queue+ implementation only, as an example of the kind of capability and API that we intend to produce.+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-task++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.1 && < 0.2,+ distributed-process-async >= 0.2 && < 0.3,+ distributed-process-client-server >= 0.1.1 && <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+ hs-source-dirs: src+ ghc-options: -Wall+ exposed-modules:+ Control.Distributed.Process.Task,+ Control.Distributed.Process.Task.Queue.BlockingQueue++test-suite TaskQueueTests+ type: exitcode-stdio-1.0+-- x-uses-tf: true+ build-depends:+ base >= 4.4 && < 5,+ ansi-terminal >= 0.5 && < 0.7,+ containers,+ hashable,+ unordered-containers >= 0.2.3.0 && < 0.3,+ distributed-process >= 0.5.2 && < 0.6,+ distributed-process-task,+ distributed-process-extras,+ distributed-process-async,+ distributed-process-client-server,+ distributed-static,+ bytestring,+ data-accessor,+ fingertree < 0.2,+ network-transport >= 0.4 && < 0.5,+ deepseq >= 1.3.0.1 && < 1.4,+ mtl,+ network-transport-tcp >= 0.4 && < 0.5,+ binary >= 0.6.3.0 && < 0.8,+ network >= 2.3 && < 2.7,+ HUnit >= 1.2 && < 2,+ stm >= 2.3 && < 2.5,+ time > 1.4 && < 1.5,+ test-framework >= 0.6 && < 0.9,+ test-framework-hunit,+ QuickCheck >= 2.4,+ test-framework-quickcheck2,+ transformers,+ rematch >= 0.2.0.0,+ ghc-prim+ hs-source-dirs:+ tests+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog+ extensions: CPP+ main-is: TestTaskQueues.hs+ other-modules: TestUtils
+ src/Control/Distributed/Process/Task.hs view
@@ -0,0 +1,30 @@+-----------------------------------------------------------------------------+-- |+-- Module : Control.Distributed.Process.Platform.Task+-- Copyright : (c) Tim Watson 2013 - 2014+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Tim Watson <watson.timothy@gmail.com>+-- Stability : experimental+-- Portability : non-portable (requires concurrency)+--+-- The /Task Framework/ intends to provide tools for task management, work+-- scheduling and distributed task coordination. These capabilities build on the+-- /Execution Framework/ as well as other tools and libraries. The framework is+-- currently a work in progress. The current release includes a simple bounded+-- blocking queue implementation only, as an example of the kind of capability+-- and API that we intend to produce.+--+-- The /Task Framework/ will be broken down by the task scheduling and management+-- algorithms it provides, e.g., at a low level providing work queues, worker pools+-- and the like, whilst at a high level allowing the user to choose between work+-- stealing, sharing, distributed coordination, user defined sensor based bounds/limits+-- and so on.+--+-----------------------------------------------------------------------------+module Control.Distributed.Process.Task+ ( -- * Task Queues+ module Control.Distributed.Process.Task.Queue.BlockingQueue+ ) where++import Control.Distributed.Process.Task.Queue.BlockingQueue
+ src/Control/Distributed/Process/Task/Queue/BlockingQueue.hs view
@@ -0,0 +1,216 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DeriveGeneric #-}++-----------------------------------------------------------------------------+-- |+-- Module : Control.Distributed.Process.Task.Queue.BlockingQueue+-- Copyright : (c) Tim Watson 2012 - 2013+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Tim Watson <watson.timothy@gmail.com>+-- Stability : experimental+-- Portability : non-portable (requires concurrency)+--+-- A simple bounded (size) task queue, which accepts requests and blocks the+-- sender until they're completed. The size limit is applied to the number+-- of concurrent tasks that are allowed to execute - if the limit is 3, then+-- the first three tasks will be executed immediately, but further tasks will+-- then be queued (internally) until one or more tasks completes and+-- the number of active/running tasks falls within the concurrency limit.+--+-- Note that the process calling 'executeTask' will be blocked for _at least_+-- the duration of the task itself, regardless of whether or not the queue has+-- reached its concurrency limit. This provides a simple means to prevent work+-- from being submitted faster than the server can handle, at the expense of+-- flexible scheduling.+--+-----------------------------------------------------------------------------++module Control.Distributed.Process.Task.Queue.BlockingQueue+ ( BlockingQueue()+ , SizeLimit+ , BlockingQueueStats(..)+ , start+ , pool+ , executeTask+ , stats+ ) where++import Control.Distributed.Process hiding (call)+import Control.Distributed.Process.Closure()+import Control.Distributed.Process.Extras.Internal.Types+import Control.Distributed.Process.Async+import Control.Distributed.Process.ManagedProcess+import qualified Control.Distributed.Process.ManagedProcess as ManagedProcess+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Serializable+import Data.Binary+import Data.List+ ( deleteBy+ , find+ )+import Data.Sequence+ ( Seq+ , ViewR(..)+ , (<|)+ , viewr+ )+import qualified Data.Sequence as Seq (empty, length)+import Data.Typeable++import GHC.Generics (Generic)++-- | Limit for the number of concurrent tasks.+--+type SizeLimit = Int++data GetStats = GetStats+ deriving (Typeable, Generic)+instance Binary GetStats where++data BlockingQueueStats = BlockingQueueStats {+ maxJobs :: Int+ , activeJobs :: Int+ , queuedJobs :: Int+ } deriving (Typeable, Generic)++instance Binary BlockingQueueStats where++data BlockingQueue a = BlockingQueue {+ poolSize :: SizeLimit+ , active :: [(MonitorRef, CallRef (Either ExitReason a), Async a)]+ , accepted :: Seq (CallRef (Either ExitReason a), Closure (Process a))+ } deriving (Typeable)++-- Client facing API++-- | Start a queue with an upper bound on the # of concurrent tasks.+--+start :: forall a . (Serializable a)+ => Process (InitResult (BlockingQueue a))+ -> Process ()+start init' = ManagedProcess.serve () (\() -> init') poolServer+ where poolServer =+ defaultProcess {+ apiHandlers = [+ handleCallFrom (\s f (p :: Closure (Process a)) -> storeTask s f p)+ , handleCall poolStatsRequest+ ]+ , infoHandlers = [ handleInfo taskComplete ]+ } :: ProcessDefinition (BlockingQueue a)++-- | Define a pool of a given size.+--+pool :: forall a . Serializable a+ => SizeLimit+ -> Process (InitResult (BlockingQueue a))+pool sz' = return $ InitOk (BlockingQueue sz' [] Seq.empty) Infinity++-- | Enqueue a task in the pool and block until it is complete.+--+executeTask :: forall s a . (Addressable s, Serializable a)+ => s+ -> Closure (Process a)+ -> Process (Either ExitReason a)+executeTask sid t = call sid t++-- | Fetch statistics for a queue.+--+stats :: forall s . Addressable s => s -> Process (Maybe BlockingQueueStats)+stats sid = tryCall sid GetStats++-- internal / server-side API++poolStatsRequest :: (Serializable a)+ => BlockingQueue a+ -> GetStats+ -> Process (ProcessReply BlockingQueueStats (BlockingQueue a))+poolStatsRequest st GetStats =+ let sz = poolSize st+ ac = length (active st)+ pj = Seq.length (accepted st)+ in reply (BlockingQueueStats sz ac pj) st++storeTask :: Serializable a+ => BlockingQueue a+ -> CallRef (Either ExitReason a)+ -> Closure (Process a)+ -> Process (ProcessReply (Either ExitReason a) (BlockingQueue a))+storeTask s r c = acceptTask s r c >>= noReply_++acceptTask :: Serializable a+ => BlockingQueue a+ -> CallRef (Either ExitReason a)+ -> Closure (Process a)+ -> Process (BlockingQueue a)+acceptTask s@(BlockingQueue sz' runQueue taskQueue) from task' =+ let currentSz = length runQueue+ in case currentSz >= sz' of+ True -> do+ return $ s { accepted = enqueue taskQueue (from, task') }+ False -> do+ proc <- unClosure task'+ asyncHandle <- async $ task proc+ ref <- monitorAsync asyncHandle+ taskEntry <- return (ref, from, asyncHandle)+ return s { active = (taskEntry:runQueue) }++-- a worker has exited, process the AsyncResult and send a reply to the+-- waiting client (who is still stuck in 'call' awaiting a response).+taskComplete :: forall a . Serializable a+ => BlockingQueue a+ -> ProcessMonitorNotification+ -> Process (ProcessAction (BlockingQueue a))+taskComplete s@(BlockingQueue _ runQ _)+ (ProcessMonitorNotification ref _ _) =+ let worker = findWorker ref runQ in+ case worker of+ Just t@(_, c, h) -> wait h >>= respond c >> bump s t >>= continue+ Nothing -> continue s++ where+ respond :: CallRef (Either ExitReason a)+ -> AsyncResult a+ -> Process ()+ respond c (AsyncDone r) = replyTo c ((Right r) :: (Either ExitReason a))+ respond c (AsyncFailed d) = replyTo c ((Left (ExitOther $ show d)) :: (Either ExitReason a))+ respond c (AsyncLinkFailed d) = replyTo c ((Left (ExitOther $ show d)) :: (Either ExitReason a))+ respond _ _ = die $ ExitOther "IllegalState"++ bump :: BlockingQueue a+ -> (MonitorRef, CallRef (Either ExitReason a), Async a)+ -> Process (BlockingQueue a)+ bump st@(BlockingQueue _ runQueue acc) worker =+ let runQ2 = deleteFromRunQueue worker runQueue+ accQ = dequeue acc in+ case accQ of+ Nothing -> return st { active = runQ2 }+ Just ((tr,tc), ts) -> acceptTask (st { accepted = ts, active = runQ2 }) tr tc++findWorker :: MonitorRef+ -> [(MonitorRef, CallRef (Either ExitReason a), Async a)]+ -> Maybe (MonitorRef, CallRef (Either ExitReason a), Async a)+findWorker key = find (\(ref,_,_) -> ref == key)++deleteFromRunQueue :: (MonitorRef, CallRef (Either ExitReason a), Async a)+ -> [(MonitorRef, CallRef (Either ExitReason a), Async a)]+ -> [(MonitorRef, CallRef (Either ExitReason a), Async a)]+deleteFromRunQueue c@(p, _, _) runQ = deleteBy (\_ (b, _, _) -> b == p) c runQ++{-# INLINE enqueue #-}+enqueue :: Seq a -> a -> Seq a+enqueue s a = a <| s++{-# INLINE dequeue #-}+dequeue :: Seq a -> Maybe (a, Seq a)+dequeue s = maybe Nothing (\(s' :> a) -> Just (a, s')) $ getR s++getR :: Seq a -> Maybe (ViewR a)+getR s =+ case (viewr s) of+ EmptyR -> Nothing+ a -> Just a+
+ tests/TestTaskQueues.hs view
@@ -0,0 +1,137 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE TemplateHaskell #-}++module Main where++import Control.Distributed.Process hiding (call)+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node+import Control.Distributed.Process.Async+import Control.Distributed.Process.ManagedProcess+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer+import Control.Distributed.Process.Extras.Internal.Types+import Control.Distributed.Process.Serializable()++import Control.Distributed.Process.Task.Queue.BlockingQueue hiding (start)+import qualified Control.Distributed.Process.Task.Queue.BlockingQueue as Pool (start)++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif++import Test.Framework (Test, testGroup)+import Test.Framework.Providers.HUnit (testCase)+import TestUtils++import qualified Network.Transport as NT++-- utilities++sampleTask :: (TimeInterval, String) -> Process String+sampleTask (t, s) = sleep t >> return s++namedTask :: (String, String) -> Process String+namedTask (name, result) = do+ self <- getSelfPid+ register name self+ () <- expect+ return result++crashingTask :: SendPort ProcessId -> Process String+crashingTask sp = getSelfPid >>= sendChan sp >> die "Boom"++$(remotable ['sampleTask, 'namedTask, 'crashingTask])++-- SimplePool tests++startPool :: SizeLimit -> Process ProcessId+startPool sz = spawnLocal $ do+ Pool.start (pool sz :: Process (InitResult (BlockingQueue String)))++testSimplePoolJobBlocksCaller :: TestResult (AsyncResult (Either ExitReason String))+ -> Process ()+testSimplePoolJobBlocksCaller result = do+ pid <- startPool 1+ -- we do a non-blocking test first+ job <- return $ ($(mkClosure 'sampleTask) (seconds 2, "foobar"))+ callAsync pid job >>= wait >>= stash result++testJobQueueSizeLimiting ::+ TestResult (Maybe (AsyncResult (Either ExitReason String)),+ Maybe (AsyncResult (Either ExitReason String)))+ -> Process ()+testJobQueueSizeLimiting result = do+ pid <- startPool 1+ job1 <- return $ ($(mkClosure 'namedTask) ("job1", "foo"))+ job2 <- return $ ($(mkClosure 'namedTask) ("job2", "bar"))+ h1 <- callAsync pid job1 :: Process (Async (Either ExitReason String))+ h2 <- callAsync pid job2 :: Process (Async (Either ExitReason String))++ -- despite the fact that we tell job2 to proceed first,+ -- the size limit (of 1) will ensure that only job1 can+ -- proceed successfully!+ nsend "job2" ()+ AsyncPending <- poll h2+ Nothing <- whereis "job2"++ -- we can get here *very* fast, so give the registration time to kick in+ sleep $ milliSeconds 250+ j1p <- whereis "job1"+ case j1p of+ Nothing -> die $ "timing is out - job1 isn't registered yet"+ Just p -> send p ()++ -- once job1 completes, we *should* be able to proceed with job2+ -- but we allow a little time for things to catch up+ sleep $ milliSeconds 250+ nsend "job2" ()++ r2 <- waitTimeout (within 2 Seconds) h2+ r1 <- waitTimeout (within 2 Seconds) h1+ stash result (r1, r2)++testExecutionErrors :: TestResult Bool -> Process ()+testExecutionErrors result = do+ pid <- startPool 1+ (sp, rp) <- newChan :: Process (SendPort ProcessId,+ ReceivePort ProcessId)+ job <- return $ ($(mkClosure 'crashingTask) sp)+ res <- executeTask pid job+ rpid <- receiveChan rp+-- liftIO $ putStrLn (show res)+ stash result (expectedErrorMessage rpid == res)+ where+ expectedErrorMessage p =+ Left $ ExitOther $ "DiedException \"exit-from=" ++ (show p) ++ ",reason=Boom\""++myRemoteTable :: RemoteTable+myRemoteTable = Main.__remoteTable initRemoteTable++tests :: NT.Transport -> IO [Test]+tests transport = do+ localNode <- newLocalNode transport myRemoteTable+ return [+ testGroup "Task Execution And Prioritisation" [+ testCase "Each execution blocks the submitter"+ (delayedAssertion+ "expected the server to return the task outcome"+ localNode (AsyncDone (Right "foobar")) testSimplePoolJobBlocksCaller)+ , testCase "Only 'max' tasks can proceed at any time"+ (delayedAssertion+ "expected the server to block the second job until the first was released"+ localNode+ (Just (AsyncDone (Right "foo")),+ Just (AsyncDone (Right "bar"))) testJobQueueSizeLimiting)+ , testCase "Crashing Tasks are Reported Properly"+ (delayedAssertion+ "expected the server to report an error"+ localNode True testExecutionErrors)+ ]+ ]++main :: IO ()+main = testMain $ tests+
+ tests/TestUtils.hs view
@@ -0,0 +1,234 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveGeneric #-}++module TestUtils+ ( TestResult+ -- ping !+ , Ping(Ping)+ , ping+ , shouldBe+ , shouldMatch+ , shouldContain+ , shouldNotContain+ , shouldExitWith+ , expectThat+ -- test process utilities+ , TestProcessControl+ , startTestProcess+ , runTestProcess+ , testProcessGo+ , testProcessStop+ , testProcessReport+ , delayedAssertion+ , assertComplete+ , waitForExit+ -- logging+ , Logger()+ , newLogger+ , putLogMsg+ , stopLogger+ -- runners+ , mkNode+ , tryRunProcess+ , testMain+ , stash+ ) where++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif+import Control.Concurrent+ ( ThreadId+ , myThreadId+ , forkIO+ )+import Control.Concurrent.STM+ ( TQueue+ , newTQueueIO+ , readTQueue+ , writeTQueue+ )+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , takeMVar+ , putMVar+ )++import Control.Distributed.Process+import Control.Distributed.Process.Node+import Control.Distributed.Process.Serializable()+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer+import Control.Distributed.Process.Extras.Internal.Types+import Control.Exception (SomeException)+import qualified Control.Exception as Exception+import Control.Monad (forever)+import Control.Monad.STM (atomically)+import Control.Rematch hiding (match)+import Control.Rematch.Run+import Test.HUnit (Assertion, assertFailure)+import Test.HUnit.Base (assertBool)+import Test.Framework (Test, defaultMain)+import Control.DeepSeq++import Network.Transport.TCP+import qualified Network.Transport as NT++import Data.Binary+import Data.Typeable+import GHC.Generics++--expect :: a -> Matcher a -> Process ()+--expect a m = liftIO $ Rematch.expect a m++expectThat :: a -> Matcher a -> Process ()+expectThat a matcher = case res of+ MatchSuccess -> return ()+ (MatchFailure msg) -> liftIO $ assertFailure msg+ where res = runMatch matcher a++shouldBe :: a -> Matcher a -> Process ()+shouldBe = expectThat++shouldContain :: (Show a, Eq a) => [a] -> a -> Process ()+shouldContain xs x = expectThat xs $ hasItem (equalTo x)++shouldNotContain :: (Show a, Eq a) => [a] -> a -> Process ()+shouldNotContain xs x = expectThat xs $ isNot (hasItem (equalTo x))++shouldMatch :: a -> Matcher a -> Process ()+shouldMatch = expectThat++shouldExitWith :: (Addressable a) => a -> DiedReason -> Process ()+shouldExitWith a r = do+ _ <- resolve a+ d <- receiveWait [ match (\(ProcessMonitorNotification _ _ r') -> return r') ]+ d `shouldBe` equalTo r++waitForExit :: MVar ExitReason+ -> Process (Maybe ExitReason)+waitForExit exitReason = do+ -- we *might* end up blocked here, so ensure the test doesn't jam up!+ self <- getSelfPid+ tref <- killAfter (within 10 Seconds) self "testcast timed out"+ tr <- liftIO $ takeMVar exitReason+ cancelTimer tref+ case tr of+ ExitNormal -> return Nothing+ other -> return $ Just other++mkNode :: String -> IO LocalNode+mkNode port = do+ Right (transport1, _) <- createTransportExposeInternals+ "127.0.0.1" port defaultTCPParameters+ newLocalNode transport1 initRemoteTable++-- | Run the supplied @testProc@ using an @MVar@ to collect and assert+-- against its result. Uses the supplied @note@ if the assertion fails.+delayedAssertion :: (Eq a) => String -> LocalNode -> a ->+ (TestResult a -> Process ()) -> Assertion+delayedAssertion note localNode expected testProc = do+ result <- newEmptyMVar+ _ <- forkProcess localNode $ testProc result+ assertComplete note result expected++-- | Takes the value of @mv@ (using @takeMVar@) and asserts that it matches @a@+assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()+assertComplete msg mv a = do+ b <- takeMVar mv+ assertBool msg (a == b)++-- synchronised logging++data Logger = Logger { _tid :: ThreadId, msgs :: TQueue String }++-- | Create a new Logger.+-- Logger uses a 'TQueue' to receive and process messages on a worker thread.+newLogger :: IO Logger+newLogger = do+ tid <- liftIO $ myThreadId+ q <- liftIO $ newTQueueIO+ _ <- forkIO $ logger q+ return $ Logger tid q+ where logger q' = forever $ do+ msg <- atomically $ readTQueue q'+ putStrLn msg++-- | Send a message to the Logger+putLogMsg :: Logger -> String -> Process ()+putLogMsg logger msg = liftIO $ atomically $ writeTQueue (msgs logger) msg++-- | Stop the worker thread for the given Logger+stopLogger :: Logger -> IO ()+stopLogger = (flip Exception.throwTo) Exception.ThreadKilled . _tid++-- | 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++-- | Runs a /test process/ around the supplied @proc@, which is executed+-- whenever the outer process loop receives a 'Go' signal.+runTestProcess :: Process () -> Process ()+runTestProcess proc = do+ ctl <- expect+ case ctl of+ Stop -> return ()+ Go -> proc >> runTestProcess proc+ Report p -> receiveWait [matchAny (\m -> forward m p)] >> runTestProcess proc++-- | Starts a test process on the local node.+startTestProcess :: Process () -> Process ProcessId+startTestProcess proc =+ spawnLocal $ do+ getSelfPid >>= register "test-process"+ runTestProcess proc++-- | Control signals used to manage /test processes/+data TestProcessControl = Stop | Go | Report ProcessId+ deriving (Typeable, Generic)++instance Binary TestProcessControl where++-- | A mutable cell containing a test result.+type TestResult a = MVar a++-- | Stashes a value in our 'TestResult' using @putMVar@+stash :: TestResult a -> a -> Process ()+stash mvar x = liftIO $ putMVar mvar x++-- | Tell a /test process/ to stop (i.e., 'terminate')+testProcessStop :: ProcessId -> Process ()+testProcessStop pid = send pid Stop++-- | Tell a /test process/ to continue executing+testProcessGo :: ProcessId -> Process ()+testProcessGo pid = send pid Go++-- | A simple @Ping@ signal+data Ping = Ping+ deriving (Typeable, Generic, Eq, Show)++instance Binary Ping where+instance NFData Ping where++ping :: ProcessId -> Process ()+ping pid = send pid Ping+++tryRunProcess :: LocalNode -> Process () -> IO ()+tryRunProcess node p = do+ tid <- liftIO myThreadId+ runProcess node $ catch p (\e -> liftIO $ Exception.throwTo tid (e::SomeException))++-- | Tell a /test process/ to send a report (message)+-- back to the calling process+testProcessReport :: ProcessId -> Process ()+testProcessReport pid = do+ self <- getSelfPid+ send pid $ Report self