9c87c64 · streamly · git.iko.soy

git.iko.soy/streamly

Remove tapRate

commit9c87c644f772f21435b4929a49a4310e30e22abc authorRanjeet Kumar Ranjan <ranjeetr@composewell.com> dateThu 2 Jun 2022, 14:50 (4 years ago) parent4608fad

3 files changed+2−88

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation.hs−8 core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+2−51 src/Streamly/Internal/Data/Stream/IsStream/Transform.hs−29

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation.hs−8

@@ -23,7 +23,6 @@ module Serial.Transformation (benchmarks) where

2323import Control.DeepSeq (NFData(..))

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

2525import Data.Functor.Identity (Identity)

26−import Data.IORef (newIORef, modifyIORef')

2726import System.Random (randomRIO)

2827

2928#ifdef INSPECTION

@@ -139,12 +138,6 @@ sequence t = S.drain . t . S.sequence

139138tap :: MonadIO m => Int -> SerialT m Int -> m ()

140139tap n = composeN n $ S.tap FL.sum

141140

142−{-# INLINE tapRate #-}

143−tapRate :: Int -> SerialT IO Int -> IO ()

144−tapRate n str = do

145− cref <- newIORef 0

146− composeN n (Internal.tapRate 1 (\c -> modifyIORef' cref (c +))) str

147−

148141{-# INLINE pollCounts #-}

149142pollCounts :: Int -> SerialT IO Int -> IO ()

150143pollCounts n =

@@ -196,7 +189,6 @@ o_1_space_mapping value =

196189 sequence fromSerial (sourceUnfoldrAction value n)

197190 , benchIOSink value "mapM" (mapM fromSerial 1)

198191 , benchIOSink value "tap" (tap 1)

199− , benchIOSink value "tapRate 1 second" (tapRate 1)

200192 , benchIOSink value "pollCounts 1 second" (pollCounts 1)

201193 , benchIOSink value "timestamped" timestamped

202194

core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+2−51

@@ -27,7 +27,6 @@ module Streamly.Internal.Data.Stream.StreamD.Transform

2727 -- * Mapping Effects

2828 , tap

2929 , tapOffsetEvery

30− , tapRate

3130 , pollCounts

3231

3332 -- * Folding

@@ -126,21 +125,17 @@ where

126125

127126#include "inline.hs"

128127

129−import Control.Concurrent (killThread, threadDelay)

130−import Control.Exception (AsyncException)

128+import Control.Concurrent (killThread)

131129import Control.Monad (void, when)

132−import Control.Monad.Catch (MonadCatch, throwM)

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

134131import Control.Monad.Trans.Class (MonadTrans(lift))

135−import Data.IORef (newIORef, mkWeakIORef)

136132import Data.Maybe (fromJust, isJust)

137133import Fusion.Plugin.Types (Fuse(..))

138134import GHC.Types (SPEC(..))

139−import qualified Control.Monad.Catch as MC

140135import qualified Data.Set as Set

141136

142137import Streamly.Internal.Control.Concurrent (MonadAsync)

143−import Streamly.Internal.Control.ForkLifted (fork, forkManaged)

138+import Streamly.Internal.Control.ForkLifted (forkManaged)

144139import Streamly.Internal.Data.Fold.Type (Fold(..))

145140import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))

146141import Streamly.Internal.Data.SVar.Type (defState, adaptState)

@@ -367,50 +362,6 @@ pollCounts predicate transf fld (Stream step state) = Stream step' Nothing

367362 liftIO $ killThread tid

368363 return Stop

369364

370−{-# INLINE_NORMAL tapRate #-}

371−tapRate ::

372− (MonadAsync m, MonadCatch m)

373− => Double

374− -> (Int -> m b)

375− -> Stream m a

376− -> Stream m a

377−tapRate samplingRate action (Stream step state) = Stream step' Nothing

378− where

379− {-# NOINLINE loop #-}

380− loop countVar prev = do

381− i <-

382− MC.catch

383− (do liftIO $ threadDelay (round $ samplingRate * 1000000)

384− i <- liftIO $ Prim.readIORef countVar

385− let !diff = i - prev

386− void $ action diff

387− return i)

388− (\(e :: AsyncException) -> do

389− i <- liftIO $ Prim.readIORef countVar

390− let !diff = i - prev

391− void $ action diff

392− throwM (MC.toException e))

393− loop countVar i

394−

395− {-# INLINE_LATE step' #-}

396− step' _ Nothing = do

397− countVar <- liftIO $ Prim.newIORef 0

398− tid <- fork $ loop countVar 0

399− ref <- liftIO $ newIORef ()

400− _ <- liftIO $ mkWeakIORef ref (killThread tid)

401− return $ Skip (Just (countVar, tid, state, ref))

402−

403− step' gst (Just (countVar, tid, st, ref)) = do

404− r <- step gst st

405− case r of

406− Yield x s -> do

407− liftIO $ Prim.modifyIORef' countVar (+ 1)

408− return $ Yield x (Just (countVar, tid, s, ref))

409− Skip s -> return $ Skip (Just (countVar, tid, s, ref))

410− Stop -> do

411− liftIO $ killThread tid

412− return Stop

413−

414365------------------------------------------------------------------------------

415366-- Scanning with a Fold

416367------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/IsStream/Transform.hs−29

@@ -33,7 +33,6 @@ module Streamly.Internal.Data.Stream.IsStream.Transform

3333 , tapAsync

3434 , tapAsyncK

3535 , distributeAsync_

36− , tapRate

3736 , pollCounts

3837

3938 -- * Scanning By 'Fold'

@@ -217,7 +216,6 @@ where

217216

218217import Control.Concurrent (threadDelay)

219218import Control.Monad (void)

220−import Control.Monad.Catch (MonadCatch)

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

222220import Control.Monad.Trans.Class (MonadTrans(..))

223221import Data.Either (isLeft, isRight, fromLeft, fromRight)

@@ -585,33 +583,6 @@ pollCounts predicate transf f xs =

585583 $ D.pollCounts predicate (toStreamD . transf . fromStreamD) f

586584 $ toStreamD xs

587585

588−-- | Calls the supplied function with the number of elements consumed

589−-- every @n@ seconds. The given function is run in a separate thread

590−-- until the end of the stream. In case there is an exception in the

591−-- stream the thread is killed during the next major GC.

592−--

593−-- Note: The action is not guaranteed to run if the main thread exits.

594−--

595−-- @

596−-- > delay n = threadDelay (round $ n * 1000000) >> return n

597−-- > Stream.toList $ Stream.tapRate 2 (\n -> print $ show n ++ " elements processed") (delay 1 Stream.|: delay 0.5 Stream.|: delay 0.5 Stream.|: Stream.nil)

598−-- "2 elements processed"

599−-- [1.0,0.5,0.5]

600−-- "1 elements processed"

601−-- @

602−--

603−-- Note: This may not work correctly on 32-bit machines.

604−--

605−-- /Pre-release/

606−{-# INLINE tapRate #-}

607−tapRate ::

608− (IsStream t, MonadAsync m, MonadCatch m)

609− => Double

610− -> (Int -> m b)

611− -> t m a

612− -> t m a

613−tapRate n f xs = fromStreamD $ D.tapRate n f $ toStreamD xs

614−

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

616587-- discard the results.

617588--