git.iko.soy/streamly
on 167ea55 help sign in
streamly/src/Streamly/Internal/Data/Stream/SVar/Generate.hs
294 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 as D
47 (Stream(..), Step(..))
48import qualified Streamly.Internal.Data.StreamK 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#if __GLASGOW_HASKELL__ != 906
166-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
167-- space leak because the tuple gets allocated in every recursive call and each
168-- allocation holds on to the previous allocation. This test is to make sure
169-- that we do not use the constraint tuple type class.
170--
171inspect $ hasNoTypeClassesExcept 'fromStreamVar
172 [ ''Monad
173 , ''Applicative
174 , ''MonadThrow
175 , ''Exception
176 , ''MonadIO
177 , ''MonadBaseControl
178 , ''Typeable
179 , ''Functor
180 ]
181#endif
182#endif
184-- | Generate a stream from an SVar. An unevaluated stream can be pushed to an
185-- SVar using 'toSVar'. As we pull a stream from the SVar the input stream
186-- gets evaluated concurrently. The evaluation depends on the SVar style and
187-- the configuration parameters e.g. using the maxBuffer/maxThreads
188-- combinators.
189--
190{-# INLINE fromSVar #-}
191fromSVar :: MonadAsync m => SVar K.Stream m a -> SerialT m a
192fromSVar sv =
193 Stream.fromStreamK $ K.mkStream $ \st yld sng stp -> do
194 ref <- liftIO $ newIORef ()
195 _ <- liftIO $ mkWeakIORef ref hook
196 -- We pass a copy of sv to fromStreamVar, so that we know that it has
197 -- no other references, when that copy gets garbage collected "ref"
198 -- will get garbage collected and our hook will be called.
199 K.foldStreamShared st yld sng stp $
200 fromStreamVar sv{svarRef = Just ref}
201 where
203 hook = do
204 when (svarInspectMode sv) $ do
205 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
206 when (isNothing r) $
207 printSVar sv "SVar Garbage Collected"
208 cleanupSVar sv
209 -- If there are any SVars referenced by this SVar a GC will prompt
210 -- them to be cleaned up quickly.
211 when (svarInspectMode sv) performMajorGC
213data FromSVarState t m a =
214 FromSVarInit
215 | FromSVarRead (SVar t m a)
216 | FromSVarLoop (SVar t m a) [ChildEvent a]
217 | FromSVarDone (SVar t m a)
219-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
220--
221{-# INLINE_NORMAL fromSVarD #-}
222fromSVarD :: (MonadAsync m) => SVar t m a -> D.Stream m a
223fromSVarD svar = D.Stream step FromSVarInit
224 where
226 {-# INLINE_LATE step #-}
227 step _ FromSVarInit = do
228 ref <- liftIO $ newIORef ()
229 _ <- liftIO $ mkWeakIORef ref hook
230 -- when this copy of svar gets garbage collected "ref" will get
231 -- garbage collected and our GC hook will be called.
232 let sv = svar{svarRef = Just ref}
233 return $ D.Skip (FromSVarRead sv)
235 where
237 {-# NOINLINE hook #-}
238 hook = do
239 when (svarInspectMode svar) $ do
240 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
241 when (isNothing r) $
242 printSVar svar "SVar Garbage Collected"
243 cleanupSVar svar
244 -- If there are any SVars referenced by this SVar a GC will prompt
245 -- them to be cleaned up quickly.
246 when (svarInspectMode svar) performMajorGC
248 step _ (FromSVarRead sv) = do
249 list <- readOutputQ sv
250 -- Reversing the output is important to guarantee that we process the
251 -- outputs in the same order as they were generated by the constituent
252 -- streams.
253 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
255 step _ (FromSVarLoop sv []) = do
256 done <- postProcess sv
257 return $ D.Skip $ if done
258 then FromSVarDone sv
259 else FromSVarRead sv
261 step _ (FromSVarLoop sv (ev : es)) = do
262 case ev of
263 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
264 ChildStop tid e -> do
265 accountThread sv tid
266 case e of
267 Nothing -> do
268 stop <- shouldStop tid
269 if stop
270 then do
271 liftIO (cleanupSVar sv)
272 return $ D.Skip (FromSVarDone sv)
273 else return $ D.Skip (FromSVarLoop sv es)
274 Just ex ->
275 case fromException ex of
276 Just ThreadAbort ->
277 return $ D.Skip (FromSVarLoop sv es)
278 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
279 where
281 shouldStop tid =
282 case svarStopStyle sv of
283 StopNone -> return False
284 StopAny -> return True
285 StopBy -> do
286 sid <- liftIO $ readIORef (svarStopBy sv)
287 return $ tid == sid
289 step _ (FromSVarDone sv) = do
290 when (svarInspectMode sv) $ do
291 t <- liftIO $ getTime Monotonic
292 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
293 liftIO $ printSVar sv "SVar Done"
294 return D.Stop