streamly 0.4.0 → 0.4.1
raw patch · 15 files changed
+363/−226 lines, 15 files
Files
- Changelog.md +8/−0
- README.md +6/−0
- bench.sh +1/−1
- benchmark/BaseStreams.hs +2/−2
- benchmark/StreamDOps.hs +3/−3
- benchmark/StreamKOps.hs +14/−13
- src/Streamly.hs +81/−0
- src/Streamly/Streams/Prelude.hs +1/−87
- src/Streamly/Streams/StreamD.hs +6/−13
- src/Streamly/Streams/StreamK.hs +46/−2
- src/Streamly/Streams/Zip.hs +5/−45
- stack-7.10.yaml +2/−2
- stack.yaml +2/−6
- streamly.cabal +54/−27
- test/Main.hs +132/−25
Changelog.md view
@@ -1,3 +1,9 @@+## 0.4.1++### Bug Fixes++* foldxM was not fully strict, fixed.+ ## 0.4.0 ### Breaking changes@@ -13,6 +19,8 @@ ### Enhancements * Add concurrency control primitives `maxThreads` and `maxBuffer`.+* Concurrency of a stream with bounded concurrency when used with `take` is now+ limited by the number elements demanded by `take`. * Significant performance improvements utilizing stream fusion optimizations. * Add `yield` to construct a singleton stream from a pure value * Add `repeat` to generate an infinite stream by repeating a pure value
README.md view
@@ -1,5 +1,11 @@ # Streamly +[](https://hackage.haskell.org/package/streamly)+[](https://gitter.im/composewell/streamly)+[](https://travis-ci.org/composewell/streamly)+[](https://ci.appveyor.com/project/harendra-kumar/streamly)+[](https://coveralls.io/github/composewell/streamly?branch=master)+ ## Stream`ing` `Concurrent`ly Streamly, short for streaming concurrently, provides monadic streams, with a
bench.sh view
@@ -133,7 +133,7 @@ --include-first-iter \ --min-samples 3 \ --min-duration 0 \- --match exact+ --match exact \ --csvraw=$OUTPUT_FILE \ -v 2 \ --measure-with $BENCH_PROG $GAUGE_ARGS || die "Benchmarking failed"
benchmark/BaseStreams.hs view
@@ -37,7 +37,7 @@ , benchIO "unfoldrM" D.toNull D.sourceUnfoldrM , benchIO "fromEnum" D.toNull D.sourceFromEnum - , benchIO "fromFoldable" D.toNull D.sourceFromFoldable+ , benchIO "fromList" D.toNull D.sourceFromList -- , benchIO "fromFoldableM" D.sourceFromFoldableM ] , bgroup "elimination"@@ -64,7 +64,7 @@ -- , benchIO "fromEnum" K.toNull K.sourceFromEnum , benchIO "fromFoldable" K.toNull K.sourceFromFoldable- , benchIO "fromFoldableM" K.toNull K.sourceFromFoldableM+ -- , benchIO "fromFoldableM" K.toNull K.sourceFromFoldableM -- appends , benchIO "foldMapWith" K.toNull K.sourceFoldMapWith
benchmark/StreamDOps.hs view
@@ -96,9 +96,9 @@ sourceFromEnum :: Monad m => Int -> Stream m Int sourceFromEnum n = S.enumFromStepN n 1 value -{-# INLINE sourceFromFoldable #-}-sourceFromFoldable :: Monad m => Int -> Stream m Int-sourceFromFoldable n = S.fromList [n..n+value]+{-# INLINE sourceFromList #-}+sourceFromList :: Monad m => Int -> Stream m Int+sourceFromList n = S.fromList [n..n+value] {-# INLINE source #-} source :: Monad m => Int -> Stream m Int
benchmark/StreamKOps.hs view
@@ -13,10 +13,9 @@ (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=), subtract, undefined, Maybe(..), not) -import qualified Streamly.Streams.StreamK as S hiding (runStream)--- import qualified Streamly.Streams.Serial as S-import qualified Streamly as S-import qualified Streamly.Prelude as P+import qualified Streamly.Streams.StreamK as S+import qualified Streamly.Streams.Prelude as S+import qualified Streamly.SVar as S value, maxValue :: Int value = 1000000@@ -68,7 +67,7 @@ -- Stream generation and elimination ------------------------------------------------------------------------------- -type Stream m a = S.SerialT m a+type Stream m a = S.Stream m a {-# INLINE sourceUnfoldr #-} sourceUnfoldr :: Int -> Stream m Int@@ -98,17 +97,19 @@ sourceFromFoldable :: Int -> Stream m Int sourceFromFoldable n = S.fromFoldable [n..n+value] +{- {-# INLINE sourceFromFoldableM #-} sourceFromFoldableM :: S.MonadAsync m => Int -> Stream m Int-sourceFromFoldableM n = P.fromFoldableM (Prelude.fmap return [n..n+value])+sourceFromFoldableM n = S.fromFoldableM (Prelude.fmap return [n..n+value])+-} {-# INLINE sourceFoldMapWith #-}-sourceFoldMapWith :: Monad m => Int -> Stream m Int-sourceFoldMapWith n = S.foldMapWith (S.<>) return [n..n+value]+sourceFoldMapWith :: Int -> Stream m Int+sourceFoldMapWith n = S.foldMapWith (S.serial) S.yield [n..n+value] {-# INLINE sourceFoldMapWithM #-} sourceFoldMapWithM :: Monad m => Int -> Stream m Int-sourceFoldMapWithM n = S.foldMapWith (S.<>) (S.yieldM . return) [n..n+value]+sourceFoldMapWithM n = S.foldMapWith (S.serial) (S.yieldM . return) [n..n+value] {-# INLINE source #-} source :: S.MonadAsync m => Int -> Stream m Int@@ -160,15 +161,15 @@ filterAllIn = transform . S.filter (<= maxValue) takeOne = transform . S.take 1 takeAll = transform . S.take maxValue-takeWhileTrue = transform . P.takeWhile (<= maxValue)-dropAll = transform . P.drop maxValue-dropWhileTrue = transform . P.dropWhile (<= maxValue)+takeWhileTrue = transform . S.takeWhile (<= maxValue)+dropAll = transform . S.drop maxValue+dropWhileTrue = transform . S.dropWhile (<= maxValue) ------------------------------------------------------------------------------- -- Zipping and concat ------------------------------------------------------------------------------- -zip src = transform $ (P.zipWith (,) src src)+zip src = transform $ (S.zipWith (,) src src) concat _n = return () -------------------------------------------------------------------------------
src/Streamly.hs view
@@ -55,6 +55,14 @@ -- import Streamly -- @ +{-# LANGUAGE CPP #-}++#if __GLASGOW_HASKELL__ >= 800+{-# OPTIONS_GHC -Wno-orphans #-}+#endif++#include "Streamly/Streams/inline.h"+ module Streamly ( MonadAsync@@ -167,6 +175,79 @@ import Streamly.Streams.SVar (maxThreads, maxBuffer) import Streamly.SVar (MonadAsync) import Data.Semigroup (Semigroup(..))++import qualified Streamly.Streams.StreamD as D+import qualified Streamly.Streams.StreamK as K++-- XXX This should perhaps be moved to Prelude.++------------------------------------------------------------------------------+-- Eliminating a stream+------------------------------------------------------------------------------++-- | Run a streaming composition, discard the results. By default it interprets+-- the stream as 'SerialT', to run other types of streams use the type adapting+-- combinators for example @runStream . 'asyncly'@.+--+-- @since 0.2.0+{-# INLINE_EARLY runStream #-}+runStream :: Monad m => SerialT m a -> m ()+runStream m = D.runStream $ D.fromStreamK (toStream m)+{-# RULES "runStream fallback to CPS" [1]+ forall a. D.runStream (D.fromStreamK a) = K.runStream a #-}++-- | Same as 'runStream'+--+-- @since 0.1.0+{-# DEPRECATED runStreaming "Please use runStream instead." #-}+runStreaming :: (Monad m, IsStream t) => t m a -> m ()+runStreaming = runStream . K.adapt++-- | Same as @runStream@.+--+-- @since 0.1.0+{-# DEPRECATED runStreamT "Please use runStream instead." #-}+runStreamT :: Monad m => SerialT m a -> m ()+runStreamT = runStream++-- | Same as @runStream . wSerially@.+--+-- @since 0.1.0+{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}+runInterleavedT :: Monad m => WSerialT m a -> m ()+runInterleavedT = runStream . K.adapt++-- | Same as @runStream . parallely@.+--+-- @since 0.1.0+{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}+runParallelT :: Monad m => ParallelT m a -> m ()+runParallelT = runStream . K.adapt++-- | Same as @runStream . asyncly@.+--+-- @since 0.1.0+{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}+runAsyncT :: Monad m => AsyncT m a -> m ()+runAsyncT = runStream . K.adapt++-- | Same as @runStream . zipping@.+--+-- @since 0.1.0+{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}+runZipStream :: Monad m => ZipSerialM m a -> m ()+runZipStream = runStream . K.adapt++-- | Same as @runStream . zippingAsync@.+--+-- @since 0.1.0+{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}+runZipAsync :: Monad m => ZipAsyncM m a -> m ()+runZipAsync = runStream . K.adapt++------------------------------------------------------------------------------+-- Documentation+------------------------------------------------------------------------------ -- $serial --
src/Streamly/Streams/Prelude.hs view
@@ -9,12 +9,6 @@ {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE UndecidableInstances #-} -- XXX -#if __GLASGOW_HASKELL__ >= 800-{-# OPTIONS_GHC -Wno-orphans #-}-#endif--#include "inline.h"- -- | -- Module : Streamly.Streams.Prelude -- Copyright : (c) 2017 Harendra Kumar@@ -27,95 +21,15 @@ -- module Streamly.Streams.Prelude (- -- * Elimination- runStream- , runStreaming -- deprecated- , runStreamT -- deprecated- , runInterleavedT -- deprecated- , runParallelT -- deprecated- , runAsyncT -- deprecated- , runZipStream -- deprecated- , runZipAsync -- deprecated- -- * Fold Utilities- , foldWith+ foldWith , foldMapWith , forEachWith ) where import Streamly.Streams.StreamK (IsStream(..))-import Streamly.Streams.Serial (SerialT, WSerialT)-import Streamly.Streams.Parallel (ParallelT)-import Streamly.Streams.Async (AsyncT)-import Streamly.Streams.Zip (ZipSerialM, ZipAsyncM)--import qualified Streamly.Streams.StreamD as D import qualified Streamly.Streams.StreamK as K----------------------------------------------------------------------------------- Eliminating a stream----------------------------------------------------------------------------------- | Run a streaming composition, discard the results. By default it interprets--- the stream as 'SerialT', to run other types of streams use the type adapting--- combinators for example @runStream . 'asyncly'@.------ @since 0.2.0-{-# INLINE_EARLY runStream #-}-runStream :: Monad m => SerialT m a -> m ()-runStream m = D.runStream $ D.fromStreamK (toStream m)-{-# RULES "runStream fallback to CPS" [1]- forall a. D.runStream (D.fromStreamK a) = K.runStream a #-}---- | Same as 'runStream'------ @since 0.1.0-{-# DEPRECATED runStreaming "Please use runStream instead." #-}-runStreaming :: (Monad m, IsStream t) => t m a -> m ()-runStreaming = runStream . K.adapt---- | Same as @runStream@.------ @since 0.1.0-{-# DEPRECATED runStreamT "Please use runStream instead." #-}-runStreamT :: Monad m => SerialT m a -> m ()-runStreamT = runStream---- | Same as @runStream . wSerially@.------ @since 0.1.0-{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}-runInterleavedT :: Monad m => WSerialT m a -> m ()-runInterleavedT = runStream . K.adapt---- | Same as @runStream . parallely@.------ @since 0.1.0-{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}-runParallelT :: Monad m => ParallelT m a -> m ()-runParallelT = runStream . K.adapt---- | Same as @runStream . asyncly@.------ @since 0.1.0-{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}-runAsyncT :: Monad m => AsyncT m a -> m ()-runAsyncT = runStream . K.adapt---- | Same as @runStream . zipping@.------ @since 0.1.0-{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}-runZipStream :: Monad m => ZipSerialM m a -> m ()-runZipStream = runStream . K.adapt---- | Same as @runStream . zippingAsync@.------ @since 0.1.0-{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}-runZipAsync :: Monad m => ZipAsyncM m a -> m ()-runZipAsync = runStream . K.adapt ------------------------------------------------------------------------------ -- Fold Utilities
src/Streamly/Streams/StreamD.hs view
@@ -538,22 +538,15 @@ {-# INLINE_NORMAL drop #-} drop :: Monad m => Int -> Stream m a -> Stream m a-drop n (Stream step state) = Stream step' (state, Just n)+drop n (Stream step state) = Stream step' (state, n) where {-# INLINE_LATE step' #-}- step' gst (st, Just i)- | i > 0 = do- r <- step (rstState gst) st- case r of- Yield _ s -> step' (rstState gst) (s, Just (i - 1))- Stop -> return Stop- | otherwise = step' gst (st, Nothing)-- step' gst (st, Nothing) = do+ step' gst (st, i) = do r <- step (rstState gst) st- return $ case r of- Yield x s -> Yield x (s, Nothing)- Stop -> Stop+ case r of+ Yield _ s | i > 0 -> step' gst (s, i - 1)+ Yield x s -> return $ Yield x (s, 0)+ Stop -> return Stop data DropWhileState s a = DropWhileDrop s
src/Streamly/Streams/StreamK.hs view
@@ -116,6 +116,10 @@ -- ** Map and Filter , mapMaybe + -- ** Zipping+ , zipWith+ , zipWithM+ -- * Semigroup Style Composition , serial @@ -137,7 +141,7 @@ import Prelude hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence, take, filter, all, any, takeWhile, drop, dropWhile, minimum,- maximum, elem, notElem, null, head, tail)+ maximum, elem, notElem, null, head, tail, zipWith) import qualified Prelude import Streamly.SVar@@ -521,7 +525,7 @@ go !acc m1 = let stop = acc >>= done single a = acc >>= \b -> step b a >>= done- yieldk a r = acc >>= \b -> go (step b a) r+ yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r in (unStream m1) defState stop single yieldk -- | Like 'foldl'' but with a monadic step function.@@ -810,6 +814,46 @@ Just b -> yld b (toStream $ go r) Nothing -> (unStream r) (rstState st) stp single yieldk in unStream m1 (rstState st) stp single yieldk++------------------------------------------------------------------------------+-- Serial Zipping+------------------------------------------------------------------------------++{-# INLINE zipWithS #-}+zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c+zipWithS f m1 m2 = go m1 m2+ where+ go mx my = Stream $ \st stp sng yld -> do+ let merge a ra =+ let single2 b = sng (f a b)+ yield2 b rb = yld (f a b) (go ra rb)+ in unStream my (rstState st) stp single2 yield2+ let single1 a = merge a nil+ yield1 a ra = merge a ra+ unStream mx (rstState st) stp single1 yield1++-- | Zip two streams serially using a pure zipping function.+--+-- @since 0.1.0+{-# INLINABLE zipWith #-}+zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c+zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)++-- | Zip two streams serially using a monadic zipping function.+--+-- @since 0.1.0+zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c+zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)+ where+ go mx my = Stream $ \st stp sng yld -> do+ let merge a ra =+ let runIt x = unStream x (rstState st) stp sng yld+ single2 b = f a b >>= sng+ yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)+ in unStream my (rstState st) stp single2 yield2+ let single1 a = merge a nil+ yield1 a ra = merge a ra+ unStream mx (rstState st) stp single1 yield1 ------------------------------------------------------------------------------ -- Semigroup
src/Streamly/Streams/Zip.hs view
@@ -20,8 +20,8 @@ -- module Streamly.Streams.Zip (- zipWith- , zipWithM+ K.zipWith+ , K.zipWithM , zipAsyncWith , zipAsyncWithM @@ -51,46 +51,6 @@ #include "Instances.hs" --------------------------------------------------------------------------------- Serial Zipping---------------------------------------------------------------------------------{-# INLINE zipWithS #-}-zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c-zipWithS f m1 m2 = go m1 m2- where- go mx my = Stream $ \st stp sng yld -> do- let merge a ra =- let single2 b = sng (f a b)- yield2 b rb = yld (f a b) (go ra rb)- in unStream my (rstState st) stp single2 yield2- let single1 a = merge a K.nil- yield1 a ra = merge a ra- unStream mx (rstState st) stp single1 yield1---- | Zip two streams serially using a pure zipping function.------ @since 0.1.0-{-# INLINABLE zipWith #-}-zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c-zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)---- | Zip two streams serially using a monadic zipping function.------ @since 0.1.0-zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c-zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)- where- go mx my = Stream $ \st stp sng yld -> do- let merge a ra =- let runIt x = unStream x (rstState st) stp sng yld- single2 b = f a b >>= sng- yield2 b rb = f a b >>= \x -> runIt (x `K.cons` go ra rb)- in unStream my (rstState st) stp single2 yield2- let single1 a = merge a K.nil- yield1 a ra = merge a ra- unStream mx (rstState st) stp single1 yield1-------------------------------------------------------------------------------- -- Serially Zipping Streams ------------------------------------------------------------------------------ @@ -157,7 +117,7 @@ instance Monad m => Applicative (ZipSerialM m) where pure = ZipSerialM . K.repeat- m1 <*> m2 = fromStream $ zipWith id (toStream m1) (toStream m2)+ m1 <*> m2 = fromStream $ K.zipWith id (toStream m1) (toStream m2) ------------------------------------------------------------------------------ -- Parallel Zipping@@ -172,7 +132,7 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do ma <- mkAsync' (rstState st) m1 mb <- mkAsync' (rstState st) m2- unStream (toStream (zipWith f ma mb)) (rstState st) stp sng yld+ unStream (toStream (K.zipWith f ma mb)) (rstState st) stp sng yld -- | Zip two streams asyncly (i.e. both the elements being zipped are generated -- concurrently) using a monadic zipping function.@@ -183,7 +143,7 @@ zipAsyncWithM f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do ma <- mkAsync' (rstState st) m1 mb <- mkAsync' (rstState st) m2- unStream (toStream (zipWithM f ma mb)) (rstState st) stp sng yld+ unStream (toStream (K.zipWithM f ma mb)) (rstState st) stp sng yld ------------------------------------------------------------------------------ -- Parallely Zipping Streams
stack-7.10.yaml view
@@ -15,5 +15,5 @@ flags: {} extra-package-dbs: [] # For mac ports installed SDL library on Mac OS X-#extra-include-dirs:-#- /opt/local/include+extra-include-dirs:+- /opt/local/include
stack.yaml view
@@ -1,17 +1,13 @@-#resolver: lts-11.0-resolver: nightly-2018-07-06+resolver: lts-12.0 packages: - '.'-#- location: ../bench-graph-# extra-dep: true allow-newer: true extra-deps: - SDL-0.6.6.0 - gauge-0.2.3- - bench-graph-0.1.0+ - bench-graph-0.1.1 - Chart-1.9 - Chart-diagrams-1.9- - Unique-0.4.7.2 - SVGFonts-1.6.0.3 flags: {}
streamly.cabal view
@@ -1,5 +1,5 @@ name: streamly-version: 0.4.0+version: 0.4.1 synopsis: Beautiful Streaming, Concurrent and Reactive Composition description: Streamly, short for streaming concurrently, provides monadic streams, with a@@ -274,12 +274,11 @@ -- Benchmarks ------------------------------------------------------------------------------- -benchmark base+benchmark linear type: exitcode-stdio-1.0 hs-source-dirs: benchmark- main-is: BaseStreams.hs- other-modules: StreamDOps- , StreamKOps+ main-is: Linear.hs+ other-modules: LinearOps default-language: Haskell2010 ghc-options: -O2 -Wall if flag(dev)@@ -295,22 +294,18 @@ -Wredundant-constraints -Wnoncanonical-monad-instances -Wnoncanonical-monadfail-instances- if flag(dev)- buildable: True- build-depends:- streamly- , base >= 4.8 && < 5- , deepseq >= 1.4.0 && < 1.5- , random >= 1.0 && < 2.0- , gauge >= 0.2.3 && < 0.3- else- buildable: False+ build-depends:+ streamly+ , base >= 4.8 && < 5+ , deepseq >= 1.4.0 && < 1.5+ , random >= 1.0 && < 2.0+ , gauge >= 0.2.3 && < 0.3 -benchmark linear+benchmark nested type: exitcode-stdio-1.0 hs-source-dirs: benchmark- main-is: Linear.hs- other-modules: LinearOps+ main-is: Nested.hs+ other-modules: NestedOps default-language: Haskell2010 ghc-options: -O2 -Wall if flag(dev)@@ -333,11 +328,25 @@ , random >= 1.0 && < 2.0 , gauge >= 0.2.3 && < 0.3 -benchmark nested+-------------------------------------------------------------------------------+-- Internal benchmarks for unexposed modules+-------------------------------------------------------------------------------++-- We have to copy the streamly library modules here because there is no+-- way to use unexposed modules from the library.++benchmark base type: exitcode-stdio-1.0- hs-source-dirs: benchmark- main-is: Nested.hs- other-modules: NestedOps+ hs-source-dirs: benchmark, src+ main-is: BaseStreams.hs+ other-modules: Streamly.SVar+ , Streamly.Streams.StreamK+ , Streamly.Streams.StreamD+ , Streamly.Streams.Prelude++ , StreamDOps+ , StreamKOps+ default-language: Haskell2010 ghc-options: -O2 -Wall if flag(dev)@@ -353,12 +362,30 @@ -Wredundant-constraints -Wnoncanonical-monad-instances -Wnoncanonical-monadfail-instances+ build-depends:- streamly- , base >= 4.8 && < 5- , deepseq >= 1.4.0 && < 1.5- , random >= 1.0 && < 2.0- , gauge >= 0.2.3 && < 0.3+ base >= 4.8 && < 5+ , deepseq >= 1.4.0 && < 1.5+ , random >= 1.0 && < 2.0+ , gauge >= 0.2.3 && < 0.3++ , ghc-prim >= 0.2 && < 0.6+ , containers >= 0.5 && < 0.6+ , heaps >= 0.3 && < 0.4++ -- concurrency+ , atomic-primops >= 0.8 && < 0.9+ , lockfree-queue >= 0.2.3 && < 0.3++ , exceptions >= 0.8 && < 0.11+ , monad-control >= 1.0 && < 2+ , mtl >= 2.2 && < 3+ , transformers >= 0.4 && < 0.6+ , transformers-base >= 0.4 && < 0.5++ if impl(ghc < 8.0)+ build-depends:+ semigroups >= 0.18 && < 0.19 executable chart-linear default-language: Haskell2010
test/Main.hs view
@@ -1,16 +1,19 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-} module Main (main) where import Control.Concurrent (threadDelay)-import Control.Exception (Exception, try)+import Control.Exception (Exception, try, ErrorCall(..), catch, throw) import Control.Monad.Catch (throwM, MonadThrow) import Control.Monad.Error.Class (throwError, MonadError) import Control.Monad.Trans.Except (runExceptT, ExceptT) import Data.Foldable (forM_, fold) import Data.List (sort)++import Data.IORef import Test.Hspec import Streamly@@ -89,30 +92,6 @@ `shouldReturn` ([(1,2),(1,3)] :: [(Int, Int)]) ---------------------------------------------------------------------------- -- Semigroup/Monoidal Composition strict ordering checks- ----------------------------------------------------------------------------- -- test both (<>) and mappend to make sure we are using correct instance- -- for Monoid that is using the right version of semigroup. Instance- -- deriving can cause us to pick wrong instances sometimes.-- describe "WSerial interleaved (<>) ordering check" $ interleaveCheck wSerially (<>)- describe "WSerial interleaved mappend ordering check" $ interleaveCheck wSerially mappend-- -- describe "WAsync interleaved (<>) ordering check" $ interleaveCheck wAsyncly (<>)- -- describe "WAsync interleaved mappend ordering check" $ interleaveCheck wAsyncly mappend-- describe "Async (<>) time order check" $ parallelCheck asyncly (<>)- describe "Async mappend time order check" $ parallelCheck asyncly mappend-- -- XXX this keeps failing intermittently, need to investigate- -- describe "WAsync (<>) time order check" $ parallelCheck wAsyncly (<>)- -- describe "WAsync mappend time order check" $ parallelCheck wAsyncly mappend-- describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)- describe "Parallel mappend time order check" $ parallelCheck parallely mappend-- --------------------------------------------------------------------------- -- Monoidal Compositions, multiset equality checks --------------------------------------------------------------------------- @@ -384,6 +363,50 @@ describe "take on infinite concurrent stream" $ takeInfinite wAsyncly describe "take on infinite concurrent stream" $ takeInfinite aheadly + ---------------------------------------------------------------------------+ -- Folds are strict enough+ ---------------------------------------------------------------------------++ it "foldx is strict enough" checkFoldxStrictness+ it "foldl' is strict enough" checkFoldl'Strictness+ it "scanx is strict enough" checkScanxStrictness+ it "scanl' is strict enough" checkScanl'Strictness+ it "foldxM is strict enough" (checkFoldMStrictness foldxMStrictCheck)+ it "foldlM' is strict enough" (checkFoldMStrictness foldlM'StrictCheck)+ it "scanlM' is strict enough" (checkScanlMStrictness scanlM'StrictCheck)++ ---------------------------------------------------------------------------+ -- Slower tests are at the end+ ---------------------------------------------------------------------------++ ---------------------------------------------------------------------------+ -- Semigroup/Monoidal Composition strict ordering checks+ ---------------------------------------------------------------------------++ -- test both (<>) and mappend to make sure we are using correct instance+ -- for Monoid that is using the right version of semigroup. Instance+ -- deriving can cause us to pick wrong instances sometimes.++ describe "WSerial interleaved (<>) ordering check" $ interleaveCheck wSerially (<>)+ describe "WSerial interleaved mappend ordering check" $ interleaveCheck wSerially mappend++ -- describe "WAsync interleaved (<>) ordering check" $ interleaveCheck wAsyncly (<>)+ -- describe "WAsync interleaved mappend ordering check" $ interleaveCheck wAsyncly mappend++ describe "Async (<>) time order check" $ parallelCheck asyncly (<>)+ describe "Async mappend time order check" $ parallelCheck asyncly mappend++ -- XXX this keeps failing intermittently, need to investigate+ -- describe "WAsync (<>) time order check" $ parallelCheck wAsyncly (<>)+ -- describe "WAsync mappend time order check" $ parallelCheck wAsyncly mappend++ describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)+ describe "Parallel mappend time order check" $ parallelCheck parallely mappend++ ---------------------------------------------------------------------------+ -- Thread limits+ ---------------------------------------------------------------------------+ it "asyncly crosses thread limit (2000 threads)" $ runStream (asyncly $ fold $ replicate 2000 $ S.yieldM $ threadDelay 1000000)@@ -393,6 +416,90 @@ runStream (aheadly $ fold $ replicate 4000 $ S.yieldM $ threadDelay 1000000) `shouldReturn` ()+++checkFoldxStrictness :: IO ()+checkFoldxStrictness = do+ let s = return (1 :: Int) `S.consM` error "failure"+ catch (S.foldx (\_ a -> if a == 1 then error "success" else "done")+ "begin" id s)+ (\e -> case e of+ ErrorCall err -> return err+ _ -> throw e)+ `shouldReturn` "success"++checkFoldl'Strictness :: IO ()+checkFoldl'Strictness = do+ let s = return (1 :: Int) `S.consM` error "failure"+ catch (S.foldl' (\_ a -> if a == 1 then error "success" else "done")+ "begin" s)+ (\e -> case e of+ ErrorCall err -> return err+ _ -> throw e)+ `shouldReturn` "success"++checkScanxStrictness :: IO ()+checkScanxStrictness = do+ let s = return (1 :: Int) `S.consM` error "failure"+ catch+ (runStream (+ S.scanx (\_ a ->+ if a == 1+ then error "success"+ else "done")+ "begin" id s+ )+ >> return "finished"+ )+ (\e -> case e of+ ErrorCall err -> return err+ _ -> throw e)+ `shouldReturn` "success"++checkScanl'Strictness :: IO ()+checkScanl'Strictness = do+ let s = return (1 :: Int) `S.consM` error "failure"+ catch+ (runStream+ (S.scanl'+ (\_ a ->+ if a == 1+ then error "success"+ else "done")+ "begin"+ s)+ >> return "finished"+ )+ (\e -> case e of+ ErrorCall err -> return err+ _ -> throw e)+ `shouldReturn` "success"++foldlM'StrictCheck :: IORef Int -> SerialT IO Int -> IO ()+foldlM'StrictCheck ref s =+ S.foldlM' (\_ _ -> writeIORef ref 1) () s++foldxMStrictCheck :: IORef Int -> SerialT IO Int -> IO ()+foldxMStrictCheck ref s =+ S.foldxM (\_ _ -> writeIORef ref 1) (return ()) return s++checkFoldMStrictness :: (IORef Int -> SerialT IO Int -> IO ()) -> IO ()+checkFoldMStrictness f = do+ ref <- newIORef 0+ let s = return 1 `S.consM` error "x"+ catch (f ref s) (\(_ :: ErrorCall) -> return ())+ readIORef ref `shouldReturn` 1++scanlM'StrictCheck :: IORef Int -> SerialT IO Int -> SerialT IO ()+scanlM'StrictCheck ref s =+ S.scanlM' (\_ _ -> writeIORef ref 1) () s++checkScanlMStrictness :: (IORef Int -> SerialT IO Int -> SerialT IO ()) -> IO ()+checkScanlMStrictness f = do+ ref <- newIORef 0+ let s = return 1 `S.consM` error "x"+ catch (runStream $ f ref s) (\(_ :: ErrorCall) -> return ())+ readIORef ref `shouldReturn` 1 takeInfinite :: IsStream t => (t IO Int -> SerialT IO Int) -> Spec takeInfinite t = do