3e074ef · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move pollCounts from streamly-core to streamly

3e074efHarendra Kumar4 years ago

core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+1−40

compare with master

unifiedsplit

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

27 -- * Mapping Effects

28 , tap

29 , tapOffsetEvery

30− , pollCounts

32 -- * Folding

31 -- * Folding

33 , foldrS

32 , foldrS

@@ -124,23 +123,18 @@ where

124

123

125#include "inline.hs"

124#include "inline.hs"

126

125

127−import Control.Concurrent (killThread)

126+import Control.Monad (void)

128−import Control.Monad (void, when)

129−import Control.Monad.IO.Class (MonadIO(..))

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

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

131import Data.Maybe (fromJust, isJust)

128import Data.Maybe (fromJust, isJust)

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

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

133import GHC.Types (SPEC(..))

130import GHC.Types (SPEC(..))

134

131

135−import Streamly.Internal.Control.Concurrent (MonadAsync)

136−import Streamly.Internal.Control.ForkLifted (forkManaged)

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

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

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

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

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

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

140

135

141import qualified Data.Set as Set

136import qualified Data.Set as Set

142import qualified Streamly.Internal.Data.Fold.Type as FL

137import qualified Streamly.Internal.Data.Fold.Type as FL

143−import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed

144import qualified Streamly.Internal.Data.Pipe.Type as Pipe

138import qualified Streamly.Internal.Data.Pipe.Type as Pipe

145

139

146import Prelude hiding

140import Prelude hiding

@@ -325,39 +319,6 @@ tapOffsetEvery offset n (Fold fstep initial extract) (Stream step state) =

325 Skip s -> Skip (TapOffDone s)

319 Skip s -> Skip (TapOffDone s)

326 Stop -> Stop

320 Stop -> Stop

327

321

328−{-# INLINE_NORMAL pollCounts #-}

329−pollCounts

330− :: MonadAsync m

331− => (a -> Bool)

332− -> (Stream m Int -> m b)

333− -> Stream m a

334− -> Stream m a

335−pollCounts predicate fld (Stream step state) = Stream step' Nothing

336− where

337−

338− {-# INLINE_LATE step' #-}

339− step' _ Nothing = do

340− -- As long as we are using an "Int" for counts lockfree reads from

341− -- Var should work correctly on both 32-bit and 64-bit machines.

342− -- However, an Int on a 32-bit machine may overflow quickly.

343− countVar <- liftIO $ Unboxed.newIORef (0 :: Int)

344− tid <- forkManaged

345− $ void $ fld

346− $ Unboxed.toStreamD countVar

347− return $ Skip (Just (countVar, tid, state))

348−

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

350− r <- step gst st

351− case r of

352− Yield x s -> do

353− when (predicate x)

354− $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)

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

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

357− Stop -> do

358− liftIO $ killThread tid

359− return Stop

360−

361------------------------------------------------------------------------------

322------------------------------------------------------------------------------

362-- Scanning with a Fold

323-- Scanning with a Fold

363------------------------------------------------------------------------------

324------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/Concurrent.hs+70−2

@@ -98,15 +98,18 @@ module Streamly.Internal.Data.Stream.Concurrent

99 -- ** Reactive

100 , fromCallback

101+ , pollCountsD

102+ , pollCounts

101 )

103 )

102where

104where

103

105

104#include "inline.hs"

106#include "inline.hs"

105

107

106−import Control.Concurrent (myThreadId)

108+import Control.Concurrent (myThreadId, killThread)

107−import Control.Monad (void)

109+import Control.Monad (void, when)

108import Control.Monad.IO.Class (MonadIO(liftIO))

110import Control.Monad.IO.Class (MonadIO(liftIO))

109import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

111import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

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

110import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)

113import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)

111import Streamly.Internal.Data.Stream.Channel.Types

114import Streamly.Internal.Data.Stream.Channel.Types

112 ( ChildEvent(..)

115 ( ChildEvent(..)

@@ -114,8 +117,11 @@ import Streamly.Internal.Data.Stream.Channel.Types

114 )

117 )

115import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)

118import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)

116import Streamly.Internal.Data.Stream.Type (Stream)

119import Streamly.Internal.Data.Stream.Type (Stream)

120+import Streamly.Internal.Data.Stream.StreamD (Step(..))

117

121

122+import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed

118import qualified Streamly.Internal.Data.Stream as Stream

123import qualified Streamly.Internal.Data.Stream as Stream

