concurrent-machines 0.2.0 → 0.2.1
raw patch · 7 files changed
+46/−16 lines, 7 filesdep ~asyncdep ~lifted-asyncPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: async, lifted-async
API changes (from Hackage documentation)
Files
- concurrent-machines.cabal +4/−3
- src/Data/Machine/Concurrent.hs +4/−2
- src/Data/Machine/Concurrent/Buffer.hs +5/−1
- src/Data/Machine/Concurrent/Fanout.hs +5/−1
- src/Data/Machine/Concurrent/Scatter.hs +6/−6
- src/Data/Machine/Concurrent/Wye.hs +4/−2
- tests/AllTests.hs +18/−1
concurrent-machines.cabal view
@@ -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
src/Data/Machine/Concurrent.hs view
@@ -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
src/Data/Machine/Concurrent/Buffer.hs view
@@ -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.
src/Data/Machine/Concurrent/Fanout.hs view
@@ -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
src/Data/Machine/Concurrent/Scatter.hs view
@@ -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 /
src/Data/Machine/Concurrent/Wye.hs view
@@ -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
tests/AllTests.hs view
@@ -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 ]