2888bf5 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Review/update docs, combinators in Fold module

2888bf5Harendra Kumar6 years ago

benchmark/Streamly/Benchmark/Data/Fold.hs+71−101

compare with master

unifiedsplit

@@ -10,31 +10,24 @@

10module Main (main) where

12import Control.DeepSeq (NFData(..))

13+import Data.Map.Strict (Map)

13import Data.Monoid (Last(..), Sum(..))

14import Data.Monoid (Last(..), Sum(..))

14import System.Random (randomRIO)

15import System.Random (randomRIO)

15−import Prelude (IO, Int, Double, String, (>), (<$>), (+), ($), even,

16− (<=), Monad(..), (==), Maybe(..), (.), fromIntegral,

17− compare, (>=), concat, seq, mod, fst, snd, const, Bool, Ord(..),

18− div, Num(..), odd, Either(..))

19−

20−import Data.Map.Strict (Map)

22import Streamly.Prelude (SerialT)

17import Streamly.Prelude (SerialT)

23import Streamly.Internal.Data.Fold (Fold(..))

18import Streamly.Internal.Data.Fold (Fold(..))

25import qualified Data.Map.Strict as Map

20import qualified Data.Map.Strict as Map

26−

27−import qualified Streamly.Prelude as S

28import qualified Streamly.Internal.Data.Fold as FL

21import qualified Streamly.Internal.Data.Fold as FL

29import qualified Streamly.Internal.Data.Fold.Types as FL

22import qualified Streamly.Internal.Data.Fold.Types as FL

30import qualified Streamly.Internal.Data.Pipe as Pipe

23import qualified Streamly.Internal.Data.Pipe as Pipe

31−

32import qualified Streamly.Internal.Data.Sink as Sink

24import qualified Streamly.Internal.Data.Sink as Sink

33−

34import qualified Streamly.Internal.Data.Stream.IsStream as IP

25import qualified Streamly.Internal.Data.Stream.IsStream as IP

26+import qualified Streamly.Prelude as S

36import Gauge

28import Gauge

37import Streamly.Benchmark.Common

29import Streamly.Benchmark.Common

30+import Prelude hiding (all, any, take, unzip)

39-- We need a monadic bind here to make sure that the function f does not get

32-- We need a monadic bind here to make sure that the function f does not get

40-- completely optimized out by the compiler in some cases.

33-- completely optimized out by the compiler in some cases.

@@ -75,48 +68,60 @@ all value = IP.fold (FL.all (<= value))

75take :: Monad m => Int -> SerialT m a -> m ()

68take :: Monad m => Int -> SerialT m a -> m ()

76take value = IP.fold (FL.ltake value FL.drain)

69take value = IP.fold (FL.ltake value FL.drain)

78−{-# INLINE takeSepBy #-}

71+-------------------------------------------------------------------------------

72+-- Splitting by serial application

73+-------------------------------------------------------------------------------

74+

75+{-# INLINE sliceSepBy #-}

79−takeSepBy :: Monad m => Int -> SerialT m Int -> m ()

76+sliceSepBy :: Monad m => Int -> SerialT m Int -> m ()

80−takeSepBy value = IP.fold (FL.takeSepBy (<= value) FL.drain)

77+sliceSepBy value = IP.fold (FL.sliceSepBy (>= value) FL.drain)

82{-# INLINE many #-}

79{-# INLINE many #-}

83−many :: Monad m => SerialT m Int -> m Int

80+many :: Monad m => SerialT m Int -> m ()

84−many = IP.fold (FL.many FL.length FL.sum)

81+many = IP.fold (FL.many FL.drain (FL.ltake 1 FL.drain))

86{-# INLINE splitAllAny #-}

83{-# INLINE splitAllAny #-}

87splitAllAny :: Monad m => Int -> SerialT m Int -> m (Bool, Bool)

84splitAllAny :: Monad m => Int -> SerialT m Int -> m (Bool, Bool)

88splitAllAny value =

85splitAllAny value =

86+ IP.fold

87+ (FL.splitWith (,)

89− IP.fold (FL.splitWith (,) (FL.all (<= (value `div` 2))) (FL.any (> value)))

88+ (FL.all (<= (value `div` 2)))

89+ (FL.any (> value))

90+ )

91+

92+{-

93+{-# INLINE split_ #-}

94+split_ :: Monad m

95+ => Int -> SerialT m Int -> m ()

96+split_ value =

97+ IP.fold

98+ (FL.split_

99+ (FL.all (<= (value `div` 2)))

100+ (FL.any (> value))

101+ )

102+-}

103

91−{-# INLINE teeAllAny #-}

104+-------------------------------------------------------------------------------

92−teeAllAny :: (Monad m, Ord a) => a -> SerialT m a -> m (Bool, Bool)

105+-- Distributing by parallel application

93−teeAllAny value =

106+-------------------------------------------------------------------------------

94− IP.fold (FL.teeWith (,) (FL.all (<= value)) (FL.any (> value)))

107

96{-# INLINE teeSumLength #-}

108{-# INLINE teeSumLength #-}

97teeSumLength :: Monad m => SerialT m Int -> m (Int, Int)

109teeSumLength :: Monad m => SerialT m Int -> m (Int, Int)

98teeSumLength = IP.fold (FL.teeWith (,) FL.sum FL.length)

110teeSumLength = IP.fold (FL.teeWith (,) FL.sum FL.length)

111

100−{-

112+{-# INLINE teeApplicative #-}

101−{-# INLINE applicativeAllAny #-}

102−applicativeAllAny :: (Monad m, Ord a) => a -> SerialT m a -> m (Bool, Bool)

113+teeApplicative :: (Monad m, Ord a) => a -> SerialT m a -> m (Bool, Bool)

103−applicativeAllAny value =

114+teeApplicative value =

104− IP.fold ((,) <$> FL.all (<= value) <*> (FL.any (> value)))

115+ IP.fold ((,) <$> FL.all (<= value) <*> FL.any (> value))

105−

106−{-# INLINE applicativeSumLength #-}

107−applicativeSumLength :: Monad m => SerialT m Int -> m (Int, Int)

108−applicativeSumLength = IP.fold ((,) <$> FL.sum <*> FL.length)

109−-}

110−

111−{-# INLINE distribute_ #-}

112−distribute_ :: Monad m => SerialT m Int -> m ()

113−distribute_ =

114− IP.fold (FL.distribute_ [const () <$> FL.sum, const () <$> FL.length])

115

116

116{-# INLINE distribute #-}

117{-# INLINE distribute #-}

117distribute :: Monad m => SerialT m Int -> m [Int]

118distribute :: Monad m => SerialT m Int -> m [Int]

118distribute = IP.fold (FL.distribute [FL.sum, FL.length])

119distribute = IP.fold (FL.distribute [FL.sum, FL.length])

119

120

121+-------------------------------------------------------------------------------

122+-- Partitioning

123+-------------------------------------------------------------------------------

124+

120{-# INLINE partition #-}

125{-# INLINE partition #-}

121partition :: Monad m => SerialT m Int -> m (Int, Int)

126partition :: Monad m => SerialT m Int -> m (Int, Int)

122partition =

127partition =

@@ -126,10 +131,6 @@ partition =

126 else Right a

131 else Right a

127 in IP.fold $ FL.lmap f (FL.partition FL.sum FL.length)

132 in IP.fold $ FL.lmap f (FL.partition FL.sum FL.length)

128

133

129−{-# INLINE unzip #-}

130−unzip :: Monad m => SerialT m Int -> m (Int, Int)

131−unzip = IP.fold $ FL.lmap (\a -> (a, a)) (FL.unzip FL.sum FL.length)

132−

133{-# INLINE demuxWith #-}

134{-# INLINE demuxWith #-}

134demuxWith ::

135demuxWith ::

135 (Monad m, Ord k)

136 (Monad m, Ord k)

@@ -139,29 +140,28 @@ demuxWith ::

139 -> m (Map k b)

140 -> m (Map k b)

140demuxWith f mp = S.fold (FL.demuxWith f mp)

141demuxWith f mp = S.fold (FL.demuxWith f mp)

141

142

142−{-# INLINE demuxWith_ #-}

143+{-# INLINE demuxDefaultWith #-}

143−demuxWith_ ::

144+demuxDefaultWith ::

144− (Monad m, Ord k)

145− => (a -> (k, a'))

146− -> Map k (Fold m a' b)

147− -> SerialT m a

148− -> m ()

149−demuxWith_ f mp = S.fold (FL.demuxWith_ f mp)

150−

151−{-# INLINE demuxWithDefault_ #-}

152−demuxWithDefault_ ::

153 (Monad m, Ord k, Num b)

145 (Monad m, Ord k, Num b)

154 => (a -> (k, b))

146 => (a -> (k, b))

155 -> Map k (Fold m b b)

147 -> Map k (Fold m b b)

156 -> SerialT m a

148 -> SerialT m a

157− -> m ()

149+ -> m (Map k b, b)

158−demuxWithDefault_ f mp = S.fold (FL.demuxWithDefault_ f mp (FL.lmap snd FL.sum))

150+demuxDefaultWith f mp = S.fold (FL.demuxDefaultWith f mp (FL.lmap snd FL.sum))

159

151

160{-# INLINE classifyWith #-}

152{-# INLINE classifyWith #-}

161classifyWith ::

153classifyWith ::

162 (Monad m, Ord k, Num a) => (a -> k) -> SerialT m a -> m (Map k a)

154 (Monad m, Ord k, Num a) => (a -> k) -> SerialT m a -> m (Map k a)

163classifyWith f = S.fold (FL.classifyWith f FL.sum)

155classifyWith f = S.fold (FL.classifyWith f FL.sum)

164

156

157+-------------------------------------------------------------------------------

158+-- unzip

159+-------------------------------------------------------------------------------

160+

161+{-# INLINE unzip #-}

162+unzip :: Monad m => SerialT m Int -> m (Int, Int)

163+unzip = IP.fold $ FL.lmap (\a -> (a, a)) (FL.unzip FL.sum FL.length)

164+

165-------------------------------------------------------------------------------

166-- Benchmarks

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

@@ -175,14 +175,6 @@ o_1_space_serial_elimination value =

175 [ benchIOSink value "drain" (S.fold FL.drain)

176 , benchIOSink value "drainBy" (S.fold (FL.drainBy return))

177 , benchIOSink value "drainN" (S.fold (FL.drainN value))

178− , benchIOSink

179− value

180− "drainSepByTrue"

181− (S.fold (FL.drainSepBy $ (<=) (value + 1)))

182− , benchIOSink

183− value

184− "drainSepByFalse"

185− (S.fold (FL.drainSepBy $ (>=) (value + 1)))

186 , benchIOSink value "sink" (S.fold $ Sink.toFold Sink.drain)

178 , benchIOSink value "sink" (S.fold $ Sink.toFold Sink.drain)

187 , benchIOSink value "last" (S.fold FL.last)

179 , benchIOSink value "last" (S.fold FL.last)

188 , benchIOSink value "length" (S.fold FL.length)

180 , benchIOSink value "length" (S.fold FL.length)

@@ -196,28 +188,19 @@ o_1_space_serial_elimination value =

196 , benchIOSink

188 , benchIOSink

197 value

189 value

198 "mean"

190 "mean"

199− (\s ->

200− S.fold

201− FL.mean

202− (S.map (fromIntegral :: Int -> Double) s))

191+ (S.fold FL.mean . S.map (fromIntegral :: Int -> Double))

203 , benchIOSink

192 , benchIOSink

204 value

193 value

205 "variance"

194 "variance"

206− (\s ->

207− S.fold

208− FL.variance

209− (S.map (fromIntegral :: Int -> Double) s))

195+ (S.fold FL.variance . S.map (fromIntegral :: Int -> Double))

210 , benchIOSink

196 , benchIOSink

211 value

197 value

212 "stdDev"

198 "stdDev"

213− (\s ->

214− S.fold

215− FL.stdDev

216− (S.map (fromIntegral :: Int -> Double) s))

199+ (S.fold FL.stdDev . S.map (fromIntegral :: Int -> Double))

217 , benchIOSink

200 , benchIOSink

218 value

201 value

219 "mconcat"

202 "mconcat"

220− (S.fold FL.mconcat . (S.map (Last . Just)))

203+ (S.fold FL.mconcat . S.map (Last . Just))

221 , benchIOSink

204 , benchIOSink

222 value

205 value

223 "foldMap"

206 "foldMap"

@@ -247,15 +230,9 @@ o_1_space_serial_elimination value =

247 , benchIOSink value "all" $ all value

230 , benchIOSink value "all" $ all value

248 , benchIOSink value "any" $ any value

231 , benchIOSink value "any" $ any value

249 , benchIOSink value "take" $ take value

232 , benchIOSink value "take" $ take value

250− , benchIOSink value "takeSepBy" $ takeSepBy value

233+ , benchIOSink value "sliceSepBy" $ sliceSepBy value

251− , benchIOSink

252− value

253− "and"

254− (\s -> S.fold FL.and (S.map (<= (value + 1)) s))

234+ , benchIOSink value "and" (S.fold FL.and . S.map (<= (value + 1)))

255− , benchIOSink

256− value

257− "or"

258− (\s -> S.fold FL.or (S.map (> (value + 1)) s))

235+ , benchIOSink value "or" (S.fold FL.or . S.map (> (value + 1)))

259 ]

236 ]

260 ]

237 ]

261

238

@@ -285,30 +262,24 @@ o_1_space_serial_composition :: Int -> [Benchmark]

285o_1_space_serial_composition value =

262o_1_space_serial_composition value =

286 [ bgroup

263 [ bgroup

287 "composition"

264 "composition"

288− -- Applicative

265+ [ benchIOSink value "splitWith (all, any)" $ splitAllAny value

289− [ benchIOSink value "tee (all, any)" $ teeAllAny value

266+ , benchIOSink value "teeApplicative (all, any)"

267+ $ teeApplicative value

268+ , benchIOSink value "many drain (take 1)" many

290− , benchIOSink value "tee (sum, length)" $ teeSumLength

269+ , benchIOSink value "tee (sum, length)" teeSumLength

291− -- , benchIOSink value "applicative (all, any)"

270+ , benchIOSink value "distribute [sum, length]" distribute

292− -- $ applicativeAllAny value

293− -- , benchIOSink value "applicative (sum, length)"

271+ , benchIOSink value "partition (sum, length)" partition

294− -- $ applicativeSumLength

295− , benchIOSink value "distribute_ [sum, length]" $ distribute_

272+ , benchIOSink value "unzip (sum, length)" unzip

296− , benchIOSink value "distribute [sum, length]" $ distribute

273+ , benchIOSink value "demuxDefaultWith [sum, length] sum"

297− , benchIOSink value "partition (sum, length)" $ partition

274+ $ demuxDefaultWith fn mp

298− , benchIOSink value "unzip (sum, length)" $ unzip

299 , benchIOSink value "demuxWith [sum, length]" $ demuxWith fn mp

275 , benchIOSink value "demuxWith [sum, length]" $ demuxWith fn mp

300− , benchIOSink value "demuxWith_ [sum, length]"

301− $ demuxWith_ fn mp

302− , benchIOSink value "demuxWithDefault_ [sum, length] sum"

303− $ demuxWithDefault_ fn mp

304 , benchIOSink value "classifyWith sum" $ classifyWith (fst . fn)

276 , benchIOSink value "classifyWith sum" $ classifyWith (fst . fn)

305− , benchIOSink value "many length sum" many

306− , benchIOSink value "split (all, any)" $ splitAllAny value

307 ]

277 ]

308 ]

278 ]

309

279

310 where

280 where

311

281

282+ -- We use three keys 0, 1, and 2. 0 and 1 are mapped and 3 is unmapped.

312 fn x = (x `mod` 3, x)

283 fn x = (x `mod` 3, x)

313

284

314 mp = Map.fromList [(0, FL.sum), (1, FL.length)]

285 mp = Map.fromList [(0, FL.sum), (1, FL.length)]

@@ -341,6 +312,5 @@ main = do

341 , o_1_space_serial_transformation value

312 , o_1_space_serial_transformation value

342 , o_1_space_serial_composition value

313 , o_1_space_serial_composition value

343 ]

314 ]

344− , bgroup (o_n_heap_prefix moduleName) $ concat

315+ , bgroup (o_n_heap_prefix moduleName) (o_n_heap_serial value)

345− [ o_n_heap_serial value ]

346 ]

316 ]

src/Streamly/Internal/Data/Fold.hs+320−342

@@ -6,9 +6,8 @@

6-- Maintainer : streamly@composewell.com

7-- Stability : experimental

8-- Portability : GHC

9−

9+--

10−-- Also see the "Streamly.Internal.Data.Sink" module that provides specialized

10+-- See "Streamly.Data.Fold" for an overview.

11−-- left folds that discard the outputs.

12--

11--

13-- IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude

12-- IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude

@@ -18,13 +17,7 @@ module Streamly.Internal.Data.Fold

18 Step (..)

17 Step (..)

19 , Fold (..)

18 , Fold (..)

21− , hoist

20+ -- * Creating

22− , generally

23−

24− -- , tail

25− -- , init

26−

27− -- * Fold Creation Utilities

28 , mkAccum

21 , mkAccum

29 , mkAccum_

22 , mkAccum_

30 , mkAccumM

23 , mkAccumM

@@ -34,19 +27,23 @@ module Streamly.Internal.Data.Fold

34 , mkFoldM

27 , mkFoldM

35 , mkFoldM_

28 , mkFoldM_

37− -- ** Full Folds

30+ -- * Generators

31+ , yield

32+ , yieldM

33+

34+ -- * Accumulators

35+ -- ** Semigroups and Monoids

36+ , sconcat

37+ , mconcat

38+ , foldMap

39+ , foldMapM

40+

41+ -- ** Reducers

38 , drain

42 , drain

39 , drainBy

43 , drainBy

40 , drainBy2

44 , drainBy2

41 , last

45 , last

42 , length

46 , length

43− , sum

44− , product

45− , maximumBy

46− , maximum

47− , minimumBy

48− , minimum

49− -- , the

50 , mean

47 , mean

51 , variance

48 , variance

52 , stdDev

49 , stdDev

@@ -55,22 +52,24 @@ module Streamly.Internal.Data.Fold

55 , rollingHashFirstN

52 , rollingHashFirstN

56 -- , rollingHashLastN

53 -- , rollingHashLastN

58− -- ** Fold Semigroups

55+ -- ** Saturating Reducers

59− , sconcat

56+ -- | 'product' terminates if it becomes 0. Other folds can theoretically

60−

57+ -- saturate on bounded types, and therefore terminate, however, they will

61− -- ** Full Folds (Monoidal)

58+ -- run forever on unbounded types like Integer/Double.

62− , mconcat

59+ , sum

63− , foldMap

60+ , product

64− , foldMapM

61+ , maximumBy

65−

62+ , maximum

66− -- ** Full Folds (To Containers)

63+ , minimumBy

64+ , minimum

66+ -- ** Collectors

68 , toList

67 , toList

69 , toListRevF -- experimental

68 , toListRevF -- experimental

69+ , toParallelSVar

71− -- ** Partial Folds

71+ -- * Terminating Folds

72 , drainN

73− , drainSepBy

74 -- , lastN

73 -- , lastN

75 -- , (!!)

74 -- , (!!)

76 -- , genericIndex

75 -- , genericIndex

@@ -84,18 +83,27 @@ module Streamly.Internal.Data.Fold

84 , null

83 , null

85 , elem

84 , elem

86 , notElem

85 , notElem

87− -- XXX these are slower than right folds even when full input is used

88 , all

86 , all

89 , any

87 , any

90 , and

88 , and

91 , or

89 , or

90+ -- , the

91+ , toParallelSVarLimited

93− -- * Transformations

93+ -- * Adapting

94+ , hoist

95+ , generally

96+

97+ -- * Running Incrementally

98+ , initialize

99+ , runStep

100

95− -- ** Covariant Operations

101+ -- * Output Transformations

96 , sequence

102 , sequence

97 , mapM

103 , mapM

104

105+ -- * Input Transformations

106+

99 -- ** Mapping

107 -- ** Mapping

100 , transform

108 , transform

101 , lmap

109 , lmap

@@ -127,37 +135,35 @@ module Streamly.Internal.Data.Fold

127 , lreverse

135 , lreverse

128 -}

136 -}

129

137

130− -- * Parsing

131 -- ** Trimming

138 -- ** Trimming

132 , ltake

139 , ltake

133− -- , lrunFor -- time

140+ , takeByTime

134− , takeSepBy

141+ -- By elements

142+ , sliceSepBy

143+ -- , breakOn

144+ , sliceSepByMax

145+ , sliceEndWith

146+ -- , breakAfter

135 {-

147 {-

136− , ltakeWhileM

137 , ldrop

148 , ldrop

138 , ldropWhile

149 , ldropWhile

139 , ldropWhileM

150 , ldropWhileM

140 -}

151 -}

141

152

142− , lsessionsOf

153+ -- * Splitting

143− , lchunksOf

144−

145− -- ** Breaking

146

154

147 -- Binary

155 -- Binary

156+ -- , tail

157+ -- , init

148 , splitAt -- spanN

158 , splitAt -- spanN

149 -- , splitIn -- sessionN

159 -- , splitIn -- sessionN

150

160

151− -- By elements

161+ -- XXX To be moved to parsers

152 , span -- spanWhile

162 , span -- spanWhile

153 , break -- breakBefore

163 , break -- breakBefore

154− -- , breakAfter

155− -- , breakOn

156 -- , breakAround

164 -- , breakAround

157 , spanBy

165 , spanBy

158 , spanByRolling

166 , spanByRolling

159− , sliceSepBy

160− , sliceSepWith

161

167

162 -- By sequences

168 -- By sequences

163 -- , breakOnSeq

169 -- , breakOnSeq

@@ -166,8 +172,12 @@ module Streamly.Internal.Data.Fold

166 -- * Distributing

172 -- * Distributing

167

173

168 , tee

174 , tee

175+ , teeWith

176+ , teeWithFst

177+ , teeWithMin

169 , distribute

178 , distribute

170− , distribute_

179+ -- , distributeFst

180+ -- , distributeMin

171

181

172 -- * Partitioning

182 -- * Partitioning

173

183

@@ -179,17 +189,19 @@ module Streamly.Internal.Data.Fold

179

189

180 -- * Demultiplexing

190 -- * Demultiplexing

181

191

182− , demux

192+ , demux -- XXX rename this to demux_

183 , demuxWith

193 , demuxWith

184− , demux_

194+ , demuxDefault -- XXX rename this to demux

185− , demuxDefault_

195+ , demuxDefaultWith

186− , demuxWith_

196+ -- , demuxWithSel

187− , demuxWithDefault_

197+ -- , demuxWithMin

188

198

189 -- * Classifying

199 -- * Classifying

190

200

191 , classify

201 , classify

192 , classifyWith

202 , classifyWith

203+ -- , classifyWithSel

204+ -- , classifyWithMin

193

205

194 -- * Unzipping

206 -- * Unzipping

195 , unzip

207 , unzip

@@ -198,18 +210,15 @@ module Streamly.Internal.Data.Fold

198 , unzipWithFstM

210 , unzipWithFstM

199 , unzipWithMinM

211 , unzipWithMinM

200

212

201− -- * Nested Folds

213+ -- * Nesting

202− -- , concatMap

203 , many

214 , many

204− , duplicate

215+ , lsessionsOf

205−

216+ , lchunksOf

206− -- * Running Folds

217+ , chunksBetween

207− , initialize

208− , runStep

209

218

210− -- * Folding to SVar

219+ , concatSequence

211− , toParallelSVar

220+ , concatMap

212− , toParallelSVarLimited

221+ , duplicate

213 )

222 )

214where

223where

215

224

@@ -222,15 +231,14 @@ import Data.Maybe (isJust, fromJust)

222#if __GLASGOW_HASKELL__ < 804

231#if __GLASGOW_HASKELL__ < 804

223import Data.Semigroup (Semigroup((<>)))

232import Data.Semigroup (Semigroup((<>)))

224#endif

233#endif

234+import Streamly.Internal.Data.Either.Strict (Either'(..))

235+import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

225import Streamly.Internal.Data.Pipe.Types (Pipe (..), PipeState(..))

236import Streamly.Internal.Data.Pipe.Types (Pipe (..), PipeState(..))

226import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))

237import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))

227−import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

228−import Streamly.Internal.Data.Either.Strict (Either'(..))

229

238

239+import qualified Data.Map.Strict as Map

230import qualified Streamly.Internal.Data.Fold.Types as FL

240import qualified Streamly.Internal.Data.Fold.Types as FL

231import qualified Streamly.Internal.Data.Pipe.Types as Pipe

241import qualified Streamly.Internal.Data.Pipe.Types as Pipe

232−import qualified Data.Map.Strict as Map

233−import qualified Data.Set as Set

234import qualified Prelude

242import qualified Prelude

235

243

236import Prelude hiding

244import Prelude hiding

@@ -542,7 +550,7 @@ sum = Fold (\x a -> return $ Partial $ x + a) (return 0) return

542-- > product = fmap getProduct $ FL.foldMap Product

550-- > product = fmap getProduct $ FL.foldMap Product

543--

551--

544-- @since 0.7.0

552-- @since 0.7.0

545−-- /since 0.8.0 (Added 'Eq' constraint)/

553+-- /Since 0.8.0 (Added 'Eq' constraint)/

546{-# INLINE product #-}

554{-# INLINE product #-}

547product :: (Monad m, Num a, Eq a) => Fold m a a

555product :: (Monad m, Num a, Eq a) => Fold m a a

548product = Fold step (return 1) return

556product = Fold step (return 1) return

@@ -758,7 +766,7 @@ mconcat = sconcat mempty

758-- @since 0.7.0

766-- @since 0.7.0

759{-# INLINABLE foldMap #-}

767{-# INLINABLE foldMap #-}

760foldMap :: (Monad m, Monoid b

768foldMap :: (Monad m, Monoid b

761−#if __GLASGOW_HASKELL__ < 804

769+#if !MIN_VERSION_base(4,11,0)

762 , Semigroup b

770 , Semigroup b

763#endif

771#endif

764 ) => (a -> b) -> Fold m a b

772 ) => (a -> b) -> Fold m a b

@@ -816,12 +824,6 @@ toList = Fold (\f x -> return $ Partial $ f . (x :))

816drainN :: Monad m => Int -> Fold m a ()

824drainN :: Monad m => Int -> Fold m a ()

817drainN n = ltake n drain

825drainN n = ltake n drain

818

826

819−-- | A fold that drains elements of its input as long as the predicate succeeds,

820−-- running the effects and discarding the results.

821−{-# INLINABLE drainSepBy #-}

822−drainSepBy :: Monad m => (a -> Bool) -> Fold m a ()

823−drainSepBy p = takeSepBy p drain

824−

825------------------------------------------------------------------------------

827------------------------------------------------------------------------------

826-- To Elements

828-- To Elements

827------------------------------------------------------------------------------

829------------------------------------------------------------------------------

@@ -932,12 +934,15 @@ elemIndex a = findIndex (a ==)

932null :: Monad m => Fold m a Bool

934null :: Monad m => Fold m a Bool

933null = Fold (\() _ -> return $ Done False) (return ()) (\() -> return True)

935null = Fold (\() _ -> return $ Done False) (return ()) (\() -> return True)

934

936

935−-- |

937+--

936-- > any p = lmap p or

938-- > any p = lmap p or

937-- > any p = fmap getAny . FL.foldMap (Any . p)

939-- > any p = fmap getAny . FL.foldMap (Any . p)

938--

940--

939-- | Returns 'True' if any of the elements of a stream satisfies a predicate.

941-- | Returns 'True' if any of the elements of a stream satisfies a predicate.

940--

942--

943+-- >>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]

944+-- > True

945+--

941-- @since 0.7.0

946-- @since 0.7.0

942{-# INLINE any #-}

947{-# INLINE any #-}

943any :: Monad m => (a -> Bool) -> Fold m a Bool

948any :: Monad m => (a -> Bool) -> Fold m a Bool

@@ -962,12 +967,15 @@ any predicate = Fold step initial return

962elem :: (Eq a, Monad m) => a -> Fold m a Bool

967elem :: (Eq a, Monad m) => a -> Fold m a Bool

963elem a = any (a ==)

968elem a = any (a ==)

964

969

965−-- |

970+--

966-- > all p = lmap p and

971-- > all p = lmap p and

967-- > all p = fmap getAll . FL.foldMap (All . p)

972-- > all p = fmap getAll . FL.foldMap (All . p)

968--

973--

969-- | Returns 'True' if all elements of a stream satisfy a predicate.

974-- | Returns 'True' if all elements of a stream satisfy a predicate.

970--

975--

976+-- >>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]

977+-- > False

978+--

971-- @since 0.7.0

979-- @since 0.7.0

972{-# INLINABLE all #-}

980{-# INLINABLE all #-}

973all :: Monad m => (a -> Bool) -> Fold m a Bool

981all :: Monad m => (a -> Bool) -> Fold m a Bool

@@ -1050,11 +1058,10 @@ or = any (== True)

1050-- >>> splitAt_ 4 [1,2,3]

1058-- >>> splitAt_ 4 [1,2,3]

1051-- > ([1,2,3],[])

1059-- > ([1,2,3],[])

1052--

1060--

1061+-- > splitAt n f1 f2 = splitWith (,) (ltake n f1) f2

1062+--

1053-- /Internal/

1063-- /Internal/

1054

1064

1055−-- This can be considered as a two-fold version of 'ltake' where we take both

1056−-- the segments instead of discarding the leftover.

1057−--

1058{-# INLINE splitAt #-}

1065{-# INLINE splitAt #-}

1059splitAt

1066splitAt

1060 :: Monad m

1067 :: Monad m

@@ -1062,7 +1069,7 @@ splitAt

1062 -> Fold m a b

1069 -> Fold m a b

1063 -> Fold m a c

1070 -> Fold m a c

1064 -> Fold m a (b, c)

1071 -> Fold m a (b, c)

1065−splitAt n fld1 = splitWith (,) (ltake n fld1)

1072+splitAt n fld = splitWith (,) (ltake n fld)

1066

1073

1067------------------------------------------------------------------------------

1074------------------------------------------------------------------------------

1068-- Element Aware APIs

1075-- Element Aware APIs

@@ -1072,7 +1079,18 @@ splitAt n fld1 = splitWith (,) (ltake n fld1)

1072-- Binary APIs

1079-- Binary APIs

1073------------------------------------------------------------------------------

1080------------------------------------------------------------------------------

1074

1081

1082+-- Note: Keep this consistent with S.splitOn. In fact we should eliminate

1083+-- S.splitOn in favor of the fold.

1084+--

1075−-- | Stops the fold at an infixed separator element, dropping the separator.

1085+-- | Consume the input until it encounters an infixed separator element (i.e.

1086+-- when the supplied predicate succeeds), dropping the separator.

1087+--

1088+-- Repeated applications of 'sliceSepBy' splits the stream on separator

1089+-- elements determined by the supplied predicate, separator is considered as

1090+-- infixed between two segments, if one side of the separator is missing then

1091+-- it is parsed as an empty stream. The supplied 'Fold' is applied on the

1092+-- split segments. With '-' representing non-separator elements and '.' as

1093+-- separator, repeated 'sliceSepBy' splits the stream as follows:

1076--

1094--

1077-- @

1095-- @

1078-- "--.--" => "--" "--"

1096-- "--.--" => "--" "--"

@@ -1080,12 +1098,41 @@ splitAt n fld1 = splitWith (,) (ltake n fld1)

1080-- ".--" => "" "--"

1098-- ".--" => "" "--"

1081-- @

1099-- @

1082--

1100--

1101+-- Repeated applications of @Fold.sliceSepBy (== x)@ on the input stream gives us

1102+-- an inverse of @Stream.intercalate (Stream.yield x)@

1103+--

1104+-- > Stream.splitOn pred f = Stream.foldMany (Fold.sliceSepBy pred f)

1105+--

1106+-- Let's use the following definition for illustration:

1107+--

1108+-- > splitOn p = Stream.foldMany (Fold.sliceSepBy pred Fold.toList)

1109+-- > splitOn' p = Stream.toList . splitOn p . Stream.fromList

1110+--

1111+-- >>> splitOn' (== '.') ""

1112+-- [""]

1113+--

1114+-- >>> splitOn' (== '.') "."

1115+-- ["",""]

1116+--

1117+-- >>> splitOn' (== '.') ".a"

1118+-- > ["","a"]

1119+--

1120+-- >>> splitOn' (== '.') "a."

1121+-- > ["a",""]

1122+--

1123+-- >>> splitOn' (== '.') "a.b"

1124+-- > ["a","b"]

1125+--

1126+-- >>> splitOn' (== '.') "a..b"

1127+-- > ["a","","b"]

1128+--

1083−-- * Stops - when the predicate succeeds.

1129+-- Stops - when the predicate succeeds.

1084--

1130--

1085-- /Internal/

1131-- /Internal/

1086{-# INLINE sliceSepBy #-}

1132{-# INLINE sliceSepBy #-}

1087sliceSepBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b

1133sliceSepBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b

1088−sliceSepBy predicate (Fold fstep finitial fextract) = Fold step initial fextract

1134+sliceSepBy predicate (Fold fstep finitial fextract) =

1135+ Fold step initial fextract

1089

1136

1090 where

1137 where

1091

1138

@@ -1096,20 +1143,56 @@ sliceSepBy predicate (Fold fstep finitial fextract) = Fold step initial fextract

1096 then fstep s a

1143 then fstep s a

1097 else Done <$> fextract s

1144 else Done <$> fextract s

1098

1145

1099−-- | Like 'sliceSepBy' but does not drop the seperator element.

1146+-- | Like 'sliceSepBy' but terminates a parse even before the separator

1147+-- is encountered if its size exceeds the specified maximum limit.

1100--

1148--

1101−-- @

1149+-- > take n = PR.sliceSepByMax (const True) n

1102−-- "--.--" => "--." "--"

1150+-- > sliceSepBy p = PR.sliceSepByMax p maxBound

1103−-- "--." => "--." ""

1151+--

1104−-- ".--" => "." "--"

1152+-- Let's use the following definitions for illustration:

1105−-- @

1153+--

1154+-- > splitOn p n = PR.many FL.toList $ PR.sliceSepByMax p n (FL.toList)

1155+-- > splitOn' p n = S.parse (splitOn p n) . S.fromList

1156+--

1157+-- >>> splitOn' (== '.') 0 ""

1158+-- [""]

1159+--

1160+-- >>> splitOn' (== '.') 0 "a"

1161+-- infinite list of empty strings

1162+--

1163+-- >>> splitOn' (== '.') 3 "hello.world"

1164+-- ["hel","lo","wor","ld"]

1165+--

1166+-- If the separator is found and the limit is reached at the same time then it

1167+-- behaves just like 'sliceSepBy' i.e. the separator is dropped.

1168+--

1169+-- >>> splitOn' (== '.') 0 "."

1170+-- ["",""]

1171+--

1172+-- >>> splitOn' (== '.') 0 ".."

1173+-- ["","",""]

1174+--

1175+-- Stops - when the predicate succeeds or the limit is reached.

1176+--

1177+-- /Internal/

1178+{-# INLINABLE sliceSepByMax #-}

1179+sliceSepByMax :: Monad m

1180+ => (a -> Bool) -> Int -> Fold m a b -> Fold m a b

1181+sliceSepByMax p n = sliceSepBy p . ltake n

1182+

1183+-- | Collect stream elements until an element succeeds the predicate. Also take

1184+-- the element on which the predicate succeeded. The succeeding element is

1185+-- treated as a suffix separator which is kept in the output segement.

1106--

1186--

1107-- * Stops - when the predicate succeeds.

1187-- * Stops - when the predicate succeeds.

1108--

1188--

1189+-- > Stream.splitWithSuffix pred f = Stream.foldMany (Fold.sliceEndWith pred f)

1190+--

1109-- /Internal/

1191-- /Internal/

1110−{-# INLINE sliceSepWith #-}

1192+--

1193+{-# INLINE sliceEndWith #-}

1111−sliceSepWith :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b

1194+sliceEndWith :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b

1112−sliceSepWith predicate (Fold fstep finitial fextract) =

1195+sliceEndWith predicate (Fold fstep finitial fextract) =

1113 Fold step initial fextract

1196 Fold step initial fextract

1114

1197

1115 where

1198 where

@@ -1375,7 +1458,6 @@ foldNil = Fold step begin done where

1375 step _ _ = return $ Partial []

1458 step _ _ = return $ Partial []

1376 done = return

1459 done = return

1377

1460

1378−-- XXX How is the performance?

1379{-# INLINE foldCons #-}

1461{-# INLINE foldCons #-}

1380foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]

1462foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]

1381foldCons = teeWith (:)

1463foldCons = teeWith (:)

@@ -1401,58 +1483,19 @@ foldCons = teeWith (:)

1401--

1483--

1402-- This is the consumer side dual of the producer side 'sequence' operation.

1484-- This is the consumer side dual of the producer side 'sequence' operation.

1403--

1485--

1486+-- Stops when all the folds stop.

1487+--

1404-- @since 0.7.0

1488-- @since 0.7.0

1405{-# INLINE distribute #-}

1489{-# INLINE distribute #-}

1406distribute :: Monad m => [Fold m a b] -> Fold m a [b]

1490distribute :: Monad m => [Fold m a b] -> Fold m a [b]

1407distribute = foldr foldCons foldNil

1491distribute = foldr foldCons foldNil

1408

1492

1409−-- XXX This is 2x times faster wiithout the terminating condition.

1410−-- | Like 'distribute' but for folds that return (), this can be more efficient

1411−-- than 'distribute' as it does not need to maintain state. This fold terminates

1412−-- when all the folds terminate.

1413−--

1414−{-# INLINE distribute_ #-}

1415−distribute_ :: Monad m => [Fold m a ()] -> Fold m a ()

1416−distribute_ fs = Fold step initial extract

1417−

1418− where

1419−

1420− initial = Prelude.mapM initialize fs

1421−

1422− -- XXX This itself can be a recursive function but we might face inlining

1423− -- problems.

1424− step [] _ = return $ Done ()

1425− step ss a = do

1426− !ss1 <- go ss a

1427− return

1428− $ if Prelude.null ss1

1429− then Done ()

1430− else Partial ss1

1431−

1432− go [] _ = return []

1433− go ((Fold s i d):ss) a = do

1434− i1 <- i

1435− res <- s i1 a

1436− case res of

1437− Partial i2 -> do

1438− rst <- go ss a

1439− let fld = Fold s (return i2) d

1440− return $ fld : rst

1441− Done () -> go ss a

1442−

1443− extract ss = go1 ss

1444−

1445− go1 [] = return ()

1446− go1 ((Fold _ i d):ss) = (i >>= d) >> go1 ss

1447−

1448------------------------------------------------------------------------------

1493------------------------------------------------------------------------------

1449-- Partitioning

1494-- Partitioning

1450------------------------------------------------------------------------------

1495------------------------------------------------------------------------------

1451

1496

1452-- | Partition the input over two folds using an 'Either' partitioning

1497-- | Partition the input over two folds using an 'Either' partitioning

1453−-- predicate. This fold terminates when both the folds terminate.

1498+-- predicate.

1454−--

1455−-- See also: 'partitionByFstM' and 'partitionByMinM'.

1456--

1499--

1457-- @

1500-- @

1458--

1501--

@@ -1490,6 +1533,13 @@ distribute_ fs = Fold step initial extract

1490--

1533--

1491-- This is the consumer side dual of the producer side 'mergeBy' operation.

1534-- This is the consumer side dual of the producer side 'mergeBy' operation.

1492--

1535--

1536+-- When one fold is done, any input meant for it is ignored until the other

1537+-- fold is also done.

1538+--

1539+-- Stops when both the folds stop.

1540+--

1541+-- /See also: 'partitionByFstM' and 'partitionByMinM'./

1542+--

1493-- @since 0.7.0

1543-- @since 0.7.0

1494{-# INLINE partitionByM #-}

1544{-# INLINE partitionByM #-}

1495partitionByM :: Monad m

1545partitionByM :: Monad m

@@ -1499,10 +1549,7 @@ partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =

1499

1549

1500 where

1550 where

1501

1551

1502− begin = do

1552+ begin = RunBoth <$> beginL <*> beginR

1503− sL <- beginL

1504− sR <- beginR

1505− return $ RunBoth sL sR

1506

1553

1507 step (RunBoth sL sR) a = do

1554 step (RunBoth sL sR) a = do

1508 r <- f a

1555 r <- f a

@@ -1542,16 +1589,9 @@ partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =

1542 Partial sres -> Partial $ RunRight bL sres

1589 Partial sres -> Partial $ RunRight bL sres

1543 Done bres -> Done (bL, bres)

1590 Done bres -> Done (bL, bres)

1544

1591

1545− done (RunBoth sL sR) = do

1592+ done (RunBoth sL sR) = (,) <$> doneL sL <*> doneR sR

1546− bL <- doneL sL

1547− bR <- doneR sR

1548− return (bL, bR)

1549− done (RunLeft sL bR) = do

1593+ done (RunLeft sL bR) = (,bR) <$> doneL sL

1550− bL <- doneL sL

1551− return (bL, bR)

1552− done (RunRight bL sR) = do

1594+ done (RunRight bL sR) = (bL,) <$> doneR sR

1553− bR <- doneR sR

1554− return (bL, bR)

1555

1595

1556-- | Similar to 'partitionByM' but terminates when the first fold terminates.

1596-- | Similar to 'partitionByM' but terminates when the first fold terminates.

1557--

1597--

@@ -1639,44 +1679,16 @@ partition = partitionBy id

1639-- ...

1679-- ...

1640-- @

1680-- @

1641--

1681--

1642−-- @since 0.7.0

1682+-- Any input that does not map to a fold in the input Map is silently ignored.

1683+--

1684+-- > demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain

1685+--

1686+-- /Internal/

1687+--

1643{-# INLINE demuxWith #-}

1688{-# INLINE demuxWith #-}

1644demuxWith :: (Monad m, Ord k)

1689demuxWith :: (Monad m, Ord k)

1645 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)

1690 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)

1646−demuxWith f kv = Fold step initial extract

1691+demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain

1647−

1648− where

1649−

1650− initial = Tuple' Map.empty <$> Prelude.mapM initialize kv

1651−

1652− step (Tuple' bmp mp) a =

1653− if Map.size mp > 0

1654− then do

1655− let (k, a') = f a

1656− case Map.lookup k mp of

1657− Nothing -> return $ Partial $ Tuple' bmp mp

1658− Just (Fold stp ini dn) -> do

1659− !st <- ini

1660− !res <- stp st a'

1661− return

1662− $ case res of

1663− Partial sres ->

1664− let fld = Fold stp (return sres) dn

1665− mp1 = Map.insert k fld mp

1666− in Partial $ Tuple' bmp mp1

1667− Done bres ->

1668− let mp1 = Map.delete k mp

1669− bmp1 = Map.insert k bres bmp

1670− in if Map.size mp1 == 0

1671− then Done bmp1

1672− else Partial $ Tuple' bmp1 mp1

1673− -- XXX This state is reached even for an empty Map. That will change

1674− -- once we have Step in the initial value

1675− else error "Illegal state"

1676−

1677− extract (Tuple' bmp mp) = do

1678− mpe <- Prelude.mapM (\(Fold _ i d) -> i >>= d) mp

1679− return $ bmp `Map.union` mpe

1680

1692

1681-- | Fold a stream of key value pairs using a map of specific folds for each

1693-- | Fold a stream of key value pairs using a map of specific folds for each

1682-- key into a map from keys to the results of fold outputs of the corresponding

1694-- key into a map from keys to the results of fold outputs of the corresponding

@@ -1695,164 +1707,119 @@ demux :: (Monad m, Ord k)

1695 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)

1707 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)

1696demux = demuxWith id

1708demux = demuxWith id

1697

1709

1698−-- data DemuxState m s = DemuxP !m !s | DemuxingOnlyMap !m

1710+data DemuxState s b rMap kvMap =

1699−data DemuxState s m1 m2 = DemuxingWithDefault !s !m1 !m2 | DemuxingOnlyMap !m2

1711+ DemuxMapWithDefault !s !rMap !kvMap

1712+ | DemuxMap b !rMap !kvMap

1713+ | DemuxDefault s !rMap

1700

1714

1701−{-# INLINE demuxWithDefault_ #-}

1715+-- | Like 'demuxWith' but uses a default catchall fold to handle inputs which

1702−demuxWithDefault_ :: (Monad m, Ord k)

1716+-- do not have a specific fold in the map to handle them.

1717+--

1718+-- If any fold in the map stops, inputs meant for that fold are sent to the

1719+-- catchall fold. If the catchall fold stops then inputs that do not match any

1720+-- fold are ignored.

1721+--

1722+-- Stops when all the folds, including the catchall fold, stop.

1723+--

1724+-- /Internal/

1725+--

1726+{-# INLINE demuxDefaultWith #-}

1727+demuxDefaultWith :: (Monad m, Ord k)

1728+ => (a -> (k, a'))

1703− => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m (k, a') b -> Fold m a ()

1729+ -> Map k (Fold m a' b)

1730+ -> Fold m (k, a') c

1731+ -> Fold m a (Map k b, c)

1704−demuxWithDefault_ f kv (Fold dstep dinitial dextract) =

1732+demuxDefaultWith f kv (Fold dstep dinitial dextract) =

1705 Fold step initial extract

1733 Fold step initial extract

1706

1734

1707 where

1735 where

1708

1736

1709− initial = do

1737+ initial =

1738+ DemuxMapWithDefault

1710− ini <- dinitial

1739+ <$> dinitial

1740+ <*> return Map.empty

1711− mp <- Prelude.mapM initialize kv

1741+ <*> Prelude.mapM initialize kv

1712− return $ DemuxingWithDefault ini Set.empty mp

1713

1742

1714− -- XXX Some code can probably be abstracted

1743+ step (DemuxMapWithDefault dacc results kvMap) a =

1715− step (DemuxingWithDefault dacc bst mp) a =

1716− if Map.size mp > 0

1744+ if Map.size kvMap > 0

1717 then do

1745 then do

1718 let (k, a1) = f a

1746 let (k, a1) = f a

1719− case Map.lookup k mp of

1747+ case Map.lookup k kvMap of

1720− Nothing ->

1748+ Nothing -> do

1721− if k `Set.member` bst

1722− then return $ Partial $ DemuxingWithDefault dacc bst mp

1723− else do

1724− res <- dstep dacc (k, a1)

1749+ res <- dstep dacc (k, a1)

1725− return

1726− $ Partial

1727− $ case res of

1728− Partial sres ->

1729− DemuxingWithDefault sres bst mp

1730− Done _ -> DemuxingOnlyMap mp

1731− Just (Fold stp ini dn) -> do

1732− !st <- ini

1733− !res <- stp st a1

1734 return

1750 return

1751+ $ Partial

1735 $ case res of

1752 $ case res of

1736− Partial sres ->

1753+ Partial s -> DemuxMapWithDefault s results kvMap

1737− let fld = Fold stp (return sres) dn

1754+ Done b -> DemuxMap b results kvMap

1738− mp1 = Map.insert k fld mp

1739− in Partial $ DemuxingWithDefault dacc bst mp1

1740− Done _ ->

1741− -- XXX We can skip the check here

1742− if Map.size mp == 1

1743− then Done ()

1744− else Partial

1745− $ DemuxingWithDefault

1746− dacc

1747− (Set.insert k bst)

1748− (Map.delete k mp)

1749− -- XXX This state is reached even for an empty Map. That will change

1750− -- once we have Step in the initial value

1751− else error "Illegal state"

1752− -- XXX Reduce code duplication?

1753− step (DemuxingOnlyMap mp) a =

1754− if Map.size mp > 0

1755− then do

1756− let (k, a1) = f a

1757− case Map.lookup k mp of

1758− Nothing -> return $ Partial $ DemuxingOnlyMap mp

1759− Just (Fold stp ini dn) -> do

1755+ Just (Fold step1 initial1 extract1) -> do

1760− !st <- ini

1756+ !st <- initial1

1761− !res <- stp st a1

1757+ !res <- step1 st a1

1762 return

1758 return

1763 $ case res of

1759 $ case res of

1764− Partial sres ->

1760+ Partial s ->

1765− Partial

1761+ let fld = Fold step1 (return s) extract1

1766− $ DemuxingOnlyMap

1767− $ Map.insert

1762+ kvMap1 = Map.insert k fld kvMap

1768− k

1763+ in Partial

1769− (Fold stp (return sres) dn)

1764+ $ DemuxMapWithDefault

1770− mp

1765+ dacc results kvMap1

1771− Done _ ->

1766+ Done b ->

1767+ let kvMap1 = Map.delete k kvMap

1768+ results1 = Map.insert k b results

1769+ in Partial

1772− if Map.size mp == 1

1770+ $ if Map.size kvMap1 == 0

1773− then Done ()

1771+ then DemuxDefault dacc results1

1774− else Partial

1772+ else DemuxMapWithDefault

1775− $ DemuxingOnlyMap $ Map.delete k mp

1773+ dacc results1 kvMap1

1776− -- XXX This state is reached even for an empty Map. That will change

1777− -- once we have Step in the initial value

1774+ -- XXX Remove this when we have Step in the initial value

1778− else error "Illegal state"

1775+ else error "demuxDefaultWith: empty key-value Map"

1779−

1776+ step (DemuxMap dval results kvMap) a =

1780− extract (DemuxingWithDefault dacc _ mp) = do

1777+ if Map.size kvMap > 0

1781− void $ dextract dacc

1778+ then do

1782− Prelude.mapM_ (\(Fold _ i d) -> i >>= d) mp

1783− extract (DemuxingOnlyMap mp) = do

1784− Prelude.mapM_ (\(Fold _ i d) -> i >>= d) mp

1785−

1786−-- | Split the input stream based on a key field and fold each split using a

1787−-- specific fold without collecting the results. Useful for cases like protocol

1788−-- handlers to handle different type of packets.

1789−--

1790−-- @

1791−--

1792−-- |-------Fold m a ()

1793−-- -----stream m a-----Map-----|

1794−-- |-------Fold m a ()

1795−-- |

1796−-- ...

1797−-- @

1798−--

1799−--

1800−-- @since 0.7.0

1801−

1802−-- demuxWith_ can be slightly faster than demuxWith because we do not need to

1803−-- update the Map in this case. This may be significant only if the map is

1804−-- large.

1805−{-# INLINE demuxWith_ #-}

1806−demuxWith_ :: (Monad m, Ord k)

1807− => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a ()

1808−demuxWith_ f kv = Fold step initial extract

1809−

1810− where

1811−

1812− initial = Tuple' (Map.size kv) <$> Prelude.mapM initialize kv

1813−

1814− step (Tuple' n mp) a

1815− | (k, a') <- f a =

1779+ let (k, a1) = f a

1816− case Map.lookup k mp of

1780+ case Map.lookup k kvMap of

1817− Nothing -> return $ Partial $ Tuple' n mp

1781+ Nothing -> return $ Partial $ DemuxMap dval results kvMap

1818− Just (Fold stp ini dn) -> do

1782+ Just (Fold step1 initial1 extract1) -> do

1819− !st <- ini

1783+ !st <- initial1

1820− !res <- stp st a'

1784+ !res <- step1 st a1

1821− let n1 = n - 1

1822 return

1785 return

1823 $ case res of

1786 $ case res of

1824− Partial sres ->

1787+ Partial s ->

1825− let fld = Fold stp (return sres) dn

1788+ let fld = Fold step1 (return s) extract1

1826− in Partial $ Tuple' n $ Map.insert k fld mp

1789+ kvMap1 = Map.insert k fld kvMap

1827− Done _ ->

1790+ in Partial $ DemuxMap dval results kvMap1

1828− if n1 == 0

1791+ Done b ->

1792+ let kvMap1 = Map.delete k kvMap

1793+ results1 = Map.insert k b results

1794+ in if Map.size kvMap1 == 0

1829− then Done ()

1795+ then Done (results1, dval)

1830− else Partial $ Tuple' n1 $ Map.delete k mp

1796+ else Partial

1831−

1797+ $ DemuxMap dval results1 kvMap1

1832− extract (Tuple' _ mp) = do

1798+ -- XXX Remove this when we have Step in the initial value

1833− Prelude.mapM_ (\(Fold _ i d) -> i >>= d) mp

1799+ else error "demuxDefaultWith: empty key-value Map"

1834−

1800+ step (DemuxDefault dacc results) a = do

1835−-- | Given a stream of key value pairs and a map from keys to folds, fold the

1801+ let (k, a1) = f a

1836−-- values for each key using the corresponding folds, discarding the outputs.

1802+ res <- dstep dacc (k, a1)

1837−--

1803+ return

1838−-- @

1804+ $ case res of

1839−-- > let prn = FL.drainBy print

1805+ Partial s -> Partial $ DemuxDefault s results

1840−-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]

1806+ Done b -> Done (results, b)

1841−-- input = S.fromList [(\"ONE",1),(\"TWO",2)]

1807+

1842−-- in S.fold (FL.demux_ table) input

1808+ extract (DemuxMapWithDefault dacc results kvMap) = do

1843−-- One 1

1809+ b <- dextract dacc

1844−-- Two 2

1810+ kvMap1 <- Prelude.mapM (\(Fold _ i d) -> i >>= d) kvMap

1845−-- @

1811+ return (results `Map.union` kvMap1, b)

1846−--

1812+ extract (DemuxMap dval results kvMap) = do

1847−-- @since 0.7.0

1813+ kvMap1 <- Prelude.mapM (\(Fold _ i d) -> i >>= d) kvMap

1848−{-# INLINE demux_ #-}

1814+ return (results `Map.union` kvMap1, dval)

1849−demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()

1815+ extract (DemuxDefault dacc results) = do

1850−demux_ = demuxWith_ id

1816+ b <- dextract dacc

1851−

1817+ return (results, b)

1852−{-# INLINE demuxDefault_ #-}

1818+

1819+{-# INLINE demuxDefault #-}

1853−demuxDefault_ :: (Monad m, Ord k)

1820+demuxDefault :: (Monad m, Ord k)

1854− => Map k (Fold m a ()) -> Fold m (k, a) () -> Fold m (k, a) ()

1821+ => Map k (Fold m a b) -> Fold m (k, a) b -> Fold m (k, a) (Map k b, b)

1855−demuxDefault_ = demuxWithDefault_ id

1822+demuxDefault = demuxDefaultWith id

1856

1823

1857-- TODO If the data is large we may need a map/hashmap in pinned memory instead

1824-- TODO If the data is large we may need a map/hashmap in pinned memory instead

1858-- of a regular Map. That may require a serializable constraint though. We can

1825-- of a regular Map. That may require a serializable constraint though. We can

@@ -1868,7 +1835,13 @@ demuxDefault_ = demuxWithDefault_ id

1868-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]

1835-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]

1869-- @

1836-- @

1870--

1837--

1871−-- @since 0.7.0

1838+-- If the classifier fold stops for a particular key any further inputs in that

1839+-- bucket are ignored.

1840+--

1841+-- /Stops: never/

1842+--

1843+-- /Internal/

1844+--

1872{-# INLINE classifyWith #-}

1845{-# INLINE classifyWith #-}

1873classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)

1846classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)

1874classifyWith f (Fold step initial extract) = Fold step' initial' extract'

1847classifyWith f (Fold step initial extract) = Fold step' initial' extract'

@@ -1936,7 +1909,7 @@ classify fld = classifyWith fst (lmap snd fld)

1936

1909

1937-- | Like 'unzipWith' but with a monadic splitter function.

1910-- | Like 'unzipWith' but with a monadic splitter function.

1938--

1911--

1939−-- -- @unzipWithM f fld1 fld2 = lmapM f (unzip fld1 fld2)@

1912+-- -- @unzipWithM k f1 f2 = lmapM k (unzip f1 f2)@

1940--

1913--

1941-- @since 0.7.0

1914-- @since 0.7.0

1942{-# INLINE unzipWithM #-}

1915{-# INLINE unzipWithM #-}

@@ -1981,8 +1954,8 @@ unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =

1981 Done bresR -> Done (bL, bresR)

1954 Done bresR -> Done (bL, bresR)

1982

1955

1983 done (RunBoth sL sR) = (,) <$> doneL sL <*> doneR sR

1956 done (RunBoth sL sR) = (,) <$> doneL sL <*> doneR sR

1984− done (RunLeft sL bR) = (,) <$> doneL sL <*> return bR

1957+ done (RunLeft sL bR) = (,bR) <$> doneL sL

1985− done (RunRight bL sR) = (,) bL <$> doneR sR

1958+ done (RunRight bL sR) = (bL,) <$> doneR sR

1986

1959

1987-- | Similar to 'unzipWithM' but terminates when the first fold terminates.

1960-- | Similar to 'unzipWithM' but terminates when the first fold terminates.

1988--

1961--

@@ -2037,10 +2010,17 @@ unzip = unzipWith id

2037-- Nesting

2010-- Nesting

2038------------------------------------------------------------------------------

2011------------------------------------------------------------------------------

2039

2012

2040−

2013+-- | @concatSequence f t@ applies folds from stream @t@ sequentially and

2041−{-

2014+-- collects the results using the fold @f@.

2042−-- XXX this would be an application of "many" using a terminating fold.

2043--

2015--

2016+-- /Unimplemented/

2017+--

2018+{-# INLINE concatSequence #-}

2019+concatSequence ::

2020+ -- IsStream t =>

2021+ Fold m b c -> t (Fold m a b) -> Fold m a c

2022+concatSequence _f _p = undefined

2023+

2044-- | Group the input stream into groups of elements between @low@ and @high@.

2024-- | Group the input stream into groups of elements between @low@ and @high@.

2045-- Collection starts in chunks of @low@ and then keeps doubling until we reach

2025-- Collection starts in chunks of @low@ and then keeps doubling until we reach

2046-- @high@. Each chunk is folded using the provided fold function.

2026-- @high@. Each chunk is folded using the provided fold function.

@@ -2049,16 +2029,14 @@ unzip = unzipWith id

2049-- size to a stream of arrays and we want to minimize the number of

2029-- size to a stream of arrays and we want to minimize the number of

2050-- allocations.

2030-- allocations.

2051--

2031--

2052−-- @

2032+-- NOTE: this would be an application of "many" using a terminating fold.

2053--

2033--

2054−-- XXX we should be able to implement it with parsers/terminating folds.

2034+-- /Unimplemented/

2055--

2035--

2056−{-# INLINE lchunksInRange #-}

2036+{-# INLINE chunksBetween #-}

2057−lchunksInRange :: Monad m

2037+chunksBetween :: -- Monad m =>

2058− => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c

2038+ Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c

2059−lchunksInRange low high (Fold step1 initial1 extract1)

2039+chunksBetween _low _high _f1 _f2 = undefined

2060− (Fold step2 initial2 extract2) = undefined

2061−-}

2062

2040

2063------------------------------------------------------------------------------

2041------------------------------------------------------------------------------

2064-- Fold to a Parallel SVar

2042-- Fold to a Parallel SVar

src/Streamly/Internal/Data/Fold/Types.hs−15

@@ -139,7 +139,6 @@ module Streamly.Internal.Data.Fold.Types

139 , lfilterM

140 , lcatMaybes

141 , ltake

142− , takeSepBy

143 , takeByTime

142 , takeByTime

144

143

145 -- * Distributing

144 -- * Distributing

@@ -691,20 +690,6 @@ ltake n (Fold fstep finitial fextract) = Fold step initial extract

691

690

692 extract (Tuple' _ r) = fextract r

691 extract (Tuple' _ r) = fextract r

693

692

694−-- | Takes elements from the input as long as the predicate succeeds.

695−--

696−-- @since 0.7.0

697−{-# INLINE takeSepBy #-}

698−takeSepBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b

699−takeSepBy predicate (Fold fstep finitial fextract) = Fold step finitial fextract

700−

701− where

702−

703− step s a =

704− if predicate a

705− then fstep s a

706− else Done <$> fextract s

707−

708------------------------------------------------------------------------------

693------------------------------------------------------------------------------

709-- Nesting

694-- Nesting

710------------------------------------------------------------------------------

695------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/StreamD.hs+1−1

@@ -1522,7 +1522,7 @@ reverse' m =

1522{-# INLINE_NORMAL splitSuffixWith #-}

1523splitSuffixWith :: Monad m

1524 => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b

1525−splitSuffixWith predicate f = foldMany1 (FL.sliceSepWith predicate f)

1525+splitSuffixWith predicate f = foldMany1 (FL.sliceEndWith predicate f)

1526

1527{-# INLINE_NORMAL groupsBy #-}

1528groupsBy :: Monad m

streamly.cabal+1

@@ -681,6 +681,7 @@ test-suite Data.Fold

681 import: test-options

682 type: exitcode-stdio-1.0

683 main-is: Streamly/Test/Data/Fold.hs

684+ other-modules: Streamly.Test.Common

684

685

685test-suite Data.Unfold

686test-suite Data.Unfold

686 import: test-options

687 import: test-options

test/Streamly/Test/Common.hs+10−2

@@ -10,6 +10,8 @@

10module Streamly.Test.Common

11 ( equals

12 , listEquals

13+ , checkListEqual

14+ , chooseInt

13 ) where

15 ) where

15import Control.Monad (when)

17import Control.Monad (when)

@@ -18,8 +20,8 @@ import Data.List ((\\))

18#if __GLASGOW_HASKELL__ < 808

20#if __GLASGOW_HASKELL__ < 808

19import Data.Semigroup ((<>))

21import Data.Semigroup ((<>))

20#endif

22#endif

21−import Test.QuickCheck (counterexample)

23+import Test.QuickCheck (Property, Gen, choose, counterexample)

22−import Test.QuickCheck.Monadic (PropertyM, assert, monitor)

24+import Test.QuickCheck.Monadic (PropertyM, assert, monitor, monadicIO)

24equals

26equals

25 :: (Show a, Monad m)

27 :: (Show a, Monad m)

@@ -51,3 +53,9 @@ listEquals eq stream list = do

51 <> "\nlist \\\\ stream " <> show (list \\ stream)

53 <> "\nlist \\\\ stream " <> show (list \\ stream)

52 )

54 )

53 assert (stream `eq` list)

55 assert (stream `eq` list)

56+

57+checkListEqual :: (Show a, Eq a) => [a] -> [a] -> Property

58+checkListEqual ls_1 ls_2 = monadicIO (listEquals (==) ls_1 ls_2)

59+

60+chooseInt :: (Int, Int) -> Gen Int

61+chooseInt = choose

test/Streamly/Test/Data/Fold.hs+41−1

@@ -1,5 +1,6 @@

1module Main (main) where

3+import Streamly.Test.Common (checkListEqual)

3import Prelude hiding (maximum, minimum, elem, notElem, null, product, sum, head, last)

4import Prelude hiding (maximum, minimum, elem, notElem, null, product, sum, head, last)

4import qualified Streamly.Internal.Data.Fold as F

5import qualified Streamly.Internal.Data.Fold as F

5import qualified Streamly.Prelude as S

6import qualified Streamly.Prelude as S

@@ -7,7 +8,8 @@ import qualified Streamly.Data.Fold as FL

8import Test.Hspec as H

9import Test.Hspec as H

9import Test.Hspec.QuickCheck

10import Test.Hspec.QuickCheck

11+import Test.QuickCheck

10−import Test.QuickCheck (Property, forAll, Gen, vectorOf, arbitrary, choose)

12+ (Property, property, forAll, Gen, vectorOf, arbitrary, choose, listOf)

11import Test.QuickCheck.Monadic (monadicIO, assert, run)

13import Test.QuickCheck.Monadic (monadicIO, assert, run)

13maxStreamLen :: Int

15maxStreamLen :: Int

@@ -19,6 +21,15 @@ intMin = minBound

19intMax :: Int

21intMax :: Int

20intMax = maxBound

22intMax = maxBound

24+min_value :: Int

25+min_value = 0

26+

27+max_value :: Int

28+max_value = 10000

29+

30+chooseInt :: (Int, Int) -> Gen Int

31+chooseInt = choose

32+

22{-# INLINE maxStreamLen #-}

33{-# INLINE maxStreamLen #-}

23{-# INLINE intMin #-}

34{-# INLINE intMin #-}

24{-# INLINE intMax #-}

35{-# INLINE intMax #-}

@@ -192,6 +203,32 @@ and ls = S.fold FL.and (S.fromList ls) `shouldReturn` Prelude.and ls

192or :: [Bool] -> Expectation

203or :: [Bool] -> Expectation

193or ls = S.fold FL.or (S.fromList ls) `shouldReturn` Prelude.or ls

204or ls = S.fold FL.or (S.fromList ls) `shouldReturn` Prelude.or ls

194

205

206+ltake :: [Int] -> Property

207+ltake ls =

208+ forAll (chooseInt (-1, Prelude.length ls + 2)) $ \n ->

209+ S.fold (F.ltake n FL.toList) (S.fromList ls)

210+ `shouldReturn` Prelude.take n ls

211+

212+sliceSepBy :: Property

213+sliceSepBy =

214+ forAll (listOf (chooseInt (0, 1))) $ \ls ->

215+ let p = (== 1)

216+ f = F.sliceSepBy p FL.toList

217+ ys = Prelude.takeWhile (not . p) ls

218+ in case S.fold f (S.fromList ls) of

219+ Right xs -> checkListEqual xs ys

220+ Left _ -> property False

221+

222+sliceSepByMax :: Property

223+sliceSepByMax =

224+ forAll (chooseInt (min_value, max_value)) $ \n ->

225+ forAll (listOf (chooseInt (0, 1))) $ \ls ->

226+ let p = (== 1)

227+ f = F.sliceSepByMax p n FL.toList

228+ ys = Prelude.take n (Prelude.takeWhile (not . p) ls)

229+ in case S.fold f (S.fromList ls) of

230+ Right xs -> checkListEqual xs ys

231+ Left _ -> property False

195

232

196main :: IO ()

233main :: IO ()

197main = hspec $

234main = hspec $

@@ -219,3 +256,6 @@ main = hspec $

219 prop "And" Main.and

256 prop "And" Main.and

220 prop "Or" Main.or

257 prop "Or" Main.or

221 prop "mapMaybe" mapMaybe

258 prop "mapMaybe" mapMaybe

259+ prop "ltake" ltake

260+ prop "sliceSepBy" sliceSepBy

261+ prop "sliceSepByMax" sliceSepByMax