124+import qualified Streamly.Internal.Data.Stream.StreamD as D

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

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

120import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

126import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

121

127

@@ -695,3 +701,65 @@ fromCallback setCallback = Stream.concatM $ do

695 (callback, stream) <- newCallbackStream

701 (callback, stream) <- newCallbackStream

696 setCallback callback

702 setCallback callback

697 return stream

703 return stream

704+

705+{-# INLINE_NORMAL pollCountsD #-}

706+pollCountsD

707+ :: MonadAsync m

708+ => (a -> Bool)

709+ -> (D.Stream m Int -> m b)

710+ -> D.Stream m a

711+ -> D.Stream m a

712+pollCountsD predicate fld (D.Stream step state) = D.Stream step' Nothing

713+ where

714+

715+ {-# INLINE_LATE step' #-}

716+ step' _ Nothing = do

717+ -- As long as we are using an "Int" for counts lockfree reads from

718+ -- Var should work correctly on both 32-bit and 64-bit machines.

719+ -- However, an Int on a 32-bit machine may overflow quickly.

720+ countVar <- liftIO $ Unboxed.newIORef (0 :: Int)

721+ tid <- forkManaged

722+ $ void $ fld

723+ $ Unboxed.toStreamD countVar

724+ return $ Skip (Just (countVar, tid, state))

725+

726+ step' gst (Just (countVar, tid, st)) = do

727+ r <- step gst st

728+ case r of

729+ Yield x s -> do

730+ when (predicate x)

731+ $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)

732+ return $ Yield x (Just (countVar, tid, s))

733+ Skip s -> return $ Skip (Just (countVar, tid, s))

734+ Stop -> do

735+ liftIO $ killThread tid

736+ return Stop

737+

738+-- | @pollCounts predicate transform fold stream@ counts those elements in the

739+-- stream that pass the @predicate@. The resulting count stream is sent to

740+-- another thread which transforms it using @transform@ and then folds it using

741+-- @fold@. The thread is automatically cleaned up if the stream stops or

742+-- aborts due to exception.

743+--

744+-- For example, to print the count of elements processed every second:

745+--

746+-- @

747+-- > Stream.drain $ Concur.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)

748+-- $ Stream.enumerateFrom 0

749+-- @

750+--

751+-- Note: This may not work correctly on 32-bit machines.

752+--

753+-- /Pre-release/

754+--

755+{-# INLINE pollCounts #-}

756+pollCounts ::

757+ (MonadAsync m)

758+ => (a -> Bool)

759+ -> (Stream m Int -> m b)

760+ -> Stream m a

761+ -> Stream m a

762+pollCounts predicate f xs =

763+ Stream.fromStreamD

764+ $ pollCountsD predicate (f . Stream.fromStreamD)

765+ $ Stream.toStreamD xs

src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+3−2

@@ -253,10 +253,11 @@ import Streamly.Internal.Data.SVar (Rate(..))

253import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)

254

255import qualified Streamly.Data.Fold as FL

256+import qualified Streamly.Internal.Data.Stream.Concurrent as Concur

256import qualified Streamly.Internal.Data.Stream.Parallel as Par

257import qualified Streamly.Internal.Data.Stream.Parallel as Par

257import qualified Streamly.Internal.Data.Stream.Serial as Serial

258import qualified Streamly.Internal.Data.Stream.Serial as Serial

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

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

259− (transform, foldrT, tap, tapOffsetEvery, pollCounts, mapM, scanOnce

260+ (transform, foldrT, tap, tapOffsetEvery, mapM, scanOnce

260 , scanMany, postscanOnce, scanlx', scanlM', scanl', postscanl', prescanl'

261 , scanMany, postscanOnce, scanlx', scanlM', scanl', postscanl', prescanl'

261 , prescanlM', scanl1M', scanl1', filter, filterM, uniq, deleteBy, takeWhileM

262 , prescanlM', scanl1M', scanl1', filter, filterM, uniq, deleteBy, takeWhileM

262 , dropWhile, dropWhileM, insertBy, intersperse

263 , dropWhile, dropWhileM, insertBy, intersperse

@@ -582,7 +583,7 @@ pollCounts ::

582 -> t m a

583 -> t m a

583pollCounts predicate f xs =

584pollCounts predicate f xs =

584 fromStreamD

585 fromStreamD

585− $ D.pollCounts predicate (f . fromStreamD)

586+ $ Concur.pollCountsD predicate (f . fromStreamD)

586 $ toStreamD xs

587 $ toStreamD xs

587

588

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

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