9c87c64 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Remove tapRate

9c87c64Ranjeet Kumar Ranjan4 years ago

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

compare with master

unifiedsplit

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

23import Control.DeepSeq (NFData(..))

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

25import Data.Functor.Identity (Identity)

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

27import System.Random (randomRIO)

26import System.Random (randomRIO)

29#ifdef INSPECTION

28#ifdef INSPECTION

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

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

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

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

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

141

140

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−

148{-# INLINE pollCounts #-}

141{-# INLINE pollCounts #-}

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

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

150pollCounts n =

143pollCounts n =

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

196 sequence fromSerial (sourceUnfoldrAction value n)

189 sequence fromSerial (sourceUnfoldrAction value n)

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

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

198 , benchIOSink value "tap" (tap 1)

191 , benchIOSink value "tap" (tap 1)

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

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

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

201 , benchIOSink value "timestamped" timestamped

193 , benchIOSink value "timestamped" timestamped

202

194

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

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

27 -- * Mapping Effects

28 , tap

29 , tapOffsetEvery

30− , tapRate

31 , pollCounts

30 , pollCounts

33 -- * Folding

32 -- * Folding

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

126

125

127#include "inline.hs"

126#include "inline.hs"

128

127

129−import Control.Concurrent (killThread, threadDelay)

128+import Control.Concurrent (killThread)

130−import Control.Exception (AsyncException)

131import Control.Monad (void, when)

129import Control.Monad (void, when)

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

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

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

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

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

135−import Data.IORef (newIORef, mkWeakIORef)

136import Data.Maybe (fromJust, isJust)

132import Data.Maybe (fromJust, isJust)

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

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

138import GHC.Types (SPEC(..))

134import GHC.Types (SPEC(..))

139−import qualified Control.Monad.Catch as MC

140import qualified Data.Set as Set

135import qualified Data.Set as Set

141

136

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

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

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

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

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

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

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

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

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

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

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

367 liftIO $ killThread tid

362 liftIO $ killThread tid

368 return Stop

363 return Stop

369

364

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−

414------------------------------------------------------------------------------

365------------------------------------------------------------------------------

415-- Scanning with a Fold

366-- Scanning with a Fold

416------------------------------------------------------------------------------

367------------------------------------------------------------------------------

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

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

33 , tapAsync

34 , tapAsyncK

35 , distributeAsync_

36− , tapRate

37 , pollCounts

36 , pollCounts

39 -- * Scanning By 'Fold'

38 -- * Scanning By 'Fold'

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

217

216

218import Control.Concurrent (threadDelay)

217import Control.Concurrent (threadDelay)

219import Control.Monad (void)

218import Control.Monad (void)

220−import Control.Monad.Catch (MonadCatch)

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

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

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

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

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

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

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

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

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

586 $ toStreamD xs

584 $ toStreamD xs

587

585

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−

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

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

616-- discard the results.

587-- discard the results.

617--

588--