lawless-concurrent-machines (empty) → 0.3.1
raw patch · 16 files changed
+1287/−0 lines, 16 filesdep +asyncdep +basedep +containerssetup-changed
Dependencies added: async, base, containers, lawless-concurrent-machines, lifted-async, machines, monad-control, process, semigroups, tasty, tasty-hunit, time, transformers, transformers-base
Files
- CHANGELOG.md +21/−0
- LICENSE +30/−0
- README.md +53/−0
- Setup.hs +2/−0
- examples/ExampleFanout.hs +38/−0
- lawless-concurrent-machines.cabal +108/−0
- src/Data/Machine/Concurrent.hs +123/−0
- src/Data/Machine/Concurrent/AsyncStep.hs +61/−0
- src/Data/Machine/Concurrent/Buffer.hs +136/−0
- src/Data/Machine/Concurrent/Fanout.hs +85/−0
- src/Data/Machine/Concurrent/Scatter.hs +237/−0
- src/Data/Machine/Concurrent/Tee.hs +35/−0
- src/Data/Machine/Concurrent/Wye.hs +113/−0
- src/Data/Machine/Regulated.hs +22/−0
- tests/AllTests.hs +51/−0
- tests/SPlotTests.hs +172/−0
+ CHANGELOG.md view
@@ -0,0 +1,21 @@+# 0.3.0++- Removed `buffer` and `rolling` functions++These functions did not properly use the buffers intended to mediate+the connection to a downstream consumer. This functionality is+provided by `bufferConnect` and `rollingConnect`. Thanks to Ben+Sinclair for identifying the problem.++The issue is that these are necessarily connectors between machines,+and can not be treated as modifiers of a single machine in isolation.++# 0.2.3++- GHC-8.0.1 compatibility+- Dropped support for GHC-7.8++# 0.2.0++- Fix fanout behavior (Ben SInclair)+- Make ExampleFanout a benchmark that cabal can run
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2014, Anthony Cowley++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 Anthony Cowley 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.
+ README.md view
@@ -0,0 +1,53 @@+A simple example of a pipelined computation whose throughput is improved by concurrently running distinct processing stages is given in [`examples/Pipeline.hs`](http://github.com/acowley/concurrent-machines/blob/master/examples/Pipeline.hs).++```haskell+import Data.Time.Clock (getCurrentTime, diffUTCTime)+import Control.Concurrent (threadDelay)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Machine.Concurrent+```++Suppose we have a worker that performs a computation on its input before producing output. This operation may take some time due to, say, network IO or just CPU load. Here we simulate an operation that takes some time with `threadDelay`.++```haskell+worker :: String -> Double -> ProcessT IO () ()+worker name dt = repeatedly $ do _ <- await+ liftIO $ do+ putStrLn $ name ++ " working on its input"+ threadDelay dt'+ yield ()+ where dt' = floor $ dt * 1000000+```++We will use a little helper to time two variations of our test program.++```haskell+timed :: MonadIO m => m a -> m (a, Double)+timed m = do t1 <- liftIO getCurrentTime+ r <- m+ t2 <- liftIO getCurrentTime+ return (r, realToFrac $ t2 `diffUTCTime` t1)+```++Now we will run a three-stage pipeline where each stage takes one second to process its input before yielding some output. A sequential execution strategy will thus take three seconds to pass an input through this pipeline. At the top-level, we will request three outputs, each of which will take three seconds to produce, resulting in a total execution time of approximately nine seconds.++```haskell+main :: IO ()+main = do (r,dt) <- timed . runT . supply (repeat ()) $+ worker "A" 1 ~> worker "B" 1 ~> worker "C" 1 ~> taking 3+ putStrLn $ "Sequentially produced "++show r+ putStrLn $ "Sequential processing took "++show dt++"s"+```++If we instead run the same arrangement as a pipelined computation, we allow the independent stages to run concurrently, much like the stages in a pipelined CPU.++```haskell+ (r',dt') <- timed . runT . supply (repeat ()) $+ worker "A" 1 >~> worker "B" 1 >~> worker "C" 1 >~> taking 3+ putStrLn $ "Pipeline produced "++show r'+ putStrLn $ "Pipeline processing took "++show dt'++"s"+```++With this arrangement, the first output we request takes three seconds to produce as we must wait for an input to pass through the entire length of the pipeline. However, successive outputs are following that first output through the pipeline so that we will produce more output at one second intervals. Therefore it takes is `3 + 2 = 5` seconds to produce three outputs: three seconds for the first output, and one second for each of the following two.++[](https://travis-ci.org/acowley/concurrent-machines)
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ examples/ExampleFanout.hs view
@@ -0,0 +1,38 @@+-- | A demonstration of concurrent vs serialized fanout. A single+-- input is fed to two workers who both perform a CPU intensive+-- task. The concurrent version is approximately twice as fast when+-- two CPU cores are available.+import Control.Monad (when)+import Data.List (sort)+import Data.Machine.Concurrent+import qualified Data.Machine.Fanout as M+import Data.Time.Clock+import Text.Printf++-- A slow Fibonacci sequence can occupy the CPU+fib :: Int -> Int+fib 0 = 0+fib 1 = 1+fib n = fib (n-1) + fib (n-2)++-- Build a worker process with the given name.+worker :: String -> ProcessT IO Int [String]+worker name = repeatedly $ do+ seed <- await+ let x = fib seed+ x `seq` yield $ [name++": "++show seed++" => "++show x]++main :: IO ()+main = do t1 <- getCurrentTime+ [responses] <- runT $ supply [42] (fanout workers) ~> taking 1+ t2 <- getCurrentTime+ [responses'] <- runT $ supply [42] (M.fanout workers) ~> taking 1+ t3 <- getCurrentTime+ when (map sort responses /= map sort responses') $+ putStrLn "Concurrent and serial workers gave different results!"+ let dt = realToFrac (diffUTCTime t2 t1) :: Double+ dt' = realToFrac (diffUTCTime t3 t2) :: Double+ putStrLn $ "Concurrent vs. serial ("++show responses++"): "+ putStrLn $ " " ++ printf "%0.1fs vs %0.1fs" dt dt'+ where workers = [ worker "alpha"+ , worker "beta" ]
+ lawless-concurrent-machines.cabal view
@@ -0,0 +1,108 @@+name: lawless-concurrent-machines+version: 0.3.1+synopsis: Concurrent networked stream transducers++description: A simple use-case for this library is to run the stages+ of a pipelined streaming computation concurrently. If+ data is streaming through multiple processing stages, you+ might build a machine like+ .+ @+ step1 >~> step2 >~> step3+ @+ .+ The @>~>@ operator connects the machines on+ either side with a one-element buffer. This means that+ data is pulled from upstream sources eagerly (perhaps+ pulling one more value than will be consumed by+ downstream), but it also means that each stage can be+ working simultaneously, increasing throughput of the+ entire pipeline.+ .+ A few small examples are available in the @examples@+ directory of the source repository.+ .+ NOTE: This is a temporary fork until concurrent-machines+ 0.3.1 is released.++license: BSD3+license-file: LICENSE+author: Anthony Cowley+maintainer: acowley@gmail.com+copyright: Copyright (C) 2014 Anthony Cowley+category: Concurrency, Control+build-type: Simple+extra-source-files: README.md, CHANGELOG.md+cabal-version: >=1.10+tested-with: GHC == 7.10.3, GHC == 8.0.1, GHC == 8.2.1++source-repository head+ type: git+ location: git@gitlab.com:theunixman/haskell/lawless-concurrent-machines.git+++flag splot+ description: Build test with splot visual output+ default: False+ manual: True++library+ exposed-modules: Data.Machine.Concurrent,+ Data.Machine.Regulated,+ Data.Machine.Concurrent.AsyncStep,+ Data.Machine.Concurrent.Buffer,+ Data.Machine.Concurrent.Fanout,+ Data.Machine.Concurrent.Scatter,+ Data.Machine.Concurrent.Tee,+ Data.Machine.Concurrent.Wye+ -- other-modules:+ other-extensions: GADTs, FlexibleContexts, RankNTypes, TupleSections,+ ScopedTypeVariables+ build-depends: base >= 4.8 && < 5,+ monad-control >= 1.0 && < 1.1,+ transformers >= 0.4 && < 0.6,+ time >= 1.4 && < 1.9,+ containers >= 0.5 && < 0.6,+ transformers-base >= 0.4 && < 0.6,+ machines >= 0.5 && < 0.7,+ async >= 2.0.1 && < 2.2,+ lifted-async >= 0.1 && < 0.10,+ semigroups >= 0.8 && < 0.19+ hs-source-dirs: src+ default-language: Haskell2010+ ghc-options: -Wall++test-suite tests+ type: exitcode-stdio-1.0+ hs-source-dirs: tests+ main-is: AllTests.hs+ ghc-options: -Wall -O0+ default-language: Haskell2010+ build-depends: base >= 4.6 && < 5, lawless-concurrent-machines, machines,+ tasty, tasty-hunit, transformers, time++-- This is a nice modification of the tests that generates plots for+-- visualizing the execution of parallel processes using the splot+-- tool. That program requires cairo, so the test is disabled by+-- default.+test-suite splotime+ type: exitcode-stdio-1.0+ if flag(splot)+ buildable: True+ else+ buildable: False+ hs-source-dirs: tests+ main-is: SPlotTests.hs+ ghc-options: -Wall -O0+ default-language: Haskell2010+ if flag(splot)+ build-depends: base >= 4.6 && < 5, lawless-concurrent-machines, machines,+ process, tasty, tasty-hunit, transformers, time++benchmark fanout+ type: exitcode-stdio-1.0+ hs-source-dirs: examples+ main-is: ExampleFanout.hs+ build-depends: base >= 4.8 && < 5, time, machines, lawless-concurrent-machines+ default-language: Haskell2010+ ghc-options: -threaded "-with-rtsopts=-N2"
+ src/Data/Machine/Concurrent.hs view
@@ -0,0 +1,123 @@+{-# LANGUAGE CPP, GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables,+ TupleSections #-}+-- | The primary use of concurrent machines is to establish a+-- pipelined architecture that can boost overall throughput by running+-- each stage of the pipeline at the same time. The processing, or+-- production, rate of each stage may not be identical, so facilities+-- are provided to loosen the temporal coupling between pipeline+-- stages using buffers.+--+-- This architecture also lends itself to operations where multiple+-- workers are available for procesisng inputs. If each worker is to+-- process the same set of inputs, consider 'fanout' and+-- 'fanoutSteps'. If each worker is to process a disjoint set of+-- inputs, consider 'scatter'.+module Data.Machine.Concurrent (module Data.Machine,+ -- * Concurrent connection+ (>~>), (<~<),+ -- * Buffered machines+ bufferConnect, rollingConnect,+ -- * Concurrent processing of shared inputs+ fanout, fanoutSteps,+ -- * Concurrent multiple-input machines+ wye, tee, scatter, splitSum, mergeSum,+ splitProd) where+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Control.Applicative+#endif+import Control.Concurrent.Async.Lifted+import Control.Monad (join)+import Control.Monad.Trans.Control+import Data.Machine hiding (tee, wye)+import Data.Machine.Concurrent.AsyncStep+import Data.Machine.Concurrent.Buffer+import Data.Machine.Concurrent.Fanout+import Data.Machine.Concurrent.Scatter+import Data.Machine.Concurrent.Wye+import Data.Machine.Concurrent.Tee++-- | Build a new 'Machine' by adding a 'Process' to the output of an+-- old 'Machine'. The upstream machine is run concurrently with+-- downstream with the aim that upstream will have a yielded value+-- ready as soon as downstream awaits. This effectively creates a+-- buffer between upstream and downstream, or source and sink, that+-- can contain up to one value.+--+-- @+-- ('<~<') :: 'Process' b c -> 'Process' a b -> 'Process' a c+-- ('<~<') :: 'Process' c d -> 'Data.Machine.Tee.Tee' a b c -> 'Data.Machine.Tee.Tee' a b d+-- ('<~<') :: 'Process' b c -> 'Machine' k b -> 'Machine' k c+-- @+(<~<) :: MonadBaseControl IO m+ => ProcessT m b c -> MachineT m k b -> MachineT m k c+mp <~< ma = racers ma mp++-- | Flipped ('<~<').+(>~>) :: MonadBaseControl IO m+ => MachineT m k b -> ProcessT m b c -> MachineT m k c+ma >~> mp = mp <~< ma++infixl 7 >~>++-- | We want the first available response.+waitEither' :: MonadBaseControl IO m+ => Maybe (Async (StM m a)) -> Async (StM m b)+ -> m (Either a b)+waitEither' Nothing y = Right <$> wait y+waitEither' (Just x) y = waitEither x y++-- | Let a source and a sink chase each other, providing an effective+-- one-element buffer between the two. The idea is to run both+-- concurrently at all times so that as soon as the sink 'Await's, we+-- have a source-yielded value to provide it. This, of course,+-- involves eagerly running the source, percolating its 'Await's up+-- the chain as soon as possible.+racers :: forall m k a b. MonadBaseControl IO m+ => MachineT m k a -> ProcessT m a b -> MachineT m k b+racers src snk = MachineT . join $+ go <$> (Just <$> asyncRun src) <*> asyncRun snk+ where go :: Maybe (AsyncStep m k a)+ -> AsyncStep m (Is a) b+ -> m (MachineStep m k b)+ go srcA snkA =+ waitEither' srcA snkA >>= \n -> case n of+ Left (Stop :: MachineStep m k a) -> go Nothing snkA+ Left (Yield o k) -> wait snkA >>= \m -> case m of+ (Stop :: MachineStep m (Is a) b) -> return Stop+ Yield o' k' -> return . Yield o' . MachineT . flushDown k' $+ \f -> join $ go <$> (Just <$> asyncRun k)+ <*> asyncRun (f o)+ Await f Refl _ -> join $ go <$> (Just <$> asyncRun k)+ <*> asyncRun (f o)+ Left (Await g kg fg) -> asyncAwait g kg fg $+ MachineT . flip go snkA . Just+ Right (Stop :: MachineStep m (Is a) b) -> return Stop+ Right (Yield o k) -> asyncRun k >>=+ return . Yield o . MachineT . go srcA+ Right (Await f Refl ff) -> case srcA of+ Nothing -> asyncRun ff >>= go Nothing+ Just src' -> wait src' >>= \m -> case m of+ Stop -> return Stop+ Yield o k -> join $ go <$> (Just <$> asyncRun k)+ <*> asyncRun (f o)+ a -> feedUp (encased a) $ \o k -> join $+ go <$> (Just <$> asyncRun k) <*> asyncRun (f o)+ -- If we have an upstream source value ready, we must flush+ -- all available values yielded by downstream until it awaits.+ flushDown :: ProcessT m a b+ -> ((a -> ProcessT m a b) -> m (MachineStep m k b))+ -> m (MachineStep m k b)+ flushDown m k = runMachineT m >>= \s -> case s of+ Stop -> return Stop+ Yield o m' -> return . Yield o . MachineT $ flushDown m' k+ Await f Refl _ -> k f+ -- If downstream is awaiting an input, we must pull in all+ -- necessary upstream awaits until we have a yielded value to+ -- push downstream.+ feedUp :: MachineT m k a+ -> (a -> MachineT m k a -> m (MachineStep m k b))+ -> m (MachineStep m k b)+ feedUp m k = runMachineT m >>= \s -> case s of+ Stop -> return Stop+ Yield o m' -> k o m'+ Await g kg fg -> return $ awaitStep g kg fg (MachineT . flip feedUp k)
+ src/Data/Machine/Concurrent/AsyncStep.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE FlexibleContexts, GADTs, RankNTypes, ScopedTypeVariables #-}+-- | Internal helpers for taking asynchronous machine steps.+module Data.Machine.Concurrent.AsyncStep where+import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.Machine++-- | Slightly more compact notation for a 'Step'.+type MachineStep m k o = Step k o (MachineT m k o)++-- | Compact notation for a 'Step' taken asynchronously.+type AsyncStep m k o = Async (StM m (MachineStep m k o))++-- | Build an 'Await' step given a continuation that provides+-- subsequent steps. @awaitStep f sel ff k@ is like applying the+-- 'Await' constructor directly, but the continuation @k@ is used to+-- continue the machine. +-- +-- @awaitStep f sel ff k = Await (k . f) sel (k ff)@+awaitStep :: (a -> d) -> k' a -> d -> (d -> r) -> Step k' b r+awaitStep f sel ff k = Await (k . f) sel (k ff)++-- | Run one step of a machine as an 'Async' operation.+asyncRun :: MonadBaseControl IO m => MachineT m k o -> m (AsyncStep m k o)+asyncRun = async . runMachineT++-- | Satisfy a downstream Await by blocking on an upstream step.+stepAsync :: forall m k k' a' d b.+ MonadBaseControl IO m+ => (forall c. k c -> k' c)+ -> AsyncStep m k a'+ -> (a' -> d)+ -> d+ -> d+ -> (AsyncStep m k a' -> d -> MachineT m k' b)+ -> MachineT m k' b+stepAsync sel src f def prev go = MachineT $ wait src >>= \u -> case u of+ Stop -> go' stopped def+ Yield a k -> go' k (f a)+ Await g kg fg -> return $ awaitStep g (sel kg) fg (MachineT . flip go' prev)+ where go' :: MachineT m k a' -> d -> m (MachineStep m k' b)+ go' k d = asyncRun k >>= runMachineT . flip go d++-- | @asyncEncased f x@ launches @x@ and provides the resulting+-- 'AsyncStep' to @f@. Turn a function on 'AsyncStep' to a funciton on+-- 'MachineT'.+asyncEncased :: MonadBaseControl IO m+ => (AsyncStep m k1 o1 -> MachineT m k o)+ -> MachineT m k1 o1+ -> MachineT m k o+asyncEncased f x = MachineT $ asyncRun x >>= runMachineT . f++-- | Similar to 'awaitStep', but for continuations that want their inputs+-- to be run asynchronously.+asyncAwait :: MonadBaseControl IO m+ => (a -> MachineT m k o)+ -> k' a+ -> MachineT m k o+ -> (AsyncStep m k o -> MachineT m k1 o1)+ -> m (Step k' b (MachineT m k1 o1))+asyncAwait f sel ff = return . awaitStep f sel ff . asyncEncased
+ src/Data/Machine/Concurrent/Buffer.hs view
@@ -0,0 +1,136 @@+ {-# LANGUAGE CPP, FlexibleContexts, GADTs, ScopedTypeVariables, TupleSections #-}+-- | Place buffers between two machines. This is most useful with+-- irregular production rates.+module Data.Machine.Concurrent.Buffer (+ -- * Blocking buffers+ bufferConnect,+ -- * Non-blocking (rolling) buffers+ rollingConnect,+ -- * Internal helpers+ mediatedConnect, BufferRoom(..)+ ) where+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Control.Applicative ((<$>), (<*>))+#endif+import Control.Concurrent.Async.Lifted (wait, waitEither)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad (join, (>=>))+import Data.Machine.Concurrent.AsyncStep+import Data.Machine+import Data.Sequence (ViewL(..), (|>))+import qualified Data.Sequence as S+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Data.Traversable (traverse)+#endif++-- | Drain downstream until it awaits a value, then pass the awaiting+-- step to the given function.+drain :: Monad m+ => MachineStep m k a+ -> (MachineStep m k a -> m (MachineStep m k' a))+ -> m (MachineStep m k' a)+drain z k = go z+ where go Stop = return Stop+ go (Yield o kd) = Yield o . MachineT . go <$> runMachineT kd+ go aStep = k aStep++-- | Feed upstream until it yields a value, then pass the yielded+-- value and next step to the given function.+feedToBursting :: Monad m+ => MachineStep m k a+ -> (Maybe (a, MachineT m k a) -> m (MachineStep m k b))+ -> m (MachineStep m k b)+feedToBursting z k = go z+ where go Stop = k Nothing+ go (Await f kf ff) = return $+ Await (\a -> go' (f a)) kf (go' ff)+ go (Yield o kk) = k $ Just (o, kk)+ go' step = MachineT $ runMachineT step >>= go++-- | Mediate a 'MachineT' and a 'ProcessT' with a bounded capacity+-- buffer. The source machine runs concurrently with the sink process,+-- and is only blocked when the buffer is full.+bufferConnect :: MonadBaseControl IO m+ => Int -> MachineT m k b -> ProcessT m b c -> MachineT m k c+bufferConnect n = mediatedConnect S.empty snoc view+ where snoc acc x = (if S.length acc < n - 1 then Vacancy else NoVacancy) $+ acc |> x+ view acc = case S.viewl acc of+ EmptyL -> Nothing+ x :< acc' -> Just (x, acc')++-- | Mediate a 'MachineT' and a 'ProcessT' with a rolling buffer. The+-- source machine runs concurrently with the sink process and is never+-- blocked. If the sink process can not keep up with upstream, yielded+-- values will be dropped.+rollingConnect :: MonadBaseControl IO m+ => Int -> MachineT m k b -> ProcessT m b c -> MachineT m k c+rollingConnect n = mediatedConnect S.empty snoc view+ where snoc acc x = Vacancy $ S.take (n-1) acc |> x+ view acc = case S.viewl acc of+ EmptyL -> Nothing+ x :< acc' -> Just (x, acc')++-- | Indication if the payload value is "full" or not.+data BufferRoom a = NoVacancy a | Vacancy a deriving (Eq, Ord, Show)++-- | Mediate a 'MachineT' and a 'ProcessT' with a buffer.+--+-- @mediatedConnect z snoc view source sink@ pipes @source@ into+-- @sink@ through a buffer initialized to @z@ and updated with+-- @snoc@. Upstream is blocked if @snoc@ indicates that the buffer is+-- full after adding a new element. Downstream blocks if @view@+-- indicates that the buffer is empty. Otherwise, @view@ is expected+-- to return the next element to process and an updated buffer.+mediatedConnect :: forall m t b k c. MonadBaseControl IO m+ => t -> (t -> b -> BufferRoom t) -> (t -> Maybe (b,t))+ -> MachineT m k b -> ProcessT m b c -> MachineT m k c+mediatedConnect z snoc view src0 snk0 =+ MachineT $ do srcFuture <- asyncRun src0+ snkFuture <- asyncRun snk0+ go z (Just srcFuture) snkFuture+ where -- Wait for the next available step+ go :: t+ -> Maybe (AsyncStep m k b)+ -> AsyncStep m (Is b) c+ -> m (MachineStep m k c)+ go acc src snk = maybe (Left <$> wait snk) (waitEither snk) src >>=+ goStep acc . either (Right . (,src)) (Left . (,snk))++ -- Kick off the next step of both the source and the sink+ goAsync :: t+ -> Maybe (MachineT m k b)+ -> ProcessT m b c+ -> m (MachineStep m k c)+ goAsync acc src snk =+ join $ go acc <$> traverse asyncRun src <*> asyncRun snk++ -- Handle whichever step is ready first+ goStep :: t -> Either (MachineStep m k b, AsyncStep m (Is b) c)+ (MachineStep m (Is b) c, Maybe (AsyncStep m k b))+ -> m (MachineStep m k c)+ goStep acc step = case step of+ -- @src@ stepped first+ Left (Stop, snk) -> go acc Nothing snk+ Left (Await g kg fg, snk) ->+ asyncAwait g kg fg (MachineT . flip (go acc) snk . Just)+ Left (Yield o k, snk) -> case snoc acc o of+ -- add it to the right end of the buffer+ Vacancy acc' -> asyncRun k >>= flip (go acc') snk . Just+ -- buffer was full+ NoVacancy acc' ->+ let go' snk' = do src' <- asyncRun k+ goStep acc' (Right (snk', Just src'))+ in wait snk >>= flip drain go'++ -- @snk@ stepped first+ Right (Stop, _) -> return Stop+ Right (Yield o k, src) -> do+ return $ Yield o (MachineT $ asyncRun k >>= go acc src)+ Right (Await f Refl ff, src) ->+ case view acc of+ Nothing -> maybe (goAsync acc Nothing ff) (wait >=> demandSrc) src+ Just (x, acc') -> asyncRun (f x) >>= go acc' src+ where demandSrc = flip feedToBursting go'+ go' Nothing = goAsync acc Nothing ff+ go' (Just (o, k)) = goAsync acc (Just k) (f o)
+ src/Data/Machine/Concurrent/Fanout.hs view
@@ -0,0 +1,85 @@+{-# LANGUAGE CPP, FlexibleContexts, GADTs, ScopedTypeVariables #-}+-- | Provide a notion of fanout wherein a single input is passed to+-- several consumers. The consumers are run concurrently.+module Data.Machine.Concurrent.Fanout (fanout, fanoutSteps) where+import Control.Arrow (first, second)+import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Monad (foldM)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.Machine (Step(..), MachineT(..), encased, stopped, ProcessT, Is(..))+import Data.Machine.Concurrent.AsyncStep (MachineStep)+import Data.Maybe (catMaybes)+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Data.Monoid (Monoid, mempty, mconcat)+#endif+import Data.Semigroup (Semigroup(sconcat))+import Data.List.NonEmpty (NonEmpty((:|)))+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Data.Coerce (coerce)+#endif++-- | Feed a value to a 'ProcessT' at an 'Await' 'Step'. If the+-- 'ProcessT' is awaiting a value, then its next step is+-- returned. Otherwise, the original process is returned.+feed :: forall m a b. MonadBaseControl IO m+ => a -> ProcessT m a b+ -> m (Async (StM m (MachineStep m (Is a) b)))+feed x m = async $ runMachineT m >>= \(v :: MachineStep m (Is a) b) ->+ case v of+ Await f Refl _ -> runMachineT (f x)+ s -> return s++-- | Like 'Data.List.mapAccumL' but with a monadic accumulating+-- function.+mapAccumLM :: MonadBaseControl IO m+ => (acc -> x -> m (acc, y)) -> acc -> [Async (StM m x)]+ -> m (acc, [y])+mapAccumLM f z = fmap (second ($ [])) . foldM aux (z,id)+ where aux (acc,ys) x = do (yielded, nxt) <- wait x >>= f acc+ return $ (yielded, (nxt:) . ys)++-- | Exhaust a sequence of all successive 'Yield' steps taken by a+-- 'MachineT'. Returns the list of yielded values and the next+-- (non-Yield) step of the machine.+flushYields :: Monad m+ => Step k o (MachineT m k o) -> m ([o], Maybe (MachineT m k o))+flushYields = go id+ where go rs (Yield o s) = runMachineT s >>= go ((o:) . rs)+ go rs Stop = return (rs [], Nothing)+ go rs s = return (rs [], Just $ encased s)++-- | Share inputs with each of a list of processes in lockstep. Any+-- values yielded by the processes for a given input are combined into+-- a single yield from the composite process.+fanout :: (MonadBaseControl IO m, Semigroup r)+ => [ProcessT m a r] -> ProcessT m a r+fanout [] = stopped+fanout xs = encased $ Await (MachineT . aux) Refl (fanout xs)+ where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+ let nxt = fanout $ catMaybes xs'+ case rs of+ [] -> runMachineT nxt+ (r:rs') -> return $ Yield (sconcat $ r :| rs') nxt+ yields rs Stop = return (rs,Nothing)+ yields rs y@Yield{} = first (++ rs) <$> flushYields y+ yields rs a@Await{} = return (rs, Just $ encased a)++-- | Share inputs with each of a list of processes in lockstep. If+-- none of the processes yields a value, the composite process will+-- itself yield 'mempty'. The idea is to provide a handle on steps+-- only executed for their side effects. For instance, if you want to+-- run a collection of 'ProcessT's that await but don't yield some+-- number of times, you can use 'fanOutSteps . map (fmap (const ()))'+-- followed by a 'taking' process.+fanoutSteps :: (MonadBaseControl IO m, Monoid r)+ => [ProcessT m a r] -> ProcessT m a r+fanoutSteps [] = stopped+fanoutSteps xs = encased $ Await (MachineT . aux) Refl (fanoutSteps xs)+ where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+ let nxt = fanoutSteps $ catMaybes xs'+ if null rs+ then return $ Yield mempty nxt+ else return $ Yield (mconcat rs) nxt+ yields rs Stop = return (rs,Nothing)+ yields rs y@Yield{} = first (++rs) <$> flushYields y+ yields rs a@Await{} = return (rs, Just $ encased a)
+ src/Data/Machine/Concurrent/Scatter.hs view
@@ -0,0 +1,237 @@+{-# LANGUAGE FlexibleContexts, GADTs, TupleSections, RankNTypes,+ ScopedTypeVariables #-}+-- | Routing for splitting and merging processing pipelines.+module Data.Machine.Concurrent.Scatter (+ scatter, mergeSum, splitSum, splitProd+ ) where+import Control.Arrow ((***))+import Control.Concurrent.Async (Async, waitAny)+import Control.Concurrent.Async.Lifted (wait, waitEither, waitBoth)+import Control.Monad ((>=>))+import Control.Monad.Base (liftBase)+import Control.Monad.Trans.Control (MonadBaseControl, restoreM, StM)+import Data.Machine+import Data.Machine.Concurrent.AsyncStep++holes :: [a] -> [[a]]+holes = go id+ where go _ [] = []+ go x (y:ys) = x ys : go (x . (y:)) ys++diff :: [a] -> [(a,[a])]+diff xs = zip xs (holes xs)++waitAnyHole :: MonadBaseControl IO m => [(Async (StM m a), [b])] -> m (a, [b])+waitAnyHole xs = do (_,(s,b)) <- liftBase $ waitAny xs'+ fmap (,b) (restoreM s)+ where xs' = map (\(a,b) -> fmap (,b) a) xs++-- | Produces values from whichever source 'MachineT' yields+-- first. This operation may also be viewed as a /gather/ operation in+-- that all values produced by the given machines are interleaved when+-- fed downstream. Note that inputs are /not/ shared. The composite+-- machine will await an input when any constituent machine awaits an+-- input. That input will be supplied to the awaiting constituent and+-- no other.+--+-- Some examples of more specific useful types @scatter@ may be used+-- at,+-- +-- @+-- scatter :: [ProcessT m a b] -> ProcessT m a b+-- scatter :: [SourceT m a] -> SourceT m a+-- @+--+-- The former may be used to stream data through a collection of+-- worker 'Process'es, the latter may be used to intersperse values+-- from a collection of sources.+scatter :: MonadBaseControl IO m => [MachineT m k o] -> MachineT m k o+scatter [] = stopped+scatter sinks = MachineT $ mapM asyncRun sinks+ >>= waitAnyHole . diff+ >>= uncurry go+ where go :: MonadBaseControl IO m+ => MachineStep m k o+ -> [AsyncStep m k o]+ -> m (MachineStep m k o)+ go Stop [] = return Stop+ go Stop sinks' = waitAnyHole (diff sinks') >>= uncurry go+ go (Yield o k) sinks' = + asyncRun k >>= return . Yield o . MachineT . goWait . (:sinks')+ go (Await f fk ff) sinks' =+ asyncAwait f fk ff (MachineT . goWait . (:sinks'))+ goWait :: MonadBaseControl IO m+ => [AsyncStep m k o]+ -> m (MachineStep m k o)+ goWait = waitAnyHole . diff >=> uncurry go++-- | Similar to 'Control.Arrow.|||': split the input between two+-- processes and merge their outputs.+--+-- Connect two processes to the downstream tails of a 'Machine' that+-- produces 'Either's. The two downstream consumers are run+-- concurrently when possible. When one downstream consumer stops, the+-- other is allowed to run until it stops or the upstream source+-- yields a value the remaining consumer can not handle.+--+-- @mergeSum sinkL sinkR@ produces a topology like this,+--+-- @+-- sinkL+-- / \\+-- a \\+-- / \\+-- source -- Either a b --> -- r -->+-- \\ /+-- b /+-- \\ /+-- sinkR +-- @+mergeSum :: forall m a b r. MonadBaseControl IO m+ => ProcessT m a r -> ProcessT m b r -> ProcessT m (Either a b) r+mergeSum snkL snkR = MachineT $ do sl <- asyncRun snkL+ sr <- asyncRun snkR+ go sl sr+ where go :: AsyncStep m (Is a) r+ -> AsyncStep m (Is b) r+ -> m (MachineStep m (Is (Either a b)) r)+ go sl sr = waitEither sl sr >>= + \(s :: Either (MachineStep m (Is a) r)+ (MachineStep m (Is b) r)) -> case s of+ Left Stop -> wait sr >>= runMachineT . rightOnly . encased+ Right Stop -> wait sl >>= runMachineT . leftOnly . encased++ Left (Yield o k) -> + return . Yield o . MachineT $ asyncRun k >>= flip go sr+ Right (Yield o k) -> + return . Yield o . MachineT $ asyncRun k >>= go sl+ + Left (Await f Refl ff) ->+ return $ + Await (\u -> case u of+ Left a -> MachineT $ asyncRun (f a) >>= flip go sr+ Right b -> MachineT $ + wait sr >>= forceFeed (go sl) b . encased)+ Refl+ (MachineT $ asyncRun ff >>= flip go sr)+ Right (Await g Refl gg) -> return $+ Await (\u -> case u of+ Left a -> + MachineT $+ wait sl >>= forceFeed (flip go sr) a . encased+ Right b -> MachineT $ asyncRun (g b) >>= go sl)+ Refl+ (MachineT $ asyncRun gg >>= go sl)++-- | Similar to 'Control.Arrow.+++': split the input between two+-- processes, retagging and merging their outputs.+--+-- The two processes are run concurrently whenever possible.+splitSum :: forall m a b c d. MonadBaseControl IO m+ => ProcessT m a b -> ProcessT m c d -> ProcessT m (Either a c) (Either b d)+splitSum snkL snkR = MachineT $ do sl <- asyncRun (fmap lft snkL)+ sr <- asyncRun (fmap rgt snkR)+ go sl sr+ where lft :: b -> Either b d+ lft = Left+ rgt :: d -> Either b d+ rgt = Right+ go :: AsyncStep m (Is a) (Either b d)+ -> AsyncStep m (Is c) (Either b d)+ -> m (MachineStep m (Is (Either a c)) (Either b d))+ go sl sr = waitEither sl sr >>=+ \(s :: Either (MachineStep m (Is a) (Either b d))+ (MachineStep m (Is c) (Either b d))) -> case s of+ Left Stop -> wait sr >>= runMachineT . rightOnly . encased+ Right Stop -> wait sl >>= runMachineT . leftOnly . encased++ Left (Yield o k) -> + return . Yield o . MachineT $ asyncRun k >>= flip go sr+ Right (Yield o k) -> + return . Yield o . MachineT $ asyncRun k >>= go sl+ + Left (Await f Refl ff) ->+ return $ + Await (\u -> case u of+ Left a -> MachineT $ asyncRun (f a) >>= flip go sr+ Right b -> MachineT $ + wait sr >>= forceFeed (go sl) b . encased)+ Refl+ (MachineT $ asyncRun ff >>= flip go sr)+ Right (Await g Refl gg) -> return $+ Await (\u -> case u of+ Left a -> + MachineT $+ wait sl >>= forceFeed (flip go sr) a . encased+ Right b -> MachineT $ asyncRun (g b) >>= go sl)+ Refl+ (MachineT $ asyncRun gg >>= go sl)++-- | @forceFeed k x p@ runs machine @p@ until it awaits, at which+-- point it is fed @x@. The result of that feeding is asynchronously+-- run, and supplied to the continuation @k@.+forceFeed :: forall m a k b. MonadBaseControl IO m+ => (AsyncStep m (Is a) b -> m (MachineStep m k b))+ -> a+ -> ProcessT m a b+ -> m (MachineStep m k b)+forceFeed go x = aux+ where aux p = runMachineT p >>= \v -> case v of+ -- Stop -> asyncRun stopped >>= go+ Stop -> return Stop+ Yield o k -> return . Yield o . MachineT $ aux k+ Await f Refl _ -> asyncRun (f x) >>= go++-- | We have a sink for the Right output of a source, so we want to+-- keep running it as long as upstream does not yield a 'Left' which+-- we can not handle. When upstream yields a 'Left', we 'stop'.+rightOnly :: Monad m => ProcessT m b r -> ProcessT m (Either a b) r+rightOnly snk = repeatedly (await >>= either (const stop) yield) ~> snk++-- | We have a sink for the Left output of a source, so we want to+-- keep running it as long as upstream does not yield a 'Right' which+-- we can not handle. When upstream yields a 'Right', we 'stop'.+leftOnly :: Monad m => ProcessT m a r -> ProcessT m (Either a b) r+leftOnly snk = repeatedly (await >>= either yield (const stop)) ~> snk++-- | Connect two processes to the downstream tails of a 'Machine' that+-- produces tuples. The two downstream consumers are run+-- concurrently. When one downstream consumer stops, the entire+-- pipeline is stopped.+--+-- @splitProd sink1 sink2@ produces a topology like this,+--+-- @+-- sink1+-- / \\+-- a \\+-- / \\+-- source -- (a,b) --> -- r -->+-- \\ /+-- b /+-- \\ /+-- sink2 +-- @+splitProd :: forall m a b r. MonadBaseControl IO m+ => ProcessT m a r -> ProcessT m b r -> ProcessT m (a,b) r+splitProd snk1 snk2 = MachineT $ do s1 <- asyncRun snk1+ s2 <- asyncRun snk2+ go s1 s2+ where go :: AsyncStep m (Is a) r+ -> AsyncStep m (Is b) r+ -> m (MachineStep m (Is (a,b)) r)+ go s1 s2 = waitBoth s1 s2 >>= + \(ss :: (MachineStep m (Is a) r, MachineStep m (Is b) r)) -> case ss of+ (Stop, _) -> return Stop+ (_, Stop) -> return Stop+ (Yield o1 k1, Yield o2 k2) -> + return . Yield o1 . encased $ Yield o2 $ MachineT $+ do k1' <- asyncRun k1+ k2' <- asyncRun k2+ go k1' k2'+ (Yield o k, _) ->+ return . Yield o . MachineT $ asyncRun k >>= flip go s2+ (_, Yield o k) ->+ return . Yield o . MachineT $ asyncRun k >>= go s1+ (Await f Refl ff, Await g Refl gg) ->+ return $ Await (uncurry splitProd . (f***g)) Refl (splitProd ff gg)
+ src/Data/Machine/Concurrent/Tee.hs view
@@ -0,0 +1,35 @@+{-# LANGUAGE FlexibleContexts, GADTs, ScopedTypeVariables #-}+-- | Support for machines with two inputs from which input may be+-- drawn deterministically. In contrast to "Data.Machine.Tee", the two+-- inputs are eagerly run concurrently in this implementation.+module Data.Machine.Concurrent.Tee where+import Control.Concurrent.Async.Lifted (wait)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Machine+import Data.Machine.Concurrent.AsyncStep++-- | Compose a pair of pipes onto the front of a Tee.+tee :: forall m a a' b b' c. MonadBaseControl IO m+ => ProcessT m a a' -> ProcessT m b b' -> TeeT m a' b' c -> TeeT m a b c+tee ma mb m = MachineT $ do srcL <- asyncRun ma+ srcR <- asyncRun mb+ go m (Just srcL) (Just srcR)+ where go :: TeeT m a' b' c+ -> Maybe (AsyncStep m (Is a) a')+ -> Maybe (AsyncStep m (Is b) b')+ -> m (MachineStep m (T a b) c)+ go snk srcL srcR = runMachineT snk >>= \v -> case v of+ Stop -> return Stop+ Yield o k -> return . Yield o . MachineT $ go k srcL srcR+ Await f L ff -> maybe (return Stop) wait srcL >>= + \(u :: MachineStep m (Is a) a') -> case u of+ Stop -> go ff Nothing srcR+ Yield a k -> asyncRun k >>= flip (go (f a)) srcR . Just+ Await g Refl fg -> + asyncAwait g L fg $ MachineT . flip (go (encased v)) srcR . Just+ Await f R ff -> maybe (return Stop) wait srcR >>= + \(u :: MachineStep m (Is b) b') -> case u of+ Stop -> go ff srcL Nothing+ Yield b k -> asyncRun k >>= go (f b) srcL . Just+ Await g Refl fg -> + asyncAwait g R fg $ MachineT . go (encased v) srcL . Just
+ src/Data/Machine/Concurrent/Wye.hs view
@@ -0,0 +1,113 @@+{-# LANGUAGE CPP, GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables,+ TupleSections #-}+-- | Support for machines with two inputs from which input may be+-- drawn deterministically or non-deterministically. In contrast to+-- "Data.Machine.Wye", the two inputs are eagerly run concurrently in+-- this implementation.+module Data.Machine.Concurrent.Wye (wye) where+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Control.Applicative+#endif+import Control.Concurrent.Async.Lifted (wait, waitEither)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Machine hiding (wye, (~>), (<~))+import Data.Machine.Concurrent.AsyncStep++isX :: Is a c -> Y a b c+isX Refl = X++isY :: Is b c -> Y a b c+isY Refl = Y++-- | Only the 'X' input of a 'Wye' is not yet stopped, so we may employ+-- simpler dispatch logic.+wyeOnlyX :: forall a a' b b' c m. MonadBaseControl IO m+ => AsyncStep m (Is a) a' -> WyeT m a' b' c -> WyeT m a b c+wyeOnlyX src snk = MachineT $ runMachineT snk >>= \v -> case v of+ Stop -> return Stop+ Yield o k -> return $ Yield o (wyeOnlyX src k)+ Await _ Y ff -> runMachineT $ wye stopped stopped ff+ Await f X ff -> runMachineT $ stepAsync isX src f ff (encased v) wyeOnlyX+ Await f Z ff -> runMachineT $ + stepAsync isX src (f . Left) ff (encased v) wyeOnlyX++-- | Only the 'Y' input of a 'Wye' is not yet stopped, so we may+-- employ simpler dispatch logic.+wyeOnlyY :: MonadBaseControl IO m+ => AsyncStep m (Is b) b' -> WyeT m a' b' c -> WyeT m a b c+wyeOnlyY src m = MachineT $ runMachineT m >>= \v -> case v of+ Stop -> return Stop+ Yield o k -> return $ Yield o (wyeOnlyY src k)+ Await _ X ff -> runMachineT $ wye stopped stopped ff+ Await f Y ff -> runMachineT $ stepAsync isY src f ff (encased v) wyeOnlyY+ Await f Z ff -> + runMachineT $ stepAsync isY src (f . Right) ff (encased v) wyeOnlyY++-- | Precompose a 'Process' onto each input of a 'Wye' (or 'WyeT').+--+-- When the choice of input is free (using the 'Z' input descriptor)+-- the two sources will be interleaved.+wye :: forall m a a' b b' c.+ (MonadBaseControl IO m)+ => ProcessT m a a' -> ProcessT m b b' -> WyeT m a' b' c -> WyeT m a b c+wye ma mb m = MachineT $ do srcL <- asyncRun ma+ srcR <- asyncRun mb+ go True m srcL srcR+ where go :: Bool+ -> WyeT m a' b' c+ -> AsyncStep m (Is a) a'+ -> AsyncStep m (Is b) b'+ -> m (MachineStep m (Y a b) c)+ go fair snk srcL srcR = runMachineT snk >>= \v -> case v of+ Stop -> return Stop+ Yield o k -> return . Yield o . MachineT $ go fair k srcL srcR+ Await f X ff -> wait srcL >>=+ \(u :: MachineStep m (Is a) a') -> case u of+ Stop -> runMachineT $ wyeOnlyY srcR ff+ Yield a k -> asyncRun k >>= flip (go fair (f a)) srcR+ Await g Refl fg -> + asyncAwait g X fg $ MachineT . flip (go fair (encased v)) srcR+ Await f Y ff -> wait srcR >>=+ \(u :: MachineStep m (Is b) b') -> case u of+ Stop -> runMachineT $ wyeOnlyX srcL ff+ Yield b k -> asyncRun k >>= go fair (f b) srcL+ Await h Refl fh -> + asyncAwait h Y fh $ MachineT . go fair (encased v) srcL++ -- Wait for whoever yields first+ Await f Z _ -> + waitFair fair srcL srcR+ >>= \(u :: Either (MachineStep m (Is a) a')+ (MachineStep m (Is b) b')) -> case u of+ Left (Yield a k) -> + asyncRun k >>= \srcL' -> go (not fair) (f $ Left a) srcL' srcR+ Right (Yield b k) -> + asyncRun k >>= \srcR' -> go (not fair) (f $ Right b) srcL srcR'+ Left Stop -> runMachineT $ wyeOnlyY srcR (encased v)+ Right Stop -> runMachineT $ wyeOnlyX srcL (encased v)++ -- The first source to respond wants to await, see what+ -- the other source has to offer.+ Left la@(Await g Refl fg) -> + wait srcR >>= \(w :: MachineStep m (Is b) b') -> case w of+ Stop -> asyncAwait g X fg $ \l' -> wyeOnlyX l' (encased v)+ Yield b k -> runMachineT $ wye (encased la) k (f $ Right b)+ ra@(Await h Refl fh) -> return $+ Await (\c -> case c of+ Left a -> wye (g a) (encased ra) (encased v)+ Right b -> wye (encased la) (h b) (encased v))+ Z+ (wye fg fh $ encased v)+ Right ra@(Await h Refl fh) -> + wait srcL >>= \(w :: MachineStep m (Is a) a') -> case w of+ Stop -> asyncAwait h Y fh $ \r' -> wyeOnlyY r' (encased v)+ Yield a k -> runMachineT $ wye k (encased ra) (f $ Left a)+ la@(Await g Refl fg) -> return $+ Await (\c -> case c of+ Left a -> wye (g a) (encased ra) (encased v)+ Right b -> wye (encased la) (h b) (encased v))+ Z+ (wye fg fh $ encased v)+ where waitFair True l r = waitEither l r+ waitFair False l r = either Right Left <$> waitEither r l+
+ src/Data/Machine/Regulated.hs view
@@ -0,0 +1,22 @@+-- | Slow producers down to run at desired rates.+module Data.Machine.Regulated where+import Control.Concurrent (threadDelay)+import Control.Monad (when)+import Control.Monad.IO.Class (MonadIO(..))+import Data.Machine.Plan+import Data.Machine.Process+import Data.Machine.Type+import Data.Time.Clock (getCurrentTime, diffUTCTime)++-- | A pass-through process rate-limited to the given inter-step+-- period in seconds. This may be used to slow down an upstream+-- producer; it can not speed things up.+regulated :: MonadIO m => Double -> ProcessT m a a+regulated target = construct $ liftIO getCurrentTime >>= go 0+ where go dt prevT =+ do await >>= yield+ t <- liftIO getCurrentTime+ let e = target - realToFrac (diffUTCTime t prevT)+ dt' = dt + 0.5 * e+ when (dt' > 0) (liftIO . threadDelay . round $ dt' * 1000000)+ go dt' t
+ tests/AllTests.hs view
@@ -0,0 +1,51 @@+import Data.Time.Clock (getCurrentTime, diffUTCTime)+import Control.Concurrent (threadDelay)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Machine.Concurrent+import Test.Tasty+import Test.Tasty.HUnit++worker :: String -> Double -> ProcessT IO () ()+worker _name dt = repeatedly $ do _ <- await+ liftIO $ threadDelay dt'+ yield ()+ where dt' = floor $ dt * 10000++timed :: MonadIO m => m a -> m (a, Double)+timed m = do t1 <- liftIO getCurrentTime+ r <- m+ t2 <- liftIO getCurrentTime+ return (r, realToFrac $ t2 `diffUTCTime` t1)++pipeline :: TestTree+pipeline = testCaseSteps "pipeline" $ \step -> do+ (r,dt) <- timed . runT . supply (repeat ()) $+ worker "A" 1 ~> worker "B" 1 ~> worker "C" 1 ~> taking 10+ (r',dt') <- timed . runT . supply (repeat ()) $+ worker "A" 1 >~> worker "B" 1 >~> worker "C" 1 >~> taking 10+ step "Consistent results"+ assertEqual "Results" r r'+ step "Parallelism"+ assertBool ("Pipeline faster than sequential" ++ show (dt',dt)) (dt' * 1.5 < dt)++workStealing :: TestTree+workStealing = testCaseSteps "work stealing" $ \step -> do+ (r,dt) <- timed . runT $+ source [1..32::Int] ~> scatter (replicate 4 slowDoubler)+ (r',dt') <- timed. runT $ source [1..32] ~> slowDoubler+ step "Consistent results"+ assertBool "Predicted Parallel Length" (length r == 32)+ assertBool "Predicted Serial Length" (length r' == 32)+ assertBool "Predicted Results" (all (`elem` r) (map (*2) [1..32]))+ assertBool "Results" (all (`elem` r) r')+ step "Parallelism"+ assertBool ("Work Stealing faster than sequential" ++ show (dt',dt))+ (dt * 1.5 < dt')+ where slowDoubler = repeatedly $ do x <- await+ liftIO (threadDelay 100000)+ yield (x * 2)++main :: IO ()+main = defaultMain $ + testGroup "concurrent-machines"+ [ pipeline, workStealing ]
+ tests/SPlotTests.hs view
@@ -0,0 +1,172 @@+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Exception (catch, throwIO)+import Control.Monad (when, forM_)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Writer+import Control.Monad.Trans.Class (lift)+import Data.Machine.Concurrent+import Data.Time.Clock (UTCTime, addUTCTime, diffUTCTime, getCurrentTime)+import Text.Printf (printf)+import Data.Time.Format (defaultTimeLocale, formatTime, readPTime)+import System.Exit (ExitCode (ExitSuccess), exitSuccess)+import System.IO (writeFile)+import qualified System.Process as Proc+import Text.ParserCombinators.ReadP (readP_to_S)++import qualified Test.Tasty as T+import qualified Test.Tasty.HUnit as TH++writeSPlot :: Bool+writeSPlot = True++showTime :: UTCTime -> String+showTime = formatTime defaultTimeLocale "%Y-%m-%d %H:%M:%S%Q"++worker :: (Show a, MonadIO m)+ => (a -> b) -> Int -> Double -> ProcessT (WriterT [String] m) a b+worker f i dt = repeatedly $ do+ x <- await+ t1 <- liftIO getCurrentTime+ lift $ tell [ printf "%s >%d colour%d" (showTime t1) i i+ , printf "%s !%d black %s" (showTime t1) i (show x) ]+ liftIO $ threadDelay (floor (dt * 10000))+ t2 <- liftIO getCurrentTime+ lift $ tell [printf "%s <%d" (showTime t2) i]+ yield (f x)++timed :: MonadIO m => m a -> m (a, Double)+timed m = do+ t1 <- liftIO getCurrentTime+ r <- m+ t2 <- liftIO getCurrentTime+ return (r, realToFrac $ t2 `diffUTCTime` t1)++pipeline :: T.TestTree+pipeline = TH.testCaseSteps "pipeline" $ \step -> do+ let xs = [(0::Int)..]++ ((r,dt), ls) <- runWriterT . timed . runT $+ source xs ~> worker id 1 3 ~> worker id 2 5 ~> worker id 3 10 ~> taking 10++ ((r',dt'), ls') <- runWriterT . timed . runT $+ source xs ~> worker id 1 2 >~> worker id 2 4 >~> worker id 3 8 ~> taking 10++ when writeSPlot $ do+ writeFile "pipeline-seq.splot" (unlines ls)+ writeFile "pipeline-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertEqual "Results" r r'++ step "Parallelism"+ TH.assertBool ("Pipeline faster than sequential" ++ show (dt',dt))+ (dt' * 1.5 < dt)++buffering1 :: T.TestTree+buffering1 = TH.testCaseSteps "buffering1" $ \step -> do+ let xs = [1..32::Int]++ ((r, dt), ls) <- runWriterT . timed . runT $+ source xs ~> worker (*2) 1 2 ~> worker (+1) 2 4++ ((r', dt'), ls') <- runWriterT . timed . runT $+ source xs ~> bufferConnect 5 (worker (*2) 1 2) (worker (+1) 2 4)++ when writeSPlot $ do+ writeFile "buffering1-seq.splot" (unlines ls)+ writeFile "buffering1-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertEqual "Results" r r'++ step "Parallelism"+ TH.assertBool ("Buffered pipeline faster than sequential" ++ show (dt', dt))+ (dt' * 1.4 < dt)++rolling1 :: T.TestTree+rolling1 = TH.testCaseSteps "rolling1" $ \step -> do+ let xs = [1..32::Int]++ ((r, dt), ls) <- runWriterT . timed . runT $+ source xs ~> worker (*2) 1 2 ~> worker (+1) 2 4++ ((r', dt'), ls') <- runWriterT . timed . runT $+ source xs ~> rollingConnect 5 (worker (*2) 1 2) (worker (+1) 2 4)++ when writeSPlot $ do+ writeFile "rolling1-seq.splot" (unlines ls)+ writeFile "rolling1-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertBool "Results" (all (`elem` r) r')++ step "Parallelism"+ TH.assertBool ("Rolling pipeline faster than sequential" ++ show (dt', dt))+ (dt' * 1.5 < dt)++workStealing :: T.TestTree+workStealing = TH.testCaseSteps "work stealing" $ \step -> do+ let xs = [1..32::Int]++ ((r,dt), ls) <- runWriterT . timed . runT $+ source xs ~> (worker (*2) 0 4)++ ((r',dt'), ls') <- runWriterT . timed . runT $+ source xs ~> scatter (map (\i -> worker (*2) i 4) [1..4])++ when writeSPlot $ do+ writeFile "work-stealing-seq.splot" (unlines ls)+ writeFile "work-stealing-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertBool "Predicted Serial Length" (length r == length xs)+ TH.assertBool "Predicted Parallel Length" (length r' == length xs)+ TH.assertBool "Predicted Results" (all (`elem` r') (map (*2) xs))+ TH.assertBool "Results" (all (`elem` r') r)++ step "Parallelism"+ TH.assertBool ("Work Stealing faster than sequential" ++ show (dt,dt'))+ (dt' * 1.5 < dt)++main :: IO ()+main = do+ catch+ (T.defaultMain+ (T.testGroup "concurrent-machines"+ [ pipeline, buffering1, rolling1, workStealing ]))+ (\e -> if e == ExitSuccess+ then return ()+ else throwIO e)+ when writeSPlot $ do+ let splots = [ "pipeline-seq.splot"+ , "pipeline-par.splot"+ , "buffering1-seq.splot"+ , "buffering1-par.splot"+ , "rolling1-seq.splot"+ , "rolling1-par.splot"+ , "work-stealing-seq.splot"+ , "work-stealing-par.splot"+ ]+ forM_ splots $ \path -> do+ l:_ <- lines <$> readFile path+ case readP_to_S (readPTime False defaultTimeLocale "%Y-%m-%d %H:%M:%S%Q") l of+ [] ->+ fail "Shit no."+ (fromTime,_):_ -> do+ let toTime = addUTCTime 2.5 fromTime+ let args = [ "-if", path+ , "-o", path ++ ".png"+ , "-w", "2048"+ , "-h", "200"+ , "-bh", "1"+ , "-tickInterval", "100"+ , "-legendWidth", "20"+ , "-numTracks", "4"+ , "-fromTime", showTime fromTime+ , "-toTime", showTime toTime+ ]+ Proc.readProcess "splot" args "" >>= putStr+ putStrLn $ "OK: " ++ show (fromTime :: UTCTime) ++ " " ++ path+ exitSuccess