68d2bf8 · streamly · git.iko.soy

git.iko.soy/streamly

Add parallel stream stop styles

End the composed stream: * when the first stream ends * when any constituent stream ends (race) The first one is useful to compose an infinit stream with a finite stream. For example we implement the insperseByTime combinator using it which in turn is used to implement the sessionsOf comibnator that folds the stream in chunks of time.

commit68d2bf885cd51c7bcdba84d8694604d444cddbcb authorHarendra Kumar <harendra.kumar@gmail.com> dateTue 28 May 2019, 10:14 (7 years ago) parentf1e2f03

6 files changed+152−27

compare with master

unifiedsplit

src/Streamly/Fold.hs+3−6 src/Streamly/Prelude.hs+59−4 src/Streamly/SVar.hs+31−5 src/Streamly/Streams/Async.hs+4 src/Streamly/Streams/Parallel.hs+43−11 src/Streamly/Streams/SVar.hs+12−1

src/Streamly/Fold.hs+3−6

@@ -1123,12 +1123,9 @@ lcatMaybes = lfilter isJust . lmap fromJust

11231123sessionsOf

11241124 :: (IsStream t, MonadAsync m)

11251125 => Double -> Fold m a b -> t m a -> t m b

1126−sessionsOf n f m = splitSuffixBy' isNothing (lcatMaybes f) s

1127− where

1128− s = S.map Just m `parallel` S.repeatM timeout

1129− timeout = do

1130− liftIO $ threadDelay (round $ n * 1000000)

1131− return Nothing

1126+sessionsOf n f xs =

1127+ splitSuffixBy' isNothing (lcatMaybes f)

1128+ (S.intersperseByTime n (return Nothing) (S.map Just xs))

11321129

11331130------------------------------------------------------------------------------

11341131-- Element Aware APIs

src/Streamly/Prelude.hs+59−4

@@ -367,6 +367,8 @@ module Streamly.Prelude

367367

368368 , insertBy

369369 , intersperseM

370+ -- , intersperseBySpan

371+ , intersperseByTime

370372

371373 -- ** Reordering

372374 , reverse

@@ -470,6 +472,7 @@ module Streamly.Prelude

470472 )

471473where

472474

475+import Control.Concurrent (threadDelay)

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

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

475478import Data.Maybe (isJust, fromJust)

@@ -507,6 +510,7 @@ import qualified Streamly.Streams.StreamD as S

507510#endif

508511

509512import qualified Streamly.Streams.Serial as Serial

513+import qualified Streamly.Streams.Parallel as Par

510514

511515------------------------------------------------------------------------------

512516-- Deconstruction

@@ -1858,8 +1862,8 @@ mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m

18581862-- Transformation by Reordering

18591863------------------------------------------------------------------------------

18601864

1861−-- XXX to scale this we need to use a slab allocated array backed

1862−-- representation for temporary storage.

1865+-- XXX Use a compact region list to temporarily store the list, in both reverse

1866+-- as well as in reverse'.

18631867--

18641868-- |

18651869-- > reverse = S.foldlT (flip S.cons) S.nil

@@ -1886,14 +1890,16 @@ reverse' s = fromStreamD $ D.reverse' $ toStreamD s

18861890-- Transformation by Inserting

18871891------------------------------------------------------------------------------

18881892

1893+-- intersperseM = intersperseBySpan 1

1894+

18891895-- | Generate a stream by performing a monadic action between consecutive

18901896-- elements of the given stream.

18911897--

18921898-- /Concurrent (do not use with 'parallely' on infinite streams)/

18931899--

18941900-- @

1895−-- > S.toList $ S.intersperseM (putChar \'a' >> return ',') $ S.fromList "hello"

1896−-- aaaa"h,e,l,l,o"

1901+-- > S.toList $ S.intersperseM (return ',') $ S.fromList "hello"

1902+-- "h,e,l,l,o"

18971903-- @

18981904--

18991905-- @since 0.5.0

@@ -1901,6 +1907,37 @@ reverse' s = fromStreamD $ D.reverse' $ toStreamD s

19011907intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

19021908intersperseM = K.intersperseM

19031909

