02e22d4 · streamly · git.iko.soy

git.iko.soy/streamly

Factor out IOFinalizer module

From StreamD/Exception module. This is a common functionality used in Stream as well as in Unfold and can be used generally.

commit02e22d4d25081bb089c6782519c6c5bfc80fb5c7 authorHarendra Kumar <harendra@composewell.com> dateSat 23 Jan 2021, 20:37 (6 years ago) parent9bcf1d6

4 files changed+131−101

compare with master

unifiedsplit

src/Streamly/Internal/Data/IOFinalizer.hs+103 src/Streamly/Internal/Data/Stream/StreamD/Exception.hs+9−84 src/Streamly/Internal/Data/Unfold.hs+18−17 streamly.cabal+1

src/Streamly/Internal/Data/IOFinalizer.hsnew file+103

@@ -0,0 +1,103 @@

1+-- |

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

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

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 Control.Monad.Trans.Control (MonadBaseControl, control)

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

26+import Streamly.Internal.Data.SVar (captureMonadState, runInIO)

27+

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

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

30+-- cleared prematurely.

31+--

32+-- You can hold a reference to the finalizer in your data structure, if the

33+-- data structure gets garbage collected the finalizer will be called.

34+--

35+-- It is implemented using 'mkWeakIORef'.

36+--

37+-- /Internal/

38+newtype IOFinalizer = IOFinalizer (IORef (Maybe (IO ())))

39+

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

41+mkIOFinalizer :: MonadBaseControl IO m => m b -> m (IO ())

42+mkIOFinalizer f = do

43+ mrun <- captureMonadState

44+ return $

45+ void $ do

46+ _ <- runInIO mrun f

47+ return ()

48+

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

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

51+runFinalizerGC ref = do

52+ res <- readIORef ref

53+ case res of

54+ Nothing -> return ()

55+ Just f -> f

56+

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

58+-- it is garbage collected.

59+--

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

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

62+--

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

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

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

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

67+--

68+-- /Internal/

69+newIOFinalizer :: (MonadIO m, MonadBaseControl IO m) => m a -> m IOFinalizer

70+newIOFinalizer finalizer = do

71+ f <- mkIOFinalizer finalizer

72+ ref <- liftIO $ newIORef $ Just f

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

74+ return $ IOFinalizer ref

75+

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

77+-- never runs again. Note, the finalizing action runs with async exceptions

78+-- masked.

79+--

80+-- /Internal/

81+runIOFinalizer :: MonadIO m => IOFinalizer -> m ()

82+runIOFinalizer (IOFinalizer ref) = liftIO $ do

83+ res <- readIORef ref

84+ case res of

85+ Nothing -> return ()

86+ Just action -> do

87+ -- if an async exception comes after writing 'Nothing' then the

88+ -- finalizing action will never be run. We need to do this

89+ -- atomically wrt async exceptions.

90+ mask_ $ do

91+ writeIORef ref Nothing

92+ action

93+

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

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

96+--

97+-- /Internal/

98+clearingIOFinalizer :: MonadBaseControl IO m => IOFinalizer -> m a -> m a

99+clearingIOFinalizer (IOFinalizer ref) action = do

100+ control $ \runinio ->

101+ mask_ $ do

102+ writeIORef ref Nothing

103+ runinio action

src/Streamly/Internal/Data/Stream/StreamD/Exception.hs+9−84

@@ -8,13 +8,7 @@

99module Streamly.Internal.Data.Stream.StreamD.Exception

1010 (

11− -- * Finalizers

12− newFinalizedIORef

13− , runIORefFinalizer

14− , withIORefFinalizer

15−

16− -- * Combinators

17− , gbracket_

11+ gbracket_

1812 , gbracket

1913 , before

2014 , after_

@@ -32,86 +26,17 @@ where

3226#include "inline.hs"

3327

3428import Control.Exception (Exception, SomeException, mask_)

35−import Control.Monad (void)

3629import Control.Monad.Catch (MonadCatch)

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

38−import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_, control)

39−import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, IORef)

31+import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)

32+import Streamly.Internal.Data.IOFinalizer

