babb151 · streamly · git.iko.soy

src/Streamly/Internal/Control/Concurrent.hs+47−5

compare with master

unifiedsplit

@@ -13,14 +13,29 @@

13module Streamly.Internal.Control.Concurrent

14 (

15 MonadAsync

16+ , MonadRunInIO

16 , RunInIO(..)

17 , RunInIO(..)

17− , captureMonadState

18+ , askRunInIO

19+ , withRunInIO

20+ , withRunInIONoRestore

21+ , restoreM

18 )

22 )

19where

23where

20

24

21import Control.Monad.Catch (MonadThrow)

25import Control.Monad.Catch (MonadThrow)

26+

27+#ifdef USE_UNLIFTIO

28+import Control.Monad.IO.Unlift (MonadUnliftIO(..), UnliftIO(..), askUnliftIO)

29+#else

22−import Control.Monad.IO.Class (MonadIO(..))

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

23−import Control.Monad.Trans.Control (MonadBaseControl, control, StM)

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

32+#endif

33+

34+#ifdef USE_UNLIFTIO

35+type MonadRunInIO m = MonadUnliftIO m

36+#else

37+type MonadRunInIO m = (MonadIO m, MonadBaseControl IO m)

38+#endif

24

39

25-- /Since: 0.8.0 ("Streamly.Prelude")/

40-- /Since: 0.8.0 ("Streamly.Prelude")/

26--

41--

@@ -31,9 +46,18 @@ import Control.Monad.Trans.Control (MonadBaseControl, control, StM)

31-- /Since: 0.1.0 ("Streamly")/

46-- /Since: 0.1.0 ("Streamly")/

32--

47--

33-- @since 0.8.0

48-- @since 0.8.0

49+#ifdef USE_UNLIFTIO

50+type MonadAsync m = (MonadUnliftIO m, MonadThrow m)

51+#else

34type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

52type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

53+#endif

54+

35

55

56+#ifdef USE_UNLIFTIO

57+newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO b }

58+#else

36newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

59newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

60+#endif

37

61

38-- | When we run computations concurrently, we completely isolate the state of

62-- | When we run computations concurrently, we completely isolate the state of

39-- the concurrent computations from the parent computation. The invariant is

63-- the concurrent computations from the parent computation. The invariant is

@@ -42,5 +66,23 @@ newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

42-- a concurrent computation in the parent thread, otherwise it may affect the

66-- a concurrent computation in the parent thread, otherwise it may affect the

43-- state of the parent which is against the defined semantics of concurrent

67-- state of the parent which is against the defined semantics of concurrent

44-- execution.

68-- execution.

45−captureMonadState :: MonadBaseControl IO m => m (RunInIO m)

69+askRunInIO :: MonadRunInIO m => m (RunInIO m)

70+#ifdef USE_UNLIFTIO

71+askRunInIO = fmap (\(UnliftIO run) -> RunInIO run) askUnliftIO

72+#else

46−captureMonadState = control $ \run -> run (return $ RunInIO run)

73+askRunInIO = control $ \run -> run (return $ RunInIO run)

74+#endif

75+

76+#ifdef USE_UNLIFTIO

77+withRunInIONoRestore :: MonadRunInIO m => ((forall a. m a -> IO a) -> IO b) -> m b

78+withRunInIONoRestore = withRunInIO

79+

80+restoreM :: MonadRunInIO m => a -> m a

81+restoreM = return

82+#else

83+withRunInIO :: MonadRunInIO m => ((forall a. m a -> IO (StM m a)) -> IO (StM m b)) -> m b

84+withRunInIO = control

85+

86+withRunInIONoRestore :: MonadRunInIO m => ((forall a. m a -> IO (StM m a)) -> IO b) -> m b

87+withRunInIONoRestore = liftBaseWith

88+#endif

src/Streamly/Internal/Control/ForkLifted.hs+7−9

@@ -16,26 +16,24 @@ where

16

17import Control.Concurrent (ThreadId, forkIO)

18import Control.Exception (SomeException(..), catch, mask)

19−import Control.Monad.IO.Class (MonadIO(..))

20−import Control.Monad.Trans.Control (MonadBaseControl, control, liftBaseDiscard)

21import Data.Functor (void)

19import Data.Functor (void)

22−import Streamly.Internal.Control.Concurrent (RunInIO(..))

20+import Streamly.Internal.Control.Concurrent (MonadRunInIO, RunInIO(..), withRunInIO, withRunInIONoRestore)

23import Streamly.Internal.Control.ForkIO (rawForkIO, forkManagedWith)

21import Streamly.Internal.Control.ForkIO (rawForkIO, forkManagedWith)

24

22

25-- | Fork a thread to run the given computation, installing the provided

23-- | Fork a thread to run the given computation, installing the provided

26−-- exception handler. Lifted to any monad with 'MonadBaseControl IO m'

