fc8e9df · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move lifted Unfold exceptions to "streamly" package

fc8e9dfHarendra Kumar4 years ago

benchmark/Streamly/Benchmark/Data/Stream/Exceptions.hs+1

compare with master

unifiedsplit

@@ -39,6 +39,7 @@ import System.IO (Handle, hClose, hPutChar)

39import qualified Streamly.FileSystem.Handle as FH

40import qualified Streamly.Internal.FileSystem.Handle as IFH

41import qualified Streamly.Internal.Data.Unfold as IUF

42+import qualified Streamly.Internal.Data.Unfold.Exception as IUF

42#endif

43#endif

44#ifdef USE_PRELUDE

45#ifdef USE_PRELUDE

benchmark/Streamly/Benchmark/Data/Unfold.hs+1

@@ -31,6 +31,7 @@ import qualified Prelude

31import qualified Streamly.FileSystem.Handle as FH

32import qualified Streamly.Internal.Data.Fold as FL

33import qualified Streamly.Internal.Data.Unfold as UF

34+import qualified Streamly.Internal.Data.Unfold.Exception as UF

34import qualified Streamly.Internal.Data.Stream as S

35import qualified Streamly.Internal.Data.Stream as S

35import qualified Streamly.Internal.Data.Stream.StreamD as D

36import qualified Streamly.Internal.Data.Stream.StreamD as D

36import qualified Streamly.Internal.Data.Stream.StreamK as K

37import qualified Streamly.Internal.Data.Stream.StreamK as K

core/src/Streamly/Internal/Data/IOFinalizer.hs+10−21

@@ -11,7 +11,7 @@

12module Streamly.Internal.Data.IOFinalizer

13 (

14− IOFinalizer

14+ IOFinalizer(..)

15 , newIOFinalizer

16 , runIOFinalizer

17 , clearingIOFinalizer

@@ -22,7 +22,6 @@ import Control.Exception (mask_)

22import Control.Monad (void)

23import Control.Monad.IO.Class (MonadIO(..))

24import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, IORef)

25−import Streamly.Internal.Control.Concurrent (MonadRunInIO, askRunInIO, runInIO, withRunInIO)

27-- | An 'IOFinalizer' has an associated IO action that is automatically called

26-- | An 'IOFinalizer' has an associated IO action that is automatically called

28-- whenever the finalizer is garbage collected. The action can be run and

27-- whenever the finalizer is garbage collected. The action can be run and

