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