git.iko.soy/streamly
on e95e149 help sign in
streamly/src/Streamly/Internal/Data/Stream/StreamD/Exception.hs
350 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 Streamly.Internal.Data.IOFinalizer
33 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)
35import qualified Control.Monad.Catch as MC
37import Streamly.Internal.Data.Stream.StreamD.Type
38import Streamly.Internal.Data.SVar
40data GbracketState s1 s2 v
41 = GBracketInit
42 | GBracketNormal s1 v
43 | GBracketException s2
45-- | Like 'gbracket' but with following differences:
46--
47-- * alloc action @m c@ runs with async exceptions enabled
48-- * cleanup action @c -> m d@ won't run if the stream is garbage collected
49-- after partial evaluation.
50-- * does not require a 'MonadAsync' constraint.
51--
52-- /Inhibits stream fusion/
53--
54-- /Pre-release/
55--
56{-# INLINE_NORMAL gbracket_ #-}
57gbracket_
58 :: Monad m
59 => m c -- ^ before
60 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
61 -> (c -> m d) -- ^ after, on normal stop
62 -> (c -> e -> Stream m b -> Stream m b) -- ^ on exception
63 -> (c -> Stream m b) -- ^ stream generator
64 -> Stream m b
65gbracket_ bef exc aft fexc fnormal =
66 Stream step GBracketInit
68 where
70 {-# INLINE_LATE step #-}
71 step _ GBracketInit = do
72 r <- bef
73 return $ Skip $ GBracketNormal (fnormal r) r
75 step gst (GBracketNormal (UnStream step1 st) v) = do
76 res <- exc $ step1 gst st
77 case res of
78 Right r -> case r of
79 Yield x s ->
80 return $ Yield x (GBracketNormal (Stream step1 s) v)
81 Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)
82 Stop -> aft v >> return Stop
83 -- XXX Do not handle async exceptions, just rethrow them.
84 Left e ->
85 return $ Skip (GBracketException (fexc v e (UnStream step1 st)))
86 step gst (GBracketException (UnStream step1 st)) = do
87 res <- step1 gst st
88 case res of
89 Yield x s -> return $ Yield x (GBracketException (Stream step1 s))
90 Skip s -> return $ Skip (GBracketException (Stream step1 s))
91 Stop -> return Stop
93data GbracketIOState s1 s2 v wref
94 = GBracketIOInit
95 | GBracketIONormal s1 v wref
96 | GBracketIOException s2
98-- | Run the alloc action @m c@ with async exceptions disabled but keeping
99-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
100-- output @c@ as input to @c -> Stream m b@ to generate an output stream. When
101-- generating the stream use the supplied @try@ operation @forall s. m s -> m
102-- (Either e s)@ to catch synchronous exceptions. If an exception occurs run
103-- the exception handler @c -> e -> Stream m b -> m (Stream m b)@.
104--
105-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due
106-- to a sync or async exception or if it gets garbage collected after a partial
107-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.
108--
109-- 'gbracket' can express all other exception handling combinators.
110--
111-- /Inhibits stream fusion/
112--
113-- /Pre-release/
114{-# INLINE_NORMAL gbracket #-}
115gbracket
116 :: (MonadIO m, MonadBaseControl IO m)
117 => m c -- ^ before
118 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
119 -> (c -> m d) -- ^ after, on normal stop or GC
120 -> (c -> e -> Stream m b -> m (Stream m b)) -- ^ on exception
121 -> (c -> Stream m b) -- ^ stream generator
122 -> Stream m b
123gbracket bef exc aft fexc fnormal =
124 Stream step GBracketIOInit
126 where
128 -- If the stream is never evaluated the "aft" action will never be
129 -- called. For that to occur we will need the user of this API to pass a
130 -- weak pointer to us.
131 {-# INLINE_LATE step #-}
132 step _ GBracketIOInit = do
133 -- We mask asynchronous exceptions to make the execution
134 -- of 'bef' and the registration of 'aft' atomic.
135 -- A similar thing is done in the resourcet package: https://git.io/JvKV3
136 -- Tutorial: https://markkarpov.com/tutorial/exceptions.html
137 (r, ref) <- liftBaseOp_ mask_ $ do
138 r <- bef
139 ref <- newIOFinalizer (aft r)
140 return (r, ref)
141 return $ Skip $ GBracketIONormal (fnormal r) r ref
143 step gst (GBracketIONormal (UnStream step1 st) v ref) = do
144 res <- exc $ step1 gst st
145 case res of
146 Right r -> case r of
147 Yield x s ->
148 return $ Yield x (GBracketIONormal (Stream step1 s) v ref)
149 Skip s ->
150 return $ Skip (GBracketIONormal (Stream step1 s) v ref)
151 Stop -> do
152 runIOFinalizer ref
153 return Stop
154 -- XXX Do not handle async exceptions, just rethrow them.
155 Left e -> do
156 -- Clearing of finalizer and running of exception handler must
157 -- be atomic wrt async exceptions. Otherwise if we have cleared
158 -- the finalizer and have not run the exception handler then we
159 -- may leak the resource.
160 stream <- clearingIOFinalizer ref (fexc v e (UnStream step1 st))
161 return $ Skip (GBracketIOException stream)
162 step gst (GBracketIOException (UnStream step1 st)) = do
163 res <- step1 gst st
164 case res of
165 Yield x s ->
166 return $ Yield x (GBracketIOException (Stream step1 s))
167 Skip s -> return $ Skip (GBracketIOException (Stream step1 s))
168 Stop -> return Stop
170-- | See 'Streamly.Internal.Data.Stream.IsStream.before'.
171--
172{-# INLINE_NORMAL before #-}
173before :: Monad m => m b -> Stream m a -> Stream m a
174before action (Stream step state) = Stream step' Nothing
176 where
178 {-# INLINE_LATE step' #-}
179 step' _ Nothing = action >> return (Skip (Just state))
181 step' gst (Just st) = do
182 res <- step gst st
183 case res of
184 Yield x s -> return $ Yield x (Just s)
185 Skip s -> return $ Skip (Just s)
186 Stop -> return Stop
188-- | See 'Streamly.Internal.Data.Stream.IsStream.after_'.
189--
190{-# INLINE_NORMAL after_ #-}
191after_ :: Monad m => m b -> Stream m a -> Stream m a
192after_ action (Stream step state) = Stream step' state
194 where
196 {-# INLINE_LATE step' #-}
197 step' gst st = do
198 res <- step gst st
199 case res of
200 Yield x s -> return $ Yield x s
201 Skip s -> return $ Skip s
202 Stop -> action >> return Stop
204-- | See 'Streamly.Internal.Data.Stream.IsStream.after'.
205--
206{-# INLINE_NORMAL after #-}
207after :: (MonadIO m, MonadBaseControl IO m)
208 => m b -> Stream m a -> Stream m a
209after action (Stream step state) = Stream step' Nothing
211 where
213 {-# INLINE_LATE step' #-}
214 step' _ Nothing = do
215 ref <- newIOFinalizer action
216 return $ Skip $ Just (state, ref)
217 step' gst (Just (st, ref)) = do
218 res <- step gst st
219 case res of
220 Yield x s -> return $ Yield x (Just (s, ref))
221 Skip s -> return $ Skip (Just (s, ref))
222 Stop -> do
223 runIOFinalizer ref
224 return Stop
226-- XXX For high performance error checks in busy streams we may need another
227-- Error constructor in step.
228--
229-- | See 'Streamly.Internal.Data.Stream.IsStream.onException'.
230--
231{-# INLINE_NORMAL onException #-}
232onException :: MonadCatch m => m b -> Stream m a -> Stream m a
233onException action str =
234 gbracket_ (return ()) MC.try return
235 (\_ (e :: MC.SomeException) _ -> nilM (action >> MC.throwM e))
236 (\_ -> str)
238{-# INLINE_NORMAL _onException #-}
239_onException :: MonadCatch m => m b -> Stream m a -> Stream m a
240_onException action (Stream step state) = Stream step' state
242 where
244 {-# INLINE_LATE step' #-}
245 step' gst st = do
246 res <- step gst st `MC.onException` action
247 case res of
248 Yield x s -> return $ Yield x s
249 Skip s -> return $ Skip s
250 Stop -> return Stop
252-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket_'.
253--
254{-# INLINE_NORMAL bracket_ #-}
255bracket_ :: MonadCatch m
256 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
257bracket_ bef aft bet =
258 gbracket_ bef MC.try aft
259 (\a (e :: SomeException) _ -> nilM (aft a >> MC.throwM e)) bet
261-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket'.
262--
263{-# INLINE_NORMAL bracket #-}
264bracket :: (MonadAsync m, MonadCatch m)
265 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
266bracket bef aft bet =
267 gbracket bef MC.try aft
268 (\a (e :: SomeException) _ -> aft a >> return (nilM (MC.throwM e))) bet
270data BracketState s v = BracketInit | BracketRun s v
272-- | Alternate (custom) implementation of 'bracket'.
273--
274{-# INLINE_NORMAL _bracket #-}
275_bracket :: MonadCatch m
276 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
277_bracket bef aft bet = Stream step' BracketInit
279 where
281 {-# INLINE_LATE step' #-}
282 step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))
284 -- NOTE: It is important to use UnStream instead of the Stream pattern
285 -- here, otherwise we get huge perf degradation, see note in concatMap.
286 step' gst (BracketRun (UnStream step state) v) = do
287 -- res <- step gst state `MC.onException` aft v
288 res <- MC.try $ step gst state
289 case res of
290 Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop
291 Right r -> case r of
292 Yield x s -> return $ Yield x (BracketRun (Stream step s) v)
293 Skip s -> return $ Skip (BracketRun (Stream step s) v)
294 Stop -> aft v >> return Stop
296-- | See 'Streamly.Internal.Data.Stream.IsStream.finally_'.
297--
298{-# INLINE finally_ #-}
299finally_ :: MonadCatch m => m b -> Stream m a -> Stream m a
300finally_ action xs = bracket_ (return ()) (\_ -> action) (const xs)
302-- | See 'Streamly.Internal.Data.Stream.IsStream.finally'.
303--
304-- finally action xs = after action $ onException action xs
305--
306{-# INLINE finally #-}
307finally :: (MonadAsync m, MonadCatch m) => m b -> Stream m a -> Stream m a
308finally action xs = bracket (return ()) (\_ -> action) (const xs)
310-- | See 'Streamly.Internal.Data.Stream.IsStream.ghandle'.
311--
312{-# INLINE_NORMAL ghandle #-}
313ghandle :: (MonadCatch m, Exception e)
314 => (e -> Stream m a -> Stream m a) -> Stream m a -> Stream m a
315ghandle f str =
316 gbracket_ (return ()) MC.try return (\_ -> f) (\_ -> str)
318-- | See 'Streamly.Internal.Data.Stream.IsStream.handle'.
319--
320{-# INLINE_NORMAL handle #-}
321handle :: (MonadCatch m, Exception e)
322 => (e -> Stream m a) -> Stream m a -> Stream m a
323handle f str =
324 gbracket_ (return ()) MC.try return (\_ e _ -> f e) (\_ -> str)
326-- | Alternate (custom) implementation of 'handle'.
327--
328{-# INLINE_NORMAL _handle #-}
329_handle :: (MonadCatch m, Exception e)
330 => (e -> Stream m a) -> Stream m a -> Stream m a
331_handle f (Stream step state) = Stream step' (Left state)
333 where
335 {-# INLINE_LATE step' #-}
336 step' gst (Left st) = do
337 res <- MC.try $ step gst st
338 case res of
339 Left e -> return $ Skip $ Right (f e)
340 Right r -> case r of
341 Yield x s -> return $ Yield x (Left s)
342 Skip s -> return $ Skip (Left s)
343 Stop -> return Stop
345 step' gst (Right (UnStream step1 st)) = do
346 res <- step1 gst st
347 case res of
348 Yield x s -> return $ Yield x (Right (Stream step1 s))
349 Skip s -> return $ Skip (Right (Stream step1 s))
350 Stop -> return Stop