fc8e9df · streamly · git.iko.soy

git.iko.soy/streamly

Move lifted Unfold exceptions to "streamly" package

Implement IO based combinators in streamly-core Move IOFinalizer lifted implementation to streamly

commitfc8e9dfd29fe96605964d175de4d643bb66f3f68 authorHarendra Kumar <harendra@composewell.com> dateMon 24 Oct 2022, 07:31 (4 years ago) parentfd681d6

9 files changed+350−74

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Data/Stream/Exceptions.hs+1 benchmark/Streamly/Benchmark/Data/Unfold.hs+1 core/src/Streamly/Internal/Data/IOFinalizer.hs+10−21 core/src/Streamly/Internal/Data/Unfold.hs+36−39 src/Streamly/Internal/Data/IOFinalizer/Lifted.hs+74 src/Streamly/Internal/Data/Unfold/Exception.hs+216 src/Streamly/Internal/FileSystem/File.hs+8−10 src/Streamly/Internal/Network/Inet/TCP.hs+2−4 streamly.cabal+2

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

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

3939import qualified Streamly.FileSystem.Handle as FH

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

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

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

4243#endif

4344

4445#ifdef USE_PRELUDE

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

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

3131import qualified Streamly.FileSystem.Handle as FH

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

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

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

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

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

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

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

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

1111

1212module Streamly.Internal.Data.IOFinalizer

1313 (

14− IOFinalizer

14+ IOFinalizer(..)

1515 , newIOFinalizer

1616 , runIOFinalizer

1717 , clearingIOFinalizer

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

2222import Control.Monad (void)

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

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

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

2625

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

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

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

3635-- /Pre-release/

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

3837

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−

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

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

5040runFinalizerGC ref = do

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

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

6656--

6757-- /Pre-release/

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

69−newIOFinalizer finalizer = do

70− f <- mkIOFinalizer finalizer

71− ref <- liftIO $ newIORef $ Just f

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

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

59+newIOFinalizer finalizer = liftIO $ do

60+ let f = void finalizer

61+ ref <- newIORef $ Just f

62+ _ <- mkWeakIORef ref (runFinalizerGC ref)

7363 return $ IOFinalizer ref

7464

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

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

9484-- action is run with async exceptions masked.

9585--

9686-- /Pre-release/

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

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

9888clearingIOFinalizer (IOFinalizer ref) action = do

99− withRunInIO $ \runinio ->

100− mask_ $ do

101− writeIORef ref Nothing

102− runinio action

89+ liftIO $ mask_ $ do

90+ writeIORef ref Nothing

91+ action

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

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

1−#include "inline.hs"

2−

31-- |

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

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

250248

251249 -- ** Resource Management

252250 , gbracket_

253− , gbracket

251+ , gbracketIO

254252 , before

255− , after

253+ , afterIO

256254 , after_

257− , finally

255+ , finallyIO

258256 , finally_

259− , bracket

257+ , bracketIO

260258 , bracket_

261259

262260 -- ** Exceptions

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

265263 )

266264where

267265

266+#include "inline.hs"

268267#include "ArrayMacros.h"

269268

270269import Control.Exception (Exception, mask_)

271270import Control.Monad.Catch (MonadCatch)

272271import Data.Functor (($>))

273272import GHC.Types (SPEC(..))

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

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

276274import Streamly.Internal.Data.IOFinalizer

277275 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

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

846844-- Exceptions

847845------------------------------------------------------------------------------

848846

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

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

850848--

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

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

853851-- after partial evaluation.

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

855852--

856853-- /Inhibits stream fusion/

857854--

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

911908-- /Inhibits stream fusion/

912909--

913910-- /Pre-release/

