55f9cfa · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

fix foldxM strictness

55f9cfaHarendra Kumar8 years ago

Changelog.md+6

compare with master

unifiedsplit

@@ -1,3 +1,9 @@

1+## Unreleased

3+### Bug Fixes

5+* foldxM was not fully strict, fixed.

1## 0.4.0

7## 0.4.0

3### Breaking changes

9### Breaking changes

benchmark/BaseStreams.hs+2−2

@@ -37,7 +37,7 @@ main = do

37 , benchIO "unfoldrM" D.toNull D.sourceUnfoldrM

38 , benchIO "fromEnum" D.toNull D.sourceFromEnum

40− , benchIO "fromFoldable" D.toNull D.sourceFromFoldable

40+ , benchIO "fromList" D.toNull D.sourceFromList

41 -- , benchIO "fromFoldableM" D.sourceFromFoldableM

42 ]

43 , bgroup "elimination"

@@ -64,7 +64,7 @@ main = do

64 -- , benchIO "fromEnum" K.toNull K.sourceFromEnum

66 , benchIO "fromFoldable" K.toNull K.sourceFromFoldable

67− , benchIO "fromFoldableM" K.toNull K.sourceFromFoldableM

67+ -- , benchIO "fromFoldableM" K.toNull K.sourceFromFoldableM

69 -- appends

70 , benchIO "foldMapWith" K.toNull K.sourceFoldMapWith

benchmark/StreamDOps.hs+3−3

@@ -96,9 +96,9 @@ sourceUnfoldrM n = S.unfoldrM step n

96sourceFromEnum :: Monad m => Int -> Stream m Int

97sourceFromEnum n = S.enumFromStepN n 1 value