@@ -36,15 +35,6 @@ import Streamly.Internal.Control.Concurrent (MonadRunInIO, askRunInIO, runInIO,

36-- /Pre-release/

35-- /Pre-release/

37newtype IOFinalizer = IOFinalizer (IORef (Maybe (IO ())))

36newtype IOFinalizer = IOFinalizer (IORef (Maybe (IO ())))

39−-- | Make a finalizer from a monadic action @m a@ that can run in IO monad.

40−mkIOFinalizer :: MonadRunInIO m => m b -> m (IO ())

41−mkIOFinalizer f = do

42− mrun <- askRunInIO

43− return $

44− void $ do

45− _ <- runInIO mrun f

46− return ()

47−

48-- | GC hook to run an IO action stored in a finalized IORef.

38-- | GC hook to run an IO action stored in a finalized IORef.

49runFinalizerGC :: IORef (Maybe (IO ())) -> IO ()

39runFinalizerGC :: IORef (Maybe (IO ())) -> IO ()

50runFinalizerGC ref = do

40runFinalizerGC ref = do

@@ -65,11 +55,11 @@ runFinalizerGC ref = do

65-- of the monad but we want to keep both the cases consistent.

55-- of the monad but we want to keep both the cases consistent.

66--

56--

67-- /Pre-release/

57-- /Pre-release/

68−newIOFinalizer :: MonadRunInIO m => m a -> m IOFinalizer

58+newIOFinalizer :: MonadIO m => IO a -> m IOFinalizer

69−newIOFinalizer finalizer = do

59+newIOFinalizer finalizer = liftIO $ do

70− f <- mkIOFinalizer finalizer

60+ let f = void finalizer

71− ref <- liftIO $ newIORef $ Just f

61+ ref <- newIORef $ Just f

72− _ <- liftIO $ mkWeakIORef ref (runFinalizerGC ref)

62+ _ <- mkWeakIORef ref (runFinalizerGC ref)

73 return $ IOFinalizer ref

63 return $ IOFinalizer ref

75-- | Run the action associated with the finalizer and deactivate it so that it

65-- | Run the action associated with the finalizer and deactivate it so that it

@@ -94,9 +84,8 @@ runIOFinalizer (IOFinalizer ref) = liftIO $ do

94-- action is run with async exceptions masked.

84-- action is run with async exceptions masked.

95--

85--

96-- /Pre-release/

86-- /Pre-release/

97−clearingIOFinalizer :: MonadRunInIO m => IOFinalizer -> m a -> m a

87+clearingIOFinalizer :: MonadIO m => IOFinalizer -> IO a -> m a

98clearingIOFinalizer (IOFinalizer ref) action = do

88clearingIOFinalizer (IOFinalizer ref) action = do

99− withRunInIO $ \runinio ->

100− mask_ $ do

89+ liftIO $ mask_ $ do

101− writeIORef ref Nothing

90+ writeIORef ref Nothing

102− runinio action

91+ action

core/src/Streamly/Internal/Data/Unfold.hs+36−39

@@ -1,5 +1,3 @@

1−#include "inline.hs"

2−

3-- |

1-- |

4-- Module : Streamly.Internal.Data.Unfold

2-- Module : Streamly.Internal.Data.Unfold

@@ -250,13 +248,13 @@ module Streamly.Internal.Data.Unfold

250

248

251 -- ** Resource Management

249 -- ** Resource Management

252 , gbracket_

250 , gbracket_

253− , gbracket

251+ , gbracketIO

254 , before

252 , before

255− , after

253+ , afterIO

256 , after_

254 , after_

257− , finally

255+ , finallyIO

258 , finally_

256 , finally_

259− , bracket

257+ , bracketIO

260 , bracket_

258 , bracket_

261

259

262 -- ** Exceptions

260 -- ** Exceptions

@@ -265,13 +263,13 @@ module Streamly.Internal.Data.Unfold

265 )

263 )

266where

264where

267

265

266+#include "inline.hs"

268#include "ArrayMacros.h"

267#include "ArrayMacros.h"

269

268

270import Control.Exception (Exception, mask_)

269import Control.Exception (Exception, mask_)

271import Control.Monad.Catch (MonadCatch)

270import Control.Monad.Catch (MonadCatch)

272import Data.Functor (($>))

271import Data.Functor (($>))

273import GHC.Types (SPEC(..))

272import GHC.Types (SPEC(..))

274−import Streamly.Internal.Control.Concurrent (MonadRunInIO, MonadAsync, withRunInIO)

275import Streamly.Internal.Data.Fold.Type (Fold(..))

273import Streamly.Internal.Data.Fold.Type (Fold(..))

276import Streamly.Internal.Data.IOFinalizer

274import Streamly.Internal.Data.IOFinalizer

277 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

275 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

@@ -846,12 +844,11 @@ dropWhile f = dropWhileM (return . f)

846-- Exceptions

844-- Exceptions

847------------------------------------------------------------------------------

845------------------------------------------------------------------------------

848

846

849−-- | Like 'gbracket' but with following differences:

847+-- | Like 'gbracketIO' but with following differences:

850--

848--

851-- * alloc action @a -> m c@ runs with async exceptions enabled

849-- * alloc action @a -> m c@ runs with async exceptions enabled

852-- * cleanup action @c -> m d@ won't run if the stream is garbage collected

850-- * cleanup action @c -> m d@ won't run if the stream is garbage collected

853-- after partial evaluation.

851-- after partial evaluation.

854−-- * does not require a 'MonadAsync' constraint.

855--

852--

856-- /Inhibits stream fusion/

853-- /Inhibits stream fusion/

857--

854--

@@ -911,16 +908,17 @@ gbracket_ bef exc aft (Unfold estep einject) (Unfold step1 inject1) =

911-- /Inhibits stream fusion/

