be953ba · streamly · git.iko.soy

git.iko.soy/streamly

Fix build with use-unliftio flag

commitbe953baaf2ee66295749201acc29548e13ca3524 authorHarendra Kumar <harendra@composewell.com> dateMon 24 Oct 2022, 09:09 (4 years ago) parent3db189c

7 files changed+36−32

compare with master

unifiedsplit

src/Streamly/Internal/Data/Fold/Concurrent/Channel/Type.hs+3−3 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Append.hs+3−5 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Consumer.hs+5−8 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Dispatcher.hs+5−5 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Interleave.hs+2−3 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Operations.hs+5−6 streamly.cabal+13−2

src/Streamly/Internal/Data/Fold/Concurrent/Channel/Type.hs+3−3

@@ -24,10 +24,10 @@ import Control.Exception (SomeException(..))

2424import Control.Monad (void)

2525import Control.Monad.Catch (throwM)

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

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

2827import Data.IORef (IORef, newIORef, readIORef)

2928import Data.List (intersperse)

30−import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

29+import Streamly.Internal.Control.Concurrent

30+ (MonadAsync, MonadRunInIO, askRunInIO)

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

3232import Streamly.Internal.Data.Fold (Fold(..))

3333import Streamly.Internal.Data.Stream.Channel.Dispatcher (dumpSVarStats)

@@ -233,7 +233,7 @@ mkNewChannel cfg = do

