git.iko.soy/streamly
on 56569bc help sign in
streamly/src/Streamly/Internal/Data/Unfold/SVar.hs
160 lines 5.9 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Unfold.SVar
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD3
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9module Streamly.Internal.Data.Unfold.SVar
10 (
11 fromSVar
12 , fromProducer
13 )
14where
16#include "inline.hs"
18import Control.Exception (fromException)
19import Control.Monad (when)
20import Control.Monad.IO.Class (MonadIO(..))
21import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
22import Data.Maybe (isNothing)
23import Streamly.Internal.Control.Concurrent (MonadAsync)
24import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))
25import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
26import Streamly.Internal.Data.SVar (printSVar, cleanupSVar, sendStopToProducer)
27import Streamly.Internal.Data.Unfold.Type (Unfold(..))
28import System.Mem (performMajorGC)
30import qualified Control.Monad.Catch as MC
32import Streamly.Internal.Data.SVar.Type
33-------------------------------------------------------------------------------
34-- Generation from SVar
35-------------------------------------------------------------------------------
37data FromSVarState t m a =
38 FromSVarInit (SVar t m a)
39 | FromSVarRead (SVar t m a)
40 | FromSVarLoop (SVar t m a) [ChildEvent a]
41 | FromSVarDone (SVar t m a)
43-- | /Internal/
44--
45{-# INLINE_NORMAL fromSVar #-}
46fromSVar :: MonadAsync m => Unfold m (SVar t m a) a
47fromSVar = Unfold step (return . FromSVarInit)
48 where
50 {-# INLINE_LATE step #-}
51 step (FromSVarInit svar) = do
52 ref <- liftIO $ newIORef ()
53 _ <- liftIO $ mkWeakIORef ref hook
54 -- when this copy of svar gets garbage collected "ref" will get
55 -- garbage collected and our GC hook will be called.
56 let sv = svar{svarRef = Just ref}
57 return $ Skip (FromSVarRead sv)
59 where
61 {-# NOINLINE hook #-}
62 hook = do
63 when (svarInspectMode svar) $ do
64 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
65 when (isNothing r) $
66 printSVar svar "SVar Garbage Collected"
67 cleanupSVar svar
68 -- If there are any SVars referenced by this SVar a GC will prompt
69 -- them to be cleaned up quickly.
70 when (svarInspectMode svar) performMajorGC
72 step (FromSVarRead sv) = do
73 list <- readOutputQ sv
74 -- Reversing the output is important to guarantee that we process the
75 -- outputs in the same order as they were generated by the constituent
76 -- streams.
77 return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
79 step (FromSVarLoop sv []) = do
80 done <- postProcess sv
81 return $ Skip $ if done
82 then FromSVarDone sv
83 else FromSVarRead sv
85 step (FromSVarLoop sv (ev : es)) = do
86 case ev of
87 ChildYield a -> return $ Yield a (FromSVarLoop sv es)
88 ChildStop tid e -> do
89 accountThread sv tid
90 case e of
91 Nothing -> do
92 stop <- shouldStop tid
93 if stop
94 then do
95 liftIO (cleanupSVar sv)
96 return $ Skip (FromSVarDone sv)
97 else return $ Skip (FromSVarLoop sv es)
98 Just ex ->
99 case fromException ex of
100 Just ThreadAbort ->
101 return $ Skip (FromSVarLoop sv es)
102 Nothing -> liftIO (cleanupSVar sv) >> MC.throwM ex
103 where
105 shouldStop tid =
106 case svarStopStyle sv of
107 StopNone -> return False
108 StopAny -> return True
109 StopBy -> do
110 sid <- liftIO $ readIORef (svarStopBy sv)
111 return $ tid == sid
113 step (FromSVarDone sv) = do
114 when (svarInspectMode sv) $ do
115 t <- liftIO $ getTime Monotonic
116 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
117 liftIO $ printSVar sv "SVar Done"
118 return Stop
120-------------------------------------------------------------------------------
121-- Process events received by a fold consumer from a stream producer
122-------------------------------------------------------------------------------
124-- XXX Have an error for "FromSVarInit" instead of undefined. The error can
125-- redirect the user to report the failure to the developers.
126-- | /Internal/
127--
128{-# INLINE_NORMAL fromProducer #-}
129fromProducer :: MonadAsync m => Unfold m (SVar t m a) a
130fromProducer = Unfold step (return . FromSVarRead)
131 where
133 {-# INLINE_LATE step #-}
134 step (FromSVarRead sv) = do
135 list <- readOutputQ sv
136 -- Reversing the output is important to guarantee that we process the
137 -- outputs in the same order as they were generated by the constituent
138 -- streams.
139 return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
141 step (FromSVarLoop sv []) = return $ Skip $ FromSVarRead sv
142 step (FromSVarLoop sv (ev : es)) = do
143 case ev of
144 ChildYield a -> return $ Yield a (FromSVarLoop sv es)
145 ChildStop tid e -> do
146 accountThread sv tid
147 case e of
148 Nothing -> do
149 sendStopToProducer sv
150 return $ Skip (FromSVarDone sv)
151 Just _ -> error "Bug: fromProducer: received exception"
153 step (FromSVarDone sv) = do
154 when (svarInspectMode sv) $ do
155 t <- liftIO $ getTime Monotonic
156 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
157 liftIO $ printSVar sv "SVar Done"
158 return Stop
160 step (FromSVarInit _) = undefined