diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,24 @@ # Revision history for streaming-concurrency +## 0.2.0.0 -- 2017-07-05 + +* Rename functions to match the `with...` naming scheme: + + - `merge{Streams,ByteStrings}` to `withMerged{Streams,ByteStrings}` + - `read{Stream,ByteString}Basket` to `with{Stream,ByteString}Basket` + +* Add the ability to use buffers to concurrently transform the stream; + following functions added: + + - `withBufferedTransform` + - `withStreamMap` + - `withStreamMapM` + - `withStreamTransform` + + (Note: it is not sound in the general case to transform a + streaming `ByteString`; as such, functions for this are not + implemented though it is possible to do yourself.) + ## 0.1.0.0 -- 2017-07-04 * First version. diff --git a/src/Streaming/Concurrent.hs b/src/Streaming/Concurrent.hs --- a/src/Streaming/Concurrent.hs +++ b/src/Streaming/Concurrent.hs @@ -28,16 +28,22 @@ , newest -- * Using a buffer , withBuffer + , withBufferedTransform , InBasket(..) , OutBasket(..) -- * Stream support , writeStreamBasket - , readStreamBasket - , mergeStreams + , withStreamBasket + , withMergedStreams + -- ** Mapping + , withStreamMap + , withStreamMapM + , withStreamTransform -- * ByteString support , writeByteStringBasket - , readByteStringBasket - , mergeByteStrings + , withByteStringBasket + , withMergedByteStrings + -- $bytestringtransform ) where import Data.ByteString.Streaming (ByteString, reread, unconsChunk) @@ -46,7 +52,8 @@ import Control.Applicative ((<|>)) import Control.Concurrent.Async.Lifted (concurrently, - forConcurrently_) + forConcurrently_, + replicateConcurrently_) import qualified Control.Concurrent.STM as STM import Control.Monad (when) import Control.Monad.Base (MonadBase, liftBase) @@ -63,20 +70,24 @@ -- -- Note that the monad of the resultant Stream can be different from -- the final result. -mergeStreams :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) - => Buffer a -> t (Stream (Of a) m v) - -> (Stream (Of a) n () -> m r) -> m r -mergeStreams buff strs f = withBuffer buff - (forConcurrently_ strs . flip writeStreamBasket) - (`readStreamBasket` f) +-- +-- @since 0.2.0.0 +withMergedStreams :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) + => Buffer a -> t (Stream (Of a) m v) + -> (Stream (Of a) n () -> m r) -> m r +withMergedStreams buff strs f = withBuffer buff + (forConcurrently_ strs . flip writeStreamBasket) + (`withStreamBasket` f) --- | A streaming 'ByteString' variant of 'mergeStreams'. -mergeByteStrings :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) - => Buffer B.ByteString -> t (ByteString m v) - -> (ByteString n () -> m r) -> m r -mergeByteStrings buff bss f = withBuffer buff - (forConcurrently_ bss . flip writeByteStringBasket) - (`readByteStringBasket` f) +-- | A streaming 'ByteString' variant of 'withMergedStreams'. +-- +-- @since 0.2.0.0 +withMergedByteStrings :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) + => Buffer B.ByteString -> t (ByteString m v) + -> (ByteString n () -> m r) -> m r +withMergedByteStrings buff bss f = withBuffer buff + (forConcurrently_ bss . flip writeByteStringBasket) + (`withByteStringBasket` f) -- | Write a single stream to a buffer. -- @@ -100,19 +111,111 @@ when continue (go bs') -- | Read the output of a buffer into a stream. -readStreamBasket :: (MonadBase IO m) => OutBasket a +-- +-- @since 0.2.0.0 +withStreamBasket :: (MonadBase IO m) => OutBasket a -> (Stream (Of a) m () -> r) -> r -readStreamBasket (OutBasket receive) f = f (S.untilRight getNext) +withStreamBasket (OutBasket receive) f = f (S.untilRight getNext) where getNext = maybe (Right ()) Left <$> liftBase (STM.atomically receive) --- | A streaming 'ByteString' variant of 'readStreamBasket'. -readByteStringBasket :: (MonadBase IO m) => OutBasket B.ByteString +-- | A streaming 'ByteString' variant of 'withStreamBasket'. +-- +-- @since 0.2.0.0 +withByteStringBasket :: (MonadBase IO m) => OutBasket B.ByteString -> (ByteString m () -> r) -> r -readByteStringBasket (OutBasket receive) f = +withByteStringBasket (OutBasket receive) f = f (reread (liftBase . STM.atomically) receive) + +-------------------------------------------------------------------------------- + +-- | Use buffers to concurrently transform the provided data. +-- +-- In essence, this is a @demultiplexer -> multiplexer@ +-- transformation: the incoming data is split into @n@ individual +-- segments, the results of which are then merged back together +-- again. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withBufferedTransform :: (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 i) + -- ^ Provide initial data; result is ignored. + -> (OutBasket b -> m r) -> m r +withBufferedTransform n transform feed consume = + withBuffer buff feed $ \obA -> + withBuffer buff (replicateConcurrently_ n . transform obA) + consume + where + buff :: Buffer v + buff = bounded n + +-- | Concurrently map a function over all elements of a 'Stream'. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withStreamMap :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) + => Int -- ^ How many concurrent computations to run. + -> (a -> b) + -> Stream (Of a) m i + -> (Stream (Of b) n () -> m r) -> m r +withStreamMap n = withStreamTransform n . S.map + +-- | Concurrently map a monadic function over all elements of a +-- 'Stream'. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withStreamMapM :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) + => Int -- ^ How many concurrent computations to run. + -> (a -> m b) + -> Stream (Of a) m i + -> (Stream (Of b) n () -> m r) -> m r +withStreamMapM n = withStreamTransform n . S.mapM + +-- | Concurrently split the provided stream into @n@ streams and +-- transform them all using the provided function. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withStreamTransform :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) + => Int -- ^ How many concurrent computations to run. + -> (Stream (Of a) m () -> Stream (Of b) m t) + -> Stream (Of a) m i + -> (Stream (Of b) n () -> m r) -> m r +withStreamTransform n f inp cont = + withBufferedTransform n transform feed consume + where + feed = writeStreamBasket inp + + transform obA ibB = withStreamBasket obA + (flip writeStreamBasket ibB . f) + + consume = flip withStreamBasket cont + +{- $bytestringtransform + +No 'ByteString' equivalents of 'withStreamMap', etc. are provided as +it is very rare for individual chunks of a 'ByteString' - the sizes of +which can vary - to be independent of their position within the +overall stream. + +If you can make such guarantees (e.g. you know that each chunk is a +distinct line and the ordering of these doesn't matter) then you can +use 'withBufferedTransform' to write your own. + +-} -------------------------------------------------------------------------------- -- This entire section is almost completely taken from 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 @@ -1,4 +1,4 @@ -{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses #-} +{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses, TypeFamilies #-} {- | Module : Streaming.Concurrent.Lifted @@ -21,16 +21,22 @@ , newest -- * Using a buffer , withBuffer + , withBufferedTransform , InBasket(..) , OutBasket(..) -- * Stream support , writeStreamBasket - , readStreamBasket - , mergeStreams + , withStreamBasket + , withMergedStreams + -- ** Mapping + , withStreamMap + , withStreamMapM + , withStreamTransform -- * ByteString support , writeByteStringBasket - , readByteStringBasket - , mergeByteStrings + , withByteStringBasket + , withMergedByteStrings + -- $bytestringtransform ) where import Data.ByteString.Streaming (ByteString) @@ -50,16 +56,20 @@ -- | Concurrently merge multiple streams together. -- -- The resulting order is unspecified. -mergeStreams :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO m, Foldable t) - => Buffer a -> t (Stream (Of a) (WithMonad w) v) - -> w (Stream (Of a) m ()) -mergeStreams buff strs = liftWith (SC.mergeStreams buff strs) +-- +-- @since 0.2.0.0 +withMergedStreams :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO m, Foldable t) + => Buffer a -> t (Stream (Of a) (WithMonad w) v) + -> w (Stream (Of a) m ()) +withMergedStreams buff strs = liftWith (SC.withMergedStreams buff strs) --- | A streaming 'ByteString' variant of 'mergeStreams'. -mergeByteStrings :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n, Foldable t) - => Buffer B.ByteString -> t (ByteString (WithMonad w) v) - -> w (ByteString n ()) -mergeByteStrings buff bss = liftWith (SC.mergeByteStrings buff bss) +-- | A streaming 'ByteString' variant of 'withMergedStreams'. +-- +-- @since 0.2.0.0 +withMergedByteStrings :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n, Foldable t) + => Buffer B.ByteString -> t (ByteString (WithMonad w) v) + -> w (ByteString n ()) +withMergedByteStrings buff bss = liftWith (SC.withMergedByteStrings buff bss) -- | Write a single stream to a buffer. -- @@ -77,13 +87,91 @@ -- | Read the output of a buffer into a stream. -- -- Note that there is no requirement that @m ~ WithMonad w@. -readStreamBasket :: (Withable w, MonadBase IO m) => OutBasket a -> w (Stream (Of a) m ()) -readStreamBasket ob = liftWith (SC.readStreamBasket ob) +-- +-- @since 0.2.0.0 +withStreamBasket :: (Withable w, MonadBase IO m) => OutBasket a -> w (Stream (Of a) m ()) +withStreamBasket ob = liftWith (SC.withStreamBasket ob) --- | A streaming 'ByteString' variant of 'readStreamBasket'. -readByteStringBasket :: (Withable w, MonadBase IO m) +-- | A streaming 'ByteString' variant of 'withStreamBasket'. +-- +-- @since 0.2.0.0 +withByteStringBasket :: (Withable w, MonadBase IO m) => OutBasket B.ByteString -> w (ByteString m ()) -readByteStringBasket ob = liftWith (SC.readByteStringBasket ob) +withByteStringBasket ob = liftWith (SC.withByteStringBasket ob) + +-- | Use buffers to concurrently transform the provided data. +-- +-- In essence, this is a @demultiplexer -> multiplexer@ +-- transformation: the incoming data is split into @n@ individual +-- segments, the results of which are then merged back together +-- again. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withBufferedTransform :: (Withable w, MonadBaseControl IO (WithMonad w)) + => Int + -- ^ How many concurrent computations to run. + -> (OutBasket a -> InBasket b -> WithMonad w ab) + -- ^ What to do with each individual concurrent + -- computation; result is ignored. + -> (InBasket a -> WithMonad w i) + -- ^ Provide initial data; result is ignored. + -> w (OutBasket b) +withBufferedTransform n transform feed = + liftWith (SC.withBufferedTransform n transform feed) + +-- | Concurrently map a function over all elements of a 'Stream'. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withStreamMap :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n) + => Int -- ^ How many concurrent computations to run. + -> (a -> b) + -> Stream (Of a) (WithMonad w) i + -> w (Stream (Of b) n ()) +withStreamMap n f inp = liftWith (SC.withStreamMap n f inp) + +-- | Concurrently map a monadic function over all elements of a +-- 'Stream'. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withStreamMapM :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n) + => Int -- ^ How many concurrent computations to run. + -> (a -> WithMonad w b) + -> Stream (Of a) (WithMonad w) i + -> w (Stream (Of b) n ()) +withStreamMapM n f inp = liftWith (SC.withStreamMapM n f inp) + +-- | Concurrently split the provided stream into @n@ streams and +-- transform them all using the provided function. +-- +-- Note: ordering of elements in the output is undeterministic. +-- +-- @since 0.2.0.0 +withStreamTransform :: ( Withable w, m ~ WithMonad w, MonadBaseControl IO m + , MonadBase IO n) + => Int -- ^ How many concurrent computations to run. + -> (Stream (Of a) m () -> Stream (Of b) m t) + -> Stream (Of a) m i + -> w (Stream (Of b) n ()) +withStreamTransform n f inp = liftWith (SC.withStreamTransform n f inp) + +{- $bytestringtransform + +No 'ByteString' equivalents of 'withStreamMap', etc. are provided as +it is very rare for individual chunks of a 'ByteString' - the sizes of +which can vary - to be independent of their position within the +overall stream. + +If you can make such guarantees (e.g. you know that each chunk is a +distinct line and the ordering of these doesn't matter) then you can +use 'withBufferedTransform' to write your own. + +-} -- | Use a buffer to asynchronously communicate. -- 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.1.0.0 +version: 0.2.0.0 synopsis: Concurrency support for the streaming ecosystem description: The primary purpose for this library is to be able to merge multiple @@ -50,4 +50,4 @@ , streaming-bytestring hs-source-dirs: test default-language: Haskell2010 - ghc-options: -Wall + ghc-options: -Wall -threaded diff --git a/test/merging.hs b/test/merging.hs --- a/test/merging.hs +++ b/test/merging.hs @@ -22,7 +22,7 @@ import Data.Monoid (mconcat) import Test.Hspec (describe, hspec) import Test.Hspec.QuickCheck (prop) -import Test.QuickCheck (Property, ioProperty) +import Test.QuickCheck (Positive(..), Property, ioProperty) import Test.QuickCheck.Instances () -------------------------------------------------------------------------------- @@ -32,20 +32,33 @@ describe "Merging retains all elements" $ do prop "Stream" (mergeCheck :: [[Int]] -> Property) prop "ByteString" mergeBSCheck + describe "Stream transformation" $ do + prop "map id" (streamMapCheckId :: Positive Int -> [Int] -> Property) + prop "map show" (streamMapCheck show :: Positive Int -> [Int] -> Property) mergeCheck :: (Ord a) => [[a]] -> Property -mergeCheck ass = ioProperty (mergeStreams unbounded - (map each ass) - (eqOn sort as . toList_)) +mergeCheck ass = ioProperty (withMergedStreams unbounded + (map each ass) + (eqOn sort as . toList_)) where as = concat ass mergeBSCheck :: [B.ByteString] -> Property -mergeBSCheck bss = ioProperty (mergeByteStrings unbounded - (map fromStrict bss) - (eqOn B.sort bs . toStrict_)) +mergeBSCheck bss = ioProperty (withMergedByteStrings unbounded + (map fromStrict bss) + (eqOn B.sort bs . toStrict_)) where bs = mconcat bss + +streamMapCheckId :: (Ord a) => Positive Int -> [a] -> Property +streamMapCheckId (Positive n) as = + ioProperty (withStreamMap n id (each as) (eqOn sort as . toList_)) + +streamMapCheck :: (Ord b) => (a -> b) -> Positive Int -> [a] -> Property +streamMapCheck f (Positive n) as = + ioProperty (withStreamMap n f (each as) (eqOn sort bs . toList_)) + where + bs = map f as eqOn :: (Eq b, Functor f) => (a -> b) -> a -> f a -> f Bool eqOn f a = fmap (((==)`on`f) a)