diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,38 @@ # Revision history for streaming-concurrency +## 0.3.0.0 -- 2017-07-05 + +* Removed support for streaming `ByteString`s. + + The ByteString-related functions which were previously implemented + are broken conceptually and should not be used; they will likely + be removed in a future release. + + A ByteString consists of a stream of bytes; the meaning of each byte + is dependent upon its position in the overall stream. + + In terms of implementation, these bytes are accumulated into + _chunks_, each of which may very well be of a different size. + + As such, if such a chunk is fed into a buffer with multiple readers, + then there is no guarantee which reader will actually receive it or if + it makes sense about what it is in isolation (e.g. it could be split + mid-word, or possibly even in the middle of a Unicode character). + + If, however, multiple ByteStrings are fed into a buffer with a single + reader, the order it has when coming out is similarly undeterministic: + it isn't possible to coherently ensure the sanity of the resulting + ByteString. + + As such, unless you really want to consider it as a stream of + raw 8-bit numbers, trying to do any concurrency with a ByteString + will only lead to trouble. If you do need such functionality, you + can implement it yourself using buffers containing `Word8` values + (in which case you can use `Data.ByteString.Streaming.unpack`). + +* Fix lower bound of `lifted-async` (`replicateConcurrently_` was + added in 0.9.3). + ## 0.2.0.0 -- 2017-07-05 * Rename functions to match the `with...` naming scheme: diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -5,12 +5,28 @@ > Concurrency for the [streaming] ecosystem -The primary purpose for this library is to be able to merge multiple -`Stream`s together. However, it is possible to build higher -abstractions on top of this to be able to also feed multiple streams. - [streaming]: http://hackage.haskell.org/package/streaming +There are two primary higher-level use-cases for this library: + +1. Merge multiple `Stream`s together. + +2. A conceptual `Stream`-based equivalent to [`parMap`] (albeit + utilising concurrency rather than true parallelism). + + [`parMap`]: http://hackage.haskell.org/package/parallel/docs/Control-Parallel-Strategies.html#v:parMap + +However, low-level functions are also exposed so you can construct +your own methods of concurrently using `Stream`s (and there are also +non-`Stream`-specific functions if you wish to use it with other data +types). + +Conceptually, the approach taken is to consider a typical +correspondence system with an in-basket/tray for receiving messages +for others, and an out-basket/tray to be later dealt with. Inputs are +thus provided into the `InBasket` and removed once available from the +`OutBasket`. + Thanks and recognition ---------------------- @@ -21,3 +37,9 @@ collection. [pipes-concurrency]: http://hackage.haskell.org/package/pipes-concurrency + +Another main difference is that the naming of the `input` and `output` +types has been switched around: [pipes-concurrency] seems to consider +them from the point of view of the supplying/consuming `Pipe`s, +whereas here they are considered from the point of view of the +`Buffer` itself. diff --git a/src/Streaming/Concurrent.hs b/src/Streaming/Concurrent.hs --- a/src/Streaming/Concurrent.hs +++ b/src/Streaming/Concurrent.hs @@ -39,16 +39,10 @@ , withStreamMap , withStreamMapM , withStreamTransform - -- * ByteString support - , writeByteStringBasket - , withByteStringBasket - , withMergedByteStrings - -- $bytestringtransform ) where -import Data.ByteString.Streaming (ByteString, reread, unconsChunk) -import Streaming (Of, Stream) -import qualified Streaming.Prelude as S +import Streaming (Of, Stream) +import qualified Streaming.Prelude as S import Control.Applicative ((<|>)) import Control.Concurrent.Async.Lifted (concurrently, @@ -59,7 +53,6 @@ import Control.Monad.Base (MonadBase, liftBase) import Control.Monad.Catch (MonadMask, bracket, bracket_) import Control.Monad.Trans.Control (MonadBaseControl) -import qualified Data.ByteString as B import Data.Foldable (forM_) -------------------------------------------------------------------------------- @@ -79,16 +72,6 @@ (forConcurrently_ strs . flip writeStreamBasket) (`withStreamBasket` 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. -- -- Type written to make it easier if this is the only stream being @@ -101,15 +84,6 @@ continue <- liftBase (STM.atomically (send a)) when continue (go str') --- | A streaming 'ByteString' variant of 'writeStreamBasket'. -writeByteStringBasket :: (MonadBase IO m) => ByteString m r -> InBasket B.ByteString -> m () -writeByteStringBasket bstring (InBasket send) = go bstring - where - go bs = do chNxt <- unconsChunk bs - forM_ chNxt $ \(chnk, bs') -> do - continue <- liftBase (STM.atomically (send chnk)) - when continue (go bs') - -- | Read the output of a buffer into a stream. -- -- @since 0.2.0.0 @@ -120,15 +94,6 @@ where getNext = maybe (Right ()) Left <$> liftBase (STM.atomically receive) --- | A streaming 'ByteString' variant of 'withStreamBasket'. --- --- @since 0.2.0.0 -withByteStringBasket :: (MonadBase IO m) => OutBasket B.ByteString - -> (ByteString m () -> r) - -> r -withByteStringBasket (OutBasket receive) f = - f (reread (liftBase . STM.atomically) receive) - -------------------------------------------------------------------------------- -- | Use buffers to concurrently transform the provided data. @@ -204,19 +169,6 @@ 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 -- pipes-concurrent by Gabriel Gonzalez: @@ -240,7 +192,7 @@ -- argument. -- -- A buffer size @<= 0@ will result in a permanently empty buffer, --- which could result in a system that hang. +-- which could result in a system that hangs. bounded :: Int -> Buffer a bounded 1 = Single bounded n = Bounded n 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 @@ -32,25 +32,17 @@ , withStreamMap , withStreamMapM , withStreamTransform - -- * ByteString support - , writeByteStringBasket - , withByteStringBasket - , withMergedByteStrings - -- $bytestringtransform ) where -import Data.ByteString.Streaming (ByteString) -import Streaming (Of, Stream) -import Streaming.Concurrent (Buffer, InBasket(..), OutBasket(..), - bounded, latest, newest, unbounded) -import qualified Streaming.Concurrent as SC -import Streaming.With.Lifted (Withable(..)) +import Streaming (Of, Stream) +import Streaming.Concurrent (Buffer, InBasket(..), OutBasket(..), + bounded, latest, newest, unbounded) +import qualified Streaming.Concurrent as SC +import Streaming.With.Lifted (Withable(..)) import Control.Monad.Base (MonadBase) import Control.Monad.Trans.Control (MonadBaseControl) -import qualified Data.ByteString as B - -------------------------------------------------------------------------------- -- | Concurrently merge multiple streams together. @@ -63,14 +55,6 @@ -> w (Stream (Of a) m ()) withMergedStreams buff strs = liftWith (SC.withMergedStreams buff strs) --- | 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. -- -- Type written to make it easier if this is the only stream being @@ -79,11 +63,6 @@ => Stream (Of a) (WithMonad w) r -> InBasket a -> w () writeStreamBasket stream ib = liftAction (SC.writeStreamBasket stream ib) --- | A streaming 'ByteString' variant of 'writeStreamBasket'. -writeByteStringBasket :: (Withable w, MonadBase IO (WithMonad w)) - => ByteString (WithMonad w) r -> InBasket B.ByteString -> w () -writeByteStringBasket bs ib = liftAction (SC.writeByteStringBasket bs ib) - -- | Read the output of a buffer into a stream. -- -- Note that there is no requirement that @m ~ WithMonad w@. @@ -92,13 +71,6 @@ withStreamBasket :: (Withable w, MonadBase IO m) => OutBasket a -> w (Stream (Of a) m ()) withStreamBasket ob = liftWith (SC.withStreamBasket ob) --- | A streaming 'ByteString' variant of 'withStreamBasket'. --- --- @since 0.2.0.0 -withByteStringBasket :: (Withable w, MonadBase IO m) - => OutBasket B.ByteString -> w (ByteString m ()) -withByteStringBasket ob = liftWith (SC.withByteStringBasket ob) - -- | Use buffers to concurrently transform the provided data. -- -- In essence, this is a @demultiplexer -> multiplexer@ @@ -159,19 +131,6 @@ -> 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,17 +1,24 @@ name: streaming-concurrency -version: 0.2.0.0 +version: 0.3.0.0 synopsis: Concurrency support for the streaming ecosystem description: - The primary purpose for this library is to be able to merge multiple - @Stream@s together. However, it is possible to build higher - abstractions on top of this to be able to also feed multiple - streams. + There are two primary higher-level use-cases for this library: + . + 1. Merge multiple @Stream@s together. + . + 2. A conceptual @Stream@-based equivalent to @parMap@ (albeit + utilising concurrency rather than true parallelism). + . + However, low-level functions are also exposed so you can construct + your own methods of concurrently using @Stream@s (and there are also + non-@Stream@-specific functions if you wish to use it with other data + types). license: MIT license-file: LICENSE author: Ivan Lazar Miljenovic maintainer: Ivan.Miljenovic@gmail.com copyright: Ivan Lazar Miljenovic -category: Data +category: Data, Streaming build-type: Simple extra-source-files: ChangeLog.md, README.md cabal-version: >=1.10 @@ -25,13 +32,11 @@ exposed-modules: Streaming.Concurrent , Streaming.Concurrent.Lifted build-depends: base ==4.* - , bytestring , exceptions >= 0.6 && < 0.9 - , lifted-async >= 0.9.1 && < 0.10 + , lifted-async >= 0.9.3 && < 0.10 , monad-control == 1.* , stm >= 2.4 && < 3 , streaming >= 0.1.4.0 && < 0.2 - , streaming-bytestring >= 0.1.4.5 && < 0.2 , streaming-with == 0.1.* , transformers-base hs-source-dirs: src @@ -42,12 +47,10 @@ main-is: merging.hs build-depends: streaming-concurrency , base - , bytestring , hspec == 2.4.* , QuickCheck == 2.* , quickcheck-instances , streaming - , streaming-bytestring hs-source-dirs: test default-language: Haskell2010 ghc-options: -Wall -threaded diff --git a/test/merging.hs b/test/merging.hs --- a/test/merging.hs +++ b/test/merging.hs @@ -12,14 +12,10 @@ import Streaming.Concurrent -import Data.ByteString.Streaming (fromStrict, toStrict_) -import Streaming.Prelude (each, toList_) - -import qualified Data.ByteString as B +import Streaming.Prelude (each, toList_) import Data.Function (on) import Data.List (concat, sort) -import Data.Monoid (mconcat) import Test.Hspec (describe, hspec) import Test.Hspec.QuickCheck (prop) import Test.QuickCheck (Positive(..), Property, ioProperty) @@ -29,9 +25,8 @@ main :: IO () main = hspec $ do - describe "Merging retains all elements" $ do - prop "Stream" (mergeCheck :: [[Int]] -> Property) - prop "ByteString" mergeBSCheck + describe "Merging retains all elements" $ + prop "Int Stream" (mergeCheck :: [[Int]] -> Property) describe "Stream transformation" $ do prop "map id" (streamMapCheckId :: Positive Int -> [Int] -> Property) prop "map show" (streamMapCheck show :: Positive Int -> [Int] -> Property) @@ -42,13 +37,6 @@ (eqOn sort as . toList_)) where as = concat ass - -mergeBSCheck :: [B.ByteString] -> Property -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 =