diff --git a/concurrent-machines.cabal b/concurrent-machines.cabal --- a/concurrent-machines.cabal +++ b/concurrent-machines.cabal @@ -1,5 +1,5 @@ name: concurrent-machines -version: 0.2.0 +version: 0.2.1 synopsis: Concurrent networked stream transducers description: A simple use-case for this library is to run the stages @@ -55,11 +55,12 @@ containers >= 0.5 && < 0.6, transformers-base >= 0.4 && < 0.5, machines >= 0.5 && < 0.6, - async >= 2.0.1 && < 2.1, - lifted-async >= 0.1 && < 0.8, + async >= 2.0.1 && < 2.2, + lifted-async >= 0.1 && < 0.9, semigroups >= 0.8 && < 0.19 hs-source-dirs: src default-language: Haskell2010 + ghc-options: -Wall test-suite tests type: exitcode-stdio-1.0 diff --git a/src/Data/Machine/Concurrent.hs b/src/Data/Machine/Concurrent.hs --- a/src/Data/Machine/Concurrent.hs +++ b/src/Data/Machine/Concurrent.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables, +{-# 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 @@ -21,9 +21,11 @@ -- * Concurrent processing of shared inputs fanout, fanoutSteps, -- * Concurrent multiple-input machines - wye, tee, scatter, splitSum, mergeSum, + 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 diff --git a/src/Data/Machine/Concurrent/Buffer.hs b/src/Data/Machine/Concurrent/Buffer.hs --- a/src/Data/Machine/Concurrent/Buffer.hs +++ b/src/Data/Machine/Concurrent/Buffer.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE FlexibleContexts, GADTs, ScopedTypeVariables, TupleSections #-} +{-# LANGUAGE CPP, FlexibleContexts, GADTs, ScopedTypeVariables, TupleSections #-} -- | Place buffers between two machines. This is most useful with -- irregular production rates. module Data.Machine.Concurrent.Buffer ( @@ -9,7 +9,9 @@ -- * 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, (>=>)) @@ -17,7 +19,9 @@ 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. diff --git a/src/Data/Machine/Concurrent/Fanout.hs b/src/Data/Machine/Concurrent/Fanout.hs --- a/src/Data/Machine/Concurrent/Fanout.hs +++ b/src/Data/Machine/Concurrent/Fanout.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE FlexibleContexts, GADTs, ScopedTypeVariables #-} +{-# 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 @@ -9,10 +9,14 @@ import Data.Machine (Step(..), MachineT(..), encased, 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 diff --git a/src/Data/Machine/Concurrent/Scatter.hs b/src/Data/Machine/Concurrent/Scatter.hs --- a/src/Data/Machine/Concurrent/Scatter.hs +++ b/src/Data/Machine/Concurrent/Scatter.hs @@ -78,9 +78,9 @@ -- -- @ -- sinkL --- / \ --- a \ --- / \ +-- / \\ +-- a \\ +-- / \\ -- source -- Either a b --> -- r --> -- \\ / -- b / @@ -205,9 +205,9 @@ -- -- @ -- sink1 --- / \ --- a \ --- / \ +-- / \\ +-- a \\ +-- / \\ -- source -- (a,b) --> -- r --> -- \\ / -- b / diff --git a/src/Data/Machine/Concurrent/Wye.hs b/src/Data/Machine/Concurrent/Wye.hs --- a/src/Data/Machine/Concurrent/Wye.hs +++ b/src/Data/Machine/Concurrent/Wye.hs @@ -1,13 +1,15 @@ -{-# LANGUAGE GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables, +{-# 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, StM) +import Control.Monad.Trans.Control (MonadBaseControl) import Data.Machine hiding (wye, (~>), (<~)) import Data.Machine.Concurrent.AsyncStep diff --git a/tests/AllTests.hs b/tests/AllTests.hs --- a/tests/AllTests.hs +++ b/tests/AllTests.hs @@ -28,7 +28,24 @@ 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 ] + [ pipeline, workStealing ]