3-- Module : Streamly.Internal.Data.Stream.StreamD.Exception
4-- Copyright : (c) 2020 Composewell Technologies and Contributors
5-- License : BSD-3-Clause
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
10module Streamly.Internal.Data.Stream.StreamD.Exception
30import Control.Monad.IO.Class (MonadIO(..))
31import Control.Exception (Exception, SomeException, mask_)
32import Control.Monad.Catch (MonadCatch)
33import GHC.Exts (inline)
34import Streamly.Internal.Data.IOFinalizer
35 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)
37import qualified Control.Monad.Catch as MC
39import Streamly.Internal.Data.Stream.StreamD.Type
41#include "DocTestDataStream.hs"
43data GbracketState s1 s2 v
46 | GBracketException s2
48-- | Like 'gbracket' but with following differences:
50-- * alloc action @m c@ runs with async exceptions enabled
51-- * cleanup action @c -> m d@ won't run if the stream is garbage collected
52-- after partial evaluation.
54-- /Inhibits stream fusion/
58{-# INLINE_NORMAL gbracket_ #-}
62 -> (c -> m d) -- ^ after, on normal stop
63 -> (c -> e -> Stream m b -> Stream m b) -- ^ on exception
64 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
65 -> (c -> Stream m b) -- ^ stream generator
67gbracket_ bef aft onExc ftry action =
68 Stream step GBracketInit
72 {-# INLINE_LATE step #-}
73 step _ GBracketInit = do
75 return $ Skip $ GBracketNormal (action r) r
77 step gst (GBracketNormal (UnStream step1 st) v) = do
78 res <- ftry $ step1 gst st
82 return $ Yield x (GBracketNormal (Stream step1 s) v)
83 Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)
84 Stop -> aft v >> return Stop
85 -- XXX Do not handle async exceptions, just rethrow them.
88 $ Skip (GBracketException (onExc v e (UnStream step1 st)))
89 step gst (GBracketException (UnStream step1 st)) = do
92 Yield x s -> return $ Yield x (GBracketException (Stream step1 s))
93 Skip s -> return $ Skip (GBracketException (Stream step1 s))
96data GbracketIOState s1 s2 v wref
98 | GBracketIONormal s1 v wref
99 | GBracketIOException s2
101-- | Run the alloc action @m c@ with async exceptions disabled but keeping
102-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
103-- output @c@ as input to @c -> Stream m b@ to generate an output stream. When
104-- generating the stream use the supplied @try@ operation @forall s. m s -> m
105-- (Either e s)@ to catch synchronous exceptions. If an exception occurs run
106-- the exception handler @c -> e -> Stream m b -> m (Stream m b)@. Note that
107-- 'gbracket' does not rethrow the exception, it has to be done by the
108-- exception handler if desired.
110-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due
111-- to a sync or async exception or if it gets garbage collected after a partial
112-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.
114-- 'gbracket' can express all other exception handling combinators.
116-- /Inhibits stream fusion/
119{-# INLINE_NORMAL gbracket #-}
123 -> (c -> IO d1) -- ^ on normal stop
124 -> (c -> e -> Stream m b -> IO (Stream m b)) -- ^ on exception
125 -> (c -> IO d2) -- ^ on GC without normal stop or exception
126 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
127 -> (c -> Stream m b) -- ^ stream generator
129gbracket bef aft onExc onGC ftry action =
130 Stream step GBracketIOInit
134 -- If the stream is never evaluated the "aft" action will never be
135 -- called. For that to occur we will need the user of this API to pass a
136 -- weak pointer to us.
137 {-# INLINE_LATE step #-}
138 step _ GBracketIOInit = do
139 -- We mask asynchronous exceptions to make the execution
140 -- of 'bef' and the registration of 'aft' atomic.
141 -- A similar thing is done in the resourcet package: https://git.io/JvKV3
142 -- Tutorial: https://markkarpov.com/tutorial/exceptions.html
143 (r, ref) <- liftIO $ mask_ $ do
145 ref <- newIOFinalizer (onGC r)
147 return $ Skip $ GBracketIONormal (action r) r ref
149 step gst (GBracketIONormal (UnStream step1 st) v ref) = do
150 res <- ftry $ step1 gst st
154 return $ Yield x (GBracketIONormal (Stream step1 s) v ref)
156 return $ Skip (GBracketIONormal (Stream step1 s) v ref)
158 liftIO (clearingIOFinalizer ref (aft v)) >> return Stop
159 -- XXX Do not handle async exceptions, just rethrow them.
161 -- Clearing of finalizer and running of exception handler must
162 -- be atomic wrt async exceptions. Otherwise if we have cleared
163 -- the finalizer and have not run the exception handler then we
164 -- may leak the resource.
166 liftIO (clearingIOFinalizer ref (onExc v e (UnStream step1 st)))
167 return $ Skip (GBracketIOException stream)
168 step gst (GBracketIOException (UnStream step1 st)) = do
172 return $ Yield x (GBracketIOException (Stream step1 s))
173 Skip s -> return $ Skip (GBracketIOException (Stream step1 s))
176-- | Run the action @m b@ before the stream yields its first element.
178-- Same as the following but more efficient due to fusion:
180-- >>> before action xs = Stream.nilM action <> xs
181-- >>> before action xs = Stream.concatMap (const xs) (Stream.fromEffect action)
183{-# INLINE_NORMAL before #-}
184before :: Monad m => m b -> Stream m a -> Stream m a
185before action (Stream step state) = Stream step' Nothing
189 {-# INLINE_LATE step' #-}
190 step' _ Nothing = action >> return (Skip (Just state))
192 step' gst (Just st) = do
195 Yield x s -> return $ Yield x (Just s)
196 Skip s -> return $ Skip (Just s)
199-- | Like 'after', with following differences:
201-- * action @m b@ won't run if the stream is garbage collected
202-- after partial evaluation.
203-- * Monad @m@ does not require any other constraints.
204-- * has slightly better performance than 'after'.
206-- Same as the following, but with stream fusion:
208-- >>> afterUnsafe action xs = xs <> Stream.nilM action
212{-# INLINE_NORMAL afterUnsafe #-}
213afterUnsafe :: Monad m => m b -> Stream m a -> Stream m a
214afterUnsafe action (Stream step state) = Stream step' state
218 {-# INLINE_LATE step' #-}
222 Yield x s -> return $ Yield x s
223 Skip s -> return $ Skip s
224 Stop -> action >> return Stop
226-- | Run the action @IO b@ whenever the stream is evaluated to completion, or
227-- if it is garbage collected after a partial lazy evaluation.
229-- The semantics of the action @IO b@ are similar to the semantics of cleanup
230-- action in 'bracketIO'.
232-- /See also 'afterUnsafe'/
234{-# INLINE_NORMAL afterIO #-}
236 => IO b -> Stream m a -> Stream m a
237afterIO action (Stream step state) = Stream step' Nothing
241 {-# INLINE_LATE step' #-}
243 ref <- liftIO $ newIOFinalizer action
244 return $ Skip $ Just (state, ref)
245 step' gst (Just (st, ref)) = do
248 Yield x s -> return $ Yield x (Just (s, ref))
249 Skip s -> return $ Skip (Just (s, ref))
254-- XXX For high performance error checks in busy streams we may need another
255-- Error constructor in step.
257-- | Run the action @m b@ if the stream evaluation is aborted due to an
258-- exception. The exception is not caught, simply rethrown.
260-- /Inhibits stream fusion/
262{-# INLINE_NORMAL onException #-}
263onException :: MonadCatch m => m b -> Stream m a -> Stream m a
264onException action stream =
266 (return ()) -- before
268 (\_ (e :: MC.SomeException) _ -> nilM (action >> MC.throwM e))
272{-# INLINE_NORMAL _onException #-}
273_onException :: MonadCatch m => m b -> Stream m a -> Stream m a
274_onException action (Stream step state) = Stream step' state
278 {-# INLINE_LATE step' #-}
280 res <- step gst st `MC.onException` action
282 Yield x s -> return $ Yield x s
283 Skip s -> return $ Skip s
286-- | Like 'bracket' but with following differences:
288-- * alloc action @m b@ runs with async exceptions enabled
289-- * cleanup action @b -> m c@ won't run if the stream is garbage collected
290-- after partial evaluation.
291-- * has slightly better performance than 'bracketIO'.
293-- /Inhibits stream fusion/
297{-# INLINE_NORMAL bracketUnsafe #-}
298bracketUnsafe :: MonadCatch m
299 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
300bracketUnsafe bef aft =
304 (\a (e :: SomeException) _ -> nilM (aft a >> MC.throwM e))
307-- For a use case of this see the "streamly-process" package. It needs to kill
308-- the process in case of exception or garbage collection, but waits for the
309-- process to terminate in normal cases.
311-- | Like 'bracketIO' but can use 3 separate cleanup actions depending on the
312-- mode of termination:
314-- 1. When the stream stops normally
315-- 2. When the stream is garbage collected
316-- 3. When the stream encounters an exception
318-- @bracketIO3 before onStop onGC onException action@ runs @action@ using the
319-- result of @before@. If the stream stops, @onStop@ action is executed, if the
320-- stream is abandoned @onGC@ is executed, if the stream encounters an
321-- exception @onException@ is executed.
323-- /Inhibits stream fusion/
326{-# INLINE_NORMAL bracketIO3 #-}
327bracketIO3 :: (MonadIO m, MonadCatch m) =>
334bracketIO3 bef aft onExc onGC =
338 (\a (e :: SomeException) _ -> onExc a >> return (nilM (MC.throwM e)))
342-- | Run the alloc action @IO b@ with async exceptions disabled but keeping
343-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
344-- output @b@ as input to @b -> Stream m a@ to generate an output stream.
346-- @b@ is usually a resource under the IO monad, e.g. a file handle, that
347-- requires a cleanup after use. The cleanup action @b -> IO c@, runs whenever
348-- the stream ends normally, due to a sync or async exception or if it gets
349-- garbage collected after a partial lazy evaluation.
351-- 'bracketIO' only guarantees that the cleanup action runs, and it runs with
352-- async exceptions enabled. The action must ensure that it can successfully
353-- cleanup the resource in the face of sync or async exceptions.
355-- When the stream ends normally or on a sync exception, cleanup action runs
356-- immediately in the current thread context, whereas in other cases it runs in
357-- the GC context, therefore, cleanup may be delayed until the GC gets to run.
359-- /See also: 'bracketUnsafe'/
361-- /Inhibits stream fusion/
363{-# INLINE bracketIO #-}
364bracketIO :: (MonadIO m, MonadCatch m)
365 => IO b -> (b -> IO c) -> (b -> Stream m a) -> Stream m a
366bracketIO bef aft = bracketIO3 bef aft aft aft
368data BracketState s v = BracketInit | BracketRun s v
370-- | Alternate (custom) implementation of 'bracket'.
372{-# INLINE_NORMAL _bracket #-}
373_bracket :: MonadCatch m
374 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
375_bracket bef aft bet = Stream step' BracketInit
379 {-# INLINE_LATE step' #-}
380 step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))
382 -- NOTE: It is important to use UnStream instead of the Stream pattern
383 -- here, otherwise we get huge perf degradation, see note in concatMap.
384 step' gst (BracketRun (UnStream step state) v) = do
385 -- res <- step gst state `MC.onException` aft v
386 res <- inline MC.try $ step gst state
388 Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop
390 Yield x s -> return $ Yield x (BracketRun (Stream step s) v)
391 Skip s -> return $ Skip (BracketRun (Stream step s) v)
392 Stop -> aft v >> return Stop
394-- | Like 'finally' with following differences:
396-- * action @m b@ won't run if the stream is garbage collected
397-- after partial evaluation.
398-- * has slightly better performance than 'finallyIO'.
400-- /Inhibits stream fusion/
404{-# INLINE finallyUnsafe #-}
405finallyUnsafe :: MonadCatch m => m b -> Stream m a -> Stream m a
406finallyUnsafe action xs = bracketUnsafe (return ()) (const action) (const xs)
408-- | Run the action @IO b@ whenever the stream stream stops normally, aborts
409-- due to an exception or if it is garbage collected after a partial lazy
412-- The semantics of running the action @IO b@ are similar to the cleanup action
413-- semantics described in 'bracketIO'.
415-- >>> finallyIO release = Stream.bracketIO (return ()) (const release)
417-- /See also 'finallyUnsafe'/
419-- /Inhibits stream fusion/
421{-# INLINE finallyIO #-}
422finallyIO :: (MonadIO m, MonadCatch m) => IO b -> Stream m a -> Stream m a
423finallyIO action xs = bracketIO3 (return ()) act act act (const xs)
426-- | Like 'handle' but the exception handler is also provided with the stream
427-- that generated the exception as input. The exception handler can thus
428-- re-evaluate the stream to retry the action that failed. The exception
429-- handler can again call 'ghandle' on it to retry the action multiple times.
431-- This is highly experimental. In a stream of actions we can map the stream
432-- with a retry combinator to retry each action on failure.
434-- /Inhibits stream fusion/
438{-# INLINE_NORMAL ghandle #-}
439ghandle :: (MonadCatch m, Exception e)
440 => (e -> Stream m a -> Stream m a) -> Stream m a -> Stream m a
442 gbracket_ (return ()) return (const f) (inline MC.try) (const stream)
444-- | When evaluating a stream if an exception occurs, stream evaluation aborts
445-- and the specified exception handler is run with the exception as argument.
447-- /Inhibits stream fusion/
449{-# INLINE_NORMAL handle #-}
450handle :: (MonadCatch m, Exception e)
451 => (e -> Stream m a) -> Stream m a -> Stream m a
453 gbracket_ (return ()) return (\_ e _ -> f e) (inline MC.try) (const stream)
455-- | Alternate (custom) implementation of 'handle'.
457{-# INLINE_NORMAL _handle #-}
458_handle :: (MonadCatch m, Exception e)
459 => (e -> Stream m a) -> Stream m a -> Stream m a
460_handle f (Stream step state) = Stream step' (Left state)
464 {-# INLINE_LATE step' #-}
465 step' gst (Left st) = do
466 res <- inline MC.try $ step gst st
468 Left e -> return $ Skip $ Right (f e)
470 Yield x s -> return $ Yield x (Left s)
471 Skip s -> return $ Skip (Left s)
474 step' gst (Right (UnStream step1 st)) = do
477 Yield x s -> return $ Yield x (Right (Stream step1 s))
478 Skip s -> return $ Skip (Right (Stream step1 s))
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list