233233{-# INLINABLE newChannel #-}

234234{-# SPECIALIZE newChannel ::

235235 (Config -> Config) -> Fold IO a b -> IO (Channel IO a b) #-}

236−newChannel :: (MonadIO m, MonadBaseControl IO m) =>

236+newChannel :: (MonadRunInIO m) =>

237237 (Config -> Config) -> Fold m a b -> m (Channel m a b)

238238newChannel modifier f = do

239239 sv <- liftIO $ mkNewChannel (modifier defaultConfig)

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Append.hs+3−5

@@ -24,13 +24,12 @@ import Control.Concurrent.MVar (newEmptyMVar, newMVar, putMVar, takeMVar)

2424import Control.Exception (assert)

2525import Control.Monad (when, void)

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

27−import Control.Monad.Trans.Control (MonadBaseControl, restoreM)

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

2928import Data.IORef (IORef, newIORef, readIORef, atomicModifyIORef, writeIORef)

3029import Data.Kind (Type)

3130import GHC.Exts (inline)

3231import Streamly.Internal.Control.Concurrent

33− (MonadRunInIO, RunInIO(..), askRunInIO)

32+ (MonadRunInIO, RunInIO(..), askRunInIO, restoreM)

3433import Streamly.Internal.Data.Atomics

3534 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

3635import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)

@@ -896,7 +895,7 @@ workLoopAhead q heap sv winfo = do

896895-- than 10%. Need to investigate what the root cause is.

897896-- Interestingly, the same thing does not make any difference for Ahead.

898897-- {-# INLINABLE getLifoSVar #-}

899−getLifoSVar :: forall m a. (MonadIO m, MonadBaseControl IO m) =>

898+getLifoSVar :: forall m a. MonadRunInIO m =>

900899 RunInIO m -> Config -> IO (Channel m a)

901900getLifoSVar mrun cfg = do

902901 outQ <- newIORef ([], 0)

@@ -1052,8 +1051,7 @@ getLifoSVar mrun cfg = do

10521051-- newChannel.

10531052{-# INLINABLE newChannel #-}

10541053{-# SPECIALIZE newChannel :: (Config -> Config) -> IO (Channel IO a) #-}

1055−newChannel :: (MonadIO m, MonadBaseControl IO m) =>

1056− (Config -> Config) -> m (Channel m a)

1054+newChannel :: MonadRunInIO m => (Config -> Config) -> m (Channel m a)

10571055newChannel modifier = do

10581056 mrun <- askRunInIO

10591057 liftIO $ getLifoSVar mrun (modifier defaultConfig)

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Consumer.hs+5−8

@@ -18,10 +18,10 @@ module Streamly.Internal.Data.Stream.Concurrent.Channel.Consumer

1818 )

1919where

2020

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

2221import Control.Monad (when, void)

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

2423import Data.IORef (readIORef)

24+import Streamly.Internal.Control.Concurrent (MonadRunInIO)

2525

2626import Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher

2727import Streamly.Internal.Data.Stream.Concurrent.Channel.Type

@@ -38,8 +38,7 @@ readOutputQChan sv = do

3838 let ss = if svarInspectMode sv then Just (svarStats sv) else Nothing

3939 in readOutputQRaw (outputQueue sv) ss

4040

41−readOutputQBounded :: (MonadIO m, MonadBaseControl IO m) =>

42− Bool -> Channel m a -> m [ChildEvent a]

41+readOutputQBounded :: MonadRunInIO m => Bool -> Channel m a -> m [ChildEvent a]

4342readOutputQBounded eagerEval sv = do

4443 (list, len) <- liftIO $ readOutputQChan sv

4544 -- When there is no output seen we dispatch more workers to help

@@ -67,8 +66,7 @@ readOutputQBounded eagerEval sv = do

6766 sendWorkerWait eagerEval sendWorkerDelay (dispatchWorker 0) sv

6867 liftIO (fst `fmap` readOutputQChan sv)

6968

70−readOutputQPaced :: (MonadIO m, MonadBaseControl IO m) =>

71− Channel m a -> m [ChildEvent a]

69+readOutputQPaced :: MonadRunInIO m => Channel m a -> m [ChildEvent a]

7270readOutputQPaced sv = do

7371 (list, len) <- liftIO $ readOutputQChan sv

7472 if len <= 0

@@ -86,7 +84,7 @@ readOutputQPaced sv = do

8684 sendWorkerWait False sendWorkerDelayPaced dispatchWorkerPaced sv

8785 liftIO (fst `fmap` readOutputQChan sv)

8886

89−postProcessPaced :: (MonadIO m, MonadBaseControl IO m) => Channel m a -> m Bool

87+postProcessPaced :: MonadRunInIO m => Channel m a -> m Bool

9088postProcessPaced sv = do

9189 workersDone <- allThreadsDone (workerThreads sv)

9290 -- XXX If during consumption we figure out we are getting delayed then we

@@ -105,8 +103,7 @@ postProcessPaced sv = do

105103 return r

106104 else return False

107105

108−postProcessBounded :: (MonadIO m, MonadBaseControl IO m) =>

109− Channel m a -> m Bool

106+postProcessBounded :: MonadRunInIO m => Channel m a -> m Bool

110107postProcessBounded sv = do

111108 workersDone <- allThreadsDone (workerThreads sv)

112109 -- There may still be work pending even if there are no workers pending

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Dispatcher.hs+5−5

@@ -20,13 +20,13 @@ module Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher

2020 )

2121where

2222

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

2423import Control.Concurrent (takeMVar, threadDelay)

2524import Control.Exception (assert)

2625import Control.Monad (when, void)

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

2827import Data.Maybe (fromJust, fromMaybe)

2928import Data.IORef (modifyIORef, newIORef, readIORef, writeIORef)

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

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

3131import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_, storeLoadBarrier)

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

@@ -43,7 +43,7 @@ import Streamly.Internal.Data.Stream.Channel.Worker

4343-------------------------------------------------------------------------------

4444

4545{-# NOINLINE pushWorker #-}

46−pushWorker :: (MonadIO m, MonadBaseControl IO m) => Count -> Channel m a -> m ()

46+pushWorker :: MonadRunInIO m => Count -> Channel m a -> m ()

4747pushWorker yieldMax sv = do

4848 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1

4949 when (svarInspectMode sv)

@@ -130,7 +130,7 @@ checkMaxBuffer active sv = do

130130 (_, n) <- liftIO $ readIORef (outputQueue sv)

131131 return $ fromIntegral lim > n + active

132132

133−dispatchWorker :: (MonadIO m, MonadBaseControl IO m) =>

133+dispatchWorker :: MonadRunInIO m =>

134134 Count -> Channel m a -> m Bool

135135dispatchWorker yieldCount sv = do

136136 -- XXX in case of Ahead streams we should not send more than one worker

@@ -175,7 +175,7 @@ dispatchWorker yieldCount sv = do

175175-- Returns:

176176-- True: can dispatch more

177177-- False: full, no more dispatches

178−dispatchWorkerPaced :: (MonadIO m, MonadBaseControl IO m) =>

178+dispatchWorkerPaced :: MonadRunInIO m =>

179179 Channel m a -> m Bool

180180dispatchWorkerPaced sv = do

181181 let yinfo = fromJust $ yieldRateInfo sv

@@ -382,7 +382,7 @@ sendWorkerWait eagerEval delay dispatch sv = go

382382-- | Start the evaluation of the channel's work queue by kicking off a worker.

383383-- Note: Work queue must not be empty otherwise the worker will exit without

384384-- doing anything.

385−startChannel :: (MonadIO m, MonadBaseControl IO m) =>

385+startChannel :: MonadRunInIO m =>

386386 Channel m a -> m ()

387387startChannel chan = do

388388 case yieldRateInfo chan of

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Interleave.hs+2−3

@@ -17,11 +17,10 @@ where

1717import Control.Concurrent (myThreadId)

1818import Control.Concurrent.MVar (newEmptyMVar)

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

20−import Control.Monad.Trans.Control (MonadBaseControl, restoreM)

2120import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)

2221import Data.IORef (newIORef, readIORef)

2322import Streamly.Internal.Control.Concurrent

24− (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)

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

2524import Streamly.Internal.Data.Stream.Channel.Dispatcher (delThread)

2625

2726import qualified Data.Set as Set

@@ -149,7 +148,7 @@ workLoopFIFOLimited q sv winfo = run

149148-- function argument to this function results in a perf degradation of more

150149-- than 10%. Need to investigate what the root cause is.

151150-- Interestingly, the same thing does not make any difference for Ahead.

152−getFifoSVar :: forall m a. (MonadIO m, MonadBaseControl IO m) =>

151+getFifoSVar :: forall m a. MonadRunInIO m =>

153152 RunInIO m -> Config -> IO (Channel m a)

154153getFifoSVar mrun cfg = do

155154 outQ <- newIORef ([], 0)

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Operations.hs+5−6

@@ -30,10 +30,10 @@ import Control.Exception (fromException)

3030import Control.Monad (when)

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

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

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

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

3534import Data.Maybe (isNothing)

36−import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

35+import Streamly.Internal.Control.Concurrent

36+ (MonadAsync, MonadRunInIO, askRunInIO)

3737import Streamly.Internal.Data.Stream.Type (Stream)

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

3939import System.Mem (performMajorGC)

@@ -50,6 +50,7 @@ import Prelude hiding (map, concat, concatMap)

5050

5151#ifdef INSPECTION

5252import Control.Exception (Exception)

53+import Control.Monad.Trans.Control (MonadBaseControl)

5354import Data.Typeable (Typeable)

5455import Test.Inspection (inspect, hasNoTypeClassesExcept)

5556#endif

@@ -88,8 +89,7 @@ import Test.Inspection (inspect, hasNoTypeClassesExcept)

8889-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

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

9091{-# INLINE toChannelK #-}

91−toChannelK :: (MonadIO m, MonadBaseControl IO m) =>

92− Channel m a -> K.Stream m a -> m ()

92+toChannelK :: MonadRunInIO m => Channel m a -> K.Stream m a -> m ()

9393toChannelK sv m = do

9494 runIn <- askRunInIO

9595 liftIO $ enqueue sv False (runIn, m)

@@ -98,8 +98,7 @@ toChannelK sv m = do

9898

9999-- | Send a stream to a given channel for concurrent evaluation.

100100{-# INLINE toChannel #-}

101−toChannel :: (MonadIO m, MonadBaseControl IO m) =>

102− Channel m a -> Stream m a -> m ()

101+toChannel :: MonadRunInIO m => Channel m a -> Stream m a -> m ()

103102toChannel chan = toChannelK chan . Stream.toStreamK

104103

105104{-

streamly.cabal+13−2

@@ -196,6 +196,11 @@ flag limit-build-mem

196196 manual: True

197197 default: False

198198

199+flag use-unliftio

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

201+ manual: True

202+ default: False

203+

199204-------------------------------------------------------------------------------

200205-- Common stanzas

201206-------------------------------------------------------------------------------

@@ -248,6 +253,9 @@ common compile-options

248253 if flag(limit-build-mem)

249254 ghc-options: +RTS -M1000M -RTS

250255

256+ if flag(use-unliftio)

257+ cpp-options: -DUSE_UNLIFTIO

258+

251259common default-extensions

252260 default-extensions:

253261 BangPatterns

@@ -473,8 +481,6 @@ library

473481 , mtl >= 2.2 && < 2.4

474482 , transformers >= 0.4 && < 0.7

475483

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

477−

478484 -- The core streamly package

479485 , streamly-core == 0.1.0

480486

@@ -494,6 +500,11 @@ library

494500 -- Network

495501 , network >= 2.6 && < 3.2

496502

503+ if flag(use-unliftio)

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

505+ else

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

507+

497508 if flag(dev)

498509 build-depends: primitive >= 0.5.4 && < 0.8

499510