git.iko.soy/streamly
on 43dda21 help sign in
streamly/src/Streamly/Internal/Data/Unfold/SVar.hs
164 lines 6.0 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 System.Mem (performMajorGC)
29import qualified Control.Monad.Catch as MC
31import Streamly.Internal.Data.SVar.Type
32import Streamly.Internal.Data.Unfold.Type
33import Prelude
34 hiding (map, mapM, takeWhile, take, filter, const, zipWith
35 , drop, dropWhile, either)
37-------------------------------------------------------------------------------
38-- Generation from SVar
39-------------------------------------------------------------------------------
41data FromSVarState t m a =
42 FromSVarInit (SVar t m a)
43 | FromSVarRead (SVar t m a)
44 | FromSVarLoop (SVar t m a) [ChildEvent a]
45 | FromSVarDone (SVar t m a)
47-- | /Internal/
48--
49{-# INLINE_NORMAL fromSVar #-}
50fromSVar :: MonadAsync m => Unfold m (SVar t m a) a
51fromSVar = Unfold step (return . FromSVarInit)
52 where
54 {-# INLINE_LATE step #-}
55 step (FromSVarInit svar) = do
56 ref <- liftIO $ newIORef ()
57 _ <- liftIO $ mkWeakIORef ref hook
58 -- when this copy of svar gets garbage collected "ref" will get
59 -- garbage collected and our GC hook will be called.
60 let sv = svar{svarRef = Just ref}
61 return $ Skip (FromSVarRead sv)
63 where
65 {-# NOINLINE hook #-}
66 hook = do
67 when (svarInspectMode svar) $ do
68 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
69 when (isNothing r) $
70 printSVar svar "SVar Garbage Collected"
71 cleanupSVar svar
72 -- If there are any SVars referenced by this SVar a GC will prompt
73 -- them to be cleaned up quickly.
74 when (svarInspectMode svar) performMajorGC
76 step (FromSVarRead sv) = do
77 list <- readOutputQ sv
78 -- Reversing the output is important to guarantee that we process the
79 -- outputs in the same order as they were generated by the constituent
80 -- streams.
81 return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
83 step (FromSVarLoop sv []) = do
84 done <- postProcess sv
85 return $ Skip $ if done
86 then FromSVarDone sv
87 else FromSVarRead sv
89 step (FromSVarLoop sv (ev : es)) = do
90 case ev of
91 ChildYield a -> return $ Yield a (FromSVarLoop sv es)
92 ChildStop tid e -> do
93 accountThread sv tid
94 case e of
95 Nothing -> do
96 stop <- shouldStop tid
97 if stop
98 then do
99 liftIO (cleanupSVar sv)
100 return $ Skip (FromSVarDone sv)
101 else return $ Skip (FromSVarLoop sv es)
102 Just ex ->
103 case fromException ex of
104 Just ThreadAbort ->
105 return $ Skip (FromSVarLoop sv es)
106 Nothing -> liftIO (cleanupSVar sv) >> MC.throwM ex
107 where
109 shouldStop tid =
110 case svarStopStyle sv of
111 StopNone -> return False
112 StopAny -> return True
113 StopBy -> do
114 sid <- liftIO $ readIORef (svarStopBy sv)
115 return $ tid == sid
117 step (FromSVarDone sv) = do
118 when (svarInspectMode sv) $ do
119 t <- liftIO $ getTime Monotonic
120 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
121 liftIO $ printSVar sv "SVar Done"
122 return Stop
124-------------------------------------------------------------------------------
125-- Process events received by a fold consumer from a stream producer
126-------------------------------------------------------------------------------
128-- XXX Have an error for "FromSVarInit" instead of undefined. The error can
129-- redirect the user to report the failure to the developers.
130-- | /Internal/
131--
132{-# INLINE_NORMAL fromProducer #-}
133fromProducer :: MonadAsync m => Unfold m (SVar t m a) a
134fromProducer = Unfold step (return . FromSVarRead)
135 where
137 {-# INLINE_LATE step #-}
138 step (FromSVarRead sv) = do
139 list <- readOutputQ sv
140 -- Reversing the output is important to guarantee that we process the
141 -- outputs in the same order as they were generated by the constituent
142 -- streams.
143 return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
145 step (FromSVarLoop sv []) = return $ Skip $ FromSVarRead sv
146 step (FromSVarLoop sv (ev : es)) = do
147 case ev of
148 ChildYield a -> return $ Yield a (FromSVarLoop sv es)
149 ChildStop tid e -> do
150 accountThread sv tid
151 case e of
152 Nothing -> do
153 sendStopToProducer sv
154 return $ Skip (FromSVarDone sv)
155 Just _ -> error "Bug: fromProducer: received exception"
157 step (FromSVarDone sv) = do
158 when (svarInspectMode sv) $ do
159 t <- liftIO $ getTime Monotonic
160 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
161 liftIO $ printSVar sv "SVar Done"
162 return Stop
164 step (FromSVarInit _) = undefined