24+-- exception handler. Lifted to any monad with 'MonadRunInIO m'

27-- capability.

25-- capability.

28--

26--

29-- TODO: the RunInIO argument can be removed, we can directly pass the action

27-- TODO: the RunInIO argument can be removed, we can directly pass the action

30-- as "mrun action" instead.

28-- as "mrun action" instead.

31{-# INLINE doFork #-}

29{-# INLINE doFork #-}

32−doFork :: MonadBaseControl IO m

30+doFork :: MonadRunInIO m

33 => m ()

31 => m ()

34 -> RunInIO m

32 -> RunInIO m

35 -> (SomeException -> IO ())

33 -> (SomeException -> IO ())

36 -> m ThreadId

34 -> m ThreadId

37doFork action (RunInIO mrun) exHandler =

35doFork action (RunInIO mrun) exHandler =

38− control $ \run ->

36+ withRunInIO $ \run ->

39 mask $ \restore -> do

37 mask $ \restore -> do

40 tid <- rawForkIO $ catch (restore $ void $ mrun action)

38 tid <- rawForkIO $ catch (restore $ void $ mrun action)

41 exHandler

39 exHandler

@@ -44,12 +42,12 @@ doFork action (RunInIO mrun) exHandler =

44-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.

42-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.

45--

43--

46{-# INLINABLE fork #-}

44{-# INLINABLE fork #-}

47−fork :: MonadBaseControl IO m => m () -> m ThreadId

45+fork :: MonadRunInIO m => m () -> m ThreadId

48−fork = liftBaseDiscard forkIO

46+fork m = withRunInIONoRestore $ \run -> forkIO $ void $ run m

49

47

50-- | Fork a thread that is automatically killed as soon as the reference to the

48-- | Fork a thread that is automatically killed as soon as the reference to the

51-- returned threadId is garbage collected.

49-- returned threadId is garbage collected.

52--

50--

53{-# INLINABLE forkManaged #-}

51{-# INLINABLE forkManaged #-}

54−forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId

52+forkManaged :: MonadRunInIO m => m () -> m ThreadId

55forkManaged = forkManagedWith fork

53forkManaged = forkManagedWith fork

src/Streamly/Internal/Data/Fold/Async.hs+2−3

@@ -21,8 +21,7 @@ import Control.Concurrent.MVar (MVar, newMVar, swapMVar, readMVar)

21import Control.Exception (SomeException(..), catch, mask)

22import Control.Monad (void)

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

24−import Control.Monad.Trans.Control (control)

25−import Streamly.Internal.Control.Concurrent (MonadAsync)

24+import Streamly.Internal.Control.Concurrent (MonadAsync, withRunInIO)

26import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))

25import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))

27

26

28import Streamly.Internal.Data.Fold.Type

27import Streamly.Internal.Data.Fold.Type

@@ -65,7 +64,7 @@ takeInterval n (Fold step initial done) = Fold step' initial' done'

65 Partial s -> do

64 Partial s -> do

66 mv <- liftIO $ newMVar False

65 mv <- liftIO $ newMVar False

67 t <-

66 t <-

68− control $ \run ->

67+ withRunInIO $ \run ->

69 mask $ \restore -> do

68 mask $ \restore -> do

70 tid <-

69 tid <-

71 forkIO

70 forkIO

src/Streamly/Internal/Data/IOFinalizer.hs+6−7

@@ -21,9 +21,8 @@ where

21import Control.Exception (mask_)

22import Control.Monad (void)

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

24−import Control.Monad.Trans.Control (MonadBaseControl, control)

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

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

26−import Streamly.Internal.Control.Concurrent (captureMonadState, runInIO)

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

27

26

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

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

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

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

@@ -38,9 +37,9 @@ import Streamly.Internal.Control.Concurrent (captureMonadState, runInIO)

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

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

39

38

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

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

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

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

42mkIOFinalizer f = do

41mkIOFinalizer f = do

43− mrun <- captureMonadState

42+ mrun <- askRunInIO

44 return $

43 return $

45 void $ do

44 void $ do

46 _ <- runInIO mrun f

45 _ <- runInIO mrun f

@@ -66,7 +65,7 @@ runFinalizerGC ref = do

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

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

67--

66--

68-- /Pre-release/

67-- /Pre-release/

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

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

70newIOFinalizer finalizer = do

69newIOFinalizer finalizer = do

71 f <- mkIOFinalizer finalizer

70 f <- mkIOFinalizer finalizer

72 ref <- liftIO $ newIORef $ Just f

71 ref <- liftIO $ newIORef $ Just f

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

95-- action is run with async exceptions masked.

94-- action is run with async exceptions masked.

96--

95--

97-- /Pre-release/

96-- /Pre-release/

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

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

99clearingIOFinalizer (IOFinalizer ref) action = do

98clearingIOFinalizer (IOFinalizer ref) action = do

100− control $ \runinio ->

99+ withRunInIO $ \runinio ->

101 mask_ $ do

100 mask_ $ do

102 writeIORef ref Nothing

101 writeIORef ref Nothing

103 runinio action

102 runinio action

src/Streamly/Internal/Data/SVar.hs+3−3

@@ -48,7 +48,7 @@ import Data.Heap (Heap, Entry(..))

48import Data.IORef (newIORef, readIORef)

49import Data.IORef (IORef, atomicModifyIORef)

50import Streamly.Internal.Control.Concurrent

51− (MonadAsync, captureMonadState, RunInIO(..))

51+ (MonadAsync, askRunInIO, RunInIO)

52import Streamly.Internal.Data.Atomics

53 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)

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

@@ -432,7 +432,7 @@ newAheadVar :: MonadAsync m

432 -> m ())

