git.iko.soy/streamly
on a322c07 help sign in
streamly/src/Streamly/Internal/Data/Stream/StreamD/Exception.hs
351 lines 12 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.StreamD.Exception
3-- Copyright : (c) 2020 Composewell Technologies and Contributors
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Stream.StreamD.Exception
10 (
11 gbracket_
12 , gbracket
13 , before
14 , after_
15 , after
16 , bracket_
17 , bracket
18 , onException
19 , finally_
20 , finally
21 , ghandle
22 , handle
23 )
24where
26#include "inline.hs"
28import Control.Exception (Exception, SomeException, mask_)
29import Control.Monad.Catch (MonadCatch)
30import Control.Monad.IO.Class (MonadIO(..))
31import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)
32import GHC.Exts (inline)
33import Streamly.Internal.Data.IOFinalizer
34 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)
36import qualified Control.Monad.Catch as MC
38import Streamly.Internal.Data.Stream.StreamD.Type
39import Streamly.Internal.Data.SVar
41data GbracketState s1 s2 v
42 = GBracketInit
43 | GBracketNormal s1 v
44 | GBracketException s2
46-- | Like 'gbracket' but with following differences:
47--
48-- * alloc action @m c@ runs with async exceptions enabled
49-- * cleanup action @c -> m d@ won't run if the stream is garbage collected
50-- after partial evaluation.
51-- * does not require a 'MonadAsync' constraint.
52--
53-- /Inhibits stream fusion/
54--
55-- /Pre-release/
56--
57{-# INLINE_NORMAL gbracket_ #-}
58gbracket_
59 :: Monad m
60 => m c -- ^ before
61 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
62 -> (c -> m d) -- ^ after, on normal stop
63 -> (c -> e -> Stream m b -> Stream m b) -- ^ on exception
64 -> (c -> Stream m b) -- ^ stream generator
65 -> Stream m b
66gbracket_ bef exc aft fexc fnormal =
67 Stream step GBracketInit
69 where
71 {-# INLINE_LATE step #-}
72 step _ GBracketInit = do
73 r <- bef
74 return $ Skip $ GBracketNormal (fnormal r) r
76 step gst (GBracketNormal (UnStream step1 st) v) = do
77 res <- exc $ step1 gst st
78 case res of
79 Right r -> case r of
80 Yield x s ->
81 return $ Yield x (GBracketNormal (Stream step1 s) v)
82 Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)
83 Stop -> aft v >> return Stop
84 -- XXX Do not handle async exceptions, just rethrow them.
85 Left e ->
86 return $ Skip (GBracketException (fexc v e (UnStream step1 st)))
87 step gst (GBracketException (UnStream step1 st)) = do
88 res <- step1 gst st
89 case res of
90 Yield x s -> return $ Yield x (GBracketException (Stream step1 s))
91 Skip s -> return $ Skip (GBracketException (Stream step1 s))
92 Stop -> return Stop
94data GbracketIOState s1 s2 v wref
95 = GBracketIOInit
96 | GBracketIONormal s1 v wref
97 | GBracketIOException s2
99-- | Run the alloc action @m c@ with async exceptions disabled but keeping
100-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
101-- output @c@ as input to @c -> Stream m b@ to generate an output stream. When
102-- generating the stream use the supplied @try@ operation @forall s. m s -> m
103-- (Either e s)@ to catch synchronous exceptions. If an exception occurs run
104-- the exception handler @c -> e -> Stream m b -> m (Stream m b)@.
105--
106-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due
107-- to a sync or async exception or if it gets garbage collected after a partial
108-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.
109--
110-- 'gbracket' can express all other exception handling combinators.
111--
112-- /Inhibits stream fusion/
113--
114-- /Pre-release/
115{-# INLINE_NORMAL gbracket #-}
116gbracket
117 :: (MonadIO m, MonadBaseControl IO m)
118 => m c -- ^ before
119 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
120 -> (c -> m d) -- ^ after, on normal stop or GC
121 -> (c -> e -> Stream m b -> m (Stream m b)) -- ^ on exception
122 -> (c -> Stream m b) -- ^ stream generator
123 -> Stream m b
124gbracket bef exc aft fexc fnormal =
125 Stream step GBracketIOInit
127 where
129 -- If the stream is never evaluated the "aft" action will never be
130 -- called. For that to occur we will need the user of this API to pass a
131 -- weak pointer to us.
132 {-# INLINE_LATE step #-}
133 step _ GBracketIOInit = do
134 -- We mask asynchronous exceptions to make the execution
135 -- of 'bef' and the registration of 'aft' atomic.
136 -- A similar thing is done in the resourcet package: https://git.io/JvKV3
137 -- Tutorial: https://markkarpov.com/tutorial/exceptions.html
138 (r, ref) <- liftBaseOp_ mask_ $ do
139 r <- bef
140 ref <- newIOFinalizer (aft r)
141 return (r, ref)
142 return $ Skip $ GBracketIONormal (fnormal r) r ref
144 step gst (GBracketIONormal (UnStream step1 st) v ref) = do
145 res <- exc $ step1 gst st
146 case res of
147 Right r -> case r of
148 Yield x s ->
149 return $ Yield x (GBracketIONormal (Stream step1 s) v ref)
150 Skip s ->
151 return $ Skip (GBracketIONormal (Stream step1 s) v ref)
152 Stop -> do
153 runIOFinalizer ref
154 return Stop
155 -- XXX Do not handle async exceptions, just rethrow them.
156 Left e -> do
157 -- Clearing of finalizer and running of exception handler must
158 -- be atomic wrt async exceptions. Otherwise if we have cleared
159 -- the finalizer and have not run the exception handler then we
160 -- may leak the resource.
161 stream <- clearingIOFinalizer ref (fexc v e (UnStream step1 st))
162 return $ Skip (GBracketIOException stream)
163 step gst (GBracketIOException (UnStream step1 st)) = do
164 res <- step1 gst st
165 case res of
166 Yield x s ->
167 return $ Yield x (GBracketIOException (Stream step1 s))
168 Skip s -> return $ Skip (GBracketIOException (Stream step1 s))
169 Stop -> return Stop
171-- | See 'Streamly.Internal.Data.Stream.IsStream.before'.
172--
173{-# INLINE_NORMAL before #-}
174before :: Monad m => m b -> Stream m a -> Stream m a
175before action (Stream step state) = Stream step' Nothing
177 where
179 {-# INLINE_LATE step' #-}
180 step' _ Nothing = action >> return (Skip (Just state))
182 step' gst (Just st) = do
183 res <- step gst st
184 case res of
185 Yield x s -> return $ Yield x (Just s)
186 Skip s -> return $ Skip (Just s)
187 Stop -> return Stop
189-- | See 'Streamly.Internal.Data.Stream.IsStream.after_'.
190--
191{-# INLINE_NORMAL after_ #-}
192after_ :: Monad m => m b -> Stream m a -> Stream m a
193after_ action (Stream step state) = Stream step' state
195 where
197 {-# INLINE_LATE step' #-}
198 step' gst st = do
199 res <- step gst st
200 case res of
201 Yield x s -> return $ Yield x s
202 Skip s -> return $ Skip s
203 Stop -> action >> return Stop
205-- | See 'Streamly.Internal.Data.Stream.IsStream.after'.
206--
207{-# INLINE_NORMAL after #-}
208after :: (MonadIO m, MonadBaseControl IO m)
209 => m b -> Stream m a -> Stream m a
210after action (Stream step state) = Stream step' Nothing
212 where
214 {-# INLINE_LATE step' #-}
215 step' _ Nothing = do
216 ref <- newIOFinalizer action
217 return $ Skip $ Just (state, ref)
218 step' gst (Just (st, ref)) = do
219 res <- step gst st
220 case res of
221 Yield x s -> return $ Yield x (Just (s, ref))
222 Skip s -> return $ Skip (Just (s, ref))
223 Stop -> do
224 runIOFinalizer ref
225 return Stop
227-- XXX For high performance error checks in busy streams we may need another
228-- Error constructor in step.
229--
230-- | See 'Streamly.Internal.Data.Stream.IsStream.onException'.
231--
232{-# INLINE_NORMAL onException #-}
233onException :: MonadCatch m => m b -> Stream m a -> Stream m a
234onException action str =
235 gbracket_ (return ()) (inline MC.try) return
236 (\_ (e :: MC.SomeException) _ -> nilM (action >> MC.throwM e))
237 (\_ -> str)
239{-# INLINE_NORMAL _onException #-}
240_onException :: MonadCatch m => m b -> Stream m a -> Stream m a
241_onException action (Stream step state) = Stream step' state
243 where
245 {-# INLINE_LATE step' #-}
246 step' gst st = do
247 res <- step gst st `MC.onException` action
248 case res of
249 Yield x s -> return $ Yield x s
250 Skip s -> return $ Skip s
251 Stop -> return Stop
253-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket_'.
254--
255{-# INLINE_NORMAL bracket_ #-}
256bracket_ :: MonadCatch m
257 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
258bracket_ bef aft bet =
259 gbracket_ bef (inline MC.try) aft
260 (\a (e :: SomeException) _ -> nilM (aft a >> MC.throwM e)) bet
262-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket'.
263--
264{-# INLINE_NORMAL bracket #-}
265bracket :: (MonadAsync m, MonadCatch m)
266 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
267bracket bef aft bet =
268 gbracket bef (inline MC.try) aft
269 (\a (e :: SomeException) _ -> aft a >> return (nilM (MC.throwM e))) bet
271data BracketState s v = BracketInit | BracketRun s v
273-- | Alternate (custom) implementation of 'bracket'.
274--
275{-# INLINE_NORMAL _bracket #-}
276_bracket :: MonadCatch m
277 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
278_bracket bef aft bet = Stream step' BracketInit
280 where
282 {-# INLINE_LATE step' #-}
283 step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))
285 -- NOTE: It is important to use UnStream instead of the Stream pattern
286 -- here, otherwise we get huge perf degradation, see note in concatMap.
287 step' gst (BracketRun (UnStream step state) v) = do
288 -- res <- step gst state `MC.onException` aft v
289 res <- inline MC.try $ step gst state
290 case res of
291 Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop
292 Right r -> case r of
293 Yield x s -> return $ Yield x (BracketRun (Stream step s) v)
294 Skip s -> return $ Skip (BracketRun (Stream step s) v)
295 Stop -> aft v >> return Stop
297-- | See 'Streamly.Internal.Data.Stream.IsStream.finally_'.
298--
299{-# INLINE finally_ #-}
300finally_ :: MonadCatch m => m b -> Stream m a -> Stream m a
301finally_ action xs = bracket_ (return ()) (\_ -> action) (const xs)
303-- | See 'Streamly.Internal.Data.Stream.IsStream.finally'.
304--
305-- finally action xs = after action $ onException action xs
306--
307{-# INLINE finally #-}
308finally :: (MonadAsync m, MonadCatch m) => m b -> Stream m a -> Stream m a
309finally action xs = bracket (return ()) (\_ -> action) (const xs)
311-- | See 'Streamly.Internal.Data.Stream.IsStream.ghandle'.
312--
313{-# INLINE_NORMAL ghandle #-}
314ghandle :: (MonadCatch m, Exception e)
315 => (e -> Stream m a -> Stream m a) -> Stream m a -> Stream m a
316ghandle f str =
317 gbracket_ (return ()) (inline MC.try) return (\_ -> f) (\_ -> str)
319-- | See 'Streamly.Internal.Data.Stream.IsStream.handle'.
320--
321{-# INLINE_NORMAL handle #-}
322handle :: (MonadCatch m, Exception e)
323 => (e -> Stream m a) -> Stream m a -> Stream m a
324handle f str =
325 gbracket_ (return ()) (inline MC.try) return (\_ e _ -> f e) (\_ -> str)
327-- | Alternate (custom) implementation of 'handle'.
328--
329{-# INLINE_NORMAL _handle #-}
330_handle :: (MonadCatch m, Exception e)
331 => (e -> Stream m a) -> Stream m a -> Stream m a
332_handle f (Stream step state) = Stream step' (Left state)
334 where
336 {-# INLINE_LATE step' #-}
337 step' gst (Left st) = do
338 res <- inline MC.try $ step gst st
339 case res of
340 Left e -> return $ Skip $ Right (f e)
341 Right r -> case r of
342 Yield x s -> return $ Yield x (Left s)
343 Skip s -> return $ Skip (Left s)
344 Stop -> return Stop
346 step' gst (Right (UnStream step1 st)) = do
347 res <- step1 gst st
348 case res of
349 Yield x s -> return $ Yield x (Right (Stream step1 s))
350 Skip s -> return $ Skip (Right (Stream step1 s))
351 Stop -> return Stop