8a4d965 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add trace, tap and tee combinators

8a4d965Harendra Kumar7 years ago

Changelog.md+4

compare with master

unifiedsplit

@@ -77,6 +77,10 @@ example, concat streams concurrently using this.

77 stream

78 * `insertAfterEach` to insert a monadic action after each element in th

79 stream

80+ * `trace` combinator maps a monadic function on a stream just for side

81+ effects

82+ * `tap` redirects a copy of the stream to a `Fold`

83+ * `tee` redirects a copy of the stream to a concurrent fold function

81### Deprecations

85### Deprecations

src/Streamly/Prelude.hs+49

@@ -441,6 +441,11 @@ module Streamly.Prelude

441 -- , interposeBy

442 -- , intercalate

443

444+ -- ** Distributing

445+ , trace

446+ , tap

447+ , Par.tee

448+

444 -- ** Containers of Streams

449 -- ** Containers of Streams

445 -- | These are variants of standard 'Foldable' fold functions that use a

450 -- | These are variants of standard 'Foldable' fold functions that use a

446 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a

451 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a

@@ -486,6 +491,7 @@ where

486

491

487import Control.Concurrent (threadDelay)

492import Control.Concurrent (threadDelay)

488import Control.Exception (Exception)

493import Control.Exception (Exception)

494+import Control.Monad (void)

489import Control.Monad.Catch (MonadCatch)

495import Control.Monad.Catch (MonadCatch)

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

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

491import Control.Monad.Trans (MonadTrans(..))

497import Control.Monad.Trans (MonadTrans(..))

@@ -512,6 +518,7 @@ import Streamly.Streams.StreamK (IsStream(..))

512import Streamly.Streams.Serial (SerialT)

518import Streamly.Streams.Serial (SerialT)

513import Streamly.Pipe.Types (Pipe (..))

519import Streamly.Pipe.Types (Pipe (..))

514

520

521+import qualified Streamly.Fold.Types as FL

515import qualified Streamly.Streams.Prelude as P

522import qualified Streamly.Streams.Prelude as P

516import qualified Streamly.Streams.StreamK as K

523import qualified Streamly.Streams.StreamK as K

517import qualified Streamly.Streams.StreamD as D

524import qualified Streamly.Streams.StreamD as D

@@ -2261,6 +2268,48 @@ interposeBy :: (IsStream t, Monad m)

2261intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

2268intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

2262-}

2269-}

2263

2270

2271+------------------------------------------------------------------------------

2272+-- Distributing

2273+------------------------------------------------------------------------------

2274+

2275+-- | Tap the data flowing through a stream into a 'Fold'. For example, you may

2276+-- add a tap to log the contents flowing through the stream. The fold is used

2277+-- only for effects, its result is discarded.

2278+--

2279+-- @

2280+-- Fold m a b

2281+-- |

2282+-- -----stream m a ---------------stream m a-----

2283+--

2284+-- @

2285+--

2286+-- @

2287+-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)

2288+-- 1

2289+-- 2

2290+-- @

2291+--

2292+-- Compare with 'trace'.

2293+--

2294+-- @since 0.7.0

2295+tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a

2296+tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)

2297+

2298+-- | Apply a monadic function to each element flowing through the stream and

2299+-- discard the results.

2300+--

2301+-- @

2302+-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)

2303+-- 1

2304+-- 2

2305+-- @

2306+--

2307+-- Compare with 'tap'.

2308+--

2309+-- @since 0.7.0

2310+trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a

2311+trace f = mapM (\x -> void (f x) >> return x)

2312+

2264------------------------------------------------------------------------------

2313------------------------------------------------------------------------------

2265-- Exceptions

2314-- Exceptions

2266------------------------------------------------------------------------------

2315------------------------------------------------------------------------------

src/Streamly/SVar.hs+2

@@ -98,6 +98,8 @@ module Streamly.SVar

98 , dispatchWorkerPaced

99 , sendFirstWorker

100 , delThread

101+ , modifyThread

102+ , doFork

101

103

102 , toStreamVar

104 , toStreamVar

103 , SVarStats (..)

105 , SVarStats (..)

src/Streamly/Sink.hs−26

@@ -44,9 +44,6 @@ module Streamly.Sink

44 -- -- * Comonad

45 -- , duplicate

47− -- * Utilities

48− , tap

49−

50 -- * Input Transformation

47 -- * Input Transformation

51 -- | These are contravariant operations i.e. they apply on the input of the

48 -- | These are contravariant operations i.e. they apply on the input of the

52 -- 'Sink', for this reason they are prefixed with 'l' for 'left'.

49 -- 'Sink', for this reason they are prefixed with 'l' for 'left'.

@@ -79,7 +76,6 @@ import Streamly.Streams.Serial (SerialT)

