babb151 · streamly · git.iko.soy

git.iko.soy/streamly

Allow using unliftio-core instead of monad-control (#1485)

* Allow using unliftio-core instead of monad-control * Add explcit imports to Control.Monad.Trans.Control * Expand definition of MonadAsync

commitbabb1510f605b3f746ff081ceb39a336b160d256 authorandremarianiello <andremarianiello@users.noreply.github.com> dateMon 28 Feb 2022, 00:51 (5 years ago) parentcdd6228

14 files changed+119−77

compare with master

unifiedsplit

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

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

1313module Streamly.Internal.Control.Concurrent

1414 (

1515 MonadAsync

16+ , MonadRunInIO

1617 , RunInIO(..)

17− , captureMonadState

18+ , askRunInIO

19+ , withRunInIO

20+ , withRunInIONoRestore

21+ , restoreM

1822 )

1923where

2024

2125import Control.Monad.Catch (MonadThrow)

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

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

26+

27+#ifdef USE_UNLIFTIO

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

29+#else

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

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

2439

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

2641--

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

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

3247--

3348-- @since 0.8.0

49+#ifdef USE_UNLIFTIO

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

51+#else

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

53+#endif

54+

3555

56+#ifdef USE_UNLIFTIO

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

58+#else

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

60+#endif

3761

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

3963-- 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) }

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

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

4468-- execution.

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

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

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

70+#ifdef USE_UNLIFTIO

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

72+#else

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

1616

1717import Control.Concurrent (ThreadId, forkIO)

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

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

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

2119import Data.Functor (void)

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

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

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

2422

2523-- | 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'

2725-- capability.

2826--

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

3028-- as "mrun action" instead.

3129{-# INLINE doFork #-}

32−doFork :: MonadBaseControl IO m

30+doFork :: MonadRunInIO m

3331 => m ()

3432 -> RunInIO m

3533 -> (SomeException -> IO ())

3634 -> m ThreadId

3735doFork action (RunInIO mrun) exHandler =

38− control $ \run ->

36+ withRunInIO $ \run ->

3937 mask $ \restore -> do

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

4139 exHandler

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

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

4543--

4644{-# INLINABLE fork #-}

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

48−fork = liftBaseDiscard forkIO

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

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

4947

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

5149-- returned threadId is garbage collected.

5250--

5351{-# INLINABLE forkManaged #-}

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

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

5553forkManaged = forkManagedWith fork

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

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

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

2222import Control.Monad (void)

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

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

2726

2827import Streamly.Internal.Data.Fold.Type

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

6564 Partial s -> do

6665 mv <- liftIO $ newMVar False

6766 t <-

68− control $ \run ->

67+ withRunInIO $ \run ->

6968 mask $ \restore -> do

7069 tid <-

7170 forkIO

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

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

2121import Control.Exception (mask_)

2222import Control.Monad (void)

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

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

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

2726

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

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

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

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

3938

4039-- | 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 ())

4241mkIOFinalizer f = do

43− mrun <- captureMonadState

42+ mrun <- askRunInIO

4443 return $

4544 void $ do

4645 _ <- runInIO mrun f

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

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

6766--

6867-- /Pre-release/

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

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

7069newIOFinalizer finalizer = do

7170 f <- mkIOFinalizer finalizer

7271 ref <- liftIO $ newIORef $ Just f

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

9594-- action is run with async exceptions masked.

9695--

9796-- /Pre-release/

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

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

9998clearingIOFinalizer (IOFinalizer ref) action = do

100− control $ \runinio ->

99+ withRunInIO $ \runinio ->

101100 mask_ $ do

102101 writeIORef ref Nothing

103102 runinio action

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

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

4848import Data.IORef (newIORef, readIORef)

4949import Data.IORef (IORef, atomicModifyIORef)

5050import Streamly.Internal.Control.Concurrent

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

51+ (MonadAsync, askRunInIO, RunInIO)

5252import Streamly.Internal.Data.Atomics

5353 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)

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

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

432432 -> m ())

433433 -> m (SVar t m a)

434434newAheadVar st m wloop = do

435− mrun <- captureMonadState

435+ mrun <- askRunInIO

436436 sv <- liftIO $ getAheadSVar st wloop mrun

437437 sendFirstWorker sv m

438438

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

533533newParallelVar :: MonadAsync m

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

535535newParallelVar ss st = do

536− mrun <- captureMonadState

536+ mrun <- askRunInIO

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

4949#if __GLASGOW_HASKELL__ < 804

5050import Data.Semigroup ((<>))

5151#endif

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

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

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

5454import Streamly.Internal.Data.Atomics

5555 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

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

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

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

709709 -- chance to push.

710− runIn <- captureMonadState

