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