8351c93 · streamly · git.iko.soy

git.iko.soy/streamly

Add hoist and generally for streams

commit8351c936c3b4811927e8123128a72b731000cccb authorHarendra Kumar <harendra.kumar@gmail.com> dateSat 9 Nov 2019, 17:00 (7 years ago) parent02efa2c

5 files changed+52−5

compare with master

unifiedsplit

benchmark/Linear.hs+8 src/Streamly/Internal/Data/SVar.hs+1−1 src/Streamly/Internal/Prelude.hs+11 src/Streamly/Streams/StreamD.hs+18−3 src/Streamly/Streams/StreamK.hs+14−1

benchmark/Linear.hs+8

@@ -48,6 +48,13 @@ benchIOSink

4848 => String -> (t IO Int -> IO b) -> Benchmark

4949benchIOSink name f = bench name $ nfIO $ randomRIO (1,1) >>= f . Ops.source

5050

51+{-# INLINE benchHoistSink #-}

52+benchHoistSink

53+ :: (IsStream t, NFData b)

54+ => String -> (t Identity Int -> IO b) -> Benchmark

55+benchHoistSink name f =

56+ bench name $ nfIO $ randomRIO (1,1) >>= f . Ops.sourceUnfoldr

57+

5158-- XXX once we convert all the functions to use this we can rename this to

5259-- benchIOSink

5360{-# INLINE benchIOSink1 #-}

@@ -208,6 +215,7 @@ main =

208215 , benchIOSink "or" Ops.or

209216

210217 , benchIOSink "length" Ops.length

218+ , benchHoistSink "length . generally" (Ops.length . IP.generally)

211219 , benchIOSink "sum" Ops.sum

212220 , benchIOSink "product" Ops.product

213221

src/Streamly/Internal/Data/SVar.hs+1−1

@@ -536,7 +536,7 @@ defState = State

536536-- appropriate unStream functions instead.

537537--

538538-- | Adapt the stream state from one type to another.

539−adaptState :: State t m a -> State t m b

539+adaptState :: State t m a -> State t n b

540540adaptState st = st

541541 { streamVar = Nothing

542542 , _yieldLimit = Nothing

src/Streamly/Internal/Prelude.hs+11

@@ -133,6 +133,8 @@ module Streamly.Internal.Prelude

133133

134134 -- * Transformation

135135 , transform

136+ , hoist

137+ , generally

136138

137139 -- ** Mapping

138140 , Serial.map

@@ -1469,6 +1471,15 @@ toPureRev = foldl' (flip K.cons) K.nil

14691471transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b

14701472transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)

14711473

1474+{-# INLINE hoist #-}

1475+hoist :: (Monad m, Monad n)

1476+ => (forall x. m x -> n x) -> SerialT m a -> SerialT n a

1477+hoist f xs = fromStreamS $ S.hoist f (toStreamS xs)

1478+

1479+{-# INLINE generally #-}

1480+generally :: (IsStream t, Monad m) => t Identity a -> t m a

1481+generally xs = fromStreamS $ S.hoist (return . runIdentity) (toStreamS xs)

1482+

14721483------------------------------------------------------------------------------

14731484-- Transformation by Folding (Scans)

14741485------------------------------------------------------------------------------

src/Streamly/Streams/StreamD.hs+18−3

@@ -188,6 +188,8 @@ module Streamly.Streams.StreamD

188188 , toListRev

189189 , toStreamK

190190 , toStreamD

191+ , hoist

192+ , generally

191193

192194 -- * Transformation

193195 , transform

@@ -290,9 +292,7 @@ import Control.Monad.Catch (MonadCatch)

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

291293import Control.Monad.Trans (MonadTrans(lift))

292294import Data.Bits (shiftR, shiftL, (.|.), (.&.))

293−#if __GLASGOW_HASKELL__ >= 801

294−import Data.Functor.Identity (Identity)

295−#endif

295+import Data.Functor.Identity (Identity(..))

296296import Data.Maybe (fromJust, isJust)

297297import Data.Word (Word32)

298298import Foreign.Ptr (Ptr)

@@ -635,6 +635,21 @@ fromListM = Stream step

635635toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a

636636toStreamD = fromStreamK . K.toStream

637637

638+{-# INLINE_NORMAL hoist #-}

639+hoist :: Monad n => (forall x. m x -> n x) -> Stream m a -> Stream n a

640+hoist f (Stream step state) = (Stream step' state)

641+ where

642+ step' gst st = do

643+ r <- f $ step (adaptState gst) st

644+ return $ case r of

645+ Yield x s -> Yield x s

646+ Skip s -> Skip s

647+ Stop -> Stop

648+

649+{-# INLINE_NORMAL generally #-}

650+generally :: Monad m => Stream Identity a -> Stream m a

651+generally = hoist (return . runIdentity)

652+

638653------------------------------------------------------------------------------

639654-- Elimination by Folds

640655------------------------------------------------------------------------------

src/Streamly/Streams/StreamK.hs+14−1

@@ -124,6 +124,7 @@ module Streamly.Streams.StreamK

124124 -- ** Conversions

125125 , toList

126126 , toStreamK

127+ , hoist

127128

128129 -- * Transformation

129130 -- ** By folding (scans)

@@ -187,7 +188,7 @@ module Streamly.Streams.StreamK

187188where

188189

189190import Control.Monad.Trans (MonadTrans(lift))

190−import Control.Monad (void)

191+import Control.Monad (void, join)

191192import Control.Monad.Reader.Class (MonadReader(..))

192193import Prelude

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

@@ -726,6 +727,18 @@ toList = foldr (:) []

726727toStreamK :: Stream m a -> Stream m a

727728toStreamK = id

728729

730+-- Based on suggestions by David Feuer and Pranay Sashank

731+{-# INLINE hoist #-}

732+hoist :: (IsStream t, Monad m, Monad n)

733+ => (forall x. m x -> n x) -> t m a -> t n a

734+hoist f str =

735+ mkStream $ \st yld sng stp ->

736+ let single = return . sng

737+ yieldk a s = return $ yld a (hoist f s)

738+ stop = return stp

739+ state = adaptState st

740+ in join . f $ foldStreamShared state yieldk single stop str

741+

729742-------------------------------------------------------------------------------

730743-- Transformation by folding (Scans)

731744-------------------------------------------------------------------------------