1910+{-

1911+-- | Intersperse a monadic action into the input stream after every @n@

1912+-- elements.

1913+--

1914+-- @

1915+-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"

1916+-- "he,ll,o"

1917+-- @

1918+--

1919+-- @since 0.7.0

1920+{-# INLINE intersperseBySpan #-}

1921+intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a

1922+intersperseBySpan _n _f _xs = undefined

1923+-}

1924+

1925+-- | Intersperse a monadic action into the input stream after every @n@

1926+-- seconds.

1927+--

1928+-- @

1929+-- > S.drain $ S.intersperseByTime 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"

1930+-- "h,e,l,l,o"

1931+-- @

1932+--

1933+-- @since 0.7.0

1934+{-# INLINE intersperseByTime #-}

1935+intersperseByTime

1936+ :: (IsStream t, MonadAsync m)

1937+ => Double -> m a -> t m a -> t m a

1938+intersperseByTime n f xs = xs `Par.parallelEndByFirst` repeatM timed

1939+ where timed = liftIO (threadDelay (round $ n * 1000000)) >> f

1940+

19041941-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in

19051942-- @stream@ that is less than @elem@ when compared using @cmp@.

19061943--

@@ -2071,6 +2108,24 @@ mergeByM

20712108 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

20722109mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)

20732110

2111+{-

2112+-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.

2113+{-# INLINABLE mergeUptoShortest #-}

2114+mergeEndByAny

2115+ :: (IsStream t, Monad m)

2116+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

2117+mergeEndByAny f m1 m2 = fromStreamD $

2118+ D.mergeEndByAny f (toStreamD m1) (toStreamD m2)

2119+

2120+-- Like 'mergeByM' but stops merging as soon as the first stream stops.

2121+{-# INLINABLE mergeEndByFirst #-}

2122+mergeEndByFirst

2123+ :: (IsStream t, Monad m)

2124+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

2125+mergeEndByFirst f m1 m2 = fromStreamS $

2126+ D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)

2127+-}

2128+

20742129-- Holding this back for now, we may want to use the name "merge" differently

20752130{-

20762131-- | Same as @'mergeBy' 'compare'@.

src/Streamly/SVar.hs+31−5

@@ -25,6 +25,7 @@ module Streamly.SVar

2525 (

2626 MonadAsync

2727 , SVarStyle (..)

28+ , SVarStopStyle (..)

2829 , SVar (..)

2930

3031 -- State threaded around the stream

@@ -314,11 +315,25 @@ data SVarStats = SVarStats {

314315

315316data Limit = Unlimited | Limited Word deriving Show

316317

318+-- When to stop the composed stream.

319+data SVarStopStyle =

320+ StopNone -- stops only when all streams are finished

321+ | StopAny -- stop when any stream finishes

322+ | StopBy -- stop when a specific stream finishes

323+ deriving (Eq, Show)

324+

325+-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have

326+-- references to the original SVar stored in several functions which will keep

327+-- pointing to the original data and the new updates won't reflect there.

328+-- Any updateable parts must be kept in mutable references (IORef).

329+--

317330data SVar t m a = SVar

318331 {

319332 -- Read only state

320333 svarStyle :: SVarStyle

321334 , svarMrun :: RunInIO m

335+ , svarStopStyle :: SVarStopStyle

336+ , svarStopBy :: IORef ThreadId

322337

323338 -- Shared output queue (events, length)

324339 -- XXX For better efficiency we can try a preallocated array type (perhaps

@@ -2045,6 +2060,8 @@ getAheadSVar st f mrun = do

20452060 , isQueueDone = isQueueDoneAhead sv q

20462061 , needDoorBell = wfw

20472062 , svarStyle = AheadVar

2063+ , svarStopStyle = StopNone

2064+ , svarStopBy = undefined

20482065 , svarMrun = mrun

20492066 , workerCount = active

20502067 , accountThread = delThread sv

@@ -2091,8 +2108,9 @@ getAheadSVar st f mrun = do

20912108 (xs, _) <- readIORef q

20922109 return $ null xs

20932110

2094−getParallelSVar :: MonadIO m => State t m a -> RunInIO m -> IO (SVar t m a)

2095−getParallelSVar st mrun = do

2111+getParallelSVar :: MonadIO m

2112+ => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)

2113+getParallelSVar ss st mrun = do

20962114 outQ <- newIORef ([], 0)

20972115 outQMv <- newEmptyMVar

20982116 active <- newIORef 0

@@ -2105,6 +2123,11 @@ getParallelSVar st mrun = do

21052123 stats <- newSVarStats

21062124 tid <- myThreadId

21072125

2126+ stopBy <-

2127+ case ss of

2128+ StopBy -> liftIO $ newIORef undefined

2129+ _ -> return undefined

2130+

21082131 let sv =

21092132 SVar { outputQueue = outQ

21102133 , remainingWork = yl

@@ -2122,6 +2145,8 @@ getParallelSVar st mrun = do

21222145 , isQueueDone = undefined

21232146 , needDoorBell = undefined

21242147 , svarStyle = ParallelVar

2148+ , svarStopStyle = ss

2149+ , svarStopBy = stopBy

21252150 , svarMrun = mrun

21262151 , workerCount = active

21272152 , accountThread = modifyThread sv

@@ -2176,10 +2201,11 @@ newAheadVar st m wloop = do

21762201 sendFirstWorker sv m

21772202

21782203{-# INLINABLE newParallelVar #-}

2179−newParallelVar :: MonadAsync m => State t m a -> m (SVar t m a)

2180−newParallelVar st = do

2204+newParallelVar :: MonadAsync m

2205+ => SVarStopStyle -> State t m a -> m (SVar t m a)

2206+newParallelVar ss st = do

21812207 mrun <- captureMonadState

2182− liftIO $ getParallelSVar st mrun

2208+ liftIO $ getParallelSVar ss st mrun

21832209

21842210-------------------------------------------------------------------------------

21852211-- Write a stream to an SVar

src/Streamly/Streams/Async.hs+4

@@ -306,6 +306,8 @@ getLifoSVar st mrun = do

306306 , isQueueDone = workDone sv

307307 , needDoorBell = wfw

308308 , svarStyle = AsyncVar

309+ , svarStopStyle = StopNone

310+ , svarStopBy = undefined

309311 , svarMrun = mrun

310312 , workerCount = active

311313 , accountThread = delThread sv

@@ -396,6 +398,8 @@ getFifoSVar st mrun = do

396398 , isQueueDone = workDone sv

397399 , needDoorBell = wfw

398400 , svarStyle = WAsyncVar

401+ , svarStopStyle = StopNone

402+ , svarStopBy = undefined

399403 , svarMrun = mrun

400404 , workerCount = active

401405 , accountThread = delThread sv

src/Streamly/Streams/Parallel.hs+43−11

@@ -23,6 +23,8 @@ module Streamly.Streams.Parallel

2323 , Parallel

2424 , parallely

2525 , parallel

26+ , parallelEndByFirst

27+ , parallelEndByAny

2628

2729 -- * Function application

2830 , mkParallel

@@ -42,10 +44,13 @@ import Control.Monad.Reader.Class (MonadReader(..))

4244import Control.Monad.State.Class (MonadState(..))

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

4446import Data.Functor (void)

47+import Data.IORef (readIORef, writeIORef)

4548import Data.Maybe (fromJust)

4649import Data.Semigroup (Semigroup(..))

4750import Prelude hiding (map)

4851

52+import qualified Data.Set as Set

53+

4954import Streamly.Streams.SVar (fromSVar)

5055import Streamly.Streams.Serial (map)

5156import Streamly.SVar

@@ -88,22 +93,28 @@ runOne st m winfo = foldStreamShared st yieldk single stop m

8893 >> withLimitCheck (void $ liftIO $ mrun $ runOne st r winfo)

8994

9095{-# NOINLINE forkSVarPar #-}

91−forkSVarPar :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

92−forkSVarPar m r = mkStream $ \st yld sng stp -> do

93− sv <- newParallelVar st

96+forkSVarPar :: (IsStream t, MonadAsync m)

97+ => SVarStopStyle -> t m a -> t m a -> t m a

98+forkSVarPar ss m r = mkStream $ \st yld sng stp -> do

99+ sv <- newParallelVar ss st

94100 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)

101+ case ss of

102+ StopBy -> liftIO $ do

103+ set <- readIORef (workerThreads sv)

104+ writeIORef (svarStopBy sv) $ Set.elemAt 0 set

105+ _ -> return ()

95106 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)

96107 foldStream st yld sng stp (fromSVar sv)

97108

98109{-# INLINE joinStreamVarPar #-}

99110joinStreamVarPar :: (IsStream t, MonadAsync m)

100− => SVarStyle -> t m a -> t m a -> t m a

101−joinStreamVarPar style m1 m2 = mkStream $ \st yld sng stp ->

111+ => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a

112+joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->

102113 case streamVar st of

103− Just sv | svarStyle sv == style -> do

114+ Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do

104115 pushWorkerPar sv (runOne st $ toStream m1)

105116 foldStreamShared st yld sng stp m2

106− _ -> foldStreamShared st yld sng stp (forkSVarPar m1 m2)

117+ _ -> foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)

107118

108119-- | XXX we can implement it more efficienty by directly implementing instead

109120-- of combining streams using parallel.

@@ -118,7 +129,28 @@ consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)

118129-- @since 0.2.0

119130{-# INLINE parallel #-}

120131parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

121−parallel = joinStreamVarPar ParallelVar

132+parallel = joinStreamVarPar ParallelVar StopNone

133+

134+-- This is a co-parallel like combinator for streams, where first stream is the

135+-- main stream and the rest are just supporting it, when the first ends

136+-- everything ends.

137+--

138+-- | Like `parallel` but stops the output as soon as the first stream stops.

139+--

140+-- @since 0.7.0

141+{-# INLINE parallelEndByFirst #-}

142+parallelEndByFirst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

143+parallelEndByFirst = joinStreamVarPar ParallelVar StopBy

144+

145+-- This is a race like combinator for streams.

146+--

147+-- | Like `parallel` but stops the output as soon as any of the two streams

148+-- stops.

149+--

150+-- @since 0.7.0

151+{-# INLINE parallelEndByAny #-}

152+parallelEndByAny :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

153+parallelEndByAny = joinStreamVarPar ParallelVar StopAny

122154

123155------------------------------------------------------------------------------

124156-- Convert a stream to parallel

@@ -126,7 +158,7 @@ parallel = joinStreamVarPar ParallelVar

126158

127159mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)

128160mkParallel m = do

129− sv <- newParallelVar defState

161+ sv <- newParallelVar StopNone defState

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

131163 return $ fromSVar sv

132164

@@ -137,7 +169,7 @@ mkParallel m = do

137169{-# INLINE applyParallel #-}

138170applyParallel :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

139171applyParallel f m = mkStream $ \st yld sng stp -> do

140− sv <- newParallelVar (adaptState st)

172+ sv <- newParallelVar StopNone (adaptState st)

141173 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))

142174 foldStream st yld sng stp $ f $ fromSVar sv

143175

@@ -148,7 +180,7 @@ applyParallel f m = mkStream $ \st yld sng stp -> do

148180{-# INLINE foldParallel #-}

149181foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b

150182foldParallel f m = do

151− sv <- newParallelVar defState

183+ sv <- newParallelVar StopNone defState

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

153185 f $ fromSVar sv

154186

src/Streamly/Streams/SVar.hs+12−1

@@ -68,12 +68,23 @@ fromStreamVar sv = mkStream $ \st yld sng stp -> do

6868 ChildStop tid e -> do

6969 accountThread sv tid

7070 case e of

71− Nothing -> foldStream st yld sng stp rest

71+ Nothing -> do

72+ stop <- shouldStop tid

73+ if stop

74+ then liftIO (cleanupSVar sv) >> allDone stp

75+ else foldStream st yld sng stp rest

7276 Just ex ->

7377 case fromException ex of

7478 Just ThreadAbort ->

7579 foldStream st yld sng stp rest

7680 Nothing -> liftIO (cleanupSVar sv) >> throwM ex

81+ shouldStop tid =

82+ case svarStopStyle sv of

83+ StopNone -> return False

84+ StopAny -> return True

85+ StopBy -> do

86+ sid <- liftIO $ readIORef (svarStopBy sv)

87+ return $ if tid == sid then True else False

7788

7889{-# INLINE fromSVar #-}

7990fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a