433 -> m (SVar t m a)

434newAheadVar st m wloop = do

435− mrun <- captureMonadState

435+ mrun <- askRunInIO

436 sv <- liftIO $ getAheadSVar st wloop mrun

437 sendFirstWorker sv m

438

@@ -533,5 +533,5 @@ getParallelSVar ss st mrun = do

533newParallelVar :: MonadAsync m

534 => SVarStopStyle -> State t m a -> m (SVar t m a)

535newParallelVar ss st = do

536− mrun <- captureMonadState

536+ mrun <- askRunInIO

537 liftIO $ getParallelSVar ss st mrun

src/Streamly/Internal/Data/SVar/Dispatch.hs+2−2

@@ -49,7 +49,7 @@ import Data.IORef (IORef, modifyIORef, newIORef, readIORef, writeIORef)

49#if __GLASGOW_HASKELL__ < 804

50import Data.Semigroup ((<>))

51#endif

52−import Streamly.Internal.Control.Concurrent (MonadAsync, captureMonadState)

52+import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

53import Streamly.Internal.Control.ForkLifted (doFork)

54import Streamly.Internal.Data.Atomics

55 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

@@ -707,7 +707,7 @@ sendFirstWorker sv m = do

707 -- Note: We must have all the work on the queue before sending the

708 -- pushworker, otherwise the pushworker may exit before we even get a

709 -- chance to push.

710− runIn <- captureMonadState

710+ runIn <- askRunInIO

711 liftIO $ enqueue sv (runIn, m)

712 case yieldRateInfo sv of

713 Nothing -> pushWorker 0 sv

src/Streamly/Internal/Data/SVar/Type.hs+1−1

@@ -62,7 +62,7 @@ import Data.Kind (Type)

62import Data.Set (Set)

63

64import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))

65−import Streamly.Internal.Control.Concurrent (RunInIO(..))

65+import Streamly.Internal.Control.Concurrent (RunInIO)

66

67newtype Count = Count Int64