79import qualified Data.Map.Strict as Map

76import qualified Data.Map.Strict as Map

81import qualified Streamly.Fold as FL

78import qualified Streamly.Fold as FL

82−import qualified Streamly.Streams.StreamD as D

84------------------------------------------------------------------------------

80------------------------------------------------------------------------------

85-- Conversion

81-- Conversion

@@ -223,28 +219,6 @@ unzipM f (Sink stepB) (Sink stepC) =

223unzip :: Monad m => (a -> (b,c)) -> Sink m b -> Sink m c -> Sink m a

219unzip :: Monad m => (a -> (b,c)) -> Sink m b -> Sink m c -> Sink m a

224unzip f = unzipM (return . f)

220unzip f = unzipM (return . f)

225

221

226−-- | Tap a stream to a 'Sink'. For example, you may add a tap to log the

227−-- contents flowing through the stream.

228−--

229−-- @

230−-- Sink m a

231−-- |

232−-- -----stream m a ---------------stream m a-----

233−--

234−-- @

235−--

236−-- @

237−-- > let pr x = Sink.drainM (putStrLn . ((x ++ " ") ++) . show)

238−-- > S.drain $ Sink.tap (pr \"L") (S.enumerateFromTo 1 2)

239−-- L 1

240−-- L 2

241−-- @

242−--

243−-- Compare with the 'trace' function.

244−--

245−tap :: Monad m => Sink m a -> SerialT m a -> SerialT m a

246−tap s m = D.fromStreamD $ D.tap s (D.toStreamD m)

247−

248------------------------------------------------------------------------------

222------------------------------------------------------------------------------

249-- Input transformation

223-- Input transformation

250------------------------------------------------------------------------------

224------------------------------------------------------------------------------

src/Streamly/Streams/Parallel.hs+84−1

@@ -25,6 +25,7 @@ module Streamly.Streams.Parallel

25 , parallel

26 , parallelEndByFirst

27 , parallelEndByAny

28+ , tee

29 -- * Function application

30 -- * Function application

30 , mkParallel

31 , mkParallel

@@ -35,6 +36,8 @@ module Streamly.Streams.Parallel

35 )

36 )

36where

37where

39+import Control.Concurrent (myThreadId)

40+import Control.Exception (SomeException(..), throwIO)

38import Control.Monad (ap)

41import Control.Monad (ap)

39import Control.Monad.Base (MonadBase(..), liftBaseDefault)

42import Control.Monad.Base (MonadBase(..), liftBaseDefault)

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

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

@@ -51,7 +54,7 @@ import Prelude hiding (map)

52import qualified Data.Set as Set

55import qualified Data.Set as Set

54−import Streamly.Streams.SVar (fromSVar)

57+import Streamly.Streams.SVar (fromSVar, fromStreamVar)

55import Streamly.Streams.Serial (map)

58import Streamly.Streams.Serial (map)

56import Streamly.SVar

59import Streamly.SVar

57import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,

60import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,

@@ -205,6 +208,86 @@ foldParallel f m = do

205 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)

208 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)

206 f $ fromSVar sv

209 f $ fromSVar sv

207

210