914−{-# INLINE_NORMAL gbracket #-}

915−gbracket

916− :: MonadRunInIO m

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

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

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

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

911+{-# INLINE_NORMAL gbracketIO #-}

912+gbracketIO

913+ :: MonadIO m

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

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

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

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

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

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

922920 -> 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) =

924922 Unfold step inject

925923

926924 where

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

928926 inject x = do

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

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

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

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

932930 r <- bef x

933931 ref <- newIOFinalizer (aft r)

934932 return (r, ref)

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

937935

938936 {-# INLINE_LATE step #-}

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

940− res <- ftry $ step1 st

938+ res <- liftIO $ ftry $ step1 st

941939 case res of

942940 Right r -> case r of

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

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

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

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

953951 -- may leak the resource.

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

952+ liftIO $ clearingIOFinalizer ref (onExc v)

953+ r <- einject e

955954 return $ Skip (Left r)

956955 step (Left st) = do

957956 res <- estep st

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

10131012-- /See also 'after_'/

10141013--

10151014-- /Pre-release/

1016−{-# INLINE_NORMAL after #-}

1017−after :: MonadRunInIO m

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

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

1015+{-# INLINE_NORMAL afterIO #-}

1016+afterIO :: MonadIO m

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

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

10201019

10211020 where

10221021

10231022 inject x = do

10241023 s <- inject1 x

1025− ref <- newIOFinalizer (action x)

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

10261025 return (s, ref)

10271026

10281027 {-# INLINE_LATE step #-}

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

10721071 gbracket_ return MC.try action

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

10741073

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

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

10761075--

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

10781077-- after partial evaluation.

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

10801078--

10811079-- /Inhibits stream fusion/

10821080--

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

11151113-- /Inhibits stream fusion/

11161114--

11171115-- /Pre-release/

1118−{-# INLINE_NORMAL finally #-}

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

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

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

1116+{-# INLINE_NORMAL finallyIO #-}

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

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

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

11221120

11231121 where

11241122

11251123 inject x = do

11261124 s <- inject1 x

1127− ref <- newIOFinalizer (action x)

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

11281126 return (s, ref)

11291127

11301128 {-# INLINE_LATE step #-}

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

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

11451143 MC.throwM e))

11461144

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

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

11481146--

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

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

11511149-- after partial evaluation.

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

11531150--

11541151-- /Inhibits stream fusion/

11551152--

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

11961193-- /Inhibits stream fusion/

11971194--

11981195-- /Pre-release/

1199−{-# INLINE_NORMAL bracket #-}

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

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

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

1196+{-# INLINE_NORMAL bracketIO #-}

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

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

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

12031200

12041201 where

12051202

12061203 inject x = do

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

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

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

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

12101207 r <- bef x

12111208 ref <- newIOFinalizer (aft r)

12121209 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

116116import qualified Streamly.Data.Unfold as UF

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

118118 (Step(..), snoc, initialize)

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

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

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

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

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

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

163163{-# INLINE usingFile #-}

164164usingFile :: (MonadCatch m, MonadAsync m)

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

166−usingFile =

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

168− (liftIO . hClose)

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

169167

170168{-# INLINE usingFile2 #-}

171169usingFile2 :: (MonadCatch m, MonadAsync m)

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

173−usingFile2 = UF.bracket before after

171+usingFile2 = UF.bracketIO before after

174172

175173 where

176174

177175 before (x, file) = do

178− h <- liftIO $ openFile file ReadMode

176+ h <- openFile file ReadMode

179177 return (x, h)

180178

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

179+ after (_, h) = hClose h

182180

183181{-# INLINE usingFile3 #-}

184182usingFile3 :: (MonadCatch m, MonadAsync m)

185183 => 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

187185

188186 where

189187

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

191− h <- liftIO $ openFile file ReadMode

189+ h <- openFile file ReadMode

192190 return (x, y, z, h)

193191

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

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

195193

196194-------------------------------------------------------------------------------

197195-- Array IO (Input)

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

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

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

121121import qualified Streamly.Data.Fold as FL

122122import 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)

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

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

126126 (initialize, snoc, Step(..))

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

286286usingConnection :: (MonadCatch m, MonadAsync m)

287287 => Unfold m Socket a

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

289−usingConnection =

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

291− (liftIO . Net.close)

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

292290

293291-------------------------------------------------------------------------------

294292-- Connect (streams)

streamly.cabal+2

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

333333

334334 -- streamly-concurrent

335335 , Streamly.Internal.Data.Atomics

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

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

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

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

339340

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

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

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

342344 , Streamly.Internal.Data.Fold.Concurrent