git.iko.soy/streamly
on 4e34693 help sign in
streamly/src/Streamly/Internal/Data/Unfold/Exception.hs
214 lines 7.4 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Unfold.Exception
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD3
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9-- Lifted resource management primitives.
11module Streamly.Internal.Data.Unfold.Exception
12 (
13 gbracket
14 , after
15 , finally
16 , bracket
17 )
18where
20#include "inline.hs"
22import Control.Exception (mask_)
23import Control.Monad.Catch (MonadCatch)
24import Streamly.Internal.Control.Concurrent
25 (MonadRunInIO, MonadAsync, withRunInIO)
26import Streamly.Internal.Data.IOFinalizer.Lifted
27 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)
28import qualified Control.Monad.Catch as MC
30import Streamly.Internal.Data.Unfold
32-- | Run the alloc action @a -> m c@ with async exceptions disabled but keeping
33-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
34-- output @c@ as input to @Unfold m c b@ to generate an output stream. When
35-- unfolding use the supplied @try@ operation @forall s. m s -> m (Either e s)@
36-- to catch synchronous exceptions. If an exception occurs run the exception
37-- handling unfold @Unfold m (c, e) b@.
38--
39-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due
40-- to a sync or async exception or if it gets garbage collected after a partial
41-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.
42--
43-- 'gbracket' can express all other exception handling combinators.
44--
45-- /Inhibits stream fusion/
46--
47-- /Pre-release/
48{-# INLINE_NORMAL gbracket #-}
49gbracket
50 :: MonadRunInIO m
51 => (a -> m c) -- ^ before
52 -> (c -> m d) -- ^ after, on normal stop, or GC
53 -> Unfold m (c, e) b -- ^ on exception
54 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
55 -> Unfold m c b -- ^ unfold to run
56 -> Unfold m a b
57gbracket bef aft (Unfold estep einject) ftry (Unfold step1 inject1) =
58 Unfold step inject
60 where
62 inject x = do
63 -- Mask asynchronous exceptions to make the execution of 'bef' and
64 -- the registration of 'aft' atomic. See comment in 'D.gbracketIO'.
65 (r, ref) <- withRunInIO $ \run -> mask_ $ run $ do
66 r <- bef x
67 ref <- newIOFinalizer (aft r)
68 return (r, ref)
69 s <- inject1 r
70 return $ Right (s, r, ref)
72 {-# INLINE_LATE step #-}
73 step (Right (st, v, ref)) = do
74 res <- ftry $ step1 st
75 case res of
76 Right r -> case r of
77 Yield x s -> return $ Yield x (Right (s, v, ref))
78 Skip s -> return $ Skip (Right (s, v, ref))
79 Stop -> do
80 runIOFinalizer ref
81 return Stop
82 -- XXX Do not handle async exceptions, just rethrow them.
83 Left e -> do
84 -- Clearing of finalizer and running of exception handler must
85 -- be atomic wrt async exceptions. Otherwise if we have cleared
86 -- the finalizer and have not run the exception handler then we
87 -- may leak the resource.
88 r <- clearingIOFinalizer ref (einject (v, e))
89 return $ Skip (Left r)
90 step (Left st) = do
91 res <- estep st
92 return $ case res of
93 Yield x s -> Yield x (Left s)
94 Skip s -> Skip (Left s)
95 Stop -> Stop
97-- | Unfold the input @a@ using @Unfold m a b@, run an action on @a@ whenever
98-- the unfold stops normally, or if it is garbage collected after a partial
99-- lazy evaluation.
100--
101-- The semantics of the action @a -> m c@ are similar to the cleanup action
102-- semantics in 'bracket'.
103--
104-- /See also 'after_'/
105--
106-- /Pre-release/
107{-# INLINE_NORMAL after #-}
108after :: MonadRunInIO m
109 => (a -> m c) -> Unfold m a b -> Unfold m a b
110after action (Unfold step1 inject1) = Unfold step inject
112 where
114 inject x = do
115 s <- inject1 x
116 ref <- newIOFinalizer (action x)
117 return (s, ref)
119 {-# INLINE_LATE step #-}
120 step (st, ref) = do
121 res <- step1 st
122 case res of
123 Yield x s -> return $ Yield x (s, ref)
124 Skip s -> return $ Skip (s, ref)
125 Stop -> do
126 runIOFinalizer ref
127 return Stop
129-- | Unfold the input @a@ using @Unfold m a b@, run an action on @a@ whenever
130-- the unfold stops normally, aborts due to an exception or if it is garbage
131-- collected after a partial lazy evaluation.
132--
133-- The semantics of the action @a -> m c@ are similar to the cleanup action
134-- semantics in 'bracket'.
135--
136-- @
137-- finally release = bracket return release
138-- @
139--
140-- /See also 'finally_'/
141--
142-- /Inhibits stream fusion/
143--
144-- /Pre-release/
145{-# INLINE_NORMAL finally #-}
146finally :: (MonadAsync m, MonadCatch m)
147 => (a -> m c) -> Unfold m a b -> Unfold m a b
148finally action (Unfold step1 inject1) = Unfold step inject
150 where
152 inject x = do
153 s <- inject1 x
154 ref <- newIOFinalizer (action x)
155 return (s, ref)
157 {-# INLINE_LATE step #-}
158 step (st, ref) = do
159 res <- step1 st `MC.onException` runIOFinalizer ref
160 case res of
161 Yield x s -> return $ Yield x (s, ref)
162 Skip s -> return $ Skip (s, ref)
163 Stop -> do
164 runIOFinalizer ref
165 return Stop
167-- | Run the alloc action @a -> m c@ with async exceptions disabled but keeping
168-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
169-- output @c@ as input to @Unfold m c b@ to generate an output stream.
170--
171-- @c@ is usually a resource under the state of monad @m@, e.g. a file
172-- handle, that requires a cleanup after use. The cleanup action @c -> m d@,
173-- runs whenever the stream ends normally, due to a sync or async exception or
174-- if it gets garbage collected after a partial lazy evaluation.
175--
176-- 'bracket' only guarantees that the cleanup action runs, and it runs with
177-- async exceptions enabled. The action must ensure that it can successfully
178-- cleanup the resource in the face of sync or async exceptions.
179--
180-- When the stream ends normally or on a sync exception, cleanup action runs
181-- immediately in the current thread context, whereas in other cases it runs in
182-- the GC context, therefore, cleanup may be delayed until the GC gets to run.
183--
184-- /See also: 'bracket_', 'gbracket'/
185--
186-- /Inhibits stream fusion/
187--
188-- /Pre-release/
189{-# INLINE_NORMAL bracket #-}
190bracket :: (MonadAsync m, MonadCatch m)
191 => (a -> m c) -> (c -> m d) -> Unfold m c b -> Unfold m a b
192bracket bef aft (Unfold step1 inject1) = Unfold step inject
194 where
196 inject x = do
197 -- Mask asynchronous exceptions to make the execution of 'bef' and
198 -- the registration of 'aft' atomic. See comment in 'D.gbracketIO'.
199 (r, ref) <- withRunInIO $ \run -> mask_ $ run $ do
200 r <- bef x
201 ref <- newIOFinalizer (aft r)
202 return (r, ref)
203 s <- inject1 r
204 return (s, ref)
206 {-# INLINE_LATE step #-}
207 step (st, ref) = do
208 res <- step1 st `MC.onException` runIOFinalizer ref
209 case res of
210 Yield x s -> return $ Yield x (s, ref)
211 Skip s -> return $ Skip (s, ref)
212 Stop -> do
213 runIOFinalizer ref
214 return Stop