908-- /Inhibits stream fusion/

912--

909--

913-- /Pre-release/

910-- /Pre-release/

914−{-# INLINE_NORMAL gbracket #-}

911+{-# INLINE_NORMAL gbracketIO #-}

915−gbracket

912+gbracketIO

916− :: MonadRunInIO m

913+ :: MonadIO m

917− => (a -> m c) -- ^ before

914+ => (a -> IO c) -- ^ before

918− -> (c -> m d) -- ^ after, on normal stop, or GC

915+ -> (c -> IO d) -- ^ after, on normal stop, or GC

916+ -> (c -> IO ()) -- ^ action on exception

919− -> Unfold m (c, e) b -- ^ on exception

917+ -> Unfold m e b -- ^ stream on exception

920− -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)

918+ -> (forall s. m s -> IO (Either e s)) -- ^ try (exception handling)

921 -> Unfold m c b -- ^ unfold to run

919 -> Unfold m c b -- ^ unfold to run

922 -> Unfold m a b

920 -> Unfold m a b

923−gbracket bef aft (Unfold estep einject) ftry (Unfold step1 inject1) =

921+gbracketIO bef aft onExc (Unfold estep einject) ftry (Unfold step1 inject1) =

924 Unfold step inject

922 Unfold step inject

925

923

926 where

924 where

@@ -928,7 +926,7 @@ gbracket bef aft (Unfold estep einject) ftry (Unfold step1 inject1) =

928 inject x = do

926 inject x = do

929 -- Mask asynchronous exceptions to make the execution of 'bef' and

927 -- Mask asynchronous exceptions to make the execution of 'bef' and

930 -- the registration of 'aft' atomic. See comment in 'D.gbracketIO'.

928 -- the registration of 'aft' atomic. See comment in 'D.gbracketIO'.

931− (r, ref) <- withRunInIO $ \run -> mask_ $ run $ do

929+ (r, ref) <- liftIO $ mask_ $ do

932 r <- bef x

930 r <- bef x

933 ref <- newIOFinalizer (aft r)

931 ref <- newIOFinalizer (aft r)

934 return (r, ref)

932 return (r, ref)

@@ -937,7 +935,7 @@ gbracket bef aft (Unfold estep einject) ftry (Unfold step1 inject1) =

937

935

938 {-# INLINE_LATE step #-}

936 {-# INLINE_LATE step #-}

939 step (Right (st, v, ref)) = do

937 step (Right (st, v, ref)) = do

940− res <- ftry $ step1 st

938+ res <- liftIO $ ftry $ step1 st

941 case res of

939 case res of

942 Right r -> case r of

940 Right r -> case r of

943 Yield x s -> return $ Yield x (Right (s, v, ref))

941 Yield x s -> return $ Yield x (Right (s, v, ref))

@@ -951,7 +949,8 @@ gbracket bef aft (Unfold estep einject) ftry (Unfold step1 inject1) =

951 -- be atomic wrt async exceptions. Otherwise if we have cleared

949 -- be atomic wrt async exceptions. Otherwise if we have cleared

952 -- the finalizer and have not run the exception handler then we

950 -- the finalizer and have not run the exception handler then we

953 -- may leak the resource.

951 -- may leak the resource.

952+ liftIO $ clearingIOFinalizer ref (onExc v)

954− r <- clearingIOFinalizer ref (einject (v, e))

953+ r <- einject e

955 return $ Skip (Left r)

954 return $ Skip (Left r)

956 step (Left st) = do

955 step (Left st) = do

957 res <- estep st

956 res <- estep st

@@ -1013,16 +1012,16 @@ after_ action (Unfold step1 inject1) = Unfold step inject

1013-- /See also 'after_'/

1012-- /See also 'after_'/

1014--

1013--

1015-- /Pre-release/

1014-- /Pre-release/

1016−{-# INLINE_NORMAL after #-}

1015+{-# INLINE_NORMAL afterIO #-}

1017−after :: MonadRunInIO m

1016+afterIO :: MonadIO m

1018− => (a -> m c) -> Unfold m a b -> Unfold m a b

1017+ => (a -> IO c) -> Unfold m a b -> Unfold m a b

1019−after action (Unfold step1 inject1) = Unfold step inject

1018+afterIO action (Unfold step1 inject1) = Unfold step inject

1020

1019

1021 where

1020 where

1022

1021

1023 inject x = do

1022 inject x = do

1024 s <- inject1 x

1023 s <- inject1 x

1025− ref <- newIOFinalizer (action x)

1024+ ref <- liftIO $ newIOFinalizer (action x)

1026 return (s, ref)

1025 return (s, ref)

1027

1026

1028 {-# INLINE_LATE step #-}

1027 {-# INLINE_LATE step #-}

@@ -1072,11 +1071,10 @@ _finally action =

1072 gbracket_ return MC.try action

1071 gbracket_ return MC.try action

1073 (nilM (\(a, e :: MC.SomeException) -> action a >> MC.throwM e))

1072 (nilM (\(a, e :: MC.SomeException) -> action a >> MC.throwM e))

1074

1073

1075−-- | Like 'finally' with following differences:

1074+-- | Like 'finallyIO' with following differences:

1076--

1075--

1077-- * action @a -> m c@ won't run if the stream is garbage collected

1076-- * action @a -> m c@ won't run if the stream is garbage collected

1078-- after partial evaluation.

1077-- after partial evaluation.

1079−-- * does not require a 'MonadAsync' constraint.

1080--

1078--

1081-- /Inhibits stream fusion/

1079-- /Inhibits stream fusion/

1082--

1080--

@@ -1115,16 +1113,16 @@ finally_ action (Unfold step1 inject1) = Unfold step inject

1115-- /Inhibits stream fusion/

1113-- /Inhibits stream fusion/

1116--

1114--

1117-- /Pre-release/

1115-- /Pre-release/

1118−{-# INLINE_NORMAL finally #-}

1116+{-# INLINE_NORMAL finallyIO #-}

1119−finally :: (MonadAsync m, MonadCatch m)

1117+finallyIO :: (MonadIO m, MonadCatch m)

1120− => (a -> m c) -> Unfold m a b -> Unfold m a b

1118+ => (a -> IO c) -> Unfold m a b -> Unfold m a b

1121−finally action (Unfold step1 inject1) = Unfold step inject

1119+finallyIO action (Unfold step1 inject1) = Unfold step inject

1122

1120

1123 where

1121 where

1124

1122

1125 inject x = do

1123 inject x = do

1126 s <- inject1 x

1124 s <- inject1 x

1127− ref <- newIOFinalizer (action x)

1125+ ref <- liftIO $ newIOFinalizer (action x)

1128 return (s, ref)

1126 return (s, ref)

1129

1127

1130 {-# INLINE_LATE step #-}

1128 {-# INLINE_LATE step #-}

@@ -1144,12 +1142,11 @@ _bracket bef aft =

1144 gbracket_ bef MC.try aft (nilM (\(a, e :: MC.SomeException) -> aft a >>

1142 gbracket_ bef MC.try aft (nilM (\(a, e :: MC.SomeException) -> aft a >>

1145 MC.throwM e))

1143 MC.throwM e))

1146

1144

1147−-- | Like 'bracket' but with following differences:

1145+-- | Like 'bracketIO' but with following differences:

1148--

1146--

1149-- * alloc action @a -> m c@ runs with async exceptions enabled

1147-- * alloc action @a -> m c@ runs with async exceptions enabled

1150-- * cleanup action @c -> m d@ won't run if the stream is garbage collected

1148-- * cleanup action @c -> m d@ won't run if the stream is garbage collected

1151-- after partial evaluation.

1149-- after partial evaluation.

1152−-- * does not require a 'MonadAsync' constraint.

1153--

1150--

1154-- /Inhibits stream fusion/

1151-- /Inhibits stream fusion/

1155--

1152--

@@ -1196,17 +1193,17 @@ bracket_ bef aft (Unfold step1 inject1) = Unfold step inject

1196-- /Inhibits stream fusion/

1193-- /Inhibits stream fusion/

1197--

1194--

1198-- /Pre-release/

1195-- /Pre-release/

1199−{-# INLINE_NORMAL bracket #-}

1196+{-# INLINE_NORMAL bracketIO #-}

1200−bracket :: (MonadAsync m, MonadCatch m)

1197+bracketIO :: (MonadIO m, MonadCatch m)

1201− => (a -> m c) -> (c -> m d) -> Unfold m c b -> Unfold m a b

1198+ => (a -> IO c) -> (c -> IO d) -> Unfold m c b -> Unfold m a b

1202−bracket bef aft (Unfold step1 inject1) = Unfold step inject

1199+bracketIO bef aft (Unfold step1 inject1) = Unfold step inject

1203

1200

1204 where

1201 where

1205

1202

1206 inject x = do

1203 inject x = do

1207 -- Mask asynchronous exceptions to make the execution of 'bef' and

1204 -- Mask asynchronous exceptions to make the execution of 'bef' and

1208 -- the registration of 'aft' atomic. See comment in 'D.gbracketIO'.

1205 -- the registration of 'aft' atomic. See comment in 'D.gbracketIO'.

1209− (r, ref) <- withRunInIO $ \run -> mask_ $ run $ do

1206+ (r, ref) <- liftIO $ mask_ $ do

1210 r <- bef x

1207 r <- bef x

1211 ref <- newIOFinalizer (aft r)

1208 ref <- newIOFinalizer (aft r)

1212 return (r, ref)

1209 return (r, ref)

src/Streamly/Internal/Data/IOFinalizer/Lifted.hsnew file+74

@@ -0,0 +1,74 @@

1+-- |

2+-- Module : Streamly.Internal.Data.IOFinalizer.Lifted

4+-- License : BSD-3-Clause

5+-- Maintainer : streamly@composewell.com

6+-- Stability : experimental

7+-- Portability : GHC

8+--

9+-- A value associated with an IO action that is automatically called whenever

10+-- the value is garbage collected.

11+

12+module Streamly.Internal.Data.IOFinalizer.Lifted

13+ (

14+ IOFinalizer

15+ , newIOFinalizer

16+ , runIOFinalizer

17+ , clearingIOFinalizer

18+ )

19+where

20+

21+import Control.Exception (mask_)

22+import Control.Monad (void)

23+import Control.Monad.IO.Class (MonadIO(..))

24+import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, IORef)

25+import Streamly.Internal.Control.Concurrent

26+ (MonadRunInIO, askRunInIO, runInIO, withRunInIO)

27+import Streamly.Internal.Data.IOFinalizer (IOFinalizer(..), runIOFinalizer)

28+

29+-- | Make a finalizer from a monadic action @m a@ that can run in IO monad.

30+mkIOFinalizer :: MonadRunInIO m => m b -> m (IO ())

31+mkIOFinalizer f = do

32+ mrun <- askRunInIO

33+ return $

34+ void $ do

35+ _ <- runInIO mrun f

36+ return ()

37+

38+-- | GC hook to run an IO action stored in a finalized IORef.

39+runFinalizerGC :: IORef (Maybe (IO ())) -> IO ()

40+runFinalizerGC ref = do

41+ res <- readIORef ref

42+ case res of

43+ Nothing -> return ()

44+ Just f -> f

45+

46+-- | Create a finalizer that calls the supplied function automatically when the

47+-- it is garbage collected.

48+--

49+-- /The finalizer is always run using the state of the monad that is captured

50+-- at the time of calling 'newFinalizer'./

51+--

52+-- Note: To run it on garbage collection we have no option but to use the monad

53+-- state captured at some earlier point of time. For the case when the

54+-- finalizer is run manually before GC we could run it with the current state

55+-- of the monad but we want to keep both the cases consistent.

56+--

57+-- /Pre-release/

58+newIOFinalizer :: MonadRunInIO m => m a -> m IOFinalizer

59+newIOFinalizer finalizer = do

60+ f <- mkIOFinalizer finalizer

61+ ref <- liftIO $ newIORef $ Just f

62+ _ <- liftIO $ mkWeakIORef ref (runFinalizerGC ref)

63+ return $ IOFinalizer ref

64+

65+-- | Run an action clearing the finalizer atomically wrt async exceptions. The

66+-- action is run with async exceptions masked.

67+--

68+-- /Pre-release/

69+clearingIOFinalizer :: MonadRunInIO m => IOFinalizer -> m a -> m a

70+clearingIOFinalizer (IOFinalizer ref) action = do

71+ withRunInIO $ \runinio ->

72+ mask_ $ do

73+ writeIORef ref Nothing

74+ runinio action

src/Streamly/Internal/Data/Unfold/Exception.hsnew file+216

@@ -0,0 +1,216 @@

1+-- |

2+-- Module : Streamly.Internal.Data.Unfold.Exception

4+-- License : BSD3

5+-- Maintainer : streamly@composewell.com

6+-- Stability : experimental

7+-- Portability : GHC

8+--

9+-- Lifted resource management primitives.

10+

11+module Streamly.Internal.Data.Unfold.Exception

12+ (

13+ gbracket

14+ , after

15+ , finally

16+ , bracket

17+ )

18+where

19+

20+#include "inline.hs"

21+

22+import Control.Exception (mask_)

23+import Control.Monad.Catch (MonadCatch)

24+import Streamly.Internal.Control.Concurrent

25+ (MonadRunInIO, MonadAsync, withRunInIO)

26+import Streamly.Internal.Data.IOFinalizer.Lifted

27+ (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

28+import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))

29+

30+import qualified Control.Monad.Catch as MC

31+

32+import Streamly.Internal.Data.Unfold.Type

33+

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 #-}

51+gbracket

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

59+gbracket bef aft (Unfold estep einject) ftry (Unfold step1 inject1) =

60+ Unfold step inject

61+

62+ where

63+

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)

73+

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

98+

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 #-}

110+after :: MonadRunInIO m

111+ => (a -> m c) -> Unfold m a b -> Unfold m a b

112+after action (Unfold step1 inject1) = Unfold step inject

113+

114+ where

115+

116+ inject x = do

117+ s <- inject1 x

118+ ref <- newIOFinalizer (action x)

119+ return (s, ref)

120+

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

130+

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 #-}

148+finally :: (MonadAsync m, MonadCatch m)

149+ => (a -> m c) -> Unfold m a b -> Unfold m a b

150+finally action (Unfold step1 inject1) = Unfold step inject

151+

152+ where

153+

154+ inject x = do

155+ s <- inject1 x

156+ ref <- newIOFinalizer (action x)

157+ return (s, ref)

158+

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

168+

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 #-}

192+bracket :: (MonadAsync m, MonadCatch m)

193+ => (a -> m c) -> (c -> m d) -> Unfold m c b -> Unfold m a b

194+bracket bef aft (Unfold step1 inject1) = Unfold step inject

195+

196+ where

197+

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)

207+

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

src/Streamly/Internal/FileSystem/File.hs+8−10

@@ -116,7 +116,7 @@ import qualified Streamly.Data.Array.Unboxed as A

116import qualified Streamly.Data.Unfold as UF

117import qualified Streamly.Internal.Data.Fold.Type as FL

118 (Step(..), snoc, initialize)

119−import qualified Streamly.Internal.Data.Unfold as UF (bracket)

119+import qualified Streamly.Internal.Data.Unfold as UF (bracketIO)

120import qualified Streamly.Internal.FileSystem.Handle as FH

121import qualified Streamly.Internal.Data.Array.Unboxed.Stream as AS

122import qualified Streamly.Data.Stream as S (fold, bracketIO, mapM)

@@ -163,35 +163,33 @@ withFile file mode = S.bracketIO (openFile file mode) hClose

163{-# INLINE usingFile #-}

164usingFile :: (MonadCatch m, MonadAsync m)

165 => Unfold m Handle a -> Unfold m FilePath a

166−usingFile =

167− UF.bracket (\file -> liftIO $ openFile file ReadMode)

166+usingFile = UF.bracketIO (`openFile` ReadMode) hClose

168− (liftIO . hClose)

169

167

170{-# INLINE usingFile2 #-}

168{-# INLINE usingFile2 #-}

171usingFile2 :: (MonadCatch m, MonadAsync m)

169usingFile2 :: (MonadCatch m, MonadAsync m)

172 => Unfold m (x, Handle) a -> Unfold m (x, FilePath) a

170 => Unfold m (x, Handle) a -> Unfold m (x, FilePath) a

173−usingFile2 = UF.bracket before after

171+usingFile2 = UF.bracketIO before after

174

172

175 where

173 where

176

174

177 before (x, file) = do

175 before (x, file) = do

178− h <- liftIO $ openFile file ReadMode

176+ h <- openFile file ReadMode

179 return (x, h)

177 return (x, h)

180

178

181− after (_, h) = liftIO $ hClose h

179+ after (_, h) = hClose h

182

180

183{-# INLINE usingFile3 #-}

181{-# INLINE usingFile3 #-}

184usingFile3 :: (MonadCatch m, MonadAsync m)

182usingFile3 :: (MonadCatch m, MonadAsync m)

185 => Unfold m (x, y, z, Handle) a -> Unfold m (x, y, z, FilePath) a

183 => Unfold m (x, y, z, Handle) a -> Unfold m (x, y, z, FilePath) a

186−usingFile3 = UF.bracket before after

184+usingFile3 = UF.bracketIO before after

187

185

188 where

186 where

189

187

190 before (x, y, z, file) = do

188 before (x, y, z, file) = do

191− h <- liftIO $ openFile file ReadMode

189+ h <- openFile file ReadMode

192 return (x, y, z, h)

190 return (x, y, z, h)

193

191

194− after (_, _, _, h) = liftIO $ hClose h

192+ after (_, _, _, h) = hClose h

195

193

196-------------------------------------------------------------------------------

194-------------------------------------------------------------------------------

197-- Array IO (Input)

195-- Array IO (Input)

src/Streamly/Internal/Network/Inet/TCP.hs+2−4

@@ -120,7 +120,7 @@ import qualified Network.Socket as Net

120import qualified Streamly.Data.Array.Unboxed as A

121import qualified Streamly.Data.Fold as FL

122import qualified Streamly.Data.Unfold as UF

123−import qualified Streamly.Internal.Data.Unfold as UF (bracket, first)

123+import qualified Streamly.Internal.Data.Unfold as UF (bracketIO, first)

124import qualified Streamly.Internal.Data.Array.Unboxed.Stream as AS

125import qualified Streamly.Internal.Data.Fold.Type as FL

126 (initialize, snoc, Step(..))

@@ -286,9 +286,7 @@ withConnectionM addr port =

286usingConnection :: (MonadCatch m, MonadAsync m)

287 => Unfold m Socket a

288 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a

289−usingConnection =

290− UF.bracket (\(addr, port) -> liftIO $ connect addr port)

289+usingConnection = UF.bracketIO (\(addr, port) -> connect addr port) Net.close

291− (liftIO . Net.close)

292

290

293-------------------------------------------------------------------------------

291-------------------------------------------------------------------------------

294-- Connect (streams)

292-- Connect (streams)

streamly.cabal+2

@@ -333,10 +333,12 @@ library

333

334 -- streamly-concurrent

335 , Streamly.Internal.Data.Atomics

336+ , Streamly.Internal.Data.IOFinalizer.Lifted

336 , Streamly.Internal.Data.Stream.Channel.Types

337 , Streamly.Internal.Data.Stream.Channel.Types

337 , Streamly.Internal.Data.Stream.Channel.Dispatcher

338 , Streamly.Internal.Data.Stream.Channel.Dispatcher

338 , Streamly.Internal.Data.Stream.Channel.Worker

339 , Streamly.Internal.Data.Stream.Channel.Worker

339

340

341+ , Streamly.Internal.Data.Unfold.Exception

340 , Streamly.Internal.Data.Fold.Concurrent.Channel.Type

342 , Streamly.Internal.Data.Fold.Concurrent.Channel.Type

341 , Streamly.Internal.Data.Fold.Concurrent.Channel

343 , Streamly.Internal.Data.Fold.Concurrent.Channel

342 , Streamly.Internal.Data.Fold.Concurrent

344 , Streamly.Internal.Data.Fold.Concurrent