211+{-# INLINE teeToSVar #-}

212+teeToSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> t m a

213+teeToSVar svr m = mkStream $ \st yld sng stp -> do

214+ foldStreamShared st yld sng stp (go svr m)

215+

216+ where

217+

218+ go sv m0 = mkStream $ \st yld sng stp -> do

219+ let stop = do

220+ liftIO $ do

221+ incrementBufferLimit sv

222+ sendStop sv Nothing

223+ -- XXX wait for Stop response event on exception channel,

224+ -- drain the exception channel.

225+ stp

226+ sendit a = liftIO $ do

227+ -- XXX check for exceptions before decrement so that we do not

228+ -- block forever if the child already exited with an exception.

229+ decrementBufferLimit sv

230+ void $ send sv (ChildYield a)

231+ single a = sendit a >> (liftIO $ sendStop sv Nothing) >> sng a

232+ yieldk a r = sendit a >> yld a (go sv r)

233+ in foldStreamShared st yieldk single stop m0

234+

235+-- In case of folds the roles of worker and parent on an SVar are reversed. The

236+-- parent stream pushes values to an SVar instead of pulling from it and a

237+-- worker thread running the fold pulls from the SVar and folds the stream. We

238+-- keep a separate channel for pushing exceptions in the reverse direction i.e.

239+-- from the fold to the parent stream.

240+--

241+-- NOTE: If we use fromSVar here it will kill the main computation (the parent)

242+-- when the SVar goes away so we use fromStreamVar instead.

243+

244+{-# NOINLINE handleChildException #-}

245+handleChildException :: SVar t m a -> SomeException -> IO ()

246+handleChildException _sv e = do

247+ -- tid <- myThreadId

248+ -- void $ sendReverse sv (ChildStop tid (Just e))

249+ throwIO e

250+

251+-- | Redirect a copy of the stream to a supplied fold and run it concurrently

252+-- in an independent thread. The fold may buffer some elements. The buffer size

253+-- is determined by the prevailing 'maxBuffer' setting.

254+--

255+-- @

256+-- Stream m a -> m b

257+-- |

258+-- -----stream m a ---------------stream m a-----

259+--

260+-- @

261+--

262+-- @

263+-- > S.drain $ S.tee (S.mapM_ print) (S.enumerateFromTo 1 2)

264+-- 1

265+-- 2

266+-- @

267+--

268+-- Exceptions from the concurrently running fold are propagated to the current

269+-- computation. Note that, because of buffering in the fold, exceptions may be

270+-- delayed and may not correspond to the current element being processed in the

271+-- parent stream, but we guarantee that the tap finishes and all exceptions

272+-- from it are drained before the parent stream stops.

273+--

274+--

275+-- Compare with 'tap'.

276+--

277+-- @since 0.7.0

278+{-# INLINE tee #-}

279+tee :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a

280+tee f m = mkStream $ \st yld sng stp -> do

281+ -- Buffer size for the SVar is derived from the current state

282+ sv <- newParallelVar StopNone (adaptState st)

283+ -- XXX exception handling

284+ -- XXX if we terminate due to an exception, do we need to actively

285+ -- terminate the fold?

286+ liftIO myThreadId >>= modifyThread sv

287+ void $ doFork (void $ f $ fromStream $ fromStreamVar sv)

288+ (svarMrun sv) (handleChildException sv)

289+ foldStreamShared st yld sng stp (teeToSVar sv m)

290+

208------------------------------------------------------------------------------

291------------------------------------------------------------------------------

209-- Concurrent Application

292-- Concurrent Application

210------------------------------------------------------------------------------

293------------------------------------------------------------------------------

src/Streamly/Streams/SVar.hs+1

@@ -12,6 +12,7 @@

12--

13module Streamly.Streams.SVar

14 ( fromSVar

15+ , fromStreamVar

15 , toSVar

16 , toSVar

16 )

17 )

17where

18where

src/Streamly/Streams/StreamD.hs+14−8

@@ -247,6 +247,7 @@ module Streamly.Streams.StreamD

247where

248

249import Control.Exception (Exception)

250+import Control.Monad (void)

250import Control.Monad.Catch (MonadCatch)

251import Control.Monad.Catch (MonadCatch)

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

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

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

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

@@ -273,7 +274,6 @@ import Streamly.Mem.Array.Types (Array(..))

273import Streamly.Fold.Types (Fold(..))

274import Streamly.Fold.Types (Fold(..))

274import Streamly.Pipe.Types (Pipe(..), PipeState(..))

275import Streamly.Pipe.Types (Pipe(..), PipeState(..))

275import Streamly.SVar (MonadAsync, defState, adaptState)

276import Streamly.SVar (MonadAsync, defState, adaptState)

276−import Streamly.Sink.Types (Sink(..))

277

278import Streamly.Streams.StreamD.Type

279

@@ -2016,19 +2016,25 @@ scanl1' :: Monad m => (a -> a -> a) -> Stream m a -> Stream m a

2016scanl1' f = scanl1M' (\x y -> return (f x y))

2017

2018{-# INLINE tap #-}

2019−tap :: Monad m => Sink m a -> Stream m a -> Stream m a

2019+tap :: Monad m => Fold m a b -> Stream m a -> Stream m a

2020−tap (Sink fstep) (Stream step state) = Stream step' state

2020+tap (Fold fstep initial extract) (Stream step state) = Stream step' Nothing

2021

2022 where

2023

2024+ step' _ Nothing = do

2025+ r <- initial

2026+ return $ Skip (Just (r, state))

2027+

2024− step' gst st = do

2028+ step' gst (Just (acc, st)) = do

2025 r <- step gst st

2029 r <- step gst st

2026 case r of

2030 case r of

2027 Yield x s -> do

2031 Yield x s -> do

2028− fstep x

2032+ acc' <- fstep acc x

2029− return $ Yield x s

2033+ return $ Yield x (Just (acc', s))

2030− Skip s -> return $ Skip s

2034+ Skip s -> return $ Skip (Just (acc, s))

2035+ Stop -> do

2036+ void $ extract acc

2031− Stop -> return $ Stop

2037+ return $ Stop

2032

2038

2033-------------------------------------------------------------------------------

2039-------------------------------------------------------------------------------

2034-- Filtering

2040-- Filtering