git.iko.soy/streamly
on ffcb268 help sign in
streamly/src/Streamly/Internal/Data/Stream/SVar/Generate.hs
291 lines 10 kB history raw
1#ifdef __HADDOCK_VERSION__
2#undef INSPECTION
3#endif
5#ifdef INSPECTION
6{-# LANGUAGE TemplateHaskell #-}
7{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
8#endif
10-- |
11-- Module : Streamly.Internal.Data.Stream.SVar.Generate
12-- Copyright : (c) 2017 Composewell Technologies
13-- License : BSD-3-Clause
14-- Maintainer : streamly@composewell.com
15-- Stability : experimental
16-- Portability : GHC
17--
18--
19module Streamly.Internal.Data.Stream.SVar.Generate
20 (
21 -- * Write to SVar
22 toSVar
24 -- * Read from SVar
25 -- $concurrentEval
26 , fromSVar
27 , fromSVarD
28 -- , fromStreamVar
29 )
30where
32#include "inline.hs"
34import Control.Exception (fromException)
35import Control.Monad (when)
36import Control.Monad.Catch (throwM)
37import Control.Monad.IO.Class (MonadIO(liftIO))
38import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
39import Data.Maybe (isNothing)
40import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
41import Streamly.Internal.Data.Stream.Type (Stream)
42import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
43import System.Mem (performMajorGC)
45import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
46 (Stream(..), Step(..))
47import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
48 (Stream(..), foldStreamShared, mkStream, foldStream)
49import qualified Streamly.Internal.Data.Stream.Type as Stream (fromStreamK)
51import Streamly.Internal.Data.SVar
53#ifdef INSPECTION
54import Control.Exception (Exception)
55import Control.Monad.Catch (MonadThrow)
56import Control.Monad.Trans.Control (MonadBaseControl)
57import Data.Typeable (Typeable)
58import Test.Inspection (inspect, hasNoTypeClassesExcept)
59#endif
62------------------------------------------------------------------------------
63-- Generating streams from SVar
64------------------------------------------------------------------------------
66-- $concurrentEval
67--
68-- Usually the SVar is used to concurrently evaluate multiple actions in a
69-- stream using many worker threads that push the results to the SVar and a
70-- single puller that pulls them from SVar generating the evaluated stream.
71--
72-- @
73-- input stream
74-- |
75-- <-----------------|<--------worker
76-- | exceptions |
77-- output stream <------SVar<------worker
78-- |
79-- |<--------worker
80--
81-- @
82--
83-- The puller itself schedules the worker threads based on demand.
84-- Exceptions are propagated from the worker threads to the puller.
86-------------------------------------------------------------------------------
87-- Write a stream to an SVar
88-------------------------------------------------------------------------------
90-- XXX this errors out for Parallel/Ahead SVars
91-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
92-- be read back from the SVar using 'fromSVar'.
93toSVar :: MonadAsync m => SVar Stream m a -> Stream m a -> m ()
94toSVar sv m = do
95 runIn <- askRunInIO
96 liftIO $ enqueue sv (runIn, m)
97 done <- allThreadsDone sv
98 -- XXX This is safe only when called from the consumer thread or when no
99 -- consumer is present. There may be a race if we are not running in the
100 -- consumer thread.
101 -- XXX do this only if the work queue is not empty. The work may have been
102 -- carried out by existing workers.
103 when done $
104 case yieldRateInfo sv of
105 Nothing -> pushWorker 0 sv
106 Just _ -> pushWorker 1 sv
108-------------------------------------------------------------------------------
109-- Read a stream from an SVar
110-------------------------------------------------------------------------------
112-- | Pull a stream from an SVar.
113{-# NOINLINE fromStreamVar #-}
114fromStreamVar :: MonadAsync m => SVar K.Stream m a -> K.Stream m a
115fromStreamVar sv = K.MkStream $ \st yld sng stp -> do
116 list <- readOutputQ sv
117 -- Reversing the output is important to guarantee that we process the
118 -- outputs in the same order as they were generated by the constituent
119 -- streams.
120 K.foldStream st yld sng stp $ processEvents $ reverse list
122 where
124 allDone stp = do
125 when (svarInspectMode sv) $ do
126 t <- liftIO $ getTime Monotonic
127 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
128 liftIO $ printSVar sv "SVar Done"
129 stp
131 {-# INLINE processEvents #-}
132 processEvents [] = K.MkStream $ \st yld sng stp -> do
133 done <- postProcess sv
134 if done
135 then allDone stp
136 else K.foldStream st yld sng stp $ fromStreamVar sv
138 processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
139 let rest = processEvents es
140 case ev of
141 ChildYield a -> yld a rest
142 ChildStop tid e -> do
143 accountThread sv tid
144 case e of
145 Nothing -> do
146 stop <- shouldStop tid
147 if stop
148 then liftIO (cleanupSVar sv) >> allDone stp
149 else K.foldStream st yld sng stp rest
150 Just ex ->
151 case fromException ex of
152 Just ThreadAbort ->
153 K.foldStream st yld sng stp rest
154 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
155 shouldStop tid =
156 case svarStopStyle sv of
157 StopNone -> return False
158 StopAny -> return True
159 StopBy -> do
160 sid <- liftIO $ readIORef (svarStopBy sv)
161 return $ tid == sid
163#ifdef INSPECTION
164-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
165-- space leak because the tuple gets allocated in every recursive call and each
166-- allocation holds on to the previous allocation. This test is to make sure
167-- that we do not use the constraint tuple type class.
168--
169inspect $ hasNoTypeClassesExcept 'fromStreamVar
170 [ ''Monad
171 , ''Applicative
172 , ''MonadThrow
173 , ''Exception
174 , ''MonadIO
175 , ''MonadBaseControl
176 , ''Typeable
177 , ''Functor
178 ]
179#endif
181-- | Generate a stream from an SVar. An unevaluated stream can be pushed to an
182-- SVar using 'toSVar'. As we pull a stream from the SVar the input stream
183-- gets evaluated concurrently. The evaluation depends on the SVar style and
184-- the configuration parameters e.g. using the maxBuffer/maxThreads
185-- combinators.
186--
187{-# INLINE fromSVar #-}
188fromSVar :: MonadAsync m => SVar K.Stream m a -> Stream m a
189fromSVar sv =
190 Stream.fromStreamK $ K.mkStream $ \st yld sng stp -> do
191 ref <- liftIO $ newIORef ()
192 _ <- liftIO $ mkWeakIORef ref hook
193 -- We pass a copy of sv to fromStreamVar, so that we know that it has
194 -- no other references, when that copy gets garbage collected "ref"
195 -- will get garbage collected and our hook will be called.
196 K.foldStreamShared st yld sng stp $
197 fromStreamVar sv{svarRef = Just ref}
198 where
200 hook = do
201 when (svarInspectMode sv) $ do
202 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
203 when (isNothing r) $
204 printSVar sv "SVar Garbage Collected"
205 cleanupSVar sv
206 -- If there are any SVars referenced by this SVar a GC will prompt
207 -- them to be cleaned up quickly.
208 when (svarInspectMode sv) performMajorGC
210data FromSVarState t m a =
211 FromSVarInit
212 | FromSVarRead (SVar t m a)
213 | FromSVarLoop (SVar t m a) [ChildEvent a]
214 | FromSVarDone (SVar t m a)
216-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
217--
218{-# INLINE_NORMAL fromSVarD #-}
219fromSVarD :: (MonadAsync m) => SVar t m a -> D.Stream m a
220fromSVarD svar = D.Stream step FromSVarInit
221 where
223 {-# INLINE_LATE step #-}
224 step _ FromSVarInit = do
225 ref <- liftIO $ newIORef ()
226 _ <- liftIO $ mkWeakIORef ref hook
227 -- when this copy of svar gets garbage collected "ref" will get
228 -- garbage collected and our GC hook will be called.
229 let sv = svar{svarRef = Just ref}
230 return $ D.Skip (FromSVarRead sv)
232 where
234 {-# NOINLINE hook #-}
235 hook = do
236 when (svarInspectMode svar) $ do
237 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
238 when (isNothing r) $
239 printSVar svar "SVar Garbage Collected"
240 cleanupSVar svar
241 -- If there are any SVars referenced by this SVar a GC will prompt
242 -- them to be cleaned up quickly.
243 when (svarInspectMode svar) performMajorGC
245 step _ (FromSVarRead sv) = do
246 list <- readOutputQ sv
247 -- Reversing the output is important to guarantee that we process the
248 -- outputs in the same order as they were generated by the constituent
249 -- streams.
250 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
252 step _ (FromSVarLoop sv []) = do
253 done <- postProcess sv
254 return $ D.Skip $ if done
255 then FromSVarDone sv
256 else FromSVarRead sv
258 step _ (FromSVarLoop sv (ev : es)) = do
259 case ev of
260 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
261 ChildStop tid e -> do
262 accountThread sv tid
263 case e of
264 Nothing -> do
265 stop <- shouldStop tid
266 if stop
267 then do
268 liftIO (cleanupSVar sv)
269 return $ D.Skip (FromSVarDone sv)
270 else return $ D.Skip (FromSVarLoop sv es)
271 Just ex ->
272 case fromException ex of
273 Just ThreadAbort ->
274 return $ D.Skip (FromSVarLoop sv es)
275 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
276 where
278 shouldStop tid =
279 case svarStopStyle sv of
280 StopNone -> return False
281 StopAny -> return True
282 StopBy -> do
283 sid <- liftIO $ readIORef (svarStopBy sv)
284 return $ tid == sid
286 step _ (FromSVarDone sv) = do
287 when (svarInspectMode sv) $ do
288 t <- liftIO $ getTime Monotonic
289 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
290 liftIO $ printSVar sv "SVar Done"
291 return D.Stop