710+ runIn <- askRunInIO

711711 liftIO $ enqueue sv (runIn, m)

712712 case yieldRateInfo sv of

713713 Nothing -> pushWorker 0 sv

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

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

6262import Data.Set (Set)

6363

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

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

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

6666

6767newtype Count = Count Int64

6868 deriving ( Eq

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

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

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

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

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

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

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

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

4544import Data.Maybe (fromJust)

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

5150import qualified Data.Heap as H

5251

5352import Streamly.Internal.Control.Concurrent

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

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

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

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

5756

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

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

230229--

231230processHeap

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

231+ :: MonadRunInIO m

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

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

235234 -> State Stream m a

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

326325 stop

327326 r

328327 else do

329− runIn <- captureMonadState

328+ runIn <- askRunInIO

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

331330 liftIO $ do

332331 requeueOnHeapTop heap ent seqNo

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

339338

340339{-# NOINLINE drainHeap #-}

341340drainHeap

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

341+ :: MonadRunInIO m

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

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

345344 -> State Stream m a

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

357356data WorkerStatus = Continue | Suspend

358357

359358processWithoutToken

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

359+ :: MonadRunInIO m

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

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

363362 -> State Stream m a

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

381380 r <- liftIO $ mrun $

382381 K.foldStreamShared st

383382 (\a r -> do

384− runIn <- captureMonadState

383+ runIn <- askRunInIO

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

386385 (toHeap . AheadEntryPure)

387386 stop

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

432431data TokenWorkerStatus = TokenContinue Int | TokenSuspend

433432

434433processWithToken

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

434+ :: MonadRunInIO m

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

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

438437 -> State Stream m a

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

484483 stop

485484 r

486485 else do

487− runIn <- captureMonadState

486+ runIn <- askRunInIO

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

489488 liftIO $ requeueOnHeapTop heap ent seqNo

490489 liftIO $ incrementYieldLimit sv

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

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

547546

548547workLoopAhead

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

548+ :: MonadRunInIO m

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

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

552551 -> State Stream m a

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

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

611610 where

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

613− runInIO <- captureMonadState

612+ runInIO <- askRunInIO

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

615614 K.foldStream st yld sng stp ma

616615

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

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

620619 case streamVar st of

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

622− runInIO <- captureMonadState

621+ runInIO <- askRunInIO

623622 liftIO $ enqueue sv (runInIO, m2)

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

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

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

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

4141import Control.Concurrent (myThreadId)

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

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

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

4544import Control.Concurrent.MVar (newEmptyMVar)

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

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

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

5958import qualified Data.Set as S

6059

6160import Streamly.Internal.Control.Concurrent

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

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

6362import Streamly.Internal.Data.Atomics

6463 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

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

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

104103

105104{-# INLINE workLoopLIFO #-}

106105workLoopLIFO

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

106+ :: MonadRunInIO m

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

109108 -> State Stream m a

110109 -> SVar Stream m a

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

143142 if res

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

145144 else do

146− runInIO <- captureMonadState

145+ runInIO <- askRunInIO

147146 liftIO $ enqueueLIFO sv q (runInIO, r)

148147 return Suspend

149148

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

158157-- make a check every time.

159158{-# INLINE workLoopLIFOLimited #-}

160159workLoopLIFOLimited

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

160+ :: forall m a. MonadRunInIO m

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

163162 -> State Stream m a

164163 -> SVar Stream m a

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

209208 if res && yieldLimitOk

210209 then K.foldStreamShared st yieldk single incrContinue r

211210 else do

212− runInIO <- captureMonadState

211+ runInIO <- askRunInIO

213212 liftIO $ incrementYieldLimit sv

214213 liftIO $ enqueueLIFO sv q (runInIO, r)

215214 return Suspend

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

240239

241240{-# INLINE workLoopFIFO #-}

242241workLoopFIFO

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

242+ :: MonadRunInIO m

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

245244 -> State Stream m a

246245 -> SVar Stream m a

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

273272 -- yielding.

274273 yieldk a r = do

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

276− runInIO <- captureMonadState

275+ runInIO <- askRunInIO

277276 liftIO $ enqueueFIFO sv q (runInIO, r)

278277 return $ if res then Continue else Suspend

279278

280279{-# INLINE workLoopFIFOLimited #-}

281280workLoopFIFOLimited

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

281+ :: forall m a. MonadRunInIO m

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

284283 -> State Stream m a

285284 -> SVar Stream m a

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

316315

317316 yieldk a r = do

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

319− runInIO <- captureMonadState

318+ runInIO <- askRunInIO

320319 liftIO $ enqueueFIFO sv q (runInIO, r)

321320 yieldLimitOk <- liftIO $ decrementYieldLimit sv

322321 if res && yieldLimitOk

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

533532newAsyncVar :: MonadAsync m

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

535534newAsyncVar st m = do

536− mrun <- captureMonadState

535+ mrun <- askRunInIO

537536 sv <- liftIO $ getLifoSVar st mrun

538537 sendFirstWorker sv m

539538

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

573572newWAsyncVar :: MonadAsync m

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

575574newWAsyncVar st m = do

576− mrun <- captureMonadState

575+ mrun <- askRunInIO

577576 sv <- liftIO $ getFifoSVar st mrun

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

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

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

654653 where

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

656− runInIO <- captureMonadState

655+ runInIO <- askRunInIO

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

658657 K.foldStreamShared st yld sng stp ma

659658

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

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

664663 case streamVar st of

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

666− runInIO <- captureMonadState

665+ runInIO <- askRunInIO

667666 liftIO $ enqueue sv (runInIO, m2)

668667 K.foldStreamShared st yld sng stp m1

669668 _ -> 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

2525

2626import Control.Exception (Exception)

2727import Control.Monad.Catch (MonadCatch)

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

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

3028import Data.Map.Strict (Map)

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

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

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

3331 (IsStream(..), fromStreamD, toStreamD)

3432

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

8381-- @since 0.7.0

8482--

8583{-# INLINE after #-}

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

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

8785 => m b -> t m a -> t m a

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

8987

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

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

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

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

3939import Data.Maybe (isNothing)

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

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

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

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

4343import System.Mem (performMajorGC)

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

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

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

9292toSVar sv m = do

93− runIn <- captureMonadState

93+ runIn <- askRunInIO

9494 liftIO $ enqueue sv (runIn, m)

9595 done <- allThreadsDone sv

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

2828

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

3030import Control.Monad.Catch (MonadCatch)

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

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

3331import Data.Map.Strict (Map)

3432import GHC.Exts (inline)

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

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

3634import Streamly.Internal.Data.IOFinalizer

3735 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

3836

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

119117-- /Pre-release/

120118{-# INLINE_NORMAL gbracket #-}

121119gbracket

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

120+ :: MonadRunInIO m

123121 => m c -- ^ before

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

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

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

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

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

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

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

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

145143 r <- bef

146144 ref <- newIOFinalizer (gc r)

147145 return (r, ref)

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

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

211209--

212210{-# INLINE_NORMAL after #-}

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

211+after :: MonadRunInIO m

214212 => m b -> Stream m a -> Stream m a

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

216214

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

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

248248

249249import Control.Exception (Exception, mask_)

250250import Control.Monad.Catch (MonadCatch)

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

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

253251import Data.Functor (($>))

254252import GHC.Types (SPEC(..))

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

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

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

257255import Streamly.Internal.Data.IOFinalizer

258256 (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

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

830828-- /Pre-release/

831829{-# INLINE_NORMAL gbracket #-}

832830gbracket

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

831+ :: MonadRunInIO m

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

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

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

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

845843 inject x = do

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

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

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

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

849847 r <- bef x

850848 ref <- newIOFinalizer (aft r)

851849 return (r, ref)

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

931929--

932930-- /Pre-release/

933931{-# INLINE_NORMAL after #-}

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

932+after :: MonadRunInIO m

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

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

937935

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

11231121 inject x = do

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

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

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

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

11271125 r <- bef x

11281126 ref <- newIOFinalizer (aft r)

11291127 return (r, ref)

streamly.cabal+14−2

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

207207 manual: True

208208 default: False

209209

210+flag use-unliftio

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

212+ manual: True

213+ default: False

214+

210215-------------------------------------------------------------------------------

211216-- Common stanzas

212217-------------------------------------------------------------------------------

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

258263 if flag(limit-build-mem)

259264 ghc-options: +RTS -M1000M -RTS

260265

266+ if flag(use-unliftio)

267+ cpp-options: -DUSE_UNLIFTIO

268+

261269common default-extensions

262270 default-extensions:

263271 BangPatterns

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

360368 , Streamly.Internal.Data.Maybe.Strict

361369 , Streamly.Internal.Data.Either.Strict

362370

363− -- XXX Depends on monad-control

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

364372 , Streamly.Internal.Control.Concurrent

365373 , Streamly.Internal.Control.ForkLifted

366374 , Streamly.Internal.Data.IOFinalizer

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

578586 , filepath >= 1.2.0.0 && < 1.5

579587

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

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

582589

583590 -- XXX to be removed

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

585592 , primitive >= 0.5.4 && < 0.8

586593 , 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

587598

588599 if !flag(streamly-core)

589600 build-depends:

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

596607 -- Network

597608 , network >= 2.6 && < 3.2

598609

610+

599611 if os(windows)

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

601613