git.iko.soy/streamly
on master help sign in
Merge StreamD.Common and StreamD.Type modules.
c339342Pranay Sashank
src/Streamly/Internal/Data/Stream/StreamD.hs−1
compare with master
unifiedsplit
@@ -308,7 +308,6 @@ import Prelude hiding
308 , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat
308 , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat
309 , reverse, iterate, splitAt)
309 , reverse, iterate, splitAt)
310import Streamly.Internal.Data.Stream.StreamD.Type
310import Streamly.Internal.Data.Stream.StreamD.Type
311−import Streamly.Internal.Data.Stream.StreamD.Common (uncons)
312import Streamly.Internal.Data.Stream.StreamD.Generate
311import Streamly.Internal.Data.Stream.StreamD.Generate
313import Streamly.Internal.Data.Stream.StreamD.Eliminate
312import Streamly.Internal.Data.Stream.StreamD.Eliminate
314import Streamly.Internal.Data.Stream.StreamD.Exceptions
313import Streamly.Internal.Data.Stream.StreamD.Exceptions
src/Streamly/Internal/Data/Stream/StreamD/Common.hsdeleted−338
@@ -1,338 +0,0 @@
1−#include "inline.hs"
2−
3−-- |
4−-- Module : Streamly.Internal.Data.Stream.StreamD.Common
5−-- Copyright : (c) 2018 Composewell Technologies
6−-- (c) Roman Leshchinskiy 2008-2010
7−-- (c) The University of Glasgow, 2009
8−-- License : BSD3
9−-- Maintainer : streamly@composewell.com
10−-- Stability : experimental
11−-- Portability : GHC
12−
13−module Streamly.Internal.Data.Stream.StreamD.Common
14− (
15− fromProducer
16− , fromPrimIORef
17− , fromSVar
18− , foldOnce
19− , toSVarParallel
20− , toStreamD
21− , uncons
22− , nilM
23− , consM
24− , takeWhileM
25− , takeWhile
26− )
27−where
28−
29−
30−import Control.Exception (fromException)
31−import Control.Monad (when)
32−import Control.Monad.Catch (throwM)
33−import Control.Monad.IO.Class (MonadIO(..))
34−import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
35−import Data.Maybe (isNothing)
36−import GHC.Types (SPEC(..))
37−import System.Mem (performMajorGC)
38−import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
39−import Streamly.Internal.Data.IORef.Prim (Prim)
40−import Streamly.Internal.Data.Fold.Types (Fold(..))
41−import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
42−
43−import qualified Streamly.Internal.Data.Stream.StreamK as K
44−import qualified Streamly.Internal.Data.Fold as FL
45−import qualified Streamly.Internal.Data.IORef.Prim as Prim
46−
47−import Prelude hiding (takeWhile)
48−import Streamly.Internal.Data.Stream.StreamD.Type
49−import Streamly.Internal.Data.SVar
50−
51−-------------------------------------------------------------------------------
52−-- Deconstruction
53−-------------------------------------------------------------------------------
54−
55−-- Does not fuse, has the same performance as the StreamK version.
56−{-# INLINE_NORMAL uncons #-}
57−uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
58−uncons (UnStream step state) = go state
59− where
60− go st = do
61− r <- step defState st
62− case r of
63− Yield x s -> return $ Just (x, Stream step s)
64− Skip s -> go s
65− Stop -> return Nothing
66−
67−-------------------------------------------------------------------------------
68−-- Construction
69−-------------------------------------------------------------------------------
70−
71−-- | An empty 'Stream' with a side effect.
72−{-# INLINE_NORMAL nilM #-}
73−nilM :: Monad m => m b -> Stream m a
74−nilM m = Stream (\_ _ -> m >> return Stop) ()
75−
76−{-# INLINE_NORMAL consM #-}
77−consM :: Monad m => m a -> Stream m a -> Stream m a
78−consM m (Stream step state) = Stream step1 Nothing
79− where
80− {-# INLINE_LATE step1 #-}
81− step1 _ Nothing = m >>= \x -> return $ Yield x (Just state)
82− step1 gst (Just st) = do
83− r <- step gst st
84− return $
85− case r of
86− Yield a s -> Yield a (Just s)
87− Skip s -> Skip (Just s)
88− Stop -> Stop
89−
90−{-# INLINE toStreamD #-}
91−toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a
92−toStreamD = fromStreamK . K.toStream
93−data FromSVarState t m a =
94− FromSVarInit
95− | FromSVarRead (SVar t m a)
96− | FromSVarLoop (SVar t m a) [ChildEvent a]
97− | FromSVarDone (SVar t m a)
98−
99−-------------------------------------------------------------------------------
100−-- Generation from SVar
101−-------------------------------------------------------------------------------
102−
103−{-# INLINE_NORMAL fromSVar #-}
104−fromSVar :: (MonadAsync m) => SVar t m a -> Stream m a
105−fromSVar svar = Stream step FromSVarInit
106− where
107−
108− {-# INLINE_LATE step #-}
109− step _ FromSVarInit = do
110− ref <- liftIO $ newIORef ()
111− _ <- liftIO $ mkWeakIORef ref hook
112− -- when this copy of svar gets garbage collected "ref" will get
113− -- garbage collected and our GC hook will be called.
114− let sv = svar{svarRef = Just ref}
115− return $ Skip (FromSVarRead sv)
116−
117− where
118−
119− {-# NOINLINE hook #-}
120− hook = do
121− when (svarInspectMode svar) $ do
122− r <- liftIO $ readIORef (svarStopTime (svarStats svar))
123− when (isNothing r) $
124− printSVar svar "SVar Garbage Collected"
125− cleanupSVar svar
126− -- If there are any SVars referenced by this SVar a GC will prompt
127− -- them to be cleaned up quickly.
128− when (svarInspectMode svar) performMajorGC
129−
130− step _ (FromSVarRead sv) = do
131− list <- readOutputQ sv
132− -- Reversing the output is important to guarantee that we process the
133− -- outputs in the same order as they were generated by the constituent
134− -- streams.
135− return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
136−
137− step _ (FromSVarLoop sv []) = do
138− done <- postProcess sv
139− return $ Skip $ if done
140− then (FromSVarDone sv)
141− else (FromSVarRead sv)
142−
143− step _ (FromSVarLoop sv (ev : es)) = do
144− case ev of
145− ChildYield a -> return $ Yield a (FromSVarLoop sv es)
146− ChildStop tid e -> do
147− accountThread sv tid
148− case e of
149− Nothing -> do
150− stop <- shouldStop tid
151− if stop
152− then do
153− liftIO (cleanupSVar sv)
154− return $ Skip (FromSVarDone sv)
155− else return $ Skip (FromSVarLoop sv es)
156− Just ex ->
157− case fromException ex of
158− Just ThreadAbort ->
159− return $ Skip (FromSVarLoop sv es)
160− Nothing -> liftIO (cleanupSVar sv) >> throwM ex
161− where
162−
163− shouldStop tid =
164− case svarStopStyle sv of
165− StopNone -> return False
166− StopAny -> return True
167− StopBy -> do
168− sid <- liftIO $ readIORef (svarStopBy sv)
169− return $ if tid == sid then True else False
170−
171− step _ (FromSVarDone sv) = do
172− when (svarInspectMode sv) $ do
173− t <- liftIO $ getTime Monotonic
174− liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
175− liftIO $ printSVar sv "SVar Done"
176− return Stop
177−
178−{-# INLINE_NORMAL fromPrimIORef #-}
179−fromPrimIORef :: (MonadIO m, Prim a) => Prim.IORef a -> Stream m a
180−fromPrimIORef var = Stream step ()
181− where
182− {-# INLINE_LATE step #-}
183− step _ () = liftIO (Prim.readIORef var) >>= \x -> return $ Yield x ()
184−
185−-------------------------------------------------------------------------------
186−-- Process events received by a fold consumer from a stream producer
187−-------------------------------------------------------------------------------
188−
189−{-# INLINE_NORMAL fromProducer #-}
190−fromProducer :: (MonadAsync m) => SVar t m a -> Stream m a
191−fromProducer svar = Stream step (FromSVarRead svar)
192− where
193−
194− {-# INLINE_LATE step #-}
195− step _ (FromSVarRead sv) = do
196− list <- readOutputQ sv
197− -- Reversing the output is important to guarantee that we process the
198− -- outputs in the same order as they were generated by the constituent
199− -- streams.
200− return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
201−
202− step _ (FromSVarLoop sv []) = return $ Skip $ FromSVarRead sv
203− step _ (FromSVarLoop sv (ev : es)) = do
204− case ev of
205− ChildYield a -> return $ Yield a (FromSVarLoop sv es)
206− ChildStop tid e -> do
207− accountThread sv tid
208− case e of
209− Nothing -> do
210− sendStopToProducer sv
211− return $ Skip (FromSVarDone sv)
212− Just _ -> error "Bug: fromProducer: received exception"
213−
214− step _ (FromSVarDone sv) = do
215− when (svarInspectMode sv) $ do
216− t <- liftIO $ getTime Monotonic
217− liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
218− liftIO $ printSVar sv "SVar Done"
219− return Stop
220−
221− step _ FromSVarInit = undefined
222−
223−------------------------------------------------------------------------------
224−-- Left Folds
225−------------------------------------------------------------------------------
226−
227−{-# INLINE_NORMAL foldOnce #-}
228−foldOnce :: (Monad m) => Fold m a b -> Stream m a -> m b
229−foldOnce (Fold fstep begin done) (Stream step state) =
230− begin >>= \x -> go SPEC x state
231−
232− where
233−
234− {-# INLINE go #-}
235− go !_ !fs st = do
236− r <- step defState st
237− case r of
238− Yield x s -> do
239− res <- fstep fs x
240− case res of
241− FL.Done b -> return b
242− FL.Partial fs1 -> go SPEC fs1 s
243− Skip s -> go SPEC fs s
244− Stop -> done fs
245−
246−-------------------------------------------------------------------------------
247−-- Concurrent application and fold
248−-------------------------------------------------------------------------------
249−
250−-- XXX These functions should be moved to Stream/Parallel.hs
251−--
252−-- Using StreamD the worker stream producing code can fuse with the code to
253−-- queue output to the SVar giving some perf boost.
254−--
255−-- Note that StreamD can only be used in limited situations, specifically, we
256−-- cannot implement joinStreamVarPar using this.
257−--
258−-- XXX make sure that the SVar passed is a Parallel style SVar.
259−
260−-- | Fold the supplied stream to the SVar asynchronously using Parallel
261−-- concurrency style.
262−-- {-# INLINE_NORMAL toSVarParallel #-}
263−{-# INLINE toSVarParallel #-}
264−toSVarParallel :: MonadAsync m
265− => State t m a -> SVar t m a -> Stream m a -> m ()
266−toSVarParallel st sv xs =
267− if svarInspectMode sv
268− then forkWithDiag
269− else do
270− tid <-
271− case getYieldLimit st of
272− Nothing -> doFork (work Nothing)
273− (svarMrun sv)
274− (handleChildException sv)
275− Just _ -> doFork (workLim Nothing)
276− (svarMrun sv)
277− (handleChildException sv)
278− modifyThread sv tid
279−
280− where
281−
282− {-# NOINLINE work #-}
283− work info = (foldOnce (FL.toParallelSVar sv info) xs)
284−
285− {-# NOINLINE workLim #-}
286− workLim info = foldOnce (FL.toParallelSVarLimited sv info) xs
287−
288− {-# NOINLINE forkWithDiag #-}
289− forkWithDiag = do
290− -- We do not use workerCount in case of ParallelVar but still there is
291− -- no harm in maintaining it correctly.
292− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
293− recordMaxWorkers sv
294− -- This allocation matters when significant number of workers are being
295− -- sent. We allocate it only when needed. The overhead increases by 4x.
296− winfo <-
297− case yieldRateInfo sv of
298− Nothing -> return Nothing
299− Just _ -> liftIO $ do
300− cntRef <- newIORef 0
301− t <- getTime Monotonic
302− lat <- newIORef (0, t)
303− return $ Just WorkerInfo
304− { workerYieldMax = 0
305− , workerYieldCount = cntRef
306− , workerLatencyStart = lat
307− }
308− tid <-
309− case getYieldLimit st of
310− Nothing -> doFork (work winfo)
311− (svarMrun sv)
312− (handleChildException sv)
313− Just _ -> doFork (workLim winfo)
314− (svarMrun sv)
315− (handleChildException sv)
316− modifyThread sv tid
317−
318−-------------------------------------------------------------------------------
319−-- Filtering
320−-------------------------------------------------------------------------------
321−
322−{-# INLINE_NORMAL takeWhileM #-}
323−takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
324−takeWhileM f (Stream step state) = Stream step' state
325− where
326− {-# INLINE_LATE step' #-}
327− step' gst st = do
328− r <- step gst st
329− case r of
330− Yield x s -> do
331− b <- f x
332− return $ if b then Yield x s else Stop
333− Skip s -> return $ Skip s
334− Stop -> return Stop
335−
336−{-# INLINE takeWhile #-}
337−takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
338−takeWhile f = takeWhileM (return . f)
src/Streamly/Internal/Data/Stream/StreamD/Eliminate.hs−2
@@ -122,10 +122,8 @@ import Prelude hiding
122 , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat
122 , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat
123 , reverse, iterate, splitAt)
123 , reverse, iterate, splitAt)
124import Streamly.Internal.Data.Stream.StreamD.Type
124import Streamly.Internal.Data.Stream.StreamD.Type
125−import Streamly.Internal.Data.Stream.StreamD.Common
126import Streamly.Internal.Data.SVar
125import Streamly.Internal.Data.SVar
127
126
128−
129------------------------------------------------------------------------------
127------------------------------------------------------------------------------
130-- Elimination by Folds
128-- Elimination by Folds
131------------------------------------------------------------------------------
129------------------------------------------------------------------------------
src/Streamly/Internal/Data/Stream/StreamD/Exceptions.hs−2
@@ -47,10 +47,8 @@ import Prelude hiding
47 , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat
47 , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat
48 , reverse, iterate, splitAt)
48 , reverse, iterate, splitAt)
49import Streamly.Internal.Data.Stream.StreamD.Type
49import Streamly.Internal.Data.Stream.StreamD.Type
50−import Streamly.Internal.Data.Stream.StreamD.Common
51import Streamly.Internal.Data.SVar
50import Streamly.Internal.Data.SVar
52
51
53−
54------------------------------------------------------------------------------
52------------------------------------------------------------------------------
55-- Exceptions
53-- Exceptions
56------------------------------------------------------------------------------
54------------------------------------------------------------------------------
src/Streamly/Internal/Data/Stream/StreamD/Generate.hs−1
@@ -82,7 +82,6 @@ import qualified Streamly.Internal.Data.IORef.Prim as Prim
82
82
83import Prelude hiding (iterate, repeat, replicate, takeWhile)
83import Prelude hiding (iterate, repeat, replicate, takeWhile)
84import Streamly.Internal.Data.Stream.StreamD.Type
84import Streamly.Internal.Data.Stream.StreamD.Type
85−import Streamly.Internal.Data.Stream.StreamD.Common
86import Streamly.Internal.Data.SVar
85import Streamly.Internal.Data.SVar
87
86
88------------------------------------------------------------------------------
87------------------------------------------------------------------------------
src/Streamly/Internal/Data/Stream/StreamD/Transform.hs−1
@@ -118,7 +118,6 @@ import Prelude hiding
118 ( drop, dropWhile, filter, map, mapM
118 ( drop, dropWhile, filter, map, mapM
119 , scanl, scanl1, sequence, take, takeWhile, zipWith)
119 , scanl, scanl1, sequence, take, takeWhile, zipWith)
120import Streamly.Internal.Data.Stream.StreamD.Type
120import Streamly.Internal.Data.Stream.StreamD.Type
121−import Streamly.Internal.Data.Stream.StreamD.Common
122import Streamly.Internal.Data.SVar
121import Streamly.Internal.Data.SVar
123
122
124-------------------------------------------------------------------------------
123-------------------------------------------------------------------------------
src/Streamly/Internal/Data/Stream/StreamD/Type.hs+317−7
@@ -21,10 +21,19 @@ module Streamly.Internal.Data.Stream.StreamD.Type
21 , fromStreamK
21 , fromStreamK
22 , toStreamK
22 , toStreamK
23 , fromStreamD
23 , fromStreamD
24− , map
24+ , toStreamD
25− , mapM
25+ , fromProducer
26+ , fromPrimIORef
27+ , fromSVar
28+ , foldOnce
29+ , toSVarParallel
30+ , uncons
31+ , nilM
32+ , consM
26 , yield
33 , yield
27 , yieldM
34 , yieldM
35+ , map
36+ , mapM
28 , concatMap
37 , concatMap
29 , concatMapM
38 , concatMapM
30
39
@@ -45,6 +54,8 @@ module Streamly.Internal.Data.Stream.StreamD.Type
45 , eqBy
54 , eqBy
46 , cmpBy
55 , cmpBy
47 , take
56 , take
57+ , takeWhileM
58+ , takeWhile
48 , GroupState (..) -- for inspection testing
59 , GroupState (..) -- for inspection testing
49 , foldMany
60 , foldMany
50 , foldMany1
61 , foldMany1
@@ -52,21 +63,31 @@ module Streamly.Internal.Data.Stream.StreamD.Type
52 )
63 )
53where
64where
54
65
66+
55import Control.Applicative (liftA2)
67import Control.Applicative (liftA2)
68+import Control.Exception (fromException)
56import Control.Monad (when)
69import Control.Monad (when)
57import Control.Monad.Catch (MonadThrow, throwM)
70import Control.Monad.Catch (MonadThrow, throwM)
71+import Control.Monad.IO.Class (MonadIO(..))
58import Control.Monad.Trans.Class (lift, MonadTrans)
72import Control.Monad.Trans.Class (lift, MonadTrans)
73+import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
59import Data.Functor.Identity (Identity(..))
74import Data.Functor.Identity (Identity(..))
75+import Data.Maybe (isNothing)
76+import Fusion.Plugin.Types (Fuse(..))
60import GHC.Base (build)
77import GHC.Base (build)
61import GHC.Types (SPEC(..))
78import GHC.Types (SPEC(..))
62−import Prelude hiding (map, mapM, foldr, take, concatMap)
79+import Prelude hiding (map, mapM, foldr, take, concatMap, takeWhile)
63−import Fusion.Plugin.Types (Fuse(..))
80+import System.Mem (performMajorGC)
64
81
65−import Streamly.Internal.Data.SVar (State(..), adaptState, defState)
82+import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
83+import Streamly.Internal.Data.IORef.Prim (Prim)
84+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
66import Streamly.Internal.Data.Fold.Types (Fold(..), Fold2(..))
85import Streamly.Internal.Data.Fold.Types (Fold(..), Fold2(..))
86+import Streamly.Internal.Data.SVar
67
87
68−import qualified Streamly.Internal.Data.Fold.Types as FL
88+import qualified Streamly.Internal.Data.Fold as FL
69import qualified Streamly.Internal.Data.Stream.StreamK as K
89import qualified Streamly.Internal.Data.Stream.StreamK as K
90+import qualified Streamly.Internal.Data.IORef.Prim as Prim
70
91
71------------------------------------------------------------------------------
92------------------------------------------------------------------------------
72-- The direct style stream type
93-- The direct style stream type
@@ -141,6 +162,49 @@ toStreamK (Stream step state) = go state
141fromStreamD :: (K.IsStream t, Monad m) => Stream m a -> t m a
162fromStreamD :: (K.IsStream t, Monad m) => Stream m a -> t m a
142fromStreamD = K.fromStream . toStreamK
163fromStreamD = K.fromStream . toStreamK
143
164
165+{-# INLINE toStreamD #-}
166+toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a
167+toStreamD = fromStreamK . K.toStream
168+
169+-------------------------------------------------------------------------------
170+-- Deconstruction
171+-------------------------------------------------------------------------------
172+
173+-- Does not fuse, has the same performance as the StreamK version.
174+{-# INLINE_NORMAL uncons #-}
175+uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
176+uncons (UnStream step state) = go state
177+ where
178+ go st = do
179+ r <- step defState st
180+ case r of
181+ Yield x s -> return $ Just (x, Stream step s)
182+ Skip s -> go s
183+ Stop -> return Nothing
184+
185+-------------------------------------------------------------------------------
186+-- Construction
187+-------------------------------------------------------------------------------
188+
189+-- | An empty 'Stream' with a side effect.
190+{-# INLINE_NORMAL nilM #-}
191+nilM :: Monad m => m b -> Stream m a
192+nilM m = Stream (\_ _ -> m >> return Stop) ()
193+
194+{-# INLINE_NORMAL consM #-}
195+consM :: Monad m => m a -> Stream m a -> Stream m a
196+consM m (Stream step state) = Stream step1 Nothing
197+ where
198+ {-# INLINE_LATE step1 #-}
199+ step1 _ Nothing = m >>= \x -> return $ Yield x (Just state)
200+ step1 gst (Just st) = do
201+ r <- step gst st
202+ return $
203+ case r of
204+ Yield a s -> Yield a (Just s)
205+ Skip s -> Skip (Just s)
206+ Stop -> Stop
207+
144------------------------------------------------------------------------------
208------------------------------------------------------------------------------
145-- Instances
209-- Instances
146------------------------------------------------------------------------------
210------------------------------------------------------------------------------
@@ -567,7 +631,9 @@ cmpBy cmp (Stream step1 t1) (Stream step2 t2) = cmp_loop0 SPEC t1 t2
567 Skip s2' -> cmp_null s2'
631 Skip s2' -> cmp_null s2'
568 Stop -> return EQ
632 Stop -> return EQ
569
633
570−-- XXX The take/drop combinators above should be moved to filtering section.
634+-------------------------------------------------------------------------------
635+-- Filtering
636+-------------------------------------------------------------------------------
571
637
572{-# INLINE_NORMAL take #-}
638{-# INLINE_NORMAL take #-}
573take :: Monad m => Int -> Stream m a -> Stream m a
639take :: Monad m => Int -> Stream m a -> Stream m a
@@ -582,6 +648,25 @@ take n (Stream step state) = n `seq` Stream step' (state, 0)
582 Stop -> Stop
648 Stop -> Stop
583 step' _ (_, _) = return Stop
649 step' _ (_, _) = return Stop
584
650
651+
652+{-# INLINE_NORMAL takeWhileM #-}
653+takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
654+takeWhileM f (Stream step state) = Stream step' state
655+ where
656+ {-# INLINE_LATE step' #-}
657+ step' gst st = do
658+ r <- step gst st
659+ case r of
660+ Yield x s -> do
661+ b <- f x
662+ return $ if b then Yield x s else Stop
663+ Skip s -> return $ Skip s
664+ Stop -> return Stop
665+
666+{-# INLINE takeWhile #-}
667+takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
668+takeWhile f = takeWhileM (return . f)
669+
585------------------------------------------------------------------------------
670------------------------------------------------------------------------------
586-- Grouping/Splitting
671-- Grouping/Splitting
587------------------------------------------------------------------------------
672------------------------------------------------------------------------------
@@ -706,3 +791,228 @@ groupsOf2 n input (Fold2 fstep inject extract) (Stream step state) =
706 return $ Yield r next
791 return $ Yield r next
707
792
708 step' _ GroupFinish2 = return Stop
793 step' _ GroupFinish2 = return Stop
794+
795+data FromSVarState t m a =
796+ FromSVarInit
797+ | FromSVarRead (SVar t m a)
798+ | FromSVarLoop (SVar t m a) [ChildEvent a]
799+ | FromSVarDone (SVar t m a)
800+
801+-------------------------------------------------------------------------------
802+-- Generation from SVar
803+-------------------------------------------------------------------------------
804+
805+{-# INLINE_NORMAL fromSVar #-}
806+fromSVar :: (MonadAsync m) => SVar t m a -> Stream m a
807+fromSVar svar = Stream step FromSVarInit
808+ where
809+
810+ {-# INLINE_LATE step #-}
811+ step _ FromSVarInit = do
812+ ref <- liftIO $ newIORef ()
813+ _ <- liftIO $ mkWeakIORef ref hook
814+ -- when this copy of svar gets garbage collected "ref" will get
815+ -- garbage collected and our GC hook will be called.
816+ let sv = svar{svarRef = Just ref}
817+ return $ Skip (FromSVarRead sv)
818+
819+ where
820+
821+ {-# NOINLINE hook #-}
822+ hook = do
823+ when (svarInspectMode svar) $ do
824+ r <- liftIO $ readIORef (svarStopTime (svarStats svar))
825+ when (isNothing r) $
826+ printSVar svar "SVar Garbage Collected"
827+ cleanupSVar svar
828+ -- If there are any SVars referenced by this SVar a GC will prompt
829+ -- them to be cleaned up quickly.
830+ when (svarInspectMode svar) performMajorGC
831+
832+ step _ (FromSVarRead sv) = do
833+ list <- readOutputQ sv
834+ -- Reversing the output is important to guarantee that we process the
835+ -- outputs in the same order as they were generated by the constituent
836+ -- streams.
837+ return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
838+
839+ step _ (FromSVarLoop sv []) = do
840+ done <- postProcess sv
841+ return $ Skip $ if done
842+ then (FromSVarDone sv)
843+ else (FromSVarRead sv)
844+
845+ step _ (FromSVarLoop sv (ev : es)) = do
846+ case ev of
847+ ChildYield a -> return $ Yield a (FromSVarLoop sv es)
848+ ChildStop tid e -> do
849+ accountThread sv tid
850+ case e of
851+ Nothing -> do
852+ stop <- shouldStop tid
853+ if stop
854+ then do
855+ liftIO (cleanupSVar sv)
856+ return $ Skip (FromSVarDone sv)
857+ else return $ Skip (FromSVarLoop sv es)
858+ Just ex ->
859+ case fromException ex of
860+ Just ThreadAbort ->
861+ return $ Skip (FromSVarLoop sv es)
862+ Nothing -> liftIO (cleanupSVar sv) >> throwM ex
863+ where
864+
865+ shouldStop tid =
866+ case svarStopStyle sv of
867+ StopNone -> return False
868+ StopAny -> return True
869+ StopBy -> do
870+ sid <- liftIO $ readIORef (svarStopBy sv)
871+ return $ if tid == sid then True else False
872+
873+ step _ (FromSVarDone sv) = do
874+ when (svarInspectMode sv) $ do
875+ t <- liftIO $ getTime Monotonic
876+ liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
877+ liftIO $ printSVar sv "SVar Done"
878+ return Stop
879+
880+{-# INLINE_NORMAL fromPrimIORef #-}
881+fromPrimIORef :: (MonadIO m, Prim a) => Prim.IORef a -> Stream m a
882+fromPrimIORef var = Stream step ()
883+ where
884+ {-# INLINE_LATE step #-}
885+ step _ () = liftIO (Prim.readIORef var) >>= \x -> return $ Yield x ()
886+
887+-------------------------------------------------------------------------------
888+-- Process events received by a fold consumer from a stream producer
889+-------------------------------------------------------------------------------
890+
891+{-# INLINE_NORMAL fromProducer #-}
892+fromProducer :: (MonadAsync m) => SVar t m a -> Stream m a
893+fromProducer svar = Stream step (FromSVarRead svar)
894+ where
895+
896+ {-# INLINE_LATE step #-}
897+ step _ (FromSVarRead sv) = do
898+ list <- readOutputQ sv
899+ -- Reversing the output is important to guarantee that we process the
900+ -- outputs in the same order as they were generated by the constituent
901+ -- streams.
902+ return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
903+
904+ step _ (FromSVarLoop sv []) = return $ Skip $ FromSVarRead sv
905+ step _ (FromSVarLoop sv (ev : es)) = do
906+ case ev of
907+ ChildYield a -> return $ Yield a (FromSVarLoop sv es)
908+ ChildStop tid e -> do
909+ accountThread sv tid
910+ case e of
911+ Nothing -> do
912+ sendStopToProducer sv
913+ return $ Skip (FromSVarDone sv)
914+ Just _ -> error "Bug: fromProducer: received exception"
915+
916+ step _ (FromSVarDone sv) = do
917+ when (svarInspectMode sv) $ do
918+ t <- liftIO $ getTime Monotonic
919+ liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
920+ liftIO $ printSVar sv "SVar Done"
921+ return Stop
922+
923+ step _ FromSVarInit = undefined
924+
925+------------------------------------------------------------------------------
926+-- Left Folds
927+------------------------------------------------------------------------------
928+
929+{-# INLINE_NORMAL foldOnce #-}
930+foldOnce :: (Monad m) => Fold m a b -> Stream m a -> m b
931+foldOnce (Fold fstep begin done) (Stream step state) =
932+ begin >>= \x -> go SPEC x state
933+
934+ where
935+
936+ {-# INLINE go #-}
937+ go !_ !fs st = do
938+ r <- step defState st
939+ case r of
940+ Yield x s -> do
941+ res <- fstep fs x
942+ case res of
943+ FL.Done b -> return b
944+ FL.Partial fs1 -> go SPEC fs1 s
945+ Skip s -> go SPEC fs s
946+ Stop -> done fs
947+
948+-------------------------------------------------------------------------------
949+-- Concurrent application and fold
950+-------------------------------------------------------------------------------
951+
952+-- XXX These functions should be moved to Stream/Parallel.hs
953+--
954+-- Using StreamD the worker stream producing code can fuse with the code to
955+-- queue output to the SVar giving some perf boost.
956+--
957+-- Note that StreamD can only be used in limited situations, specifically, we
958+-- cannot implement joinStreamVarPar using this.
959+--
960+-- XXX make sure that the SVar passed is a Parallel style SVar.
961+
962+-- | Fold the supplied stream to the SVar asynchronously using Parallel
963+-- concurrency style.
964+-- {-# INLINE_NORMAL toSVarParallel #-}
965+{-# INLINE toSVarParallel #-}
966+toSVarParallel :: MonadAsync m
967+ => State t m a -> SVar t m a -> Stream m a -> m ()
968+toSVarParallel st sv xs =
969+ if svarInspectMode sv
970+ then forkWithDiag
971+ else do
972+ tid <-
973+ case getYieldLimit st of
974+ Nothing -> doFork (work Nothing)
975+ (svarMrun sv)
976+ (handleChildException sv)
977+ Just _ -> doFork (workLim Nothing)
978+ (svarMrun sv)
979+ (handleChildException sv)
980+ modifyThread sv tid
981+
982+ where
983+
984+ {-# NOINLINE work #-}
985+ work info = (foldOnce (FL.toParallelSVar sv info) xs)
986+
987+ {-# NOINLINE workLim #-}
988+ workLim info = foldOnce (FL.toParallelSVarLimited sv info) xs
989+
990+ {-# NOINLINE forkWithDiag #-}
991+ forkWithDiag = do
992+ -- We do not use workerCount in case of ParallelVar but still there is
993+ -- no harm in maintaining it correctly.
994+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
995+ recordMaxWorkers sv
996+ -- This allocation matters when significant number of workers are being
997+ -- sent. We allocate it only when needed. The overhead increases by 4x.
998+ winfo <-
999+ case yieldRateInfo sv of
1000+ Nothing -> return Nothing
1001+ Just _ -> liftIO $ do
1002+ cntRef <- newIORef 0
1003+ t <- getTime Monotonic
1004+ lat <- newIORef (0, t)
1005+ return $ Just WorkerInfo
1006+ { workerYieldMax = 0
1007+ , workerYieldCount = cntRef
1008+ , workerLatencyStart = lat
1009+ }
1010+ tid <-
1011+ case getYieldLimit st of
1012+ Nothing -> doFork (work winfo)
1013+ (svarMrun sv)
1014+ (handleChildException sv)
1015+ Just _ -> doFork (workLim winfo)
1016+ (svarMrun sv)
1017+ (handleChildException sv)
1018+ modifyThread sv tid
streamly.cabal−1
@@ -452,7 +452,6 @@ library
452 , Streamly.Internal.Data.Stream.StreamK
452 , Streamly.Internal.Data.Stream.StreamK
453 , Streamly.Internal.Data.Stream.StreamD.Type
453 , Streamly.Internal.Data.Stream.StreamD.Type
454 , Streamly.Internal.Data.Stream.StreamD
454 , Streamly.Internal.Data.Stream.StreamD
455− , Streamly.Internal.Data.Stream.StreamD.Common
456 , Streamly.Internal.Data.Stream.StreamD.Generate
455 , Streamly.Internal.Data.Stream.StreamD.Generate
457 , Streamly.Internal.Data.Stream.StreamD.SplitGroup
456 , Streamly.Internal.Data.Stream.StreamD.SplitGroup
458 , Streamly.Internal.Data.Stream.StreamD.Eliminate
457 , Streamly.Internal.Data.Stream.StreamD.Eliminate