d0d4850 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Refactor StreamD's exception ops to StreamD.Exceptions.

d0d4850Pranay Sashank6 years ago

src/Streamly/Internal/Data/Stream/StreamD.hs+1−389

compare with master

unifiedsplit

@@ -302,17 +302,12 @@ module Streamly.Internal.Data.Stream.StreamD

302where

303

304import Control.Concurrent (myThreadId)

305−import Control.Exception

306− (Exception, SomeException, mask_)

307import Control.Monad (void)

305import Control.Monad (void)

308−import Control.Monad.Catch (MonadCatch)

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

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

310import Control.Monad.Trans.Reader (ReaderT)

307import Control.Monad.Trans.Reader (ReaderT)

311import Control.Monad.Trans.State.Strict (StateT)

308import Control.Monad.Trans.State.Strict (StateT)

312import Control.Monad.Trans.Class (MonadTrans(lift))

309import Control.Monad.Trans.Class (MonadTrans(lift))

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

314import Data.Functor.Identity (Identity(..))

310import Data.Functor.Identity (Identity(..))

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

316import Data.Maybe (fromJust, isJust)

311import Data.Maybe (fromJust, isJust)

317import Streamly.Internal.Data.Time.Units

312import Streamly.Internal.Data.Time.Units

318 (TimeUnit64, toRelTime64, diffAbsTime64)

313 (TimeUnit64, toRelTime64, diffAbsTime64)

@@ -322,7 +317,6 @@ import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

322import qualified Streamly.Internal.Data.Pipe.Types as Pipe

317import qualified Streamly.Internal.Data.Pipe.Types as Pipe

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

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

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

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

325−import qualified Control.Monad.Catch as MC

326import qualified Control.Monad.Trans.Reader as Reader

320import qualified Control.Monad.Trans.Reader as Reader

327import qualified Control.Monad.Trans.State.Strict as State

321import qualified Control.Monad.Trans.State.Strict as State

328

322

@@ -336,6 +330,7 @@ import Streamly.Internal.Data.Stream.StreamD.Type

336import Streamly.Internal.Data.Stream.StreamD.Common (takeWhileM, takeWhile)

330import Streamly.Internal.Data.Stream.StreamD.Common (takeWhileM, takeWhile)

337import Streamly.Internal.Data.Stream.StreamD.Generate

331import Streamly.Internal.Data.Stream.StreamD.Generate

338import Streamly.Internal.Data.Stream.StreamD.Eliminate

332import Streamly.Internal.Data.Stream.StreamD.Eliminate

333+import Streamly.Internal.Data.Stream.StreamD.Exceptions

339import Streamly.Internal.Data.Stream.StreamD.Flatten

334import Streamly.Internal.Data.Stream.StreamD.Flatten

340import Streamly.Internal.Data.Stream.StreamD.SplitGroup

335import Streamly.Internal.Data.Stream.StreamD.SplitGroup

341import Streamly.Internal.Data.SVar

336import Streamly.Internal.Data.SVar

@@ -522,389 +517,6 @@ stripPrefix (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)

522mapM_ :: Monad m => (a -> m b) -> Stream m a -> m ()

517mapM_ :: Monad m => (a -> m b) -> Stream m a -> m ()

523mapM_ m = drain . mapM m

518mapM_ m = drain . mapM m

524

519

525−------------------------------------------------------------------------------

526−-- Exceptions

527−------------------------------------------------------------------------------

528−

529−data GbracketState s1 s2 v

530− = GBracketInit

531− | GBracketNormal s1 v

532− | GBracketException s2

533−

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

535−--

536−-- * alloc action @m c@ runs with async exceptions enabled

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

538−-- after partial evaluation.

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

540−--

541−-- /Inhibits stream fusion/

542−--

543−-- /Internal/

544−--