99−{-# INLINE sourceFromFoldable #-}

99+{-# INLINE sourceFromList #-}

100−sourceFromFoldable :: Monad m => Int -> Stream m Int

100+sourceFromList :: Monad m => Int -> Stream m Int

101−sourceFromFoldable n = S.fromList [n..n+value]

101+sourceFromList n = S.fromList [n..n+value]

102

103{-# INLINE source #-}

104source :: Monad m => Int -> Stream m Int

benchmark/StreamKOps.hs+14−13

@@ -13,10 +13,9 @@ import Prelude

13 (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=),

14 subtract, undefined, Maybe(..), not)

16−import qualified Streamly.Streams.StreamK as S hiding (runStream)

16+import qualified Streamly.Streams.StreamK as S

17−-- import qualified Streamly.Streams.Serial as S

17+import qualified Streamly.Streams.Prelude as S

18−import qualified Streamly as S

18+import qualified Streamly.SVar as S

19−import qualified Streamly.Prelude as P

21value, maxValue :: Int

20value, maxValue :: Int

22value = 1000000

21value = 1000000

@@ -68,7 +67,7 @@ mapM :: S.MonadAsync m => Stream m Int -> m ()

68-- Stream generation and elimination

67-- Stream generation and elimination

69-------------------------------------------------------------------------------

68-------------------------------------------------------------------------------

71−type Stream m a = S.SerialT m a

70+type Stream m a = S.Stream m a

73{-# INLINE sourceUnfoldr #-}

72{-# INLINE sourceUnfoldr #-}

74sourceUnfoldr :: Int -> Stream m Int

73sourceUnfoldr :: Int -> Stream m Int

@@ -98,17 +97,19 @@ sourceFromEnum n = S.enumFromStepN n 1 value

98sourceFromFoldable :: Int -> Stream m Int

97sourceFromFoldable :: Int -> Stream m Int

99sourceFromFoldable n = S.fromFoldable [n..n+value]

98sourceFromFoldable n = S.fromFoldable [n..n+value]

100

100+{-

101{-# INLINE sourceFromFoldableM #-}

102sourceFromFoldableM :: S.MonadAsync m => Int -> Stream m Int

103−sourceFromFoldableM n = P.fromFoldableM (Prelude.fmap return [n..n+value])

103+sourceFromFoldableM n = S.fromFoldableM (Prelude.fmap return [n..n+value])

104+-}

104

105

105{-# INLINE sourceFoldMapWith #-}

106{-# INLINE sourceFoldMapWith #-}

106−sourceFoldMapWith :: Monad m => Int -> Stream m Int

107+sourceFoldMapWith :: Int -> Stream m Int

107−sourceFoldMapWith n = S.foldMapWith (S.<>) return [n..n+value]

108+sourceFoldMapWith n = S.foldMapWith (S.serial) S.yield [n..n+value]

108

109

109{-# INLINE sourceFoldMapWithM #-}

110{-# INLINE sourceFoldMapWithM #-}

110sourceFoldMapWithM :: Monad m => Int -> Stream m Int

111sourceFoldMapWithM :: Monad m => Int -> Stream m Int

111−sourceFoldMapWithM n = S.foldMapWith (S.<>) (S.yieldM . return) [n..n+value]

112+sourceFoldMapWithM n = S.foldMapWith (S.serial) (S.yieldM . return) [n..n+value]

112

113

113{-# INLINE source #-}

114{-# INLINE source #-}

114source :: S.MonadAsync m => Int -> Stream m Int

115source :: S.MonadAsync m => Int -> Stream m Int

@@ -160,15 +161,15 @@ filterAllOut = transform . S.filter (> maxValue)

160filterAllIn = transform . S.filter (<= maxValue)

161filterAllIn = transform . S.filter (<= maxValue)

161takeOne = transform . S.take 1

162takeOne = transform . S.take 1

162takeAll = transform . S.take maxValue

163takeAll = transform . S.take maxValue

163−takeWhileTrue = transform . P.takeWhile (<= maxValue)

164+takeWhileTrue = transform . S.takeWhile (<= maxValue)

164−dropAll = transform . P.drop maxValue

165+dropAll = transform . S.drop maxValue

165−dropWhileTrue = transform . P.dropWhile (<= maxValue)

166+dropWhileTrue = transform . S.dropWhile (<= maxValue)

166

167

167-------------------------------------------------------------------------------

168-------------------------------------------------------------------------------

168-- Zipping and concat

169-- Zipping and concat

169-------------------------------------------------------------------------------

170-------------------------------------------------------------------------------

170

171

171−zip src = transform $ (P.zipWith (,) src src)

172+zip src = transform $ (S.zipWith (,) src src)

172concat _n = return ()

173concat _n = return ()

173

174

174-------------------------------------------------------------------------------

175-------------------------------------------------------------------------------

src/Streamly.hs+81

@@ -55,6 +55,14 @@

55-- import Streamly

56-- @

58+{-# LANGUAGE CPP #-}

59+

60+#if __GLASGOW_HASKELL__ >= 800

61+{-# OPTIONS_GHC -Wno-orphans #-}

62+#endif

63+

64+#include "Streamly/Streams/inline.h"

65+

58module Streamly

66module Streamly

59 (

67 (

60 MonadAsync

68 MonadAsync

@@ -168,6 +176,79 @@ import Streamly.Streams.SVar (maxThreads, maxBuffer)

168import Streamly.SVar (MonadAsync)

176import Streamly.SVar (MonadAsync)

169import Data.Semigroup (Semigroup(..))

177import Data.Semigroup (Semigroup(..))

170

178

179+import qualified Streamly.Streams.StreamD as D

180+import qualified Streamly.Streams.StreamK as K

181+

182+-- XXX This should perhaps be moved to Prelude.

183+

184+------------------------------------------------------------------------------

185+-- Eliminating a stream

186+------------------------------------------------------------------------------

187+

188+-- | Run a streaming composition, discard the results. By default it interprets

189+-- the stream as 'SerialT', to run other types of streams use the type adapting

190+-- combinators for example @runStream . 'asyncly'@.

191+--

192+-- @since 0.2.0

193+{-# INLINE_EARLY runStream #-}

194+runStream :: Monad m => SerialT m a -> m ()

195+runStream m = D.runStream $ D.fromStreamK (toStream m)

196+{-# RULES "runStream fallback to CPS" [1]

197+ forall a. D.runStream (D.fromStreamK a) = K.runStream a #-}

198+

199+-- | Same as 'runStream'

200+--

201+-- @since 0.1.0

202+{-# DEPRECATED runStreaming "Please use runStream instead." #-}

203+runStreaming :: (Monad m, IsStream t) => t m a -> m ()

204+runStreaming = runStream . K.adapt

205+

206+-- | Same as @runStream@.

207+--

208+-- @since 0.1.0

209+{-# DEPRECATED runStreamT "Please use runStream instead." #-}

210+runStreamT :: Monad m => SerialT m a -> m ()

211+runStreamT = runStream

212+

213+-- | Same as @runStream . wSerially@.

214+--

215+-- @since 0.1.0

216+{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}

217+runInterleavedT :: Monad m => WSerialT m a -> m ()

218+runInterleavedT = runStream . K.adapt

219+

220+-- | Same as @runStream . parallely@.

221+--

222+-- @since 0.1.0

223+{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}

224+runParallelT :: Monad m => ParallelT m a -> m ()

225+runParallelT = runStream . K.adapt

226+

227+-- | Same as @runStream . asyncly@.

228+--

229+-- @since 0.1.0

230+{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}

231+runAsyncT :: Monad m => AsyncT m a -> m ()

232+runAsyncT = runStream . K.adapt

233+

234+-- | Same as @runStream . zipping@.

235+--

236+-- @since 0.1.0

237+{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}

238+runZipStream :: Monad m => ZipSerialM m a -> m ()

239+runZipStream = runStream . K.adapt

240+

241+-- | Same as @runStream . zippingAsync@.

242+--

243+-- @since 0.1.0

244+{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}

245+runZipAsync :: Monad m => ZipAsyncM m a -> m ()

246+runZipAsync = runStream . K.adapt

247+

248+------------------------------------------------------------------------------

249+-- Documentation

250+------------------------------------------------------------------------------

251+

171-- $serial

252-- $serial

172--

253--

173-- Serial streams compose serially or non-concurrently. In a composed stream,

254-- Serial streams compose serially or non-concurrently. In a composed stream,

src/Streamly/Streams/Prelude.hs+1−87

@@ -9,12 +9,6 @@

9{-# LANGUAGE StandaloneDeriving #-}

10{-# LANGUAGE UndecidableInstances #-} -- XXX

12−#if __GLASGOW_HASKELL__ >= 800

13−{-# OPTIONS_GHC -Wno-orphans #-}

14−#endif

15−

16−#include "inline.h"

17−

18-- |

12-- |

19-- Module : Streamly.Streams.Prelude

13-- Module : Streamly.Streams.Prelude

@@ -27,96 +21,16 @@

27--

21--

28module Streamly.Streams.Prelude

22module Streamly.Streams.Prelude

29 (

23 (

30− -- * Elimination

31− runStream

32− , runStreaming -- deprecated

33− , runStreamT -- deprecated

34− , runInterleavedT -- deprecated

35− , runParallelT -- deprecated

36− , runAsyncT -- deprecated

37− , runZipStream -- deprecated

38− , runZipAsync -- deprecated

39−

40 -- * Fold Utilities

24 -- * Fold Utilities

41− , foldWith

25+ foldWith

42 , foldMapWith

26 , foldMapWith

43 , forEachWith

27 , forEachWith

44 )

28 )

45where

29where

47import Streamly.Streams.StreamK (IsStream(..))

31import Streamly.Streams.StreamK (IsStream(..))

48−import Streamly.Streams.Serial (SerialT, WSerialT)

49−import Streamly.Streams.Parallel (ParallelT)

50−import Streamly.Streams.Async (AsyncT)

51−import Streamly.Streams.Zip (ZipSerialM, ZipAsyncM)

52−

53−import qualified Streamly.Streams.StreamD as D

54import qualified Streamly.Streams.StreamK as K

32import qualified Streamly.Streams.StreamK as K

56−------------------------------------------------------------------------------

57−-- Eliminating a stream

58−------------------------------------------------------------------------------

59−

60−-- | Run a streaming composition, discard the results. By default it interprets

61−-- the stream as 'SerialT', to run other types of streams use the type adapting

62−-- combinators for example @runStream . 'asyncly'@.

63−--

64−-- @since 0.2.0

65−{-# INLINE_EARLY runStream #-}

66−runStream :: Monad m => SerialT m a -> m ()

67−runStream m = D.runStream $ D.fromStreamK (toStream m)

68−{-# RULES "runStream fallback to CPS" [1]

69− forall a. D.runStream (D.fromStreamK a) = K.runStream a #-}

70−

71−-- | Same as 'runStream'

72−--

73−-- @since 0.1.0

74−{-# DEPRECATED runStreaming "Please use runStream instead." #-}

75−runStreaming :: (Monad m, IsStream t) => t m a -> m ()

76−runStreaming = runStream . K.adapt

77−

78−-- | Same as @runStream@.

79−--

80−-- @since 0.1.0

81−{-# DEPRECATED runStreamT "Please use runStream instead." #-}

82−runStreamT :: Monad m => SerialT m a -> m ()

83−runStreamT = runStream

84−

85−-- | Same as @runStream . wSerially@.

86−--

87−-- @since 0.1.0

88−{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}

89−runInterleavedT :: Monad m => WSerialT m a -> m ()

90−runInterleavedT = runStream . K.adapt

91−

92−-- | Same as @runStream . parallely@.

93−--

94−-- @since 0.1.0

95−{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}

96−runParallelT :: Monad m => ParallelT m a -> m ()

97−runParallelT = runStream . K.adapt

98−

99−-- | Same as @runStream . asyncly@.

100−--

101−-- @since 0.1.0

102−{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}

103−runAsyncT :: Monad m => AsyncT m a -> m ()

104−runAsyncT = runStream . K.adapt

105−

106−-- | Same as @runStream . zipping@.

107−--

108−-- @since 0.1.0

109−{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}

110−runZipStream :: Monad m => ZipSerialM m a -> m ()

111−runZipStream = runStream . K.adapt

112−

113−-- | Same as @runStream . zippingAsync@.

114−--

115−-- @since 0.1.0

116−{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}

117−runZipAsync :: Monad m => ZipAsyncM m a -> m ()

118−runZipAsync = runStream . K.adapt

119−

120------------------------------------------------------------------------------

34------------------------------------------------------------------------------

121-- Fold Utilities

35-- Fold Utilities

122------------------------------------------------------------------------------

36------------------------------------------------------------------------------

src/Streamly/Streams/StreamK.hs+46−2

@@ -116,6 +116,10 @@ module Streamly.Streams.StreamK

116 -- ** Map and Filter

117 , mapMaybe

118

119+ -- ** Zipping

120+ , zipWith

121+ , zipWithM

122+

119 -- * Semigroup Style Composition

123 -- * Semigroup Style Composition

120 , serial

124 , serial

121

125

@@ -137,7 +141,7 @@ import Data.Semigroup (Semigroup(..))

137import Prelude

141import Prelude

138 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,

142 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,

139 take, filter, all, any, takeWhile, drop, dropWhile, minimum,

143 take, filter, all, any, takeWhile, drop, dropWhile, minimum,

140− maximum, elem, notElem, null, head, tail)

144+ maximum, elem, notElem, null, head, tail, zipWith)

141import qualified Prelude

145import qualified Prelude

142

146

143import Streamly.SVar

147import Streamly.SVar

@@ -521,7 +525,7 @@ foldxM step begin done m = go begin (toStream m)

521 go !acc m1 =

525 go !acc m1 =

522 let stop = acc >>= done

526 let stop = acc >>= done

523 single a = acc >>= \b -> step b a >>= done

527 single a = acc >>= \b -> step b a >>= done

524− yieldk a r = acc >>= \b -> go (step b a) r

528+ yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r

525 in (unStream m1) defState stop single yieldk

529 in (unStream m1) defState stop single yieldk

526

530

527-- | Like 'foldl'' but with a monadic step function.

531-- | Like 'foldl'' but with a monadic step function.

@@ -811,6 +815,46 @@ mapMaybe f m = go (toStream m)

811 Nothing -> (unStream r) (rstState st) stp single yieldk

815 Nothing -> (unStream r) (rstState st) stp single yieldk

812 in unStream m1 (rstState st) stp single yieldk

816 in unStream m1 (rstState st) stp single yieldk

813

817

818+------------------------------------------------------------------------------

819+-- Serial Zipping

820+------------------------------------------------------------------------------

821+

822+{-# INLINE zipWithS #-}

823+zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

824+zipWithS f m1 m2 = go m1 m2

825+ where

826+ go mx my = Stream $ \st stp sng yld -> do

827+ let merge a ra =

828+ let single2 b = sng (f a b)

829+ yield2 b rb = yld (f a b) (go ra rb)

830+ in unStream my (rstState st) stp single2 yield2

831+ let single1 a = merge a nil

832+ yield1 a ra = merge a ra

833+ unStream mx (rstState st) stp single1 yield1

834+

835+-- | Zip two streams serially using a pure zipping function.

836+--

837+-- @since 0.1.0

838+{-# INLINABLE zipWith #-}

839+zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

840+zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)

841+

842+-- | Zip two streams serially using a monadic zipping function.

843+--

844+-- @since 0.1.0

845+zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

846+zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

847+ where

848+ go mx my = Stream $ \st stp sng yld -> do

849+ let merge a ra =

850+ let runIt x = unStream x (rstState st) stp sng yld

851+ single2 b = f a b >>= sng

852+ yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)

853+ in unStream my (rstState st) stp single2 yield2

854+ let single1 a = merge a nil

855+ yield1 a ra = merge a ra

856+ unStream mx (rstState st) stp single1 yield1

857+

814------------------------------------------------------------------------------

858------------------------------------------------------------------------------

815-- Semigroup

859-- Semigroup

816------------------------------------------------------------------------------

860------------------------------------------------------------------------------

src/Streamly/Streams/Zip.hs+5−45

@@ -20,8 +20,8 @@

20--

21module Streamly.Streams.Zip

22 (

23− zipWith

23+ K.zipWith

24− , zipWithM

24+ , K.zipWithM

25 , zipAsyncWith

26 , zipAsyncWithM

@@ -50,46 +50,6 @@ import qualified Streamly.Streams.StreamK as K

51#include "Instances.hs"

53−------------------------------------------------------------------------------

54−-- Serial Zipping

55−------------------------------------------------------------------------------

56−

57−{-# INLINE zipWithS #-}

58−zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

59−zipWithS f m1 m2 = go m1 m2

60− where

61− go mx my = Stream $ \st stp sng yld -> do

62− let merge a ra =

63− let single2 b = sng (f a b)

64− yield2 b rb = yld (f a b) (go ra rb)

65− in unStream my (rstState st) stp single2 yield2

66− let single1 a = merge a K.nil

67− yield1 a ra = merge a ra

68− unStream mx (rstState st) stp single1 yield1

69−

70−-- | Zip two streams serially using a pure zipping function.

71−--

72−-- @since 0.1.0

73−{-# INLINABLE zipWith #-}

74−zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

75−zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)

76−

77−-- | Zip two streams serially using a monadic zipping function.

78−--

79−-- @since 0.1.0

80−zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

81−zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

82− where

83− go mx my = Stream $ \st stp sng yld -> do

84− let merge a ra =

85− let runIt x = unStream x (rstState st) stp sng yld

86− single2 b = f a b >>= sng

87− yield2 b rb = f a b >>= \x -> runIt (x `K.cons` go ra rb)

88− in unStream my (rstState st) stp single2 yield2

89− let single1 a = merge a K.nil

90− yield1 a ra = merge a ra

91− unStream mx (rstState st) stp single1 yield1

92−

93------------------------------------------------------------------------------

53------------------------------------------------------------------------------

94-- Serially Zipping Streams

54-- Serially Zipping Streams

95------------------------------------------------------------------------------

55------------------------------------------------------------------------------

@@ -157,7 +117,7 @@ instance Monad m => Functor (ZipSerialM m) where

157

117

158instance Monad m => Applicative (ZipSerialM m) where

118instance Monad m => Applicative (ZipSerialM m) where

159 pure = ZipSerialM . K.repeat

119 pure = ZipSerialM . K.repeat

160− m1 <*> m2 = fromStream $ zipWith id (toStream m1) (toStream m2)

120+ m1 <*> m2 = fromStream $ K.zipWith id (toStream m1) (toStream m2)

161

121

162------------------------------------------------------------------------------

122------------------------------------------------------------------------------

163-- Parallel Zipping

123-- Parallel Zipping

@@ -172,7 +132,7 @@ zipAsyncWith :: (IsStream t, MonadAsync m)

172zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do

132zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do

173 ma <- mkAsync' (rstState st) m1

133 ma <- mkAsync' (rstState st) m1

174 mb <- mkAsync' (rstState st) m2

134 mb <- mkAsync' (rstState st) m2

175− unStream (toStream (zipWith f ma mb)) (rstState st) stp sng yld

135+ unStream (toStream (K.zipWith f ma mb)) (rstState st) stp sng yld

176

136

177-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

137-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

178-- concurrently) using a monadic zipping function.

138-- concurrently) using a monadic zipping function.

@@ -183,7 +143,7 @@ zipAsyncWithM :: (IsStream t, MonadAsync m)

183zipAsyncWithM f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do

143zipAsyncWithM f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do

184 ma <- mkAsync' (rstState st) m1

144 ma <- mkAsync' (rstState st) m1

185 mb <- mkAsync' (rstState st) m2

145 mb <- mkAsync' (rstState st) m2

186− unStream (toStream (zipWithM f ma mb)) (rstState st) stp sng yld

146+ unStream (toStream (K.zipWithM f ma mb)) (rstState st) stp sng yld

187

147

188------------------------------------------------------------------------------

148------------------------------------------------------------------------------

189-- Parallely Zipping Streams

149-- Parallely Zipping Streams

stack-7.10.yaml+2−2

@@ -15,5 +15,5 @@ extra-deps:

15flags: {}

16extra-package-dbs: []

17# For mac ports installed SDL library on Mac OS X

18−#extra-include-dirs:

18+extra-include-dirs:

19−#- /opt/local/include

19+- /opt/local/include

streamly.cabal+53−26

@@ -274,12 +274,11 @@ test-suite parallel-loops

274-- Benchmarks

275-------------------------------------------------------------------------------

276

277−benchmark base

277+benchmark linear

278 type: exitcode-stdio-1.0

279 hs-source-dirs: benchmark

280− main-is: BaseStreams.hs

280+ main-is: Linear.hs

281− other-modules: StreamDOps

281+ other-modules: LinearOps

282− , StreamKOps

283 default-language: Haskell2010

282 default-language: Haskell2010

284 ghc-options: -O2 -Wall

283 ghc-options: -O2 -Wall

285 if flag(dev)

284 if flag(dev)

@@ -295,22 +294,18 @@ benchmark base

295 -Wredundant-constraints

294 -Wredundant-constraints

296 -Wnoncanonical-monad-instances

295 -Wnoncanonical-monad-instances

297 -Wnoncanonical-monadfail-instances

296 -Wnoncanonical-monadfail-instances

298− if flag(dev)

299− buildable: True

300− build-depends:

297+ build-depends:

301− streamly

298+ streamly

302− , base >= 4.8 && < 5

299+ , base >= 4.8 && < 5

303− , deepseq >= 1.4.0 && < 1.5

300+ , deepseq >= 1.4.0 && < 1.5

304− , random >= 1.0 && < 2.0

301+ , random >= 1.0 && < 2.0

305− , gauge >= 0.2.3 && < 0.3

302+ , gauge >= 0.2.3 && < 0.3

306− else

307− buildable: False

308

303

309−benchmark linear

304+benchmark nested

310 type: exitcode-stdio-1.0

305 type: exitcode-stdio-1.0

311 hs-source-dirs: benchmark

306 hs-source-dirs: benchmark

312− main-is: Linear.hs

307+ main-is: Nested.hs

313− other-modules: LinearOps

308+ other-modules: NestedOps

314 default-language: Haskell2010

309 default-language: Haskell2010

315 ghc-options: -O2 -Wall

310 ghc-options: -O2 -Wall

316 if flag(dev)

311 if flag(dev)

@@ -333,11 +328,25 @@ benchmark linear

333 , random >= 1.0 && < 2.0

328 , random >= 1.0 && < 2.0

334 , gauge >= 0.2.3 && < 0.3

329 , gauge >= 0.2.3 && < 0.3

335

330

331+-------------------------------------------------------------------------------

332+-- Internal benchmarks for unexposed modules

333+-------------------------------------------------------------------------------

334+

335+-- We have to copy the streamly library modules here because there is no

336+-- way to use unexposed modules from the library.

337+

336−benchmark nested

338+benchmark base

337 type: exitcode-stdio-1.0

339 type: exitcode-stdio-1.0

338− hs-source-dirs: benchmark

340+ hs-source-dirs: benchmark, src

339− main-is: Nested.hs

341+ main-is: BaseStreams.hs

340− other-modules: NestedOps

342+ other-modules: Streamly.SVar

343+ , Streamly.Streams.StreamK

344+ , Streamly.Streams.StreamD

345+ , Streamly.Streams.Prelude

346+

347+ , StreamDOps

348+ , StreamKOps

349+

341 default-language: Haskell2010

350 default-language: Haskell2010

342 ghc-options: -O2 -Wall

351 ghc-options: -O2 -Wall

343 if flag(dev)

352 if flag(dev)

@@ -353,12 +362,30 @@ benchmark nested

353 -Wredundant-constraints

362 -Wredundant-constraints

354 -Wnoncanonical-monad-instances

363 -Wnoncanonical-monad-instances

355 -Wnoncanonical-monadfail-instances

364 -Wnoncanonical-monadfail-instances

365+

356 build-depends:

366 build-depends:

357− streamly

358− , base >= 4.8 && < 5

367+ base >= 4.8 && < 5

359− , deepseq >= 1.4.0 && < 1.5

368+ , deepseq >= 1.4.0 && < 1.5

360− , random >= 1.0 && < 2.0

369+ , random >= 1.0 && < 2.0

361− , gauge >= 0.2.3 && < 0.3

370+ , gauge >= 0.2.3 && < 0.3

371+

372+ , ghc-prim >= 0.2 && < 0.6

373+ , containers >= 0.5 && < 0.6

374+ , heaps >= 0.3 && < 0.4

375+

376+ -- concurrency

377+ , atomic-primops >= 0.8 && < 0.9

378+ , lockfree-queue >= 0.2.3 && < 0.3

379+

380+ , exceptions >= 0.8 && < 0.11

381+ , monad-control >= 1.0 && < 2

382+ , mtl >= 2.2 && < 3

383+ , transformers >= 0.4 && < 0.6

384+ , transformers-base >= 0.4 && < 0.5

385+

386+ if impl(ghc < 8.0)

387+ build-depends:

388+ semigroups >= 0.18 && < 0.19

362

389

363executable chart-linear

390executable chart-linear

364 default-language: Haskell2010

391 default-language: Haskell2010

test/Main.hs+132−25

@@ -1,16 +1,19 @@

1{-# LANGUAGE CPP #-}

2{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE RankNTypes #-}

4+{-# LANGUAGE ScopedTypeVariables #-}

5module Main (main) where

6module Main (main) where

7import Control.Concurrent (threadDelay)

8import Control.Concurrent (threadDelay)

8−import Control.Exception (Exception, try)

9+import Control.Exception (Exception, try, ErrorCall(..), catch, throw)

9import Control.Monad.Catch (throwM, MonadThrow)

10import Control.Monad.Catch (throwM, MonadThrow)

10import Control.Monad.Error.Class (throwError, MonadError)

11import Control.Monad.Error.Class (throwError, MonadError)

11import Control.Monad.Trans.Except (runExceptT, ExceptT)

12import Control.Monad.Trans.Except (runExceptT, ExceptT)

12import Data.Foldable (forM_, fold)

13import Data.Foldable (forM_, fold)

13import Data.List (sort)

14import Data.List (sort)

15+

16+import Data.IORef

14import Test.Hspec

17import Test.Hspec

16import Streamly

19import Streamly

@@ -88,30 +91,6 @@ main = hspec $ do

88 (toListParallel ((,) <$> (return 1) <*> (return 2 <> return 3)) >>= return . sort)

91 (toListParallel ((,) <$> (return 1) <*> (return 2 <> return 3)) >>= return . sort)

89 `shouldReturn` ([(1,2),(1,3)] :: [(Int, Int)])

92 `shouldReturn` ([(1,2),(1,3)] :: [(Int, Int)])

91− ---------------------------------------------------------------------------

92− -- Semigroup/Monoidal Composition strict ordering checks

93− ---------------------------------------------------------------------------

94−

95− -- test both (<>) and mappend to make sure we are using correct instance

96− -- for Monoid that is using the right version of semigroup. Instance

97− -- deriving can cause us to pick wrong instances sometimes.

98−

99− describe "WSerial interleaved (<>) ordering check" $ interleaveCheck wSerially (<>)

100− describe "WSerial interleaved mappend ordering check" $ interleaveCheck wSerially mappend

101−

102− -- describe "WAsync interleaved (<>) ordering check" $ interleaveCheck wAsyncly (<>)

103− -- describe "WAsync interleaved mappend ordering check" $ interleaveCheck wAsyncly mappend

104−

105− describe "Async (<>) time order check" $ parallelCheck asyncly (<>)

106− describe "Async mappend time order check" $ parallelCheck asyncly mappend

107−

108− -- XXX this keeps failing intermittently, need to investigate

109− -- describe "WAsync (<>) time order check" $ parallelCheck wAsyncly (<>)

110− -- describe "WAsync mappend time order check" $ parallelCheck wAsyncly mappend

111−

112− describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)

113− describe "Parallel mappend time order check" $ parallelCheck parallely mappend

114−

115 ---------------------------------------------------------------------------

94 ---------------------------------------------------------------------------

116 -- Monoidal Compositions, multiset equality checks

95 -- Monoidal Compositions, multiset equality checks

117 ---------------------------------------------------------------------------

96 ---------------------------------------------------------------------------

@@ -384,6 +363,50 @@ main = hspec $ do

384 describe "take on infinite concurrent stream" $ takeInfinite wAsyncly

363 describe "take on infinite concurrent stream" $ takeInfinite wAsyncly

385 describe "take on infinite concurrent stream" $ takeInfinite aheadly

364 describe "take on infinite concurrent stream" $ takeInfinite aheadly

386

365

366+ ---------------------------------------------------------------------------

367+ -- Folds are strict enough

368+ ---------------------------------------------------------------------------

369+

370+ it "foldx is strict enough" checkFoldxStrictness

371+ it "foldl' is strict enough" checkFoldl'Strictness

372+ it "scanx is strict enough" checkScanxStrictness

373+ it "scanl' is strict enough" checkScanl'Strictness

374+ it "foldxM is strict enough" (checkFoldMStrictness foldxMStrictCheck)

375+ it "foldlM' is strict enough" (checkFoldMStrictness foldlM'StrictCheck)

376+ it "scanlM' is strict enough" (checkScanlMStrictness scanlM'StrictCheck)

377+

378+ ---------------------------------------------------------------------------

379+ -- Slower tests are at the end

380+ ---------------------------------------------------------------------------

381+

382+ ---------------------------------------------------------------------------

383+ -- Semigroup/Monoidal Composition strict ordering checks

384+ ---------------------------------------------------------------------------

385+

386+ -- test both (<>) and mappend to make sure we are using correct instance

387+ -- for Monoid that is using the right version of semigroup. Instance

388+ -- deriving can cause us to pick wrong instances sometimes.

389+

390+ describe "WSerial interleaved (<>) ordering check" $ interleaveCheck wSerially (<>)

391+ describe "WSerial interleaved mappend ordering check" $ interleaveCheck wSerially mappend

392+

393+ -- describe "WAsync interleaved (<>) ordering check" $ interleaveCheck wAsyncly (<>)

394+ -- describe "WAsync interleaved mappend ordering check" $ interleaveCheck wAsyncly mappend

395+

396+ describe "Async (<>) time order check" $ parallelCheck asyncly (<>)

397+ describe "Async mappend time order check" $ parallelCheck asyncly mappend

398+

399+ -- XXX this keeps failing intermittently, need to investigate

400+ -- describe "WAsync (<>) time order check" $ parallelCheck wAsyncly (<>)

401+ -- describe "WAsync mappend time order check" $ parallelCheck wAsyncly mappend

402+

403+ describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)

404+ describe "Parallel mappend time order check" $ parallelCheck parallely mappend

405+

406+ ---------------------------------------------------------------------------

407+ -- Thread limits

408+ ---------------------------------------------------------------------------

409+

387 it "asyncly crosses thread limit (2000 threads)" $

410 it "asyncly crosses thread limit (2000 threads)" $

388 runStream (asyncly $ fold $

411 runStream (asyncly $ fold $

389 replicate 2000 $ S.yieldM $ threadDelay 1000000)

412 replicate 2000 $ S.yieldM $ threadDelay 1000000)

@@ -394,6 +417,90 @@ main = hspec $ do

394 replicate 4000 $ S.yieldM $ threadDelay 1000000)

417 replicate 4000 $ S.yieldM $ threadDelay 1000000)

395 `shouldReturn` ()

418 `shouldReturn` ()

396

419

420+

421+checkFoldxStrictness :: IO ()

422+checkFoldxStrictness = do

423+ let s = return (1 :: Int) `S.consM` error "failure"

424+ catch (S.foldx (\_ a -> if a == 1 then error "success" else "done")

425+ "begin" id s)

426+ (\e -> case e of

427+ ErrorCall err -> return err

428+ _ -> throw e)

429+ `shouldReturn` "success"

430+

431+checkFoldl'Strictness :: IO ()

432+checkFoldl'Strictness = do

433+ let s = return (1 :: Int) `S.consM` error "failure"

434+ catch (S.foldl' (\_ a -> if a == 1 then error "success" else "done")

435+ "begin" s)

436+ (\e -> case e of

437+ ErrorCall err -> return err

438+ _ -> throw e)

439+ `shouldReturn` "success"

440+

441+checkScanxStrictness :: IO ()

442+checkScanxStrictness = do

443+ let s = return (1 :: Int) `S.consM` error "failure"

444+ catch

445+ (runStream (

446+ S.scanx (\_ a ->

447+ if a == 1

448+ then error "success"

449+ else "done")

450+ "begin" id s

451+ )

452+ >> return "finished"

453+ )

454+ (\e -> case e of

455+ ErrorCall err -> return err

456+ _ -> throw e)

457+ `shouldReturn` "success"

458+

459+checkScanl'Strictness :: IO ()

460+checkScanl'Strictness = do

461+ let s = return (1 :: Int) `S.consM` error "failure"

462+ catch

463+ (runStream

464+ (S.scanl'

465+ (\_ a ->

466+ if a == 1

467+ then error "success"

468+ else "done")

469+ "begin"

470+ s)

471+ >> return "finished"

472+ )

473+ (\e -> case e of

474+ ErrorCall err -> return err

475+ _ -> throw e)

476+ `shouldReturn` "success"

477+

478+foldlM'StrictCheck :: IORef Int -> SerialT IO Int -> IO ()

479+foldlM'StrictCheck ref s =

480+ S.foldlM' (\_ _ -> writeIORef ref 1) () s

481+

482+foldxMStrictCheck :: IORef Int -> SerialT IO Int -> IO ()

483+foldxMStrictCheck ref s =

484+ S.foldxM (\_ _ -> writeIORef ref 1) (return ()) return s

485+

486+checkFoldMStrictness :: (IORef Int -> SerialT IO Int -> IO ()) -> IO ()

487+checkFoldMStrictness f = do

488+ ref <- newIORef 0

489+ let s = return 1 `S.consM` error "x"

490+ catch (f ref s) (\(_ :: ErrorCall) -> return ())

491+ readIORef ref `shouldReturn` 1

492+

493+scanlM'StrictCheck :: IORef Int -> SerialT IO Int -> SerialT IO ()

494+scanlM'StrictCheck ref s =

495+ S.scanlM' (\_ _ -> writeIORef ref 1) () s

496+

497+checkScanlMStrictness :: (IORef Int -> SerialT IO Int -> SerialT IO ()) -> IO ()

498+checkScanlMStrictness f = do

499+ ref <- newIORef 0

500+ let s = return 1 `S.consM` error "x"

501+ catch (runStream $ f ref s) (\(_ :: ErrorCall) -> return ())

502+ readIORef ref `shouldReturn` 1

503+

397takeInfinite :: IsStream t => (t IO Int -> SerialT IO Int) -> Spec

504takeInfinite :: IsStream t => (t IO Int -> SerialT IO Int) -> Spec

398takeInfinite t = do

505takeInfinite t = do

399 it "take 1" $

506 it "take 1" $