0d7dad8 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement <*> and >> for SerialT via StreamD.

0d7dad8Pranay Sashank7 years ago

src/Streamly/Internal/Data/Stream/Serial.hs+9−4

compare with master

unifiedsplit

@@ -76,7 +76,6 @@ import Streamly.Internal.Data.Stream.StreamK (IsStream(..), adapt, Stream, mkStr

76import qualified Streamly.Internal.Data.Stream.Prelude as P

77import qualified Streamly.Internal.Data.Stream.StreamK as K

78import qualified Streamly.Internal.Data.Stream.StreamD as D

79−import qualified Streamly.Internal.Data.Unfold as UF

81#include "Instances.hs"

80#include "Instances.hs"

82#include "inline.hs"

81#include "inline.hs"

@@ -181,6 +180,8 @@ instance Monad m => Monad (SerialT m) where

181 return = pure

180 return = pure

182 {-# INLINE (>>=) #-}

181 {-# INLINE (>>=) #-}

183 (>>=) = K.bindWith K.serial

182 (>>=) = K.bindWith K.serial

183+ {-# INLINE (>>) #-}

184+ (>>) = (*>)

184

185

185 -- StreamD based implementation

186 -- StreamD based implementation

186 -- return = SerialT . D.fromStreamD . D.yield

187 -- return = SerialT . D.fromStreamD . D.yield

@@ -213,15 +214,19 @@ map f = mapM (return . f)

213

214

214{-# INLINE apSerial #-}

215{-# INLINE apSerial #-}

215apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b

216apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b

216−apSerial (SerialT m1) (SerialT m2) =

217+apSerial (SerialT m1) (SerialT m2) = D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2

217− let f x1 = D.concatMapU (UF.singleton (pure . x1)) (D.toStreamD m2)

218+

219+{-# INLINE apSequence #-}

220+apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b

218− in D.fromStreamD $ D.concatMap f (D.toStreamD m1)

221+apSequence (SerialT m1) (SerialT m2) = D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2

219

222

220instance Monad m => Applicative (SerialT m) where

223instance Monad m => Applicative (SerialT m) where

221 {-# INLINE pure #-}

224 {-# INLINE pure #-}

222 pure = SerialT . K.yield

225 pure = SerialT . K.yield

223 {-# INLINE (<*>) #-}

226 {-# INLINE (<*>) #-}

224 (<*>) = apSerial

227 (<*>) = apSerial

228+ {-# INLINE (*>) #-}

229+ (*>) = apSequence

225

230

226MONAD_COMMON_INSTANCES(SerialT,)

231MONAD_COMMON_INSTANCES(SerialT,)

227LIST_INSTANCES(SerialT)

232LIST_INSTANCES(SerialT)

src/Streamly/Internal/Data/Stream/StreamD/Type.hs+57−16

@@ -90,13 +90,11 @@ import qualified Streamly.Internal.Data.Stream.StreamK as K

90-- the stream.

91data Step s a = Yield a s | Skip s | Stop

93−{-

94instance Functor (Step s) where

93instance Functor (Step s) where

95 {-# INLINE fmap #-}

94 {-# INLINE fmap #-}

96 fmap f (Yield x s) = Yield (f x) s

95 fmap f (Yield x s) = Yield (f x) s

97 fmap _ (Skip s) = Skip s

96 fmap _ (Skip s) = Skip s

98 fmap _ Stop = Stop

97 fmap _ Stop = Stop

99−-}

100

101-- gst = global state

99-- gst = global state

102-- | A stream consists of a step function that generates the next step given a

100-- | A stream consists of a step function that generates the next step given a

@@ -176,9 +174,12 @@ mapM f (Stream step state) = Stream step' state

176map :: Monad m => (a -> b) -> Stream m a -> Stream m b

174map :: Monad m => (a -> b) -> Stream m a -> Stream m b

177map f = mapM (return . f)

175map f = mapM (return . f)

178

176

179−instance Monad m => Functor (Stream m) where

177+instance Functor m => Functor (Stream m) where

180 {-# INLINE fmap #-}

178 {-# INLINE fmap #-}

181− fmap = map

179+ fmap f (Stream step state) = Stream step' state

180+ where

181+ {-# INLINE_LATE step' #-}

182+ step' gst st = fmap (fmap f) (step (adaptState gst) st)

182

183

183------------------------------------------------------------------------------

184------------------------------------------------------------------------------

184-- concatMap

185-- concatMap

@@ -231,23 +232,61 @@ concatMap f = concatMapM (return . f)

231

232

232-- | Create a singleton 'Stream' from a pure value.

233-- | Create a singleton 'Stream' from a pure value.

233{-# INLINE_NORMAL yield #-}

234{-# INLINE_NORMAL yield #-}

234−yield :: Monad m => a -> Stream m a

235+yield :: Applicative m => a -> Stream m a

235−yield x = Stream (\_ s -> return $ step undefined s) True

236+yield x = Stream (\_ s -> pure $ step undefined s) True

236 where

237 where

237 {-# INLINE_LATE step #-}

238 {-# INLINE_LATE step #-}

238 step _ True = Yield x False

239 step _ True = Yield x False

239 step _ False = Stop

240 step _ False = Stop

240

241

242+{-# INLINE_NORMAL concatAp #-}

243+concatAp :: Functor f => Stream f (a -> b) -> Stream f a -> Stream f b

244+concatAp (Stream stepa statea) (Stream stepb stateb) = Stream step' (Left statea)

245+ where

246+ {-# INLINE_LATE step' #-}

247+ step' gst (Left st) = fmap

248+ (\r -> case r of

249+ Yield f s -> Skip (Right (f, s, stateb))

250+ Skip s -> Skip (Left s)

251+ Stop -> Stop)

252+ (stepa (adaptState gst) st)

253+ step' gst (Right (f, os, st)) = fmap

254+ (\r -> case r of

255+ Yield a s -> Yield (f a) (Right (f, os, s))

256+ Skip s -> Skip (Right (f,os, s))

257+ Stop -> Skip (Left os))

258+ (stepb (adaptState gst) st)

259+

260+{-# INLINE_NORMAL apSequence #-}

261+apSequence :: Functor f => Stream f a -> Stream f b -> Stream f b

262+apSequence (Stream stepa statea) (Stream stepb stateb) = Stream step (Left statea)

263+ where

264+ {-# INLINE_LATE step #-}

265+ step gst (Left st) =

266+ fmap

267+ (\r ->

268+ case r of

269+ Yield _ s -> Skip (Right (s, stateb))

270+ Skip s -> Skip (Left s)

271+ Stop -> Stop)

272+ (stepa (adaptState gst) st)

273+ step gst (Right (ostate, st)) =

274+ fmap

275+ (\r ->

276+ case r of

277+ Yield b s -> Yield b (Right (ostate, s))

278+ Skip s -> Skip (Right (ostate, s))

279+ Stop -> Skip (Left ostate))

280+ (stepb gst st)

281+

241−instance Monad m => Applicative (Stream m) where

282+instance Applicative f => Applicative (Stream f) where

242 {-# INLINE pure #-}

283 {-# INLINE pure #-}

243 pure = yield

284 pure = yield

244 {-# INLINE (<*>) #-}

285 {-# INLINE (<*>) #-}

245− m1 <*> m2 =

286+ (<*>) = concatAp

246− -- XXX concatMapU is faster but it would require us to merge

287+ {-# INLINE (*>) #-}

247− -- Unfold/Types.hs into this file so that we can use UF.singleton here.

288+ (*>) = apSequence

248− -- let f x1 = concatMapU (UF.singleton (pure . x1)) m2

289+

249− let f x1 = concatMap (\x2 -> pure (x1 x2)) m2

250− in concatMap f m1

251

290

252-- NOTE: even though concatMap for StreamD is 4x faster compared to StreamK,

291-- NOTE: even though concatMap for StreamD is 4x faster compared to StreamK,

253-- the monad instance does not seem to be significantly faster.

292-- the monad instance does not seem to be significantly faster.

@@ -256,6 +295,8 @@ instance Monad m => Monad (Stream m) where

256 return = pure

295 return = pure

257 {-# INLINE (>>=) #-}

296 {-# INLINE (>>=) #-}

258 (>>=) = flip concatMap

297 (>>=) = flip concatMap

298+ {-# INLINE (>>) #-}

299+ (>>) = (*>)

259

300

260instance MonadTrans Stream where

301instance MonadTrans Stream where

261 lift = yieldM

302 lift = yieldM

@@ -425,12 +466,12 @@ foldl' fstep = foldlM' (\b a -> return (fstep b a))

425

466

426-- | Convert a list of pure values to a 'Stream'

467-- | Convert a list of pure values to a 'Stream'

427{-# INLINE_LATE fromList #-}

468{-# INLINE_LATE fromList #-}

428−fromList :: Monad m => [a] -> Stream m a

469+fromList :: Applicative m => [a] -> Stream m a

429fromList = Stream step

470fromList = Stream step

430 where

471 where

431 {-# INLINE_LATE step #-}

472 {-# INLINE_LATE step #-}

432− step _ (x:xs) = return $ Yield x xs

473+ step _ (x:xs) = pure $ Yield x xs

433− step _ [] = return Stop

474+ step _ [] = pure Stop

434

475

435------------------------------------------------------------------------------

476------------------------------------------------------------------------------

436-- Comparisons

477-- Comparisons