streaming-concurrency 0.3.0.1 → 0.3.1.0
raw patch · 6 files changed
+516/−17 lines, 6 filesdep +HUnitdep +testbenchdep ~basedep ~exceptionsdep ~lifted-asyncPVP ok
version bump matches the API change (PVP)
Dependencies added: HUnit, testbench
Dependency ranges changed: base, exceptions, lifted-async, monad-control, streaming
API changes (from Hackage documentation)
+ Streaming.Concurrent: joinBuffers :: (MonadBase IO m) => (a -> b) -> OutBasket a -> InBasket b -> m ()
+ Streaming.Concurrent: joinBuffersM :: (MonadBase IO m) => (a -> m b) -> OutBasket a -> InBasket b -> m ()
+ Streaming.Concurrent: joinBuffersStream :: (MonadBase IO m) => (Stream (Of a) m () -> Stream (Of b) m t) -> OutBasket a -> InBasket b -> m ()
+ Streaming.Concurrent.Lifted: joinBuffers :: (MonadBase IO m) => (a -> b) -> OutBasket a -> InBasket b -> m ()
+ Streaming.Concurrent.Lifted: joinBuffersM :: (MonadBase IO m) => (a -> m b) -> OutBasket a -> InBasket b -> m ()
+ Streaming.Concurrent.Lifted: joinBuffersStream :: (MonadBase IO m) => (Stream (Of a) m () -> Stream (Of b) m t) -> OutBasket a -> InBasket b -> m ()
Files
- ChangeLog.md +14/−0
- README.md +1/−1
- bench/mapping.hs +369/−0
- src/Streaming/Concurrent.hs +83/−11
- src/Streaming/Concurrent/Lifted.hs +30/−1
- streaming-concurrency.cabal +19/−4
ChangeLog.md view
@@ -1,5 +1,19 @@ # Revision history for streaming-concurrency +## 0.3.1.0 -- 2018-03-14++* Allow `streaming-0.2.*`++* Improved performance for `withStreamMap` and `withStreamMapM`.++ - Previously used additional `Stream`s within the concurrent+ buffers which isn't actually needed.++ - A benchmark is available to compare various implementations.++* Primitives to help defining custom stream-mapping functions are now+ exported.+ ## 0.3.0.1 -- 2017-07-07 * Allow `streaming-with-0.2.*`
README.md view
@@ -1,7 +1,7 @@ streaming-concurrency ============== -[](https://hackage.haskell.org/package/streaming-concurrency) [](https://travis-ci.org/ivan-m/streaming-concurrency)+[](https://hackage.haskell.org/package/streaming-concurrency) [](https://travis-ci.org/haskell-streaming/streaming-concurrency) > Concurrency for the [streaming] ecosystem
+ bench/mapping.hs view
@@ -0,0 +1,369 @@+{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses #-}++{- |+ Module : Main+ Description : Benchmark how to concurrently map a function+ Copyright : Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++ This duplicates a lot of functionality from Streaming.Concurrent to+ help verify that the right design choices were made (we don't+ import existing definitions in case of locality\/inlining\/etc.).++ To be also able to better match the behaviour of the non-concurrent+ variants, the type signatures of all functions have been+ specialised slightly.++ Note that for the cases of @n = 1@, the concurrent variants have a+ slight advantage in that - despite the overhead of concurrency -+ three threads (one for writing, one for mapping and one for+ reading) are used. If @replicateConcurrently_ n@ is removed and+ the benchmarks manually re-run, one of these threads are removed+ and the overhead becomes more apparent.++ Suffixes used:++ [@B@] Uses bounded buffers (buffers are the same size as the number of+ concurrent tasks).++ [@D@] As with @B@ but uses buffers that are double the size of the+ number of concurrent tasks.++ [@U@] Uses unbounded buffers.++ [@S@] Uses 'Stream's between the two buffers.++ [@I@] Does not use 'Stream's between the two buffers (read a value,+ operates, immediately writes to the next buffer).++ [@N@] Non-concurrent (i.e. uses definition from "Streaming.Prelude")++ -}+module Main (main) where++import Streaming.Concurrent (Buffer, InBasket(..), OutBasket(..), bounded,+ joinBuffers, joinBuffersM, joinBuffersStream,+ unbounded, withBuffer, withStreamBasket,+ writeStreamBasket)++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async.Lifted (replicateConcurrently_)+import Control.Monad.Catch (MonadMask)+import Control.Monad.Trans.Control (MonadBaseControl)+import Streaming+import qualified Streaming.Prelude as S++import Test.HUnit ((@?))+import TestBench++--------------------------------------------------------------------------------++main :: IO ()+main = testBench $ do+ collection "Pure maps" $ do+ compareFuncAllIO "show" (pureMap 10 show inputs S.toList_) normalFormIO+ collection "Fibonacci" $+ mapM_ compFib numThreads+ collection "Monadic maps" $ do+ collection "Fibonacci (return'ed)" $+ mapM_ compFibM numThreads+ collection "Identical sleep" $+ mapM_ compDelaySame numThreads+ collection "Different sleep" $+ mapM_ compDelayDiffer numThreads+ where+ compFib n = compareFuncAllIO (show n ++ " tasks")+ (pureMap n fib inputs S.toList_)+ normalFormIO++ compFibM n = compareFuncAllIO (show n ++ " tasks")+ (monadicMap n (return . fib) inputs S.toList_)+ normalFormIO++ compDelaySame n = compareFuncAllIO (show n ++ " tasks")+ (monadicMap n delayReturn inputs S.toList_)+ normalFormIO++ compDelayDiffer n = compareFuncAllIO (show n ++ " tasks")+ (monadicMap n ensureDelay mixedInputs S.length_)+ normalFormIO++ numThreads = [1, 5, 10]++-- | We use the same value repeated to avoid having to sort the+-- results, as that would give the non-concurrent variants an+-- advantage.+inputs :: Stream (Of Int) IO ()+inputs = S.replicate 1000 20++delayReturn :: Int -> IO Int+delayReturn n = threadDelay n `seq` return n++-- For when ordering can't be enforced.+ensureDelay :: Int -> IO ()+ensureDelay n = threadDelay n `seq` return ()++mixedInputs :: Stream (Of Int) IO ()+mixedInputs = S.concat (S.replicate 123 [10..17])++--------------------------------------------------------------------------------++data MapType = NonConcurrent+ | BoundedImmediate+ | BoundedStreamed+ | DoubleImmediate+ | DoubleStreamed+ | UnboundedImmediate+ | UnboundedStreamed+ deriving (Eq, Ord, Show, Read, Bounded, Enum)++pureMap :: Int -> (a -> b) -> Stream (Of a) IO () -> (Stream (Of b) IO () -> IO r)+ -> MapType -> IO r+pureMap n f inp cont pm = go pm n f inp cont+ where+ go NonConcurrent = withStreamMapN+ go BoundedImmediate = withStreamMapBI+ go BoundedStreamed = withStreamMapBS+ go DoubleImmediate = withStreamMapDI+ go DoubleStreamed = withStreamMapDS+ go UnboundedImmediate = withStreamMapUI+ go UnboundedStreamed = withStreamMapUS++monadicMap :: Int -> (a -> IO b) -> Stream (Of a) IO () -> (Stream (Of b) IO () -> IO r)+ -> MapType -> IO r+monadicMap n f inp cont pm = go pm n f inp cont+ where+ go NonConcurrent = withStreamMapMN+ go BoundedImmediate = withStreamMapMBI+ go BoundedStreamed = withStreamMapMBS+ go DoubleImmediate = withStreamMapMDI+ go DoubleStreamed = withStreamMapMDS+ go UnboundedImmediate = withStreamMapMUI+ go UnboundedStreamed = withStreamMapMUS++-- Naive fibonacci implementation+fib :: Int -> Int+fib 0 = 1+fib 1 = 1+fib n = fib (n-1) + fib (n-2)++--------------------------------------------------------------------------------+-- Non-concurrent variants. Take an unused Int just to match the types.++withStreamMapN :: (Monad m) => Int -> (a -> b) -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapN _ f str cont = cont (S.map f str)++withStreamMapMN :: (Monad m) => Int -> (a -> m b) -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapMN _ f str cont = cont (S.mapM f str)++--------------------------------------------------------------------------------+-- Bounded buffer variants++withStreamMapBI :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapBI n f inp cont =+ withBufferedTransformB n (joinBuffers f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withStreamMapMBI :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> m b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapMBI n f inp cont =+ withBufferedTransformB n (joinBuffersM f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withStreamMapBS :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapBS n = withStreamTransformB n . S.map++withStreamMapMBS :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> m b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapMBS n = withStreamTransformB n . S.mapM++withStreamTransformB :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (Stream (Of a) m () -> Stream (Of b) m t)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamTransformB n f inp cont =+ withBufferedTransformB n (joinBuffersStream f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withBufferedTransformB :: (MonadMask m, MonadBaseControl IO m)+ => Int+ -- ^ How many concurrent computations to run.+ -> (OutBasket a -> InBasket b -> m ab)+ -- ^ What to do with each individual concurrent+ -- computation; result is ignored.+ -> (InBasket a -> m ())+ -- ^ Provide initial data; result is ignored.+ -> (OutBasket b -> m r) -> m r+withBufferedTransformB n transform feed consume =+ withBuffer buff feed $ \obA ->+ withBuffer buff (replicateConcurrently_ n . transform obA)+ consume+ where+ buff :: Buffer v+ buff = bounded n++--------------------------------------------------------------------------------+-- Double-Bounded buffer variants++withStreamMapDI :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapDI n f inp cont =+ withBufferedTransformD n (joinBuffers f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withStreamMapMDI :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> m b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapMDI n f inp cont =+ withBufferedTransformD n (joinBuffersM f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withStreamMapDS :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapDS n = withStreamTransformD n . S.map++withStreamMapMDS :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> m b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapMDS n = withStreamTransformD n . S.mapM++withStreamTransformD :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (Stream (Of a) m () -> Stream (Of b) m t)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamTransformD n f inp cont =+ withBufferedTransformD n (joinBuffersStream f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withBufferedTransformD :: (MonadMask m, MonadBaseControl IO m)+ => Int+ -- ^ How many concurrent computations to run.+ -> (OutBasket a -> InBasket b -> m ab)+ -- ^ What to do with each individual concurrent+ -- computation; result is ignored.+ -> (InBasket a -> m ())+ -- ^ Provide initial data; result is ignored.+ -> (OutBasket b -> m r) -> m r+withBufferedTransformD n transform feed consume =+ withBuffer buff feed $ \obA ->+ withBuffer buff (replicateConcurrently_ n . transform obA)+ consume+ where+ buff :: Buffer v+ buff = bounded (2*n)++--------------------------------------------------------------------------------++withStreamMapUI :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapUI n f inp cont =+ withBufferedTransformU n (joinBuffers f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withStreamMapMUI :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> m b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapMUI n f inp cont =+ withBufferedTransformU n (joinBuffersM f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withStreamMapUS :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapUS n = withStreamTransformU n . S.map++withStreamMapMUS :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (a -> m b)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamMapMUS n = withStreamTransformU n . S.mapM++withStreamTransformU :: (MonadMask m, MonadBaseControl IO m)+ => Int -- ^ How many concurrent computations to run.+ -> (Stream (Of a) m () -> Stream (Of b) m t)+ -> Stream (Of a) m ()+ -> (Stream (Of b) m () -> m r) -> m r+withStreamTransformU n f inp cont =+ withBufferedTransformU n (joinBuffersStream f) feed consume+ where+ feed = writeStreamBasket inp++ consume = flip withStreamBasket cont++withBufferedTransformU :: (MonadMask m, MonadBaseControl IO m)+ => Int+ -- ^ How many concurrent computations to run.+ -> (OutBasket a -> InBasket b -> m ab)+ -- ^ What to do with each individual concurrent+ -- computation; result is ignored.+ -> (InBasket a -> m ())+ -- ^ Provide initial data; result is ignored.+ -> (OutBasket b -> m r) -> m r+withBufferedTransformU n transform feed consume =+ withBuffer buff feed $ \obA ->+ withBuffer buff (replicateConcurrently_ n . transform obA)+ consume+ where+ buff :: Buffer v+ buff = unbounded
src/Streaming/Concurrent.hs view
@@ -36,9 +36,14 @@ , withStreamBasket , withMergedStreams -- ** Mapping+ -- $mapping , withStreamMap , withStreamMapM , withStreamTransform+ -- *** Primitives+ , joinBuffers+ , joinBuffersM+ , joinBuffersStream ) where import Streaming (Of, Stream)@@ -51,7 +56,7 @@ import qualified Control.Concurrent.STM as STM import Control.Monad (when) import Control.Monad.Base (MonadBase, liftBase)-import Control.Monad.Catch (MonadMask, bracket, bracket_)+import Control.Monad.Catch (MonadMask, bracket, finally) import Control.Monad.Trans.Control (MonadBaseControl) import Data.Foldable (forM_) @@ -96,6 +101,28 @@ -------------------------------------------------------------------------------- +{- $mapping++These functions provide (concurrency-based rather than+parallelism-based) pseudo-equivalents to+<http://hackage.haskell.org/package/parallel/docs/Control-Parallel-Strategies.html#v:parMap parMap>.++Note however that in practice, these seem to be no better than - and+indeed often worse - than using 'S.map' and 'S.mapM'. A benchmarking+suite is available with this library that tries to compare different+scenarios.++These implementations try to be relatively conservative in terms of+memory usage; it is possible to get better performance by using an+'unbounded' 'Buffer' but if you feed elements into a 'Buffer' much+faster than you can consume them then memory usage will increase.++The \"Primitives\" available below can assist you with defining your+own custom mapping function in conjunction with+'withBufferedTransform'.++-}+ -- | Use buffers to concurrently transform the provided data. -- -- In essence, this is a @demultiplexer -> multiplexer@@@ -133,8 +160,15 @@ -> (a -> b) -> Stream (Of a) m i -> (Stream (Of b) n () -> m r) -> m r-withStreamMap n = withStreamTransform n . S.map+withStreamMap n f inp cont =+ withBufferedTransform n transform feed consume+ where+ feed = writeStreamBasket inp + transform = joinBuffers f++ consume = flip withStreamBasket cont+ -- | Concurrently map a monadic function over all elements of a -- 'Stream'. --@@ -146,8 +180,15 @@ -> (a -> m b) -> Stream (Of a) m i -> (Stream (Of b) n () -> m r) -> m r-withStreamMapM n = withStreamTransform n . S.mapM+withStreamMapM n f inp cont =+ withBufferedTransform n transform feed consume+ where+ feed = writeStreamBasket inp + transform = joinBuffersM f++ consume = flip withStreamBasket cont+ -- | Concurrently split the provided stream into @n@ streams and -- transform them all using the provided function. --@@ -164,11 +205,42 @@ where feed = writeStreamBasket inp - transform obA ibB = withStreamBasket obA- (flip writeStreamBasket ibB . f)+ transform = joinBuffersStream f consume = flip withStreamBasket cont +-- | Take an item out of one 'Buffer', apply a function to it and then+-- place it into another 'Buffer.+--+-- @since 0.3.1.0+joinBuffers :: (MonadBase IO m) => (a -> b) -> OutBasket a -> InBasket b -> m ()+joinBuffers f obA ibB = liftBase go+ where+ go = do ma <- STM.atomically (receiveMsg obA)+ forM_ ma $ \a ->+ do s <- STM.atomically (sendMsg ibB (f a))+ when s go++-- | As with 'joinBuffers' but apply a monadic function.+--+-- @since 0.3.1.0+joinBuffersM :: (MonadBase IO m) => (a -> m b) -> OutBasket a -> InBasket b -> m ()+joinBuffersM f obA ibB = go+ where+ go = do ma <- liftBase (STM.atomically (receiveMsg obA))+ forM_ ma $ \a ->+ do b <- f a+ s <- liftBase (STM.atomically (sendMsg ibB b))+ when s go++-- | As with 'joinBuffers' but read and write the values as 'Stream's.+--+-- @since 0.3.1.0+joinBuffersStream :: (MonadBase IO m) => (Stream (Of a) m () -> Stream (Of b) m t)+ -> OutBasket a -> InBasket b -> m ()+joinBuffersStream f obA ibB = withStreamBasket obA+ (flip writeStreamBasket ibB . f)+ -------------------------------------------------------------------------------- -- This entire section is almost completely taken from -- pipes-concurrent by Gabriel Gonzalez:@@ -287,9 +359,9 @@ return (writeB, readB, sealed, seal) withIn writeB sealed seal =- bracket_ (return ())- (liftBase (STM.atomically seal))- (sendIn (InBasket sendOrEnd))+ sendIn (InBasket sendOrEnd)+ `finally`+ liftBase (STM.atomically seal) where sendOrEnd a = do canWrite <- not <$> STM.readTVar sealed@@ -297,9 +369,9 @@ return canWrite withOut readB sealed seal =- bracket_ (return ())- (liftBase (STM.atomically seal))- (readOut (OutBasket readOrEnd))+ readOut (OutBasket readOrEnd)+ `finally`+ liftBase (STM.atomically seal) where readOrEnd = (Just <$> readB) <|> (do b <- STM.readTVar sealed
src/Streaming/Concurrent/Lifted.hs view
@@ -29,14 +29,21 @@ , withStreamBasket , withMergedStreams -- ** Mapping+ -- $mapping , withStreamMap , withStreamMapM , withStreamTransform+ -- *** Primitives+ , joinBuffers+ , joinBuffersM+ , joinBuffersStream ) where import Streaming (Of, Stream) import Streaming.Concurrent (Buffer, InBasket(..), OutBasket(..),- bounded, latest, newest, unbounded)+ bounded, joinBuffers, joinBuffersM,+ joinBuffersStream, latest, newest,+ unbounded) import qualified Streaming.Concurrent as SC import Streaming.With.Lifted (Withable(..)) @@ -92,6 +99,28 @@ -> w (OutBasket b) withBufferedTransform n transform feed = liftWith (SC.withBufferedTransform n transform feed)++{- $mapping++These functions provide (concurrency-based rather than+parallelism-based) pseudo-equivalents to+<http://hackage.haskell.org/package/parallel/docs/Control-Parallel-Strategies.html#v:parMap parMap>.++Note however that in practice, these seem to be no better than - and+indeed often worse - than using 'S.map' and 'S.mapM'. A benchmarking+suite is available with this library that tries to compare different+scenarios.++These implementations try to be relatively conservative in terms of+memory usage; it is possible to get better performance by using an+'unbounded' 'Buffer' but if you feed elements into a 'Buffer' much+faster than you can consume them then memory usage will increase.++The \"Primitives\" available below can assist you with defining your+own custom mapping function in conjunction with+'withBufferedTransform'.++-} -- | Concurrently map a function over all elements of a 'Stream'. --
streaming-concurrency.cabal view
@@ -1,5 +1,5 @@ name: streaming-concurrency-version: 0.3.0.1+version: 0.3.1.0 synopsis: Concurrency support for the streaming ecosystem description: There are two primary higher-level use-cases for this library:@@ -22,11 +22,11 @@ build-type: Simple extra-source-files: ChangeLog.md, README.md cabal-version: >=1.10-tested-with: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.1.*+tested-with: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.3.* source-repository head type: git- location: https://github.com/ivan-m/streaming-concurrency.git+ location: https://github.com/haskell-streaming/streaming-concurrency.git library exposed-modules: Streaming.Concurrent@@ -36,7 +36,7 @@ , lifted-async >= 0.9.3 && < 0.10 , monad-control == 1.* , stm >= 2.4 && < 3- , streaming >= 0.1.4.0 && < 0.2+ , streaming >= 0.1.4.0 && < 0.3 , streaming-with >= 0.1.0.0 && < 0.3 , transformers-base hs-source-dirs: src@@ -54,3 +54,18 @@ hs-source-dirs: test default-language: Haskell2010 ghc-options: -Wall -threaded++benchmark mapping+ type: exitcode-stdio-1.0+ main-is: mapping.hs+ build-depends: streaming-concurrency+ , base+ , exceptions+ , HUnit+ , lifted-async+ , monad-control+ , streaming+ , testbench >= 0.2.1.0 && < 0.3+ hs-source-dirs: bench+ default-language: Haskell2010+ ghc-options: -Wall -threaded +RTS -N -RTS