68 deriving ( Eq

src/Streamly/Internal/Data/Stream/Ahead.hs+11−12

@@ -39,7 +39,6 @@ import Control.Monad.IO.Class (MonadIO(..))

39import Control.Monad.Reader.Class (MonadReader(..))

40import Control.Monad.State.Class (MonadState(..))

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

42−import Control.Monad.Trans.Control (MonadBaseControl (..))

43import Data.Heap (Heap, Entry(..))

42import Data.Heap (Heap, Entry(..))

44import Data.IORef (IORef, readIORef, atomicModifyIORef, writeIORef)

43import Data.IORef (IORef, readIORef, atomicModifyIORef, writeIORef)

45import Data.Maybe (fromJust)

44import Data.Maybe (fromJust)

@@ -51,7 +50,7 @@ import GHC.Exts (inline)

51import qualified Data.Heap as H

50import qualified Data.Heap as H

52

51

53import Streamly.Internal.Control.Concurrent

52import Streamly.Internal.Control.Concurrent

54− (MonadAsync, RunInIO(..), captureMonadState)

53+ (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)

55import Streamly.Internal.Data.Stream.Serial (SerialT(..))

54import Streamly.Internal.Data.Stream.Serial (SerialT(..))

56import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

55import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

57

56

@@ -229,7 +228,7 @@ abortExecution q sv winfo m = do

229-- In both cases we give the drainer a chance to run more often.

228-- In both cases we give the drainer a chance to run more often.

230--

229--

231processHeap

230processHeap

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

231+ :: MonadRunInIO m

233 => IORef ([Stream m a], Int)

232 => IORef ([Stream m a], Int)

234 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

233 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

235 -> State Stream m a

234 -> State Stream m a

@@ -326,7 +325,7 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

326 stop

325 stop

327 r

326 r

328 else do

327 else do

329− runIn <- captureMonadState

328+ runIn <- askRunInIO

330 let ent = Entry seqNo (AheadEntryStream (runIn, r))

329 let ent = Entry seqNo (AheadEntryStream (runIn, r))

331 liftIO $ do

330 liftIO $ do

332 requeueOnHeapTop heap ent seqNo

331 requeueOnHeapTop heap ent seqNo

@@ -339,7 +338,7 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

339

338

340{-# NOINLINE drainHeap #-}

339{-# NOINLINE drainHeap #-}

341drainHeap

340drainHeap

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

341+ :: MonadRunInIO m

343 => IORef ([Stream m a], Int)

342 => IORef ([Stream m a], Int)

344 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

343 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

345 -> State Stream m a

344 -> State Stream m a

@@ -357,7 +356,7 @@ data HeapStatus = HContinue | HStop

357data WorkerStatus = Continue | Suspend

356data WorkerStatus = Continue | Suspend

358

357

359processWithoutToken

358processWithoutToken

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

359+ :: MonadRunInIO m

361 => IORef ([Stream m a], Int)

360 => IORef ([Stream m a], Int)

362 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

361 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

363 -> State Stream m a

362 -> State Stream m a

@@ -381,7 +380,7 @@ processWithoutToken q heap st sv winfo m seqNo = do

381 r <- liftIO $ mrun $

380 r <- liftIO $ mrun $

382 K.foldStreamShared st

381 K.foldStreamShared st

383 (\a r -> do

382 (\a r -> do

384− runIn <- captureMonadState

383+ runIn <- askRunInIO

385 toHeap $ AheadEntryStream (runIn, K.cons a r))

384 toHeap $ AheadEntryStream (runIn, K.cons a r))

386 (toHeap . AheadEntryPure)

385 (toHeap . AheadEntryPure)

387 stop

386 stop

@@ -432,7 +431,7 @@ processWithoutToken q heap st sv winfo m seqNo = do

432data TokenWorkerStatus = TokenContinue Int | TokenSuspend

431data TokenWorkerStatus = TokenContinue Int | TokenSuspend

433

432

434processWithToken

433processWithToken

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

434+ :: MonadRunInIO m

436 => IORef ([Stream m a], Int)

435 => IORef ([Stream m a], Int)

437 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

436 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

438 -> State Stream m a

437 -> State Stream m a

@@ -484,7 +483,7 @@ processWithToken q heap st sv winfo action sno = do

484 stop

483 stop

485 r

484 r

486 else do

485 else do

487− runIn <- captureMonadState

486+ runIn <- askRunInIO

488 let ent = Entry seqNo (AheadEntryStream (runIn, r))

487 let ent = Entry seqNo (AheadEntryStream (runIn, r))

489 liftIO $ requeueOnHeapTop heap ent seqNo

488 liftIO $ requeueOnHeapTop heap ent seqNo

490 liftIO $ incrementYieldLimit sv

489 liftIO $ incrementYieldLimit sv

@@ -546,7 +545,7 @@ processWithToken q heap st sv winfo action sno = do

546-- XXX we can remove the sv parameter as it can be derived from st

545-- XXX we can remove the sv parameter as it can be derived from st

547

546

548workLoopAhead

547workLoopAhead

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

548+ :: MonadRunInIO m

550 => IORef ([Stream m a], Int)

549 => IORef ([Stream m a], Int)

551 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

550 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

552 -> State Stream m a

551 -> State Stream m a

@@ -610,7 +609,7 @@ forkSVarAhead m1 m2 = K.mkStream $ \st yld sng stp -> do

610 K.foldStream st yld sng stp $ getSerialT (fromSVar sv)

609 K.foldStream st yld sng stp $ getSerialT (fromSVar sv)

611 where

610 where

612 concurrently ma mb = K.mkStream $ \st yld sng stp -> do

611 concurrently ma mb = K.mkStream $ \st yld sng stp -> do

613− runInIO <- captureMonadState

612+ runInIO <- askRunInIO

614 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)

613 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)

615 K.foldStream st yld sng stp ma

614 K.foldStream st yld sng stp ma

616

615

@@ -619,7 +618,7 @@ aheadK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

619aheadK m1 m2 = K.mkStream $ \st yld sng stp ->

618aheadK m1 m2 = K.mkStream $ \st yld sng stp ->

620 case streamVar st of

619 case streamVar st of

621 Just sv | svarStyle sv == AheadVar -> do

620 Just sv | svarStyle sv == AheadVar -> do

622− runInIO <- captureMonadState

621+ runInIO <- askRunInIO

623 liftIO $ enqueue sv (runInIO, m2)

622 liftIO $ enqueue sv (runInIO, m2)

624 -- Always run the left side on a new SVar to avoid complexity in

623 -- Always run the left side on a new SVar to avoid complexity in

625 -- sequencing results. This means the left side cannot further

624 -- sequencing results. This means the left side cannot further

src/Streamly/Internal/Data/Stream/Async.hs+13−14

@@ -41,7 +41,6 @@ where

41import Control.Concurrent (myThreadId)

42import Control.Monad.Base (MonadBase(..), liftBaseDefault)

43import Control.Monad.Catch (MonadThrow, throwM)

44−import Control.Monad.Trans.Control (MonadBaseControl (..))

45import Control.Concurrent.MVar (newEmptyMVar)

44import Control.Concurrent.MVar (newEmptyMVar)

46-- import Control.Monad.Error.Class (MonadError(..))

45-- import Control.Monad.Error.Class (MonadError(..))

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

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

@@ -59,7 +58,7 @@ import Prelude hiding (map)

59import qualified Data.Set as S

58import qualified Data.Set as S

60

59

61import Streamly.Internal.Control.Concurrent

60import Streamly.Internal.Control.Concurrent

62− (MonadAsync, RunInIO(..), captureMonadState)

61+ (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)

63import Streamly.Internal.Data.Atomics

62import Streamly.Internal.Data.Atomics

64 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

63 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

65import Streamly.Internal.Data.Stream.Serial (SerialT (..))

64import Streamly.Internal.Data.Stream.Serial (SerialT (..))

@@ -104,7 +103,7 @@ data WorkerStatus = Continue | Suspend

104

103

105{-# INLINE workLoopLIFO #-}

104{-# INLINE workLoopLIFO #-}

106workLoopLIFO

105workLoopLIFO

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

106+ :: MonadRunInIO m

108 => IORef [(RunInIO m, Stream m a)]

107 => IORef [(RunInIO m, Stream m a)]

109 -> State Stream m a

108 -> State Stream m a

110 -> SVar Stream m a

109 -> SVar Stream m a

@@ -143,7 +142,7 @@ workLoopLIFO q st sv winfo = run

143 if res

142 if res

144 then K.foldStreamShared st yieldk single (return Continue) r

143 then K.foldStreamShared st yieldk single (return Continue) r

145 else do

144 else do

146− runInIO <- captureMonadState

145+ runInIO <- askRunInIO

147 liftIO $ enqueueLIFO sv q (runInIO, r)

146 liftIO $ enqueueLIFO sv q (runInIO, r)

148 return Suspend

147 return Suspend

149

148

@@ -158,7 +157,7 @@ workLoopLIFO q st sv winfo = run

158-- make a check every time.

157-- make a check every time.

159{-# INLINE workLoopLIFOLimited #-}

158{-# INLINE workLoopLIFOLimited #-}

160workLoopLIFOLimited

159workLoopLIFOLimited

161− :: forall m a. (MonadIO m, MonadBaseControl IO m)

160+ :: forall m a. MonadRunInIO m

162 => IORef [(RunInIO m, Stream m a)]

161 => IORef [(RunInIO m, Stream m a)]

163 -> State Stream m a

162 -> State Stream m a

164 -> SVar Stream m a

163 -> SVar Stream m a

@@ -209,7 +208,7 @@ workLoopLIFOLimited q st sv winfo = run

209 if res && yieldLimitOk

208 if res && yieldLimitOk

210 then K.foldStreamShared st yieldk single incrContinue r

209 then K.foldStreamShared st yieldk single incrContinue r

211 else do

210 else do

212− runInIO <- captureMonadState

211+ runInIO <- askRunInIO

213 liftIO $ incrementYieldLimit sv

212 liftIO $ incrementYieldLimit sv

214 liftIO $ enqueueLIFO sv q (runInIO, r)

213 liftIO $ enqueueLIFO sv q (runInIO, r)

215 return Suspend

214 return Suspend

@@ -240,7 +239,7 @@ enqueueFIFO sv q m = do

240

239

241{-# INLINE workLoopFIFO #-}

240{-# INLINE workLoopFIFO #-}

242workLoopFIFO

241workLoopFIFO

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

242+ :: MonadRunInIO m

244 => LinkedQueue (RunInIO m, Stream m a)

243 => LinkedQueue (RunInIO m, Stream m a)

245 -> State Stream m a

244 -> State Stream m a

246 -> SVar Stream m a

245 -> SVar Stream m a

@@ -273,13 +272,13 @@ workLoopFIFO q st sv winfo = run

273 -- yielding.

272 -- yielding.

274 yieldk a r = do

273 yieldk a r = do

275 res <- liftIO $ sendYield sv winfo (ChildYield a)

274 res <- liftIO $ sendYield sv winfo (ChildYield a)

276− runInIO <- captureMonadState

275+ runInIO <- askRunInIO

277 liftIO $ enqueueFIFO sv q (runInIO, r)

276 liftIO $ enqueueFIFO sv q (runInIO, r)

278 return $ if res then Continue else Suspend

277 return $ if res then Continue else Suspend

279

278

280{-# INLINE workLoopFIFOLimited #-}

279{-# INLINE workLoopFIFOLimited #-}

281workLoopFIFOLimited

280workLoopFIFOLimited

282− :: forall m a. (MonadIO m, MonadBaseControl IO m)

281+ :: forall m a. MonadRunInIO m

283 => LinkedQueue (RunInIO m, Stream m a)

282 => LinkedQueue (RunInIO m, Stream m a)

284 -> State Stream m a

283 -> State Stream m a

285 -> SVar Stream m a

284 -> SVar Stream m a

@@ -316,7 +315,7 @@ workLoopFIFOLimited q st sv winfo = run

316

315

317 yieldk a r = do

316 yieldk a r = do

318 res <- liftIO $ sendYield sv winfo (ChildYield a)

317 res <- liftIO $ sendYield sv winfo (ChildYield a)

319− runInIO <- captureMonadState

318+ runInIO <- askRunInIO

320 liftIO $ enqueueFIFO sv q (runInIO, r)

319 liftIO $ enqueueFIFO sv q (runInIO, r)

321 yieldLimitOk <- liftIO $ decrementYieldLimit sv

320 yieldLimitOk <- liftIO $ decrementYieldLimit sv

322 if res && yieldLimitOk

321 if res && yieldLimitOk

@@ -533,7 +532,7 @@ getFifoSVar st mrun = do

533newAsyncVar :: MonadAsync m

532newAsyncVar :: MonadAsync m

534 => State Stream m a -> Stream m a -> m (SVar Stream m a)

533 => State Stream m a -> Stream m a -> m (SVar Stream m a)

535newAsyncVar st m = do

534newAsyncVar st m = do

536− mrun <- captureMonadState

535+ mrun <- askRunInIO

537 sv <- liftIO $ getLifoSVar st mrun

536 sv <- liftIO $ getLifoSVar st mrun

538 sendFirstWorker sv m

537 sendFirstWorker sv m

539

538

@@ -573,7 +572,7 @@ mkAsyncD m = D.Stream step Nothing

573newWAsyncVar :: MonadAsync m

572newWAsyncVar :: MonadAsync m

574 => State Stream m a -> Stream m a -> m (SVar Stream m a)

573 => State Stream m a -> Stream m a -> m (SVar Stream m a)

575newWAsyncVar st m = do

574newWAsyncVar st m = do

576− mrun <- captureMonadState

575+ mrun <- askRunInIO

577 sv <- liftIO $ getFifoSVar st mrun

576 sv <- liftIO $ getFifoSVar st mrun

578 -- XXX Use just Stream and IO in all the functions below

577 -- XXX Use just Stream and IO in all the functions below

579 -- XXX pass mrun instead of calling captureMonadState again inside it

578 -- XXX pass mrun instead of calling captureMonadState again inside it

@@ -653,7 +652,7 @@ forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do

653 K.foldStream st yld sng stp $ getSerialT $ fromSVar sv

652 K.foldStream st yld sng stp $ getSerialT $ fromSVar sv

654 where

653 where

655 concurrently ma mb = K.mkStream $ \st yld sng stp -> do

654 concurrently ma mb = K.mkStream $ \st yld sng stp -> do

656− runInIO <- captureMonadState

655+ runInIO <- askRunInIO

657 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)

656 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)

658 K.foldStreamShared st yld sng stp ma

657 K.foldStreamShared st yld sng stp ma

659

658

@@ -663,7 +662,7 @@ joinStreamVarAsync :: MonadAsync m

663joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->

662joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->

664 case streamVar st of

663 case streamVar st of

665 Just sv | svarStyle sv == style -> do

664 Just sv | svarStyle sv == style -> do

666− runInIO <- captureMonadState

665+ runInIO <- askRunInIO

667 liftIO $ enqueue sv (runInIO, m2)

666 liftIO $ enqueue sv (runInIO, m2)

668 K.foldStreamShared st yld sng stp m1

667 K.foldStreamShared st yld sng stp m1

669 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)

668 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)

src/Streamly/Internal/Data/Stream/IsStream/Exception.hs+2−4

@@ -25,10 +25,8 @@ where

25

26import Control.Exception (Exception)

27import Control.Monad.Catch (MonadCatch)

28−import Control.Monad.IO.Class (MonadIO(..))

29−import Control.Monad.Trans.Control (MonadBaseControl)

30import Data.Map.Strict (Map)

28import Data.Map.Strict (Map)

31−import Streamly.Internal.Control.Concurrent (MonadAsync)

29+import Streamly.Internal.Control.Concurrent (MonadRunInIO, MonadAsync)

32import Streamly.Internal.Data.Stream.IsStream.Type

30import Streamly.Internal.Data.Stream.IsStream.Type

33 (IsStream(..), fromStreamD, toStreamD)

31 (IsStream(..), fromStreamD, toStreamD)

34

32

@@ -83,7 +81,7 @@ after_ action xs = fromStreamD $ D.after_ action $ toStreamD xs

83-- @since 0.7.0

81-- @since 0.7.0

84--

82--

85{-# INLINE after #-}

83{-# INLINE after #-}

86−after :: (IsStream t, MonadIO m, MonadBaseControl IO m)

84+after :: (IsStream t, MonadRunInIO m)

87 => m b -> t m a -> t m a

85 => m b -> t m a -> t m a

88after action xs = fromStreamD $ D.after action $ toStreamD xs

86after action xs = fromStreamD $ D.after action $ toStreamD xs

89

87

src/Streamly/Internal/Data/Stream/SVar/Generate.hs+2−2

@@ -37,7 +37,7 @@ import Control.Monad.Catch (throwM)

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

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

39import Data.Maybe (isNothing)

40−import Streamly.Internal.Control.Concurrent (MonadAsync, captureMonadState)

40+import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

41import Streamly.Internal.Data.Stream.Serial (SerialT(..))

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

43import System.Mem (performMajorGC)

@@ -90,7 +90,7 @@ import Test.Inspection (inspect, hasNoTypeClassesExcept)

90-- be read back from the SVar using 'fromSVar'.

91toSVar :: MonadAsync m => SVar SerialT m a -> SerialT m a -> m ()

92toSVar sv m = do

93− runIn <- captureMonadState

93+ runIn <- askRunInIO

94 liftIO $ enqueue sv (runIn, m)

95 done <- allThreadsDone sv

96 -- XXX This is safe only when called from the consumer thread or when no

src/Streamly/Internal/Data/Stream/StreamD/Exception.hs+4−6

@@ -28,11 +28,9 @@ where

28

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

30import Control.Monad.Catch (MonadCatch)

31−import Control.Monad.IO.Class (MonadIO(..))

32−import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)

33import Data.Map.Strict (Map)

31import Data.Map.Strict (Map)

34import GHC.Exts (inline)

32import GHC.Exts (inline)

35−import Streamly.Internal.Control.Concurrent (MonadAsync)

33+import Streamly.Internal.Control.Concurrent (MonadRunInIO, MonadAsync, withRunInIO)

36import Streamly.Internal.Data.IOFinalizer

34import Streamly.Internal.Data.IOFinalizer

37 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

35 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

38

36

@@ -119,7 +117,7 @@ data GbracketIOState s1 s2 v wref

119-- /Pre-release/

117-- /Pre-release/

120{-# INLINE_NORMAL gbracket #-}

118{-# INLINE_NORMAL gbracket #-}

121gbracket

119gbracket

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

120+ :: MonadRunInIO m

123 => m c -- ^ before

121 => m c -- ^ before

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

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

125 -> (c -> m d1) -- ^ on normal stop

123 -> (c -> m d1) -- ^ on normal stop

@@ -141,7 +139,7 @@ gbracket bef exc aft gc fexc fnormal =

141 -- of 'bef' and the registration of 'aft' atomic.

139 -- of 'bef' and the registration of 'aft' atomic.

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

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

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

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

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

142+ (r, ref) <- withRunInIO $ \run -> mask_ $ run $ do

145 r <- bef

143 r <- bef

146 ref <- newIOFinalizer (gc r)

144 ref <- newIOFinalizer (gc r)

147 return (r, ref)

145 return (r, ref)

@@ -210,7 +208,7 @@ after_ action (Stream step state) = Stream step' state

210-- | See 'Streamly.Internal.Data.Stream.IsStream.after'.

208-- | See 'Streamly.Internal.Data.Stream.IsStream.after'.

211--

209--

212{-# INLINE_NORMAL after #-}

210{-# INLINE_NORMAL after #-}

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

211+after :: MonadRunInIO m

214 => m b -> Stream m a -> Stream m a

212 => m b -> Stream m a -> Stream m a

215after action (Stream step state) = Stream step' Nothing

213after action (Stream step state) = Stream step' Nothing

216

214

src/Streamly/Internal/Data/Unfold.hs+5−7

@@ -248,11 +248,9 @@ where

248

249import Control.Exception (Exception, mask_)

250import Control.Monad.Catch (MonadCatch)

251−import Control.Monad.IO.Class (MonadIO(..))

252−import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)

253import Data.Functor (($>))

251import Data.Functor (($>))

254import GHC.Types (SPEC(..))

252import GHC.Types (SPEC(..))

255−import Streamly.Internal.Control.Concurrent (MonadAsync)

253+import Streamly.Internal.Control.Concurrent (MonadRunInIO, MonadAsync, withRunInIO)

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

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

257import Streamly.Internal.Data.IOFinalizer

255import Streamly.Internal.Data.IOFinalizer

258 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

256 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

@@ -830,7 +828,7 @@ gbracket_ bef exc aft (Unfold estep einject) (Unfold step1 inject1) =

830-- /Pre-release/

828-- /Pre-release/

831{-# INLINE_NORMAL gbracket #-}

829{-# INLINE_NORMAL gbracket #-}

832gbracket

830gbracket

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

831+ :: MonadRunInIO m

834 => (a -> m c) -- ^ before

832 => (a -> m c) -- ^ before

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

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

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

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

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

845 inject x = do

843 inject x = do

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

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

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

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

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

846+ (r, ref) <- withRunInIO $ \run -> mask_ $ run $ do

849 r <- bef x

847 r <- bef x

850 ref <- newIOFinalizer (aft r)

848 ref <- newIOFinalizer (aft r)

851 return (r, ref)

849 return (r, ref)

@@ -931,7 +929,7 @@ after_ action (Unfold step1 inject1) = Unfold step inject

931--

929--

932-- /Pre-release/

930-- /Pre-release/

933{-# INLINE_NORMAL after #-}

931{-# INLINE_NORMAL after #-}

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

932+after :: MonadRunInIO m

935 => (a -> m c) -> Unfold m a b -> Unfold m a b

933 => (a -> m c) -> Unfold m a b -> Unfold m a b

936after action (Unfold step1 inject1) = Unfold step inject

934after action (Unfold step1 inject1) = Unfold step inject

937

935

@@ -1123,7 +1121,7 @@ bracket bef aft (Unfold step1 inject1) = Unfold step inject

1123 inject x = do

1121 inject x = do

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

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

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

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

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

1124+ (r, ref) <- withRunInIO $ \run -> mask_ $ run $ do

1127 r <- bef x

1125 r <- bef x

1128 ref <- newIOFinalizer (aft r)

1126 ref <- newIOFinalizer (aft r)

1129 return (r, ref)

1127 return (r, ref)

streamly.cabal+14−2

@@ -207,6 +207,11 @@ flag streamly-core

207 manual: True

208 default: False

209

210+flag use-unliftio

211+ description: Use unliftio-core instead of monad-control

212+ manual: True

213+ default: False

214+

210-------------------------------------------------------------------------------

215-------------------------------------------------------------------------------

211-- Common stanzas

216-- Common stanzas

212-------------------------------------------------------------------------------

217-------------------------------------------------------------------------------

@@ -258,6 +263,9 @@ common compile-options

258 if flag(limit-build-mem)

263 if flag(limit-build-mem)

259 ghc-options: +RTS -M1000M -RTS

264 ghc-options: +RTS -M1000M -RTS

260

265

266+ if flag(use-unliftio)

267+ cpp-options: -DUSE_UNLIFTIO

268+

261common default-extensions

269common default-extensions

262 default-extensions:

270 default-extensions:

263 BangPatterns

271 BangPatterns

@@ -360,7 +368,7 @@ library

360 , Streamly.Internal.Data.Maybe.Strict

368 , Streamly.Internal.Data.Maybe.Strict

361 , Streamly.Internal.Data.Either.Strict

369 , Streamly.Internal.Data.Either.Strict

362

370

363− -- XXX Depends on monad-control

371+ -- XXX Depends on monad-control or unliftio-core

364 , Streamly.Internal.Control.Concurrent

372 , Streamly.Internal.Control.Concurrent

365 , Streamly.Internal.Control.ForkLifted

373 , Streamly.Internal.Control.ForkLifted

366 , Streamly.Internal.Data.IOFinalizer

374 , Streamly.Internal.Data.IOFinalizer

@@ -578,12 +586,15 @@ library

578 , filepath >= 1.2.0.0 && < 1.5

586 , filepath >= 1.2.0.0 && < 1.5

579

587

580 , fusion-plugin-types >= 0.1 && < 0.2

588 , fusion-plugin-types >= 0.1 && < 0.2

581− , monad-control >= 1.0 && < 1.1

582

589

583 -- XXX to be removed

590 -- XXX to be removed

584 , transformers-base >= 0.4 && < 0.5

591 , transformers-base >= 0.4 && < 0.5

585 , primitive >= 0.5.4 && < 0.8

592 , primitive >= 0.5.4 && < 0.8

586 , heaps >= 0.3 && < 0.5

593 , heaps >= 0.3 && < 0.5

594+ if flag(use-unliftio)

595+ build-depends: unliftio-core >= 0.2 && < 0.3

596+ else

597+ build-depends: monad-control >= 1.0 && < 1.1

587

598

588 if !flag(streamly-core)

599 if !flag(streamly-core)

589 build-depends:

600 build-depends:

@@ -596,6 +607,7 @@ library

596 -- Network

607 -- Network

597 , network >= 2.6 && < 3.2

608 , network >= 2.6 && < 3.2

598

609

610+

599 if os(windows)

611 if os(windows)

600 build-depends: Win32 >= 2.6 && < 2.13

612 build-depends: Win32 >= 2.6 && < 2.13

601

613