545−{-# INLINE_NORMAL gbracket_ #-}

546−gbracket_

547− :: Monad m

548− => m c -- ^ before

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

550− -> (c -> m d) -- ^ after, on normal stop

551− -> (c -> e -> Stream m b -> Stream m b) -- ^ on exception

552− -> (c -> Stream m b) -- ^ stream generator

553− -> Stream m b

554−gbracket_ bef exc aft fexc fnormal =

555− Stream step GBracketInit

556−

557− where

558−

559− {-# INLINE_LATE step #-}

560− step _ GBracketInit = do

561− r <- bef

562− return $ Skip $ GBracketNormal (fnormal r) r

563−

564− step gst (GBracketNormal (UnStream step1 st) v) = do

565− res <- exc $ step1 gst st

566− case res of

567− Right r -> case r of

568− Yield x s ->

569− return $ Yield x (GBracketNormal (Stream step1 s) v)

570− Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)

571− Stop -> aft v >> return Stop

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

573− Left e ->

574− return $ Skip (GBracketException (fexc v e (UnStream step1 st)))

575− step gst (GBracketException (UnStream step1 st)) = do

576− res <- step1 gst st

577− case res of

578− Yield x s -> return $ Yield x (GBracketException (Stream step1 s))

579− Skip s -> return $ Skip (GBracketException (Stream step1 s))

580− Stop -> return Stop

581−

582−------------------------------------------------------------------------------

583−-- Finalizers

584−------------------------------------------------------------------------------

585−

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

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

588−mkIOFinalizer f = do

589− mrun <- captureMonadState

590− return $

591− void $ do

592− _ <- runInIO mrun f

593− return ()

594−

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

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

597−runIORefFinalizerGC ref = do

598− res <- readIORef ref

599− case res of

600− Nothing -> return ()

601− Just f -> f

602−

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

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

605−-- value to deactivate the finalizer.

606−--

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

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

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

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

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

612−--

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

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

615−newFinalizedIORef finalizer = do

616− f <- mkIOFinalizer finalizer

617− ref <- liftIO $ newIORef $ Just f

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

619− return ref

620−

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

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

623−--

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

625−runIORefFinalizer ref = liftIO $ do

626− res <- readIORef ref

627− case res of

628− Nothing -> return ()

629− Just action -> do

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

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

632− -- atomically wrt async exceptions.

633− mask_ $ do

634− writeIORef ref Nothing

635− action

636−

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

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

639−withIORefFinalizer :: MonadBaseControl IO m

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

641−withIORefFinalizer ref action = do

642− control $ \runinio ->

643− mask_ $ do

644− writeIORef ref Nothing

645− runinio action

646−

647−------------------------------------------------------------------------------

648−

649−data GbracketIOState s1 s2 v wref

650− = GBracketIOInit

651− | GBracketIONormal s1 v wref

652− | GBracketIOException s2

653−

654−-- | Run the alloc action @m c@ with async exceptions disabled but keeping

655−-- blocking operations interruptible (see 'Control.Exception.mask'). Use the

656−-- output @c@ as input to @c -> Stream m b@ to generate an output stream. When

657−-- generating the stream use the supplied @try@ operation @forall s. m s -> m

658−-- (Either e s)@ to catch synchronous exceptions. If an exception occurs run

659−-- the exception handler @c -> e -> Stream m b -> m (Stream m b)@.

660−--

661−-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due

662−-- to a sync or async exception or if it gets garbage collected after a partial

663−-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.

664−--

665−-- 'gbracket' can express all other exception handling combinators.

666−--

667−-- /Inhibits stream fusion/

668−--

669−-- /Internal/

670−{-# INLINE_NORMAL gbracket #-}

671−gbracket

672− :: (MonadIO m, MonadBaseControl IO m)

673− => m c -- ^ before

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

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

676− -> (c -> e -> Stream m b -> m (Stream m b)) -- ^ on exception

677− -> (c -> Stream m b) -- ^ stream generator

678− -> Stream m b

679−gbracket bef exc aft fexc fnormal =

680− Stream step GBracketIOInit

681−

682− where

683−

684− -- If the stream is never evaluated the "aft" action will never be

685− -- called. For that to occur we will need the user of this API to pass a

686− -- weak pointer to us.

687− {-# INLINE_LATE step #-}

688− step _ GBracketIOInit = do

689− -- We mask asynchronous exceptions to make the execution

690− -- of 'bef' and the registration of 'aft' atomic.

691− -- A similar thing is done in the resourcet package: https://git.io/JvKV3

692− -- Tutorial: https://markkarpov.com/tutorial/exceptions.html

693− (r, ref) <- liftBaseOp_ mask_ $ do

694− r <- bef

695− ref <- newFinalizedIORef (aft r)

696− return (r, ref)

697− return $ Skip $ GBracketIONormal (fnormal r) r ref

698−

699− step gst (GBracketIONormal (UnStream step1 st) v ref) = do

700− res <- exc $ step1 gst st

701− case res of

702− Right r -> case r of

703− Yield x s ->

704− return $ Yield x (GBracketIONormal (Stream step1 s) v ref)

705− Skip s ->

706− return $ Skip (GBracketIONormal (Stream step1 s) v ref)

707− Stop -> do

708− runIORefFinalizer ref

709− return Stop

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

711− Left e -> do

712− -- Clearing of finalizer and running of exception handler must

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

714− -- the finalizer and have not run the exception handler then we

715− -- may leak the resource.

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

717− return $ Skip (GBracketIOException stream)

718− step gst (GBracketIOException (UnStream step1 st)) = do

719− res <- step1 gst st

720− case res of

721− Yield x s ->

722− return $ Yield x (GBracketIOException (Stream step1 s))

723− Skip s -> return $ Skip (GBracketIOException (Stream step1 s))

724− Stop -> return Stop

725−

726−-- | See 'Streamly.Internal.Data.Stream.IsStream.after_'.

727−--

728−{-# INLINE_NORMAL before #-}

729−before :: Monad m => m b -> Stream m a -> Stream m a

730−before action (Stream step state) = Stream step' Nothing

731−

732− where

733−

734− {-# INLINE_LATE step' #-}

735− step' _ Nothing = action >> return (Skip (Just state))

736−

737− step' gst (Just st) = do

738− res <- step gst st

739− case res of

740− Yield x s -> return $ Yield x (Just s)

741− Skip s -> return $ Skip (Just s)

742− Stop -> return Stop

743−

744−-- | See 'Streamly.Internal.Data.Stream.IsStream.after_'.

745−--

746−{-# INLINE_NORMAL after_ #-}

747−after_ :: Monad m => m b -> Stream m a -> Stream m a

748−after_ action (Stream step state) = Stream step' state

749−

750− where

751−

752− {-# INLINE_LATE step' #-}

753− step' gst st = do

754− res <- step gst st

755− case res of

756− Yield x s -> return $ Yield x s

757− Skip s -> return $ Skip s

758− Stop -> action >> return Stop

759−

760−-- | See 'Streamly.Internal.Data.Stream.IsStream.after'.

761−--

762−{-# INLINE_NORMAL after #-}

763−after :: (MonadIO m, MonadBaseControl IO m)

764− => m b -> Stream m a -> Stream m a

765−after action (Stream step state) = Stream step' Nothing

766−

767− where

768−

769− {-# INLINE_LATE step' #-}

770− step' _ Nothing = do

771− ref <- newFinalizedIORef action

772− return $ Skip $ Just (state, ref)

773− step' gst (Just (st, ref)) = do

774− res <- step gst st

775− case res of

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

777− Skip s -> return $ Skip (Just (s, ref))

778− Stop -> do

779− runIORefFinalizer ref

780− return Stop

781−

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

783−-- Error constructor in step.

784−--

785−-- | See 'Streamly.Internal.Data.Stream.IsStream.onException'.

786−--

787−{-# INLINE_NORMAL onException #-}

788−onException :: MonadCatch m => m b -> Stream m a -> Stream m a

789−onException action str =

790− gbracket_ (return ()) MC.try return

791− (\_ (e :: MC.SomeException) _ -> nilM (action >> MC.throwM e))

792− (\_ -> str)

793−

794−{-# INLINE_NORMAL _onException #-}

795−_onException :: MonadCatch m => m b -> Stream m a -> Stream m a

796−_onException action (Stream step state) = Stream step' state

797−

798− where

799−

800− {-# INLINE_LATE step' #-}

801− step' gst st = do

802− res <- step gst st `MC.onException` action

803− case res of

804− Yield x s -> return $ Yield x s

805− Skip s -> return $ Skip s

806− Stop -> return Stop

807−

808−-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket_'.

809−--

810−{-# INLINE_NORMAL bracket_ #-}

811−bracket_ :: MonadCatch m

812− => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a

813−bracket_ bef aft bet =

814− gbracket_ bef MC.try aft

815− (\a (e :: SomeException) _ -> nilM (aft a >> MC.throwM e)) bet

816−

817−-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket'.

818−--

819−{-# INLINE_NORMAL bracket #-}

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

821− => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a

822−bracket bef aft bet =

823− gbracket bef MC.try aft

824− (\a (e :: SomeException) _ -> aft a >> return (nilM (MC.throwM e))) bet

825−

826−data BracketState s v = BracketInit | BracketRun s v

827−

828−-- | Alternate (custom) implementation of 'bracket'.

829−--

830−{-# INLINE_NORMAL _bracket #-}

831−_bracket :: MonadCatch m

832− => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a

833−_bracket bef aft bet = Stream step' BracketInit

834−

835− where

836−

837− {-# INLINE_LATE step' #-}

838− step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))

839−

840− -- NOTE: It is important to use UnStream instead of the Stream pattern

841− -- here, otherwise we get huge perf degradation, see note in concatMap.

842− step' gst (BracketRun (UnStream step state) v) = do

843− -- res <- step gst state `MC.onException` aft v

844− res <- MC.try $ step gst state

845− case res of

846− Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop

847− Right r -> case r of

848− Yield x s -> return $ Yield x (BracketRun (Stream step s) v)

849− Skip s -> return $ Skip (BracketRun (Stream step s) v)

850− Stop -> aft v >> return Stop

851−

852−-- | See 'Streamly.Internal.Data.Stream.IsStream.finally_'.

853−--

854−{-# INLINE finally_ #-}

855−finally_ :: MonadCatch m => m b -> Stream m a -> Stream m a

856−finally_ action xs = bracket_ (return ()) (\_ -> action) (const xs)

857−

858−-- | See 'Streamly.Internal.Data.Stream.IsStream.finally'.

859−--

860−-- finally action xs = after action $ onException action xs

861−--

862−{-# INLINE finally #-}

863−finally :: (MonadAsync m, MonadCatch m) => m b -> Stream m a -> Stream m a

864−finally action xs = bracket (return ()) (\_ -> action) (const xs)

865−

866−-- | See 'Streamly.Internal.Data.Stream.IsStream.ghandle'.

867−--

868−{-# INLINE_NORMAL ghandle #-}

869−ghandle :: (MonadCatch m, Exception e)

870− => (e -> Stream m a -> Stream m a) -> Stream m a -> Stream m a

871−ghandle f str =

872− gbracket_ (return ()) MC.try return (\_ -> f) (\_ -> str)

873−

874−-- | See 'Streamly.Internal.Data.Stream.IsStream.handle'.

875−--

876−{-# INLINE_NORMAL handle #-}

877−handle :: (MonadCatch m, Exception e)

878− => (e -> Stream m a) -> Stream m a -> Stream m a

879−handle f str =

880− gbracket_ (return ()) MC.try return (\_ e _ -> f e) (\_ -> str)

881−

882−-- | Alternate (custom) implementation of 'handle'.

883−--

884−{-# INLINE_NORMAL _handle #-}

885−_handle :: (MonadCatch m, Exception e)

886− => (e -> Stream m a) -> Stream m a -> Stream m a

887−_handle f (Stream step state) = Stream step' (Left state)

888−

889− where

890−

891− {-# INLINE_LATE step' #-}

892− step' gst (Left st) = do

893− res <- MC.try $ step gst st

894− case res of

895− Left e -> return $ Skip $ Right (f e)

896− Right r -> case r of

897− Yield x s -> return $ Yield x (Left s)

898− Skip s -> return $ Skip (Left s)

899− Stop -> return Stop

900−

901− step' gst (Right (UnStream step1 st)) = do

902− res <- step1 gst st

903− case res of

904− Yield x s -> return $ Yield x (Right (Stream step1 s))

905− Skip s -> return $ Skip (Right (Stream step1 s))

906− Stop -> return Stop

907−

908-------------------------------------------------------------------------------

520-------------------------------------------------------------------------------

909-- General transformation

521-- General transformation

910-------------------------------------------------------------------------------

522-------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/StreamD/Common.hs+6

@@ -15,6 +15,7 @@ module Streamly.Internal.Data.Stream.StreamD.Common

15 FromSVarState (..)

16 , fromProducer

17 , fromPrimIORef

18+ , nilM

18 , takeWhileM

19 , takeWhileM

19 , takeWhile

20 , takeWhile

20 )

21 )

@@ -38,6 +39,11 @@ import Prelude hiding

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

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

39import Streamly.Internal.Data.SVar

40import Streamly.Internal.Data.SVar

42+-- | An empty 'Stream' with a side effect.

43+{-# INLINE_NORMAL nilM #-}

44+nilM :: Monad m => m b -> Stream m a

45+nilM m = Stream (\_ _ -> m >> return Stop) ()

46+

41data FromSVarState t m a =

47data FromSVarState t m a =

42 FromSVarInit

48 FromSVarInit

43 | FromSVarRead (SVar t m a)

49 | FromSVarRead (SVar t m a)

src/Streamly/Internal/Data/Stream/StreamD/Exceptions.hsnew file+435

@@ -0,0 +1,435 @@

1+#include "inline.hs"

3+-- |

4+-- Module : Streamly.Internal.Data.Stream.StreamD.Exceptions

6+-- License : BSD-3-Clause

7+-- Maintainer : streamly@composewell.com

8+-- Stability : experimental

9+-- Portability : GHC

10+

11+module Streamly.Internal.Data.Stream.StreamD.Exceptions

12+ (

13+ -- * Exceptions

14+ newFinalizedIORef

15+ , runIORefFinalizer

16+ , withIORefFinalizer

17+ , gbracket_

18+ , gbracket

19+ , before

20+ , after_

21+ , after

22+ , bracket_

23+ , bracket

24+ , onException

25+ , finally_

26+ , finally

27+ , ghandle

28+ , handle

29+ )

30+where

31+

32+

33+import Control.Exception

34+ (Exception, SomeException, mask_)

35+import Control.Monad (void)

36+import Control.Monad.Catch (MonadCatch)

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

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

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

40+

41+import qualified Control.Monad.Catch as MC

42+

43+import Prelude hiding

44+ ( map, mapM, mapM_, repeat, foldr, last, take, filter

45+ , takeWhile, drop, dropWhile, all, any, maximum, minimum, elem

46+ , notElem, null, head, tail, zipWith, lookup, foldr1, sequence

47+ , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat

48+ , reverse, iterate, splitAt)

49+import Streamly.Internal.Data.Stream.StreamD.Type

50+import Streamly.Internal.Data.Stream.StreamD.Common

51+import Streamly.Internal.Data.SVar

52+

53+

54+------------------------------------------------------------------------------

55+-- Exceptions

56+------------------------------------------------------------------------------

57+

58+data GbracketState s1 s2 v

59+ = GBracketInit

60+ | GBracketNormal s1 v

61+ | GBracketException s2

62+

63+-- | Like 'gbracket' but with following differences:

64+--

65+-- * alloc action @m c@ runs with async exceptions enabled

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

67+-- after partial evaluation.

68+-- * does not require a 'MonadAsync' constraint.

69+--

70+-- /Inhibits stream fusion/

71+--

72+-- /Internal/

73+--

74+{-# INLINE_NORMAL gbracket_ #-}

75+gbracket_

76+ :: Monad m

77+ => m c -- ^ before

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

79+ -> (c -> m d) -- ^ after, on normal stop

80+ -> (c -> e -> Stream m b -> Stream m b) -- ^ on exception

81+ -> (c -> Stream m b) -- ^ stream generator

82+ -> Stream m b

83+gbracket_ bef exc aft fexc fnormal =

84+ Stream step GBracketInit

85+

86+ where

87+

88+ {-# INLINE_LATE step #-}

89+ step _ GBracketInit = do

90+ r <- bef

91+ return $ Skip $ GBracketNormal (fnormal r) r

92+

93+ step gst (GBracketNormal (UnStream step1 st) v) = do

94+ res <- exc $ step1 gst st

95+ case res of

96+ Right r -> case r of

97+ Yield x s ->

98+ return $ Yield x (GBracketNormal (Stream step1 s) v)

99+ Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)

100+ Stop -> aft v >> return Stop

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

102+ Left e ->

103+ return $ Skip (GBracketException (fexc v e (UnStream step1 st)))

104+ step gst (GBracketException (UnStream step1 st)) = do

105+ res <- step1 gst st

106+ case res of

107+ Yield x s -> return $ Yield x (GBracketException (Stream step1 s))

108+ Skip s -> return $ Skip (GBracketException (Stream step1 s))

109+ Stop -> return Stop

110+

111+------------------------------------------------------------------------------

112+-- Finalizers

113+------------------------------------------------------------------------------

114+

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

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

117+mkIOFinalizer f = do

118+ mrun <- captureMonadState

119+ return $

120+ void $ do

121+ _ <- runInIO mrun f

122+ return ()

123+

124+-- | Run an IO action stored in a finalized IORef.

125+runIORefFinalizerGC :: IORef (Maybe (IO ())) -> IO ()

126+runIORefFinalizerGC ref = do

127+ res <- readIORef ref

128+ case res of

129+ Nothing -> return ()

130+ Just f -> f

131+

132+-- | Create an IORef holding a finalizer that is called automatically when

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

134+-- value to deactivate the finalizer.

135+--

136+-- Note: The finalizer is always run with the state of the monad captured at

137+-- the time of calling newFinalizedIORef. To run it on garbage collection we

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

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

140+-- but we want to keep both the cases consistent.

141+--

142+newFinalizedIORef :: (MonadIO m, MonadBaseControl IO m)

143+ => m a -> m (IORef (Maybe (IO ())))

144+newFinalizedIORef finalizer = do

145+ f <- mkIOFinalizer finalizer

146+ ref <- liftIO $ newIORef $ Just f

147+ _ <- liftIO $ mkWeakIORef ref (runIORefFinalizerGC ref)

148+ return ref

149+

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

151+-- only once. Note, the action runs with async exceptions masked.

152+--

153+runIORefFinalizer :: MonadIO m => IORef (Maybe (IO ())) -> m ()

154+runIORefFinalizer ref = liftIO $ do

155+ res <- readIORef ref

156+ case res of

157+ Nothing -> return ()

158+ Just action -> do

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

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

161+ -- atomically wrt async exceptions.

162+ mask_ $ do

163+ writeIORef ref Nothing

164+ action

165+

166+-- | Run an action clearing the finalizer IORef atomically wrt async

167+-- exceptions. The action is run with async exceptions masked.

168+withIORefFinalizer :: MonadBaseControl IO m

169+ => IORef (Maybe (IO ())) -> m a -> m a

170+withIORefFinalizer ref action = do

171+ control $ \runinio ->

172+ mask_ $ do

173+ writeIORef ref Nothing

174+ runinio action

175+

176+------------------------------------------------------------------------------

177+

178+data GbracketIOState s1 s2 v wref

179+ = GBracketIOInit

180+ | GBracketIONormal s1 v wref

181+ | GBracketIOException s2

182+

183+-- | Run the alloc action @m c@ with async exceptions disabled but keeping

184+-- blocking operations interruptible (see 'Control.Exception.mask'). Use the

185+-- output @c@ as input to @c -> Stream m b@ to generate an output stream. When

186+-- generating the stream use the supplied @try@ operation @forall s. m s -> m

187+-- (Either e s)@ to catch synchronous exceptions. If an exception occurs run

188+-- the exception handler @c -> e -> Stream m b -> m (Stream m b)@.

189+--

190+-- The cleanup action @c -> m d@, runs whenever the stream ends normally, due

191+-- to a sync or async exception or if it gets garbage collected after a partial

192+-- lazy evaluation. See 'bracket' for the semantics of the cleanup action.

193+--

194+-- 'gbracket' can express all other exception handling combinators.

195+--

196+-- /Inhibits stream fusion/

197+--

198+-- /Internal/

199+{-# INLINE_NORMAL gbracket #-}

200+gbracket

201+ :: (MonadIO m, MonadBaseControl IO m)

202+ => m c -- ^ before

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

204+ -> (c -> m d) -- ^ after, on normal stop or GC

205+ -> (c -> e -> Stream m b -> m (Stream m b)) -- ^ on exception

206+ -> (c -> Stream m b) -- ^ stream generator

207+ -> Stream m b

208+gbracket bef exc aft fexc fnormal =

209+ Stream step GBracketIOInit

210+

211+ where

212+

213+ -- If the stream is never evaluated the "aft" action will never be

214+ -- called. For that to occur we will need the user of this API to pass a

215+ -- weak pointer to us.

216+ {-# INLINE_LATE step #-}

217+ step _ GBracketIOInit = do

218+ -- We mask asynchronous exceptions to make the execution

219+ -- of 'bef' and the registration of 'aft' atomic.

220+ -- A similar thing is done in the resourcet package: https://git.io/JvKV3

221+ -- Tutorial: https://markkarpov.com/tutorial/exceptions.html

222+ (r, ref) <- liftBaseOp_ mask_ $ do

223+ r <- bef

224+ ref <- newFinalizedIORef (aft r)

225+ return (r, ref)

226+ return $ Skip $ GBracketIONormal (fnormal r) r ref

227+

228+ step gst (GBracketIONormal (UnStream step1 st) v ref) = do

229+ res <- exc $ step1 gst st

230+ case res of

231+ Right r -> case r of

232+ Yield x s ->

233+ return $ Yield x (GBracketIONormal (Stream step1 s) v ref)

234+ Skip s ->

235+ return $ Skip (GBracketIONormal (Stream step1 s) v ref)

236+ Stop -> do

237+ runIORefFinalizer ref

238+ return Stop

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

240+ Left e -> do

241+ -- Clearing of finalizer and running of exception handler must

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

243+ -- the finalizer and have not run the exception handler then we

244+ -- may leak the resource.

245+ stream <- withIORefFinalizer ref (fexc v e (UnStream step1 st))

246+ return $ Skip (GBracketIOException stream)

247+ step gst (GBracketIOException (UnStream step1 st)) = do

248+ res <- step1 gst st

249+ case res of

250+ Yield x s ->

251+ return $ Yield x (GBracketIOException (Stream step1 s))

252+ Skip s -> return $ Skip (GBracketIOException (Stream step1 s))

253+ Stop -> return Stop

254+

255+-- | See 'Streamly.Internal.Data.Stream.IsStream.after_'.

256+--

257+{-# INLINE_NORMAL before #-}

258+before :: Monad m => m b -> Stream m a -> Stream m a

259+before action (Stream step state) = Stream step' Nothing

260+

261+ where

262+

263+ {-# INLINE_LATE step' #-}

264+ step' _ Nothing = action >> return (Skip (Just state))

265+

266+ step' gst (Just st) = do

267+ res <- step gst st

268+ case res of

269+ Yield x s -> return $ Yield x (Just s)

270+ Skip s -> return $ Skip (Just s)

271+ Stop -> return Stop

272+

273+-- | See 'Streamly.Internal.Data.Stream.IsStream.after_'.

274+--

275+{-# INLINE_NORMAL after_ #-}

276+after_ :: Monad m => m b -> Stream m a -> Stream m a

277+after_ action (Stream step state) = Stream step' state

278+

279+ where

280+

281+ {-# INLINE_LATE step' #-}

282+ step' gst st = do

283+ res <- step gst st

284+ case res of

285+ Yield x s -> return $ Yield x s

286+ Skip s -> return $ Skip s

287+ Stop -> action >> return Stop

288+

289+-- | See 'Streamly.Internal.Data.Stream.IsStream.after'.

290+--

291+{-# INLINE_NORMAL after #-}

292+after :: (MonadIO m, MonadBaseControl IO m)

293+ => m b -> Stream m a -> Stream m a

294+after action (Stream step state) = Stream step' Nothing

295+

296+ where

297+

298+ {-# INLINE_LATE step' #-}

299+ step' _ Nothing = do

300+ ref <- newFinalizedIORef action

301+ return $ Skip $ Just (state, ref)

302+ step' gst (Just (st, ref)) = do

303+ res <- step gst st

304+ case res of

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

306+ Skip s -> return $ Skip (Just (s, ref))

307+ Stop -> do

308+ runIORefFinalizer ref

309+ return Stop

310+

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

312+-- Error constructor in step.

313+--

314+-- | See 'Streamly.Internal.Data.Stream.IsStream.onException'.

315+--

316+{-# INLINE_NORMAL onException #-}

317+onException :: MonadCatch m => m b -> Stream m a -> Stream m a

318+onException action str =

319+ gbracket_ (return ()) MC.try return

320+ (\_ (e :: MC.SomeException) _ -> nilM (action >> MC.throwM e))

321+ (\_ -> str)

322+

323+{-# INLINE_NORMAL _onException #-}

324+_onException :: MonadCatch m => m b -> Stream m a -> Stream m a

325+_onException action (Stream step state) = Stream step' state

326+

327+ where

328+

329+ {-# INLINE_LATE step' #-}

330+ step' gst st = do

331+ res <- step gst st `MC.onException` action

332+ case res of

333+ Yield x s -> return $ Yield x s

334+ Skip s -> return $ Skip s

335+ Stop -> return Stop

336+

337+-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket_'.

338+--

339+{-# INLINE_NORMAL bracket_ #-}

340+bracket_ :: MonadCatch m

341+ => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a

342+bracket_ bef aft bet =

343+ gbracket_ bef MC.try aft

344+ (\a (e :: SomeException) _ -> nilM (aft a >> MC.throwM e)) bet

345+

346+-- | See 'Streamly.Internal.Data.Stream.IsStream.bracket'.

347+--

348+{-# INLINE_NORMAL bracket #-}

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

350+ => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a

351+bracket bef aft bet =

352+ gbracket bef MC.try aft

353+ (\a (e :: SomeException) _ -> aft a >> return (nilM (MC.throwM e))) bet

354+

355+data BracketState s v = BracketInit | BracketRun s v

356+

357+-- | Alternate (custom) implementation of 'bracket'.

358+--

359+{-# INLINE_NORMAL _bracket #-}

360+_bracket :: MonadCatch m

361+ => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a

362+_bracket bef aft bet = Stream step' BracketInit

363+

364+ where

365+

366+ {-# INLINE_LATE step' #-}

367+ step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))

368+

369+ -- NOTE: It is important to use UnStream instead of the Stream pattern

370+ -- here, otherwise we get huge perf degradation, see note in concatMap.

371+ step' gst (BracketRun (UnStream step state) v) = do

372+ -- res <- step gst state `MC.onException` aft v

373+ res <- MC.try $ step gst state

374+ case res of

375+ Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop

376+ Right r -> case r of

377+ Yield x s -> return $ Yield x (BracketRun (Stream step s) v)

378+ Skip s -> return $ Skip (BracketRun (Stream step s) v)

379+ Stop -> aft v >> return Stop

380+

381+-- | See 'Streamly.Internal.Data.Stream.IsStream.finally_'.

382+--

383+{-# INLINE finally_ #-}

384+finally_ :: MonadCatch m => m b -> Stream m a -> Stream m a

385+finally_ action xs = bracket_ (return ()) (\_ -> action) (const xs)

386+

387+-- | See 'Streamly.Internal.Data.Stream.IsStream.finally'.

388+--

389+-- finally action xs = after action $ onException action xs

390+--

391+{-# INLINE finally #-}

392+finally :: (MonadAsync m, MonadCatch m) => m b -> Stream m a -> Stream m a

393+finally action xs = bracket (return ()) (\_ -> action) (const xs)

394+

395+-- | See 'Streamly.Internal.Data.Stream.IsStream.ghandle'.

396+--

397+{-# INLINE_NORMAL ghandle #-}

398+ghandle :: (MonadCatch m, Exception e)

399+ => (e -> Stream m a -> Stream m a) -> Stream m a -> Stream m a

400+ghandle f str =

401+ gbracket_ (return ()) MC.try return (\_ -> f) (\_ -> str)

402+

403+-- | See 'Streamly.Internal.Data.Stream.IsStream.handle'.

404+--

405+{-# INLINE_NORMAL handle #-}

406+handle :: (MonadCatch m, Exception e)

407+ => (e -> Stream m a) -> Stream m a -> Stream m a

408+handle f str =

409+ gbracket_ (return ()) MC.try return (\_ e _ -> f e) (\_ -> str)

410+

411+-- | Alternate (custom) implementation of 'handle'.

412+--

413+{-# INLINE_NORMAL _handle #-}

414+_handle :: (MonadCatch m, Exception e)

415+ => (e -> Stream m a) -> Stream m a -> Stream m a

416+_handle f (Stream step state) = Stream step' (Left state)

417+

418+ where

419+

420+ {-# INLINE_LATE step' #-}

421+ step' gst (Left st) = do

422+ res <- MC.try $ step gst st

423+ case res of

424+ Left e -> return $ Skip $ Right (f e)

425+ Right r -> case r of

426+ Yield x s -> return $ Yield x (Left s)

427+ Skip s -> return $ Skip (Left s)

428+ Stop -> return Stop

429+

430+ step' gst (Right (UnStream step1 st)) = do

431+ res <- step1 gst st

432+ case res of

433+ Yield x s -> return $ Yield x (Right (Stream step1 s))

434+ Skip s -> return $ Skip (Right (Stream step1 s))

435+ Stop -> return Stop

src/Streamly/Internal/Data/Stream/StreamD/Generate.hs−5

@@ -107,11 +107,6 @@ import Streamly.Internal.Data.SVar

107nil :: Monad m => Stream m a

108nil = Stream (\_ _ -> return Stop) ()

109

110−-- | An empty 'Stream' with a side effect.

111−{-# INLINE_NORMAL nilM #-}

112−nilM :: Monad m => m b -> Stream m a

113−nilM m = Stream (\_ _ -> m >> return Stop) ()

114−

115{-# INLINE_NORMAL consM #-}

110{-# INLINE_NORMAL consM #-}

116consM :: Monad m => m a -> Stream m a -> Stream m a

111consM :: Monad m => m a -> Stream m a -> Stream m a

117consM m (Stream step state) = Stream step1 Nothing

112consM m (Stream step state) = Stream step1 Nothing

streamly.cabal+1

@@ -457,6 +457,7 @@ library

457 , Streamly.Internal.Data.Stream.StreamD.SplitGroup

458 , Streamly.Internal.Data.Stream.StreamD.Eliminate

459 , Streamly.Internal.Data.Stream.StreamD.Flatten

460+ , Streamly.Internal.Data.Stream.StreamD.Exceptions

460 , Streamly.Internal.Data.Stream.StreamDK.Type

461 , Streamly.Internal.Data.Stream.StreamDK.Type

461 , Streamly.Internal.Data.Stream.StreamDK

462 , Streamly.Internal.Data.Stream.StreamDK

462 , Streamly.Internal.Data.Stream.Enumeration

463 , Streamly.Internal.Data.Stream.Enumeration