02e22d4 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Factor out IOFinalizer module

02e22d4Harendra Kumar6 years ago

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

compare with master

unifiedsplit

@@ -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 @@

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

10 (

11− -- * Finalizers

12− newFinalizedIORef

13− , runIORefFinalizer

14− , withIORefFinalizer

15−

16− -- * Combinators

17− , gbracket_

11+ gbracket_

18 , gbracket

12 , gbracket

19 , before

13 , before

20 , after_

14 , after_

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

32#include "inline.hs"

26#include "inline.hs"

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

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

35−import Control.Monad (void)

36import Control.Monad.Catch (MonadCatch)

29import Control.Monad.Catch (MonadCatch)

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

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

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

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

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

32+import Streamly.Internal.Data.IOFinalizer

33+ (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

41import qualified Control.Monad.Catch as MC

35import qualified Control.Monad.Catch as MC

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

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

44import Streamly.Internal.Data.SVar

38import Streamly.Internal.Data.SVar

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−

115data GbracketState s1 s2 v

40data GbracketState s1 s2 v

116 = GBracketInit

41 = GBracketInit

117 | GBracketNormal s1 v

42 | GBracketNormal s1 v

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

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

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

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

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

213 r <- bef

138 r <- bef

214− ref <- newFinalizedIORef (aft r)

139+ ref <- newIOFinalizer (aft r)

215 return (r, ref)

140 return (r, ref)

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

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

217

142

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

224 Skip s ->

149 Skip s ->

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

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

226 Stop -> do

151 Stop -> do

227− runIORefFinalizer ref

152+ runIOFinalizer ref

228 return Stop

153 return Stop

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

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

230 Left e -> do

155 Left e -> do

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

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

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

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

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

234 -- may leak the resource.

159 -- may leak the resource.

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

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

236 return $ Skip (GBracketIOException stream)

161 return $ Skip (GBracketIOException stream)

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

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

238 res <- step1 gst st

163 res <- step1 gst st

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

287

212

288 {-# INLINE_LATE step' #-}

213 {-# INLINE_LATE step' #-}

289 step' _ Nothing = do

214 step' _ Nothing = do

290− ref <- newFinalizedIORef action

215+ ref <- newIOFinalizer action

291 return $ Skip $ Just (state, ref)

216 return $ Skip $ Just (state, ref)

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

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

293 res <- step gst st

218 res <- step gst st

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

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

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

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

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

297 Stop -> do

222 Stop -> do

298− runIORefFinalizer ref

223+ runIOFinalizer ref

299 return Stop

224 return Stop

300

225

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

226-- 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)

149import Control.Monad.Catch (MonadCatch)

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

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

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

152import Data.Maybe (isNothing)

153import Data.Maybe (isNothing)

153import Data.Void (Void)

154import Data.Void (Void)

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

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

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

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

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

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

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

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

157import 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(..))

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

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

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

161import System.Mem (performMajorGC)

163import System.Mem (performMajorGC)

162

164

163import qualified Prelude

165import qualified Prelude

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

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

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

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

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

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

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

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

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

170− (unfold, newFinalizedIORef, runIORefFinalizer, withIORefFinalizer)

171

172

172import Streamly.Internal.Data.SVar

173import Streamly.Internal.Data.SVar

173import Prelude

174import Prelude

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

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

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

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

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

1156 r <- bef x

1157 r <- bef x

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

1158+ ref <- newIOFinalizer (aft r)

1158 return (r, ref)

1159 return (r, ref)

1159 s <- inject1 r

1160 s <- inject1 r

1160 return $ Right (s, r, ref)

1161 return $ Right (s, r, ref)

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

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

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

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

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

1169 Stop -> do

1170 Stop -> do

1170− D.runIORefFinalizer ref

1171+ runIOFinalizer ref

1171 return Stop

1172 return Stop

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

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

1173 Left e -> do

1174 Left e -> do

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

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

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

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

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

1177 -- may leak the resource.

1178 -- may leak the resource.

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

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

1179 return $ Skip (Left r)

1180 return $ Skip (Left r)

1180 step (Left st) = do

1181 step (Left st) = do

1181 res <- estep st

1182 res <- estep st

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

1263

1264

1264 inject x = do

1265 inject x = do

1265 s <- inject1 x

1266 s <- inject1 x

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

1267+ ref <- newIOFinalizer (action x)

1267 return (s, ref)

1268 return (s, ref)

1268

1269

1269 {-# INLINE_LATE step #-}

1270 {-# INLINE_LATE step #-}

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

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

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

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

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

1275 Stop -> do

1276 Stop -> do

1276− D.runIORefFinalizer ref

1277+ runIOFinalizer ref

1277 return Stop

1278 return Stop

1278

1279

1279{-# INLINE_NORMAL _onException #-}

1280{-# INLINE_NORMAL _onException #-}

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

1365

1366

1366 inject x = do

1367 inject x = do

1367 s <- inject1 x

1368 s <- inject1 x

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

1369+ ref <- newIOFinalizer (action x)

1369 return (s, ref)

1370 return (s, ref)

1370

1371

1371 {-# INLINE_LATE step #-}

1372 {-# INLINE_LATE step #-}

1372 step (st, ref) = do

1373 step (st, ref) = do

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

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

1374 case res of

1375 case res of

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

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

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

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

1377 Stop -> do

1378 Stop -> do

1378− D.runIORefFinalizer ref

1379+ runIOFinalizer ref

1379 return Stop

1380 return Stop

1380

1381

1381{-# INLINE_NORMAL _bracket #-}

1382{-# INLINE_NORMAL _bracket #-}

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

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

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

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

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

1451 r <- bef x

1452 r <- bef x

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

1453+ ref <- newIOFinalizer (aft r)

1453 return (r, ref)

1454 return (r, ref)

1454 s <- inject1 r

1455 s <- inject1 r

1455 return (s, ref)

1456 return (s, ref)

1456

1457

1457 {-# INLINE_LATE step #-}

1458 {-# INLINE_LATE step #-}

1458 step (st, ref) = do

1459 step (st, ref) = do

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

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

1460 case res of

1461 case res of

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

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

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

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

1463 Stop -> do

1464 Stop -> do

1464− D.runIORefFinalizer ref

1465+ runIOFinalizer ref

1465 return Stop

1466 return Stop

1466

1467

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

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

streamly.cabal+1

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

420 , Streamly.Internal.Data.Either.Strict

421 , Streamly.Internal.Foreign.Malloc

422 , Streamly.Internal.Data.Atomics

423+ , Streamly.Internal.Data.IOFinalizer

423 , Streamly.Internal.Data.Time

424 , Streamly.Internal.Data.Time

424 , Streamly.Internal.Data.Time.TimeSpec

425 , Streamly.Internal.Data.Time.TimeSpec

425 , Streamly.Internal.Data.Time.Units

426 , Streamly.Internal.Data.Time.Units