diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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.*` diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ streaming-concurrency ============== -[![Hackage](https://img.shields.io/hackage/v/streaming-concurrency.svg)](https://hackage.haskell.org/package/streaming-concurrency) [![Build Status](https://travis-ci.org/ivan-m/streaming-concurrency.svg)](https://travis-ci.org/ivan-m/streaming-concurrency) +[![Hackage](https://img.shields.io/hackage/v/streaming-concurrency.svg)](https://hackage.haskell.org/package/streaming-concurrency) [![Build Status](https://travis-ci.org/haskell-streaming/streaming-concurrency.svg)](https://travis-ci.org/haskell-streaming/streaming-concurrency) > Concurrency for the [streaming] ecosystem diff --git a/bench/mapping.hs b/bench/mapping.hs new file mode 100644 --- /dev/null +++ b/bench/mapping.hs @@ -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 diff --git a/src/Streaming/Concurrent.hs b/src/Streaming/Concurrent.hs --- a/src/Streaming/Concurrent.hs +++ b/src/Streaming/Concurrent.hs @@ -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 +. + +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 diff --git a/src/Streaming/Concurrent/Lifted.hs b/src/Streaming/Concurrent/Lifted.hs --- a/src/Streaming/Concurrent/Lifted.hs +++ b/src/Streaming/Concurrent/Lifted.hs @@ -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 +. + +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'. -- diff --git a/streaming-concurrency.cabal b/streaming-concurrency.cabal --- a/streaming-concurrency.cabal +++ b/streaming-concurrency.cabal @@ -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