on 32785ad help sign in
480 lines 17 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 , afterUnsafe
15 , afterIO
16 , bracketUnsafe
17 , bracketIO3
18 , bracketIO
19 , onException
20 , finallyUnsafe
21 , finallyIO
22 , ghandle
23 , handle
24 )
25where
27#include "inline.hs"
29import Control.Monad.IO.Class (MonadIO(..))
30import Control.Exception (Exception, SomeException, mask_)
31import Control.Monad.Catch (MonadCatch)
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
40-- $setup
41-- >>> :m
42-- >>> import qualified Streamly.Internal.Data.Stream as Stream
44data GbracketState s1 s2 v
45 = GBracketInit
46 | GBracketNormal s1 v
47 | GBracketException s2
49-- | Like 'gbracket' but with following differences:
50--
51-- * alloc action @m c@ runs with async exceptions enabled
52-- * cleanup action @c -> m d@ won't run if the stream is garbage collected
53-- after partial evaluation.
54--
55-- /Inhibits stream fusion/
56--
57-- /Pre-release/
58--
59{-# INLINE_NORMAL gbracket_ #-}
60gbracket_
61 :: Monad m
62 => m c -- ^ before
63 -> (c -> m d) -- ^ after, on normal stop
64 -> (c -> e -> Stream m b -> Stream m b) -- ^ on exception
65 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
66 -> (c -> Stream m b) -- ^ stream generator
67 -> Stream m b
68gbracket_ bef aft onExc ftry action =
69 Stream step GBracketInit
71 where
73 {-# INLINE_LATE step #-}
74 step _ GBracketInit = do
75 r <- bef
76 return $ Skip $ GBracketNormal (action r) r
78 step gst (GBracketNormal (UnStream step1 st) v) = do
79 res <- ftry $ step1 gst st
80 case res of
81 Right r -> case r of
82 Yield x s ->
83 return $ Yield x (GBracketNormal (Stream step1 s) v)
84 Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)
85 Stop -> aft v >> return Stop
86 -- XXX Do not handle async exceptions, just rethrow them.
87 Left e ->
88 return
89 $ Skip (GBracketException (onExc v e (UnStream step1 st)))
90 step gst (GBracketException (UnStream step1 st)) = do
91 res <- step1 gst st
92 case res of
93 Yield x s -> return $ Yield x (GBracketException (Stream step1 s))
94 Skip s -> return $ Skip (GBracketException (Stream step1 s))
95 Stop -> return Stop
97data GbracketIOState s1 s2 v wref
98 = GBracketIOInit
99 | GBracketIONormal s1 v wref
100 | GBracketIOException s2
102-- | Run the alloc action @m c@ with async exceptions disabled but keeping
103-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
104-- output @c@ as input to @c -> Stream m b@ to generate an output stream. When
105-- generating the stream use the supplied @try@ operation @forall s. m s -> m
106-- (Either e s)@ to catch synchronous exceptions. If an exception occurs run
107-- the exception handler @c -> e -> Stream m b -> m (Stream m b)@. Note that
108-- 'gbracket' does not rethrow the exception, it has to be done by the
109-- exception handler if desired.
110--
111-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due
112-- to a sync or async exception or if it gets garbage collected after a partial
113-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.
114--
115-- 'gbracket' can express all other exception handling combinators.
116--
117-- /Inhibits stream fusion/
118--
119-- /Pre-release/
120{-# INLINE_NORMAL gbracket #-}
121gbracket
122 :: MonadIO m
123 => IO c -- ^ before
124 -> (c -> IO d1) -- ^ on normal stop
125 -> (c -> e -> Stream m b -> IO (Stream m b)) -- ^ on exception
126 -> (c -> IO d2) -- ^ on GC without normal stop or exception
127 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
128 -> (c -> Stream m b) -- ^ stream generator
129 -> Stream m b
130gbracket bef aft onExc onGC ftry action =
131 Stream step GBracketIOInit
133 where
135 -- If the stream is never evaluated the "aft" action will never be
136 -- called. For that to occur we will need the user of this API to pass a
137 -- weak pointer to us.
138 {-# INLINE_LATE step #-}
139 step _ GBracketIOInit = do
140 -- We mask asynchronous exceptions to make the execution
141 -- of 'bef' and the registration of 'aft' atomic.
142 -- A similar thing is done in the resourcet package: https://git.io/JvKV3
143 -- Tutorial: https://markkarpov.com/tutorial/exceptions.html
144 (r, ref) <- liftIO $ mask_ $ do
145 r <- bef
146 ref <- newIOFinalizer (onGC r)
147 return (r, ref)
148 return $ Skip $ GBracketIONormal (action r) r ref
150 step gst (GBracketIONormal (UnStream step1 st) v ref) = do
151 res <- ftry $ step1 gst st
152 case res of
153 Right r -> case r of
154 Yield x s ->
155 return $ Yield x (GBracketIONormal (Stream step1 s) v ref)
156 Skip s ->
157 return $ Skip (GBracketIONormal (Stream step1 s) v ref)
158 Stop ->
159 liftIO (clearingIOFinalizer ref (aft v)) >> return Stop
160 -- XXX Do not handle async exceptions, just rethrow them.
161 Left e -> do
162 -- Clearing of finalizer and running of exception handler must
163 -- be atomic wrt async exceptions. Otherwise if we have cleared
164 -- the finalizer and have not run the exception handler then we
165 -- may leak the resource.
166 stream <-
167 liftIO (clearingIOFinalizer ref (onExc v e (UnStream step1 st)))
168 return $ Skip (GBracketIOException stream)
169 step gst (GBracketIOException (UnStream step1 st)) = do
170 res <- step1 gst st
171 case res of
172 Yield x s ->
173 return $ Yield x (GBracketIOException (Stream step1 s))
174 Skip s -> return $ Skip (GBracketIOException (Stream step1 s))
175 Stop -> return Stop
177-- | Run the action @m b@ before the stream yields its first element.
178--
179-- Same as the following but more efficient due to fusion:
180--
181-- >>> before action xs = Stream.nilM action <> xs
182-- >>> before action xs = Stream.concatMap (const xs) (Stream.fromEffect action)
183--
184{-# INLINE_NORMAL before #-}
185before :: Monad m => m b -> Stream m a -> Stream m a
186before action (Stream step state) = Stream step' Nothing
188 where
190 {-# INLINE_LATE step' #-}
191 step' _ Nothing = action >> return (Skip (Just state))
193 step' gst (Just st) = do
194 res <- step gst st
195 case res of
196 Yield x s -> return $ Yield x (Just s)
197 Skip s -> return $ Skip (Just s)
198 Stop -> return Stop
200-- | Like 'after', with following differences:
201--
202-- * action @m b@ won't run if the stream is garbage collected
203-- after partial evaluation.
204-- * Monad @m@ does not require any other constraints.
205-- * has slightly better performance than 'after'.
206--
207-- Same as the following, but with stream fusion:
208--
209-- >>> afterUnsafe action xs = xs <> Stream.nilM action
210--
211-- /Pre-release/
212--
213{-# INLINE_NORMAL afterUnsafe #-}
214afterUnsafe :: Monad m => m b -> Stream m a -> Stream m a
215afterUnsafe action (Stream step state) = Stream step' state
217 where
219 {-# INLINE_LATE step' #-}
220 step' gst st = do
221 res <- step gst st
222 case res of
223 Yield x s -> return $ Yield x s
224 Skip s -> return $ Skip s
225 Stop -> action >> return Stop
227-- | Run the action @IO b@ whenever the stream is evaluated to completion, or
228-- if it is garbage collected after a partial lazy evaluation.
229--
230-- The semantics of the action @IO b@ are similar to the semantics of cleanup
231-- action in 'bracketIO'.
232--
233-- /See also 'afterUnsafe'/
234--
235{-# INLINE_NORMAL afterIO #-}
236afterIO :: MonadIO m
237 => IO b -> Stream m a -> Stream m a
238afterIO action (Stream step state) = Stream step' Nothing
240 where
242 {-# INLINE_LATE step' #-}
243 step' _ Nothing = do
244 ref <- liftIO $ newIOFinalizer action
245 return $ Skip $ Just (state, ref)
246 step' gst (Just (st, ref)) = do
247 res <- step gst st
248 case res of
249 Yield x s -> return $ Yield x (Just (s, ref))
250 Skip s -> return $ Skip (Just (s, ref))
251 Stop -> do
252 runIOFinalizer ref
253 return Stop
255-- XXX For high performance error checks in busy streams we may need another
256-- Error constructor in step.
258-- | Run the action @m b@ if the stream evaluation is aborted due to an
259-- exception. The exception is not caught, simply rethrown.
260--
261-- /Inhibits stream fusion/
262--
263{-# INLINE_NORMAL onException #-}
264onException :: MonadCatch m => m b -> Stream m a -> Stream m a
265onException action stream =
266 gbracket_
267 (return ()) -- before
268 return -- after
269 (\_ (e :: MC.SomeException) _ -> nilM (action >> MC.throwM e))
270 (inline MC.try)
271 (const stream)
273{-# INLINE_NORMAL _onException #-}
274_onException :: MonadCatch m => m b -> Stream m a -> Stream m a
275_onException action (Stream step state) = Stream step' state
277 where
279 {-# INLINE_LATE step' #-}
280 step' gst st = do
281 res <- step gst st `MC.onException` action
282 case res of
283 Yield x s -> return $ Yield x s
284 Skip s -> return $ Skip s
285 Stop -> return Stop
287-- | Like 'bracket' but with following differences:
288--
289-- * alloc action @m b@ runs with async exceptions enabled
290-- * cleanup action @b -> m c@ won't run if the stream is garbage collected
291-- after partial evaluation.
292-- * has slightly better performance than 'bracketIO'.
293--
294-- /Inhibits stream fusion/
295--
296-- /Pre-release/
297--
298{-# INLINE_NORMAL bracketUnsafe #-}
299bracketUnsafe :: MonadCatch m
300 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
301bracketUnsafe bef aft =
302 gbracket_
303 bef
304 aft
305 (\a (e :: SomeException) _ -> nilM (aft a >> MC.throwM e))
306 (inline MC.try)
308-- For a use case of this see the "streamly-process" package. It needs to kill
309-- the process in case of exception or garbage collection, but waits for the
310-- process to terminate in normal cases.
312-- | Like 'bracketIO' but can use 3 separate cleanup actions depending on the
313-- mode of termination:
314--
315-- 1. When the stream stops normally
316-- 2. When the stream is garbage collected
317-- 3. When the stream encounters an exception
318--
319-- @bracketIO3 before onStop onGC onException action@ runs @action@ using the
320-- result of @before@. If the stream stops, @onStop@ action is executed, if the
321-- stream is abandoned @onGC@ is executed, if the stream encounters an
322-- exception @onException@ is executed.
323--
324-- /Inhibits stream fusion/
325--
326-- /Pre-release/
327{-# INLINE_NORMAL bracketIO3 #-}
328bracketIO3 :: (MonadIO m, MonadCatch m) =>
329 IO b
330 -> (b -> IO c)
331 -> (b -> IO d)
332 -> (b -> IO e)
333 -> (b -> Stream m a)
334 -> Stream m a
335bracketIO3 bef aft onExc onGC =
336 gbracket
337 bef
338 aft
339 (\a (e :: SomeException) _ -> onExc a >> return (nilM (MC.throwM e)))
340 onGC
341 (inline MC.try)
343-- | Run the alloc action @IO b@ with async exceptions disabled but keeping
344-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
345-- output @b@ as input to @b -> Stream m a@ to generate an output stream.
346--
347-- @b@ is usually a resource under the IO monad, e.g. a file handle, that
348-- requires a cleanup after use. The cleanup action @b -> IO c@, runs whenever
349-- the stream ends normally, due to a sync or async exception or if it gets
350-- garbage collected after a partial lazy evaluation.
351--
352-- 'bracketIO' only guarantees that the cleanup action runs, and it runs with
353-- async exceptions enabled. The action must ensure that it can successfully
354-- cleanup the resource in the face of sync or async exceptions.
355--
356-- When the stream ends normally or on a sync exception, cleanup action runs
357-- immediately in the current thread context, whereas in other cases it runs in
358-- the GC context, therefore, cleanup may be delayed until the GC gets to run.
359--
360-- /See also: 'bracketUnsafe'/
361--
362-- /Inhibits stream fusion/
363--
364{-# INLINE bracketIO #-}
365bracketIO :: (MonadIO m, MonadCatch m)
366 => IO b -> (b -> IO c) -> (b -> Stream m a) -> Stream m a
367bracketIO bef aft = bracketIO3 bef aft aft aft
369data BracketState s v = BracketInit | BracketRun s v
371-- | Alternate (custom) implementation of 'bracket'.
372--
373{-# INLINE_NORMAL _bracket #-}
374_bracket :: MonadCatch m
375 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
376_bracket bef aft bet = Stream step' BracketInit
378 where
380 {-# INLINE_LATE step' #-}
381 step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))
383 -- NOTE: It is important to use UnStream instead of the Stream pattern
384 -- here, otherwise we get huge perf degradation, see note in concatMap.
385 step' gst (BracketRun (UnStream step state) v) = do
386 -- res <- step gst state `MC.onException` aft v
387 res <- inline MC.try $ step gst state
388 case res of
389 Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop
390 Right r -> case r of
391 Yield x s -> return $ Yield x (BracketRun (Stream step s) v)
392 Skip s -> return $ Skip (BracketRun (Stream step s) v)
393 Stop -> aft v >> return Stop
395-- | Like 'finally' with following differences:
396--
397-- * action @m b@ won't run if the stream is garbage collected
398-- after partial evaluation.
399-- * has slightly better performance than 'finallyIO'.
400--
401-- /Inhibits stream fusion/
402--
403-- /Pre-release/
404--
405{-# INLINE finallyUnsafe #-}
406finallyUnsafe :: MonadCatch m => m b -> Stream m a -> Stream m a
407finallyUnsafe action xs = bracketUnsafe (return ()) (const action) (const xs)
409-- | Run the action @IO b@ whenever the stream stream stops normally, aborts
410-- due to an exception or if it is garbage collected after a partial lazy
411-- evaluation.
412--
413-- The semantics of running the action @IO b@ are similar to the cleanup action
414-- semantics described in 'bracketIO'.
415--
416-- >>> finallyIO release = Stream.bracketIO (return ()) (const release)
417--
418-- /See also 'finallyUnsafe'/
419--
420-- /Inhibits stream fusion/
421--
422{-# INLINE finallyIO #-}
423finallyIO :: (MonadIO m, MonadCatch m) => IO b -> Stream m a -> Stream m a
424finallyIO action xs = bracketIO3 (return ()) act act act (const xs)
425 where act _ = action
427-- | Like 'handle' but the exception handler is also provided with the stream
428-- that generated the exception as input. The exception handler can thus
429-- re-evaluate the stream to retry the action that failed. The exception
430-- handler can again call 'ghandle' on it to retry the action multiple times.
431--
432-- This is highly experimental. In a stream of actions we can map the stream
433-- with a retry combinator to retry each action on failure.
434--
435-- /Inhibits stream fusion/
436--
437-- /Pre-release/
438--
439{-# INLINE_NORMAL ghandle #-}
440ghandle :: (MonadCatch m, Exception e)
441 => (e -> Stream m a -> Stream m a) -> Stream m a -> Stream m a
442ghandle f stream =
443 gbracket_ (return ()) return (const f) (inline MC.try) (const stream)
445-- | When evaluating a stream if an exception occurs, stream evaluation aborts
446-- and the specified exception handler is run with the exception as argument.
447--
448-- /Inhibits stream fusion/
449--
450{-# INLINE_NORMAL handle #-}
451handle :: (MonadCatch m, Exception e)
452 => (e -> Stream m a) -> Stream m a -> Stream m a
453handle f stream =
454 gbracket_ (return ()) return (\_ e _ -> f e) (inline MC.try) (const stream)
456-- | Alternate (custom) implementation of 'handle'.
457--
458{-# INLINE_NORMAL _handle #-}
459_handle :: (MonadCatch m, Exception e)
460 => (e -> Stream m a) -> Stream m a -> Stream m a
461_handle f (Stream step state) = Stream step' (Left state)
463 where
465 {-# INLINE_LATE step' #-}
466 step' gst (Left st) = do
467 res <- inline MC.try $ step gst st
468 case res of
469 Left e -> return $ Skip $ Right (f e)
470 Right r -> case r of
471 Yield x s -> return $ Yield x (Left s)
472 Skip s -> return $ Skip (Left s)
473 Stop -> return Stop
475 step' gst (Right (UnStream step1 st)) = do
476 res <- step1 gst st
477 case res of
478 Yield x s -> return $ Yield x (Right (Stream step1 s))
479 Skip s -> return $ Skip (Right (Stream step1 s))
480 Stop -> return Stop