33+ (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

4034

4135import qualified Control.Monad.Catch as MC

4236

4337import Streamly.Internal.Data.Stream.StreamD.Type

4438import Streamly.Internal.Data.SVar

4539

46−------------------------------------------------------------------------------

47−-- Finalizers

48−------------------------------------------------------------------------------

49−

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

51−mkIOFinalizer :: MonadBaseControl IO m => m b -> m (IO ())

52−mkIOFinalizer f = do

53− mrun <- captureMonadState

54− return $

55− void $ do

56− _ <- runInIO mrun f

57− return ()

58−

59−-- | Run an IO action stored in a finalized IORef.

60−runIORefFinalizerGC :: IORef (Maybe (IO ())) -> IO ()

61−runIORefFinalizerGC ref = do

62− res <- readIORef ref

63− case res of

64− Nothing -> return ()

65− Just f -> f

66−

67−-- | Create an IORef holding a finalizer that is called automatically when

68−-- the IORef is garbage collected. The IORef can be written to with a 'Nothing'

69−-- value to deactivate the finalizer.

70−--

71−-- Note: The finalizer is always run with the state of the monad captured at

72−-- the time of calling newFinalizedIORef. To run it on garbage collection we

73−-- have no option but to take a snapshot of the monadic state at some point of

74−-- time. For normal case we could run it with the current state of the monad

75−-- but we want to keep both the cases consistent.

76−--

77−newFinalizedIORef :: (MonadIO m, MonadBaseControl IO m)

78− => m a -> m (IORef (Maybe (IO ())))

79−newFinalizedIORef finalizer = do

80− f <- mkIOFinalizer finalizer

81− ref <- liftIO $ newIORef $ Just f

82− _ <- liftIO $ mkWeakIORef ref (runIORefFinalizerGC ref)

83− return ref

84−

85−-- | Run the finalizer stored in an IORef and deactivate it so that it is run

86−-- only once. Note, the action runs with async exceptions masked.

87−--

88−runIORefFinalizer :: MonadIO m => IORef (Maybe (IO ())) -> m ()

89−runIORefFinalizer ref = liftIO $ do

90− res <- readIORef ref

91− case res of

92− Nothing -> return ()

93− Just action -> do

94− -- if an async exception comes after writing 'Nothing' then the

95− -- finalizing action will never be run. We need to do this

96− -- atomically wrt async exceptions.

97− mask_ $ do

98− writeIORef ref Nothing

99− action

100−

101−-- | Run an action clearing the finalizer IORef atomically wrt async

102−-- exceptions. The action is run with async exceptions masked.

103−withIORefFinalizer :: MonadBaseControl IO m

104− => IORef (Maybe (IO ())) -> m a -> m a

105−withIORefFinalizer ref action = do

106− control $ \runinio ->

107− mask_ $ do

108− writeIORef ref Nothing

109− runinio action

110−

111−------------------------------------------------------------------------------

112−-- Exception handling combinators.

113−------------------------------------------------------------------------------

114−

11540data GbracketState s1 s2 v

11641 = GBracketInit

11742 | GBracketNormal s1 v

@@ -211,7 +136,7 @@ gbracket bef exc aft fexc fnormal =

211136 -- Tutorial: https://markkarpov.com/tutorial/exceptions.html

212137 (r, ref) <- liftBaseOp_ mask_ $ do

213138 r <- bef

214− ref <- newFinalizedIORef (aft r)

139+ ref <- newIOFinalizer (aft r)

215140 return (r, ref)

216141 return $ Skip $ GBracketIONormal (fnormal r) r ref

217142

@@ -224,7 +149,7 @@ gbracket bef exc aft fexc fnormal =

224149 Skip s ->

225150 return $ Skip (GBracketIONormal (Stream step1 s) v ref)

226151 Stop -> do

227− runIORefFinalizer ref

152+ runIOFinalizer ref

228153 return Stop

229154 -- XXX Do not handle async exceptions, just rethrow them.

230155 Left e -> do

@@ -232,7 +157,7 @@ gbracket bef exc aft fexc fnormal =

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

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

234159 -- may leak the resource.

235− stream <- withIORefFinalizer ref (fexc v e (UnStream step1 st))

160+ stream <- clearingIOFinalizer ref (fexc v e (UnStream step1 st))

236161 return $ Skip (GBracketIOException stream)

237162 step gst (GBracketIOException (UnStream step1 st)) = do

238163 res <- step1 gst st

@@ -287,7 +212,7 @@ after action (Stream step state) = Stream step' Nothing

287212

288213 {-# INLINE_LATE step' #-}

289214 step' _ Nothing = do

290− ref <- newFinalizedIORef action

215+ ref <- newIOFinalizer action

291216 return $ Skip $ Just (state, ref)

292217 step' gst (Just (st, ref)) = do

293218 res <- step gst st

@@ -295,7 +220,7 @@ after action (Stream step state) = Stream step' Nothing

295220 Yield x s -> return $ Yield x (Just (s, ref))

296221 Skip s -> return $ Skip (Just (s, ref))

297222 Stop -> do

298− runIORefFinalizer ref

223+ runIOFinalizer ref

299224 return Stop

300225

301226-- XXX For high performance error checks in busy streams we may need another

src/Streamly/Internal/Data/Unfold.hs+18−17

@@ -149,15 +149,17 @@ import Control.Monad ((>=>), when)

149149import Control.Monad.Catch (MonadCatch)

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

151151import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)

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

152153import Data.Maybe (isNothing)

153154import Data.Void (Void)

154−import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)

155−import GHC.Types (SPEC(..))

156155import Fusion.Plugin.Types (Fuse(..))

157−import Streamly.Internal.Data.Stream.StreamD.Type (Stream(..), Step(..))

158−import Streamly.Internal.Data.Unfold.Types (Unfold(..))

156+import GHC.Types (SPEC(..))

159157import Streamly.Internal.Data.Fold.Types (Fold(..))

158+import Streamly.Internal.Data.IOFinalizer

