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-task.cabal b/distributed-process-task.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-task.cabal @@ -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 diff --git a/src/Control/Distributed/Process/Task.hs b/src/Control/Distributed/Process/Task.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Task.hs @@ -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 +-- 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 diff --git a/src/Control/Distributed/Process/Task/Queue/BlockingQueue.hs b/src/Control/Distributed/Process/Task/Queue/BlockingQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Task/Queue/BlockingQueue.hs @@ -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 +-- 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 + diff --git a/tests/TestTaskQueues.hs b/tests/TestTaskQueues.hs new file mode 100644 --- /dev/null +++ b/tests/TestTaskQueues.hs @@ -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 + diff --git a/tests/TestUtils.hs b/tests/TestUtils.hs new file mode 100644 --- /dev/null +++ b/tests/TestUtils.hs @@ -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