3-- Module : Streamly.Internal.Data.Stream.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.Exception
33import Control.Monad.IO.Class (MonadIO(..))
34import Control.Exception (Exception, SomeException, mask_)
35import Control.Monad.Catch (MonadCatch)
36import GHC.Exts (inline)
37import Streamly.Internal.Data.IOFinalizer
38 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)
40import qualified Control.Monad.Catch as MC
42import Streamly.Internal.Data.Stream.Type
44#include "DocTestDataStream.hs"
46data GbracketState s1 s2 v
49 | GBracketException s2
51-- | Like 'gbracket' but with following differences:
53-- * alloc action @m c@ runs with async exceptions enabled
54-- * cleanup action @c -> m d@ won't run if the stream is garbage collected
55-- after partial evaluation.
57-- /Inhibits stream fusion/
61{-# INLINE_NORMAL gbracket_ #-}
65 -> (c -> m d) -- ^ after, on normal stop
66 -> (c -> e -> Stream m b -> m (Stream m b)) -- ^ on exception
67 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
68 -> (c -> Stream m b) -- ^ stream generator
70gbracket_ bef aft onExc ftry action =
71 Stream step GBracketInit
75 {-# INLINE_LATE step #-}
76 step _ GBracketInit = do
78 return $ Skip $ GBracketNormal (action r) r
80 step gst (GBracketNormal (UnStream step1 st) v) = do
81 res <- ftry $ step1 gst st
85 return $ Yield x (GBracketNormal (Stream step1 s) v)
86 Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)
87 Stop -> aft v >> return Stop
88 -- XXX Do not handle async exceptions, just rethrow them.
90 strm <- onExc v e (UnStream step1 st)
91 return $ Skip (GBracketException strm)
92 step gst (GBracketException (UnStream step1 st)) = do
95 Yield x s -> return $ Yield x (GBracketException (Stream step1 s))
96 Skip s -> return $ Skip (GBracketException (Stream step1 s))
99data GbracketIOState s1 s2 v wref
101 | GBracketIONormal s1 v wref
102 | GBracketIOException s2
104-- | Run the alloc action @m c@ with async exceptions disabled but keeping
105-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
106-- output @c@ as input to @c -> Stream m b@ to generate an output stream. When
107-- generating the stream use the supplied @try@ operation @forall s. m s -> m
108-- (Either e s)@ to catch synchronous exceptions. If an exception occurs run
109-- the exception handler @c -> e -> Stream m b -> m (Stream m b)@. Note that
110-- 'gbracket' does not rethrow the exception, it has to be done by the
111-- exception handler if desired.
113-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due
114-- to a sync or async exception or if it gets garbage collected after a partial
115-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.
117-- 'gbracket' can express all other exception handling combinators.
119-- /Inhibits stream fusion/
122{-# INLINE_NORMAL gbracket #-}
126 -> (c -> IO d1) -- ^ on normal stop
127 -> (c -> e -> Stream m b -> IO (Stream m b)) -- ^ on exception
128 -> (c -> IO d2) -- ^ on GC without normal stop or exception
129 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
130 -> (c -> Stream m b) -- ^ stream generator
132gbracket bef aft onExc onGC ftry action =
133 Stream step GBracketIOInit
137 -- If the stream is never evaluated the "aft" action will never be
138 -- called. For that to occur we will need the user of this API to pass a
139 -- weak pointer to us.
140 {-# INLINE_LATE step #-}
141 step _ GBracketIOInit = do
142 -- We mask asynchronous exceptions to make the execution
143 -- of 'bef' and the registration of 'aft' atomic.
144 -- A similar thing is done in the resourcet package: https://git.io/JvKV3
145 -- Tutorial: https://markkarpov.com/tutorial/exceptions.html
146 (r, ref) <- liftIO $ mask_ $ do
148 ref <- newIOFinalizer (onGC r)
150 return $ Skip $ GBracketIONormal (action r) r ref
152 step gst (GBracketIONormal (UnStream step1 st) v ref) = do
153 res <- ftry $ step1 gst st
157 return $ Yield x (GBracketIONormal (Stream step1 s) v ref)
159 return $ Skip (GBracketIONormal (Stream step1 s) v ref)
161 liftIO (clearingIOFinalizer ref (aft v)) >> return Stop
162 -- XXX Do not handle async exceptions, just rethrow them.
164 -- Clearing of finalizer and running of exception handler must
165 -- be atomic wrt async exceptions. Otherwise if we have cleared
166 -- the finalizer and have not run the exception handler then we
167 -- may leak the resource.
169 liftIO (clearingIOFinalizer ref (onExc v e (UnStream step1 st)))
170 return $ Skip (GBracketIOException stream)
171 step gst (GBracketIOException (UnStream step1 st)) = do
175 return $ Yield x (GBracketIOException (Stream step1 s))
176 Skip s -> return $ Skip (GBracketIOException (Stream step1 s))
179-- | Run the action @m b@ before the stream yields its first element.
181-- Same as the following but more efficient due to fusion:
183-- >>> before action xs = Stream.nilM action <> xs
184-- >>> before action xs = Stream.concatMap (const xs) (Stream.fromEffect action)
186{-# INLINE_NORMAL before #-}
187before :: Monad m => m b -> Stream m a -> Stream m a
188before action (Stream step state) = Stream step' Nothing
192 {-# INLINE_LATE step' #-}
193 step' _ Nothing = action >> return (Skip (Just state))
195 step' gst (Just st) = do
198 Yield x s -> return $ Yield x (Just s)
199 Skip s -> return $ Skip (Just s)
202-- | Like 'after', with following differences:
204-- * action @m b@ won't run if the stream is garbage collected
205-- after partial evaluation.
206-- * Monad @m@ does not require any other constraints.
207-- * has slightly better performance than 'after'.
209-- Same as the following, but with stream fusion:
211-- >>> afterUnsafe action xs = xs <> Stream.nilM action
215{-# INLINE_NORMAL afterUnsafe #-}
216afterUnsafe :: Monad m => m b -> Stream m a -> Stream m a
217afterUnsafe action (Stream step state) = Stream step' state
221 {-# INLINE_LATE step' #-}
225 Yield x s -> return $ Yield x s
226 Skip s -> return $ Skip s
227 Stop -> action >> return Stop
229-- | Run the action @IO b@ whenever the stream is evaluated to completion, or
230-- if it is garbage collected after a partial lazy evaluation.
232-- The semantics of the action @IO b@ are similar to the semantics of cleanup
233-- action in 'bracketIO'.
235-- /See also 'afterUnsafe'/
237{-# INLINE_NORMAL afterIO #-}
239 => IO b -> Stream m a -> Stream m a
240afterIO action (Stream step state) = Stream step' Nothing
244 {-# INLINE_LATE step' #-}
246 ref <- liftIO $ newIOFinalizer action
247 return $ Skip $ Just (state, ref)
248 step' gst (Just (st, ref)) = do
251 Yield x s -> return $ Yield x (Just (s, ref))
252 Skip s -> return $ Skip (Just (s, ref))
257-- XXX For high performance error checks in busy streams we may need another
258-- Error constructor in step.
260-- | Run the action @m b@ if the stream evaluation is aborted due to an
261-- exception. The exception is not caught, simply rethrown.
263-- Observes exceptions only in the stream generation, and not in stream
266-- /Inhibits stream fusion/
268{-# INLINE_NORMAL onException #-}
269onException :: MonadCatch m => m b -> Stream m a -> Stream m a
270onException action stream =
272 (return ()) -- before
274 (\_ (e :: MC.SomeException) _ -> action >> MC.throwM e)
278{-# INLINE_NORMAL _onException #-}
279_onException :: MonadCatch m => m b -> Stream m a -> Stream m a
280_onException action (Stream step state) = Stream step' state
284 {-# INLINE_LATE step' #-}
286 res <- step gst st `MC.onException` action
288 Yield x s -> return $ Yield x s
289 Skip s -> return $ Skip s
292-- | Like 'bracket' but with following differences:
294-- * alloc action @m b@ runs with async exceptions enabled
295-- * cleanup action @b -> m c@ won't run if the stream is garbage collected
296-- after partial evaluation.
297-- * has slightly better performance than 'bracketIO'.
299-- /Inhibits stream fusion/
303{-# INLINE_NORMAL bracketUnsafe #-}
304bracketUnsafe :: MonadCatch m
305 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
306bracketUnsafe bef aft =
310 (\a (e :: SomeException) _ -> aft a >> MC.throwM e)
313-- For a use case of this see the "streamly-process" package. It needs to kill
314-- the process in case of exception or garbage collection, but waits for the
315-- process to terminate in normal cases.
317-- | Like 'bracketIO' but can use 3 separate cleanup actions depending on the
318-- mode of termination:
320-- 1. When the stream stops normally
321-- 2. When the stream is garbage collected
322-- 3. When the stream encounters an exception
324-- @bracketIO3 before onStop onGC onException action@ runs @action@ using the
325-- result of @before@. If the stream stops, @onStop@ action is executed, if the
326-- stream is abandoned @onGC@ is executed, if the stream encounters an
327-- exception @onException@ is executed.
329-- The exception is not caught, it is rethrown.
331-- /Inhibits stream fusion/
334{-# INLINE_NORMAL bracketIO3 #-}
335bracketIO3 :: (MonadIO m, MonadCatch m) =>
342bracketIO3 bef aft onExc onGC =
346 (\a (e :: SomeException) _ -> onExc a >> MC.throwM e)
350-- | Run the alloc action @IO b@ with async exceptions disabled but keeping
351-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
352-- output @b@ of the IO action as input to the function @b -> Stream m a@ to
353-- generate an output stream.
355-- @b@ is usually a resource under the IO monad, e.g. a file handle, that
356-- requires a cleanup after use. The cleanup action @b -> IO c@, runs whenever
357-- (1) the stream ends normally, (2) due to a sync or async exception or, (3)
358-- if it gets garbage collected after a partial lazy evaluation. The exception
359-- is not caught, it is rethrown.
361-- 'bracketIO' only guarantees that the cleanup action runs, and it runs with
362-- async exceptions enabled. The action must ensure that it can successfully
363-- cleanup the resource in the face of sync or async exceptions.
365-- When the stream ends normally or on a sync exception, cleanup action runs
366-- immediately in the current thread context, whereas in other cases it runs in
367-- the GC context, therefore, cleanup may be delayed until the GC gets to run.
368-- An example where GC based cleanup happens is when a stream is being folded
369-- but the fold terminates without draining the entire stream or if the
370-- consumer of the stream encounters an exception.
372-- Observes exceptions only in the stream generation, and not in stream
375-- /See also: 'bracketUnsafe'/
377-- /Inhibits stream fusion/
379{-# INLINE bracketIO #-}
380bracketIO :: (MonadIO m, MonadCatch m)
381 => IO b -> (b -> IO c) -> (b -> Stream m a) -> Stream m a
382bracketIO bef aft = bracketIO3 bef aft aft aft
384data BracketState s v = BracketInit | BracketRun s v
386-- | Alternate (custom) implementation of 'bracket'.
388{-# INLINE_NORMAL _bracket #-}
389_bracket :: MonadCatch m
390 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
391_bracket bef aft bet = Stream step' BracketInit
395 {-# INLINE_LATE step' #-}
396 step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))
398 -- NOTE: It is important to use UnStream instead of the Stream pattern
399 -- here, otherwise we get huge perf degradation, see note in concatMap.
400 step' gst (BracketRun (UnStream step state) v) = do
401 -- res <- step gst state `MC.onException` aft v
402 res <- inline MC.try $ step gst state
404 Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop
406 Yield x s -> return $ Yield x (BracketRun (Stream step s) v)
407 Skip s -> return $ Skip (BracketRun (Stream step s) v)
408 Stop -> aft v >> return Stop
410-- | Like 'finally' with following differences:
412-- * action @m b@ won't run if the stream is garbage collected
413-- after partial evaluation.
414-- * has slightly better performance than 'finallyIO'.
416-- /Inhibits stream fusion/
420{-# INLINE finallyUnsafe #-}
421finallyUnsafe :: MonadCatch m => m b -> Stream m a -> Stream m a
422finallyUnsafe action xs = bracketUnsafe (return ()) (const action) (const xs)
424-- | Run the action @IO b@ whenever the stream stream stops normally, aborts
425-- due to an exception or if it is garbage collected after a partial lazy
428-- The semantics of running the action @IO b@ are similar to the cleanup action
429-- semantics described in 'bracketIO'.
431-- >>> finallyIO release = Stream.bracketIO (return ()) (const release)
433-- /See also 'finallyUnsafe'/
435-- /Inhibits stream fusion/
437{-# INLINE finallyIO #-}
438finallyIO :: (MonadIO m, MonadCatch m) => IO b -> Stream m a -> Stream m a
439finallyIO action xs = bracketIO3 (return ()) act act act (const xs)
442-- | Like 'handle' but the exception handler is also provided with the stream
443-- that generated the exception as input. The exception handler can thus
444-- re-evaluate the stream to retry the action that failed. The exception
445-- handler can again call 'ghandle' on it to retry the action multiple times.
447-- This is highly experimental. In a stream of actions we can map the stream
448-- with a retry combinator to retry each action on failure.
450-- /Inhibits stream fusion/
454{-# INLINE_NORMAL ghandle #-}
455ghandle :: (MonadCatch m, Exception e)
456 => (e -> Stream m a -> m (Stream m a)) -> Stream m a -> Stream m a
458 gbracket_ (return ()) return (const f) (inline MC.try) (const stream)
460-- | When evaluating a stream if an exception occurs, stream evaluation aborts
461-- and the specified exception handler is run with the exception as argument.
462-- The exception is caught and handled unless the handler decides to rethrow
463-- it. Note that exception handling is not applied to the stream returned by
464-- the exception handler.
466-- Observes exceptions only in the stream generation, and not in stream
469-- /Inhibits stream fusion/
471{-# INLINE_NORMAL handle #-}
472handle :: (MonadCatch m, Exception e)
473 => (e -> m (Stream m a)) -> Stream m a -> Stream m a
475 gbracket_ (return ()) return (\_ e _ -> f e) (inline MC.try) (const stream)
477-- | Alternate (custom) implementation of 'handle'.
479{-# INLINE_NORMAL _handle #-}
480_handle :: (MonadCatch m, Exception e)
481 => (e -> Stream m a) -> Stream m a -> Stream m a
482_handle f (Stream step state) = Stream step' (Left state)
486 {-# INLINE_LATE step' #-}
487 step' gst (Left st) = do
488 res <- inline MC.try $ step gst st
490 Left e -> return $ Skip $ Right (f e)
492 Yield x s -> return $ Yield x (Left s)
493 Skip s -> return $ Skip (Left s)
496 step' gst (Right (UnStream step1 st)) = do
499 Yield x s -> return $ Yield x (Right (Stream step1 s))
500 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