159+ (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

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

160161import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

162+import Streamly.Internal.Data.Unfold.Types (Unfold(..))

161163import System.Mem (performMajorGC)

162164

163165import qualified Prelude

@@ -166,8 +168,7 @@ import qualified Data.Tuple as Tuple

166168import qualified Streamly.Internal.Data.Fold.Types as FL

167169import qualified Streamly.Internal.Data.Stream.StreamK as K (uncons)

168170import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

169−import qualified Streamly.Internal.Data.Stream.StreamD as D

170− (unfold, newFinalizedIORef, runIORefFinalizer, withIORefFinalizer)

171+import qualified Streamly.Internal.Data.Stream.StreamD as D (unfold)

171172

172173import Streamly.Internal.Data.SVar

173174import Prelude

@@ -1154,7 +1155,7 @@ gbracket bef exc aft (Unfold estep einject) (Unfold step1 inject1) =

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

11551156 (r, ref) <- liftBaseOp_ mask_ $ do

11561157 r <- bef x

1157− ref <- D.newFinalizedIORef (aft r)

1158+ ref <- newIOFinalizer (aft r)

11581159 return (r, ref)

11591160 s <- inject1 r

11601161 return $ Right (s, r, ref)

@@ -1167,7 +1168,7 @@ gbracket bef exc aft (Unfold estep einject) (Unfold step1 inject1) =

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

11681169 Skip s -> return $ Skip (Right (s, v, ref))

11691170 Stop -> do

1170− D.runIORefFinalizer ref

1171+ runIOFinalizer ref

11711172 return Stop

11721173 -- XXX Do not handle async exceptions, just rethrow them.

11731174 Left e -> do

@@ -1175,7 +1176,7 @@ gbracket bef exc aft (Unfold estep einject) (Unfold step1 inject1) =

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

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

11771178 -- may leak the resource.

1178− r <- D.withIORefFinalizer ref (einject (v, e))

1179+ r <- clearingIOFinalizer ref (einject (v, e))

11791180 return $ Skip (Left r)

11801181 step (Left st) = do

11811182 res <- estep st

@@ -1263,7 +1264,7 @@ after action (Unfold step1 inject1) = Unfold step inject

12631264

12641265 inject x = do

12651266 s <- inject1 x

1266− ref <- D.newFinalizedIORef (action x)

1267+ ref <- newIOFinalizer (action x)

12671268 return (s, ref)

12681269

12691270 {-# INLINE_LATE step #-}

@@ -1273,7 +1274,7 @@ after action (Unfold step1 inject1) = Unfold step inject

12731274 Yield x s -> return $ Yield x (s, ref)

12741275 Skip s -> return $ Skip (s, ref)

12751276 Stop -> do

1276− D.runIORefFinalizer ref

1277+ runIOFinalizer ref

12771278 return Stop

12781279

12791280{-# INLINE_NORMAL _onException #-}

@@ -1365,17 +1366,17 @@ finally action (Unfold step1 inject1) = Unfold step inject

13651366

13661367 inject x = do

13671368 s <- inject1 x

1368− ref <- D.newFinalizedIORef (action x)

1369+ ref <- newIOFinalizer (action x)

13691370 return (s, ref)

13701371

13711372 {-# INLINE_LATE step #-}

13721373 step (st, ref) = do

1373− res <- step1 st `MC.onException` D.runIORefFinalizer ref

1374+ res <- step1 st `MC.onException` runIOFinalizer ref

13741375 case res of

13751376 Yield x s -> return $ Yield x (s, ref)

13761377 Skip s -> return $ Skip (s, ref)

13771378 Stop -> do

1378− D.runIORefFinalizer ref

1379+ runIOFinalizer ref

13791380 return Stop

13801381

13811382{-# INLINE_NORMAL _bracket #-}

@@ -1449,19 +1450,19 @@ bracket bef aft (Unfold step1 inject1) = Unfold step inject

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

14501451 (r, ref) <- liftBaseOp_ mask_ $ do

14511452 r <- bef x

1452− ref <- D.newFinalizedIORef (aft r)

1453+ ref <- newIOFinalizer (aft r)

14531454 return (r, ref)

14541455 s <- inject1 r

14551456 return (s, ref)

14561457

14571458 {-# INLINE_LATE step #-}

14581459 step (st, ref) = do

1459− res <- step1 st `MC.onException` D.runIORefFinalizer ref

1460+ res <- step1 st `MC.onException` runIOFinalizer ref

14601461 case res of

14611462 Yield x s -> return $ Yield x (s, ref)

14621463 Skip s -> return $ Skip (s, ref)

14631464 Stop -> do

1464− D.runIORefFinalizer ref

1465+ runIOFinalizer ref

14651466 return Stop

14661467

14671468-- | When unfolding @Unfold m a b@ if an exception @e@ occurs, unfold @e@ using

streamly.cabal+1

@@ -420,6 +420,7 @@ library

420420 , Streamly.Internal.Data.Either.Strict

421421 , Streamly.Internal.Foreign.Malloc

422422 , Streamly.Internal.Data.Atomics

423+ , Streamly.Internal.Data.IOFinalizer

423424 , Streamly.Internal.Data.Time

424425 , Streamly.Internal.Data.Time.TimeSpec

425426 , Streamly.Internal.Data.Time.Units