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