b1106ff · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Merge pull request #194 from composewell/misc-fixes

b1106ffHarendra Kumar7 years ago

Changelog.md+5

compare with master

unifiedsplit

@@ -4,6 +4,11 @@

5* Change the signature of `foldrM` to ensure that it is lazy

6* `scanx` would now require an additional `Monad m` constraint.

7+* Change the signature of `iterateM` to ensure that it is lazy.

9+### Bug Fixes

10+

11+* Fix a bug that caused `uniq` function to yield the same element twice.

8### Major Enhancements

13### Major Enhancements

benchmark/LinearOps.hs+8

@@ -85,6 +85,14 @@ sourceFromList n = S.fromList [n..n+value]

85sourceFromListM :: (S.MonadAsync m, S.IsStream t) => Int -> t m Int

86sourceFromListM n = S.fromListM (Prelude.fmap return [n..n+value])

88+{-# INLINE sourceFromIndices #-}

89+sourceFromIndices :: (Monad m, S.IsStream t) => Int -> t m Int

90+sourceFromIndices n = S.take value $ S.fromIndices (+ n)

91+

92+{-# INLINE sourceFromIndicesM #-}

93+sourceFromIndicesM :: (S.MonadAsync m, S.IsStream t) => Int -> t m Int

94+sourceFromIndicesM n = S.take value $ S.fromIndicesM (Prelude.fmap return (+ n))

95+

88{-# INLINE sourceFromFoldable #-}

96{-# INLINE sourceFromFoldable #-}

89sourceFromFoldable :: S.IsStream t => Int -> t m Int

97sourceFromFoldable :: S.IsStream t => Int -> t m Int

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

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

src/Streamly/Prelude.hs+17−11

@@ -680,9 +680,8 @@ yieldM = K.yieldM

680-- @since 0.6.0

681{-# INLINE fromIndices #-}

682fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a

683−fromIndices = fromStreamD . D.fromIndices

683+fromIndices = fromStreamS . S.fromIndices

684

685−-- XXX this needs to be concurrent

686--

685--

687-- |

686-- |

688-- @

687-- @

@@ -692,10 +691,17 @@ fromIndices = fromStreamD . D.fromIndices

692-- Generate an infinite stream, whose values are the output of a monadic

691-- Generate an infinite stream, whose values are the output of a monadic

693-- function @f@ applied on the corresponding index. Index starts at 0.

692-- function @f@ applied on the corresponding index. Index starts at 0.

694--

693--

694+-- /Concurrent/

695+--

695-- @since 0.6.0

696-- @since 0.6.0

696−{-# INLINE fromIndicesM #-}

697+{-# INLINE_EARLY fromIndicesM #-}

697−fromIndicesM :: (IsStream t, Monad m) => (Int -> m a) -> t m a

698+fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a

698−fromIndicesM = fromStreamD . D.fromIndicesM

699+fromIndicesM = K.fromIndicesM

700+

701+{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}

702+{-# INLINE fromIndicesMSerial #-}

703+fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a

704+fromIndicesMSerial = fromStreamS . S.fromIndicesM

699

705

700-- |

706-- |

701-- @

707-- @

@@ -774,7 +780,7 @@ iterate step = fromStream . go

774

780

775-- |

781-- |

776-- @

782-- @

777−-- iterateM f m = m \`consM` iterateM f m

783+-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)

778-- @

784-- @

779--

785--

780-- Generate an infinite stream with the first element generated by the action

786-- Generate an infinite stream with the first element generated by the action

@@ -788,21 +794,21 @@ iterate step = fromStream . go

788--

794--

789-- @

795-- @

790-- drain $ serially $ S.take 10 $ S.iterateM

796-- drain $ serially $ S.take 10 $ S.iterateM

791−-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0

797+-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)

792--

798--

793-- drain $ asyncly $ S.take 10 $ S.iterateM

799-- drain $ asyncly $ S.take 10 $ S.iterateM

794−-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0

800+-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)

795-- @

801-- @

796--

802--

797-- /Concurrent/

803-- /Concurrent/

798--

804--

799-- @since 0.1.2

805-- @since 0.1.2

800−iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a

806+iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a

801iterateM step = go

807iterateM step = go

802 where

808 where

803 go s = K.mkStream $ \st stp sng yld -> do

809 go s = K.mkStream $ \st stp sng yld -> do

804− next <- step s

810+ next <- s

805− K.foldStreamShared st stp sng yld (return s |: go next)

811+ K.foldStreamShared st stp sng yld (return next |: go (step next))

806

812

807------------------------------------------------------------------------------

813------------------------------------------------------------------------------

808-- Conversions

814-- Conversions

src/Streamly/Streams/StreamD.hs+1−1

@@ -2107,7 +2107,7 @@ uniq (Stream step state) = Stream step' (Nothing, state)

2107 r <- step gst st

2108 case r of

2109 Yield y s | x == y -> return $ Skip (Just x, s)

2110− | otherwise -> return $ Yield x (Just y, s)

2110+ | otherwise -> return $ Yield y (Just y, s)

2111 Skip s -> return $ Skip (Just x, s)

2112 Stop -> return Stop

2113

src/Streamly/Streams/StreamK.hs+15

@@ -63,6 +63,8 @@ module Streamly.Streams.StreamK

63 , repeat

64 , replicate

65 , replicateM

66+ , fromIndices

67+ , fromIndicesM

67 -- ** Conversions

69 -- ** Conversions

68 , yield

70 , yield

@@ -283,6 +285,19 @@ replicate n a = go n

283 where

285 where

284 go cnt = if cnt <= 0 then nil else a `cons` go (cnt - 1)

286 go cnt = if cnt <= 0 then nil else a `cons` go (cnt - 1)

285

287

288+{-# INLINE fromIndicesM #-}

289+fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a

290+fromIndicesM gen = go 0

291+ where

292+ go i = mkStream $ \st stp sng yld -> do

293+ foldStreamShared st stp sng yld (gen i |: go (i + 1))

294+

295+{-# INLINE fromIndices #-}

296+fromIndices :: IsStream t => (Int -> a) -> t m a

297+fromIndices gen = go 0

298+ where

299+ go n = (gen n) `cons` go (n + 1)

300+

286-------------------------------------------------------------------------------

301-------------------------------------------------------------------------------

287-- Conversions

302-- Conversions

288-------------------------------------------------------------------------------

303-------------------------------------------------------------------------------

test/Prop.hs+83−14

@@ -12,7 +12,7 @@ import Control.Exception

12import Control.Monad (when, forM_, replicateM, replicateM_)

13import Control.Monad.IO.Class (MonadIO(..))

14import Data.Function ((&))

15−import Data.IORef (readIORef, modifyIORef, newIORef)

15+import Data.IORef (readIORef, modifyIORef, newIORef, modifyIORef', IORef)

16import Data.List

17 (sort, foldl', scanl', findIndices, findIndex, elemIndices,

18 elemIndex, find, insertBy, intersperse, foldl1', (\\),

@@ -146,15 +146,56 @@ constructWithDoubleFromThenTo op l =

146 in constructWithLen stream list op l

147#endif

148

149+constructWithIterate ::

149−constructWithIterate :: IsStream t => (t IO Int -> SerialT IO Int) -> Spec

150+ IsStream t => (t IO Int -> SerialT IO Int) -> Word8 -> Property

150−constructWithIterate t = do

151+constructWithIterate op len =

151− it "iterate" $

152+ withMaxSuccess maxTestCount $

153+ monadicIO $ do

154+ stream <-

155+ run $

156+ (S.toList . op . S.take (fromIntegral len))

152− (S.toList . t . S.take 100) (S.iterate (+ 1) (0 :: Int))

157+ (S.iterate (+ 1) (0 :: Int))

153− `shouldReturn` take 100 (iterate (+ 1) 0)

158+ let list = take (fromIntegral len) (iterate (+ 1) 0)

154− it "iterateM" $ do

159+ listEquals (==) stream list

160+

161+constructWithIterateM ::

162+ IsStream t => (t IO Int -> SerialT IO Int) -> Word8 -> Property

163+constructWithIterateM op len =

164+ withMaxSuccess maxTestCount $

165+ monadicIO $ do

166+ mvl <- run (newIORef [] :: IO (IORef [Int]))

155− let addM y = return (y + 1)

167+ let addM mv x y = modifyIORef' mv (++ [y + x]) >> return (y + x)

168+ list = take (fromIntegral len) (iterate (+ 1) 0)

169+ run $

170+ S.drain . op $

171+ S.take (fromIntegral len) $

156− S.toList . t . S.take 100 $ S.iterateM addM (0 :: Int)

172+ S.iterateM (addM mvl 1) (addM mvl 0 0 :: IO Int)

157− `shouldReturn` take 100 (iterate (+ 1) 0)

173+ streamEffect <- run $ readIORef mvl

174+ listEquals (==) streamEffect list

175+

176+constructWithFromIndices ::

177+ IsStream t => (t IO Int -> SerialT IO Int) -> Word8 -> Property

178+constructWithFromIndices op len =

179+ withMaxSuccess maxTestCount $

180+ monadicIO $ do

181+ stream <-

182+ run $ (S.toList . op . S.take (fromIntegral len)) (S.fromIndices id)

183+ let list = take (fromIntegral len) (iterate (+ 1) 0)

184+ listEquals (==) stream list

185+

186+constructWithFromIndicesM ::

187+ IsStream t => (t IO Int -> SerialT IO Int) -> Word8 -> Property

188+constructWithFromIndicesM op len =

189+ withMaxSuccess maxTestCount $

190+ monadicIO $ do

191+ mvl <- run (newIORef [] :: IO (IORef [Int]))

192+ let addIndex mv i = modifyIORef' mv (++ [i]) >> return i

193+ list = take (fromIntegral len) (iterate (+ 1) 0)

194+ run $

195+ S.drain . op $

196+ S.take (fromIntegral len) $ S.fromIndicesM (addIndex mvl)

197+ streamEffect <- run $ readIORef mvl

198+ listEquals (==) streamEffect list

158

199

159-------------------------------------------------------------------------------

200-------------------------------------------------------------------------------

160-- Concurrent generation

201-- Concurrent generation

@@ -560,7 +601,8 @@ transformCombineOpsOrdered constr desc eq t = do

560 transform (dropWhile (> 0)) t (S.dropWhile (> 0))

601 transform (dropWhile (> 0)) t (S.dropWhile (> 0))

561 prop (desc <> " scan") $ transform (scanl' (+) 0) t (S.scanl' (+) 0)

602 prop (desc <> " scan") $ transform (scanl' (+) 0) t (S.scanl' (+) 0)

562

603

563− -- XXX add uniq

604+ prop (desc <> " uniq") $ transform referenceUniq t S.uniq

605+

564 prop (desc <> " deleteBy (<=) 0") $

606 prop (desc <> " deleteBy (<=) 0") $

565 transform (deleteBy (<=) 0) t (S.deleteBy (<=) 0)

607 transform (deleteBy (<=) 0) t (S.deleteBy (<=) 0)

566

608

@@ -599,10 +641,23 @@ wrapOutOfBounds f i x | null x = Nothing

599 | otherwise = Just (f x i)

641 | otherwise = Just (f x i)

600

642

601wrapThe :: Eq a => [a] -> Maybe a

643wrapThe :: Eq a => [a] -> Maybe a

644+wrapThe (x:xs)

602−wrapThe (x:xs) | all (x ==) xs = Just x

645+ | all (x ==) xs = Just x

603− | otherwise = Nothing

646+ | otherwise = Nothing

604wrapThe [] = Nothing

647wrapThe [] = Nothing

605

648

649+-- This is the reference uniq implementation to compare uniq against,

650+-- we can use uniq from vector package, but for now this should

651+-- suffice.

652+referenceUniq :: Eq a => [a] -> [a]

653+referenceUniq = go

654+ where

655+ go [] = []

656+ go (x:[]) = [x]

657+ go (x:y:xs)

658+ | x == y = go (x : xs)

659+ | otherwise = x : go (y : xs)

660+

606eliminationOps

661eliminationOps

607 :: ([Int] -> t IO Int)

662 :: ([Int] -> t IO Int)

608 -> String

663 -> String

@@ -971,10 +1026,24 @@ main = hspec

971 serialOps $ prop "serially DoubleFromThenTo" .

1026 serialOps $ prop "serially DoubleFromThenTo" .

972 constructWithDoubleFromThenTo

1027 constructWithDoubleFromThenTo

973#endif

1028#endif

1029+

1030+ serialOps $ prop "serially iterate" . constructWithIterate

1031+

974 -- XXX test for all types of streams

1032 -- XXX test for all types of streams

975− constructWithIterate serially

1033+ serialOps $ prop "serially iterateM" . constructWithIterateM

1034+ -- take doesn't work well on concurrent streams. Even though it

1035+ -- seems like take only has a problem when used with parallely.

1036+ -- wSerialOps $ prop "wSerially iterateM" wSerially . constructWithIterate

1037+ -- aheadOps $ prop "aheadly iterateM" aheadly . onstructWithIterate

1038+ -- asyncOps $ prop "asyncly iterateM" asyncly . constructWithIterate

1039+ -- wAsyncOps $ prop "wAsyncly iterateM" wAsyncly . onstructWithIterate

1040+ -- parallelOps $ prop "parallely iterateM" parallely . onstructWithIterate

976 -- XXX add tests for fromIndices

1041 -- XXX add tests for fromIndices

977

1042

1043+ serialOps $ prop "serially fromIndices" . constructWithFromIndices

1044+

1045+ serialOps $ prop "serially fromIndicesM" . constructWithFromIndicesM

1046+

978 describe "Functor operations" $ do

1047 describe "Functor operations" $ do

979 serialOps $ functorOps S.fromFoldable "serially" (==)

1048 serialOps $ functorOps S.fromFoldable "serially" (==)

980 serialOps $ functorOps folded "serially folded" (==)

1049 serialOps $ functorOps folded "serially folded" (==)