be953ba · streamly · git.iko.soy

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

compare with master

unifiedsplit

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

24import Control.Monad (void)

25import Control.Monad.Catch (throwM)

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

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

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

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

29import Data.List (intersperse)

28import Data.List (intersperse)

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

29+import Streamly.Internal.Control.Concurrent

30+ (MonadAsync, MonadRunInIO, askRunInIO)

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

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

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

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

233{-# INLINABLE newChannel #-}

234{-# SPECIALIZE newChannel ::

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

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

236+newChannel :: (MonadRunInIO m) =>

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

238newChannel modifier f = do

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

24import Control.Exception (assert)

25import Control.Monad (when, void)

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

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

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

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

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

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

30import Data.Kind (Type)

29import Data.Kind (Type)

31import GHC.Exts (inline)

30import GHC.Exts (inline)

32import Streamly.Internal.Control.Concurrent

31import Streamly.Internal.Control.Concurrent

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

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

34import Streamly.Internal.Data.Atomics

33import Streamly.Internal.Data.Atomics

35 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

34 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

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

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

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

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

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

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

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

898-- {-# INLINABLE getLifoSVar #-}

897-- {-# INLINABLE getLifoSVar #-}

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

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

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

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

901getLifoSVar mrun cfg = do

900getLifoSVar mrun cfg = do

902 outQ <- newIORef ([], 0)

901 outQ <- newIORef ([], 0)

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

1052-- newChannel.

1051-- newChannel.

1053{-# INLINABLE newChannel #-}

1052{-# INLINABLE newChannel #-}

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

1053{-# 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)

1057newChannel modifier = do

1055newChannel modifier = do

1058 mrun <- askRunInIO

1056 mrun <- askRunInIO

1059 liftIO $ getLifoSVar mrun (modifier defaultConfig)

1057 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

18 )

19where

20

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

22import Control.Monad (when, void)

21import Control.Monad (when, void)

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

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

24import Data.IORef (readIORef)

23import Data.IORef (readIORef)

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

25

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

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

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

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

39 in readOutputQRaw (outputQueue sv) ss

40

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]

43readOutputQBounded eagerEval sv = do

42readOutputQBounded eagerEval sv = do

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

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

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

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

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

67 sendWorkerWait eagerEval sendWorkerDelay (dispatchWorker 0) sv

66 sendWorkerWait eagerEval sendWorkerDelay (dispatchWorker 0) sv

68 liftIO (fst `fmap` readOutputQChan sv)

67 liftIO (fst `fmap` readOutputQChan sv)

69

68

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

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

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

72readOutputQPaced sv = do

70readOutputQPaced sv = do

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

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

74 if len <= 0

72 if len <= 0

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

86 sendWorkerWait False sendWorkerDelayPaced dispatchWorkerPaced sv

84 sendWorkerWait False sendWorkerDelayPaced dispatchWorkerPaced sv

87 liftIO (fst `fmap` readOutputQChan sv)

85 liftIO (fst `fmap` readOutputQChan sv)

88

86

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

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

90postProcessPaced sv = do

88postProcessPaced sv = do

91 workersDone <- allThreadsDone (workerThreads sv)

89 workersDone <- allThreadsDone (workerThreads sv)

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

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

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

105 return r

103 return r

106 else return False

104 else return False

107

105

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

109− Channel m a -> m Bool

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

110postProcessBounded sv = do

107postProcessBounded sv = do

111 workersDone <- allThreadsDone (workerThreads sv)

108 workersDone <- allThreadsDone (workerThreads sv)

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

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

20 )

21where

22

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

24import Control.Concurrent (takeMVar, threadDelay)

23import Control.Concurrent (takeMVar, threadDelay)

25import Control.Exception (assert)

24import Control.Exception (assert)

26import Control.Monad (when, void)

25import Control.Monad (when, void)

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

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

28import Data.Maybe (fromJust, fromMaybe)

27import Data.Maybe (fromJust, fromMaybe)

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

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

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

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

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

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

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

43-------------------------------------------------------------------------------

44

45{-# NOINLINE pushWorker #-}

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

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

47pushWorker yieldMax sv = do

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

49 when (svarInspectMode sv)

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

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

131 return $ fromIntegral lim > n + active

132

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

133+dispatchWorker :: MonadRunInIO m =>

134 Count -> Channel m a -> m Bool

135dispatchWorker yieldCount sv = do

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

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

175-- Returns:

176-- True: can dispatch more

177-- False: full, no more dispatches

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

178+dispatchWorkerPaced :: MonadRunInIO m =>

179 Channel m a -> m Bool

180dispatchWorkerPaced sv = do

181 let yinfo = fromJust $ yieldRateInfo sv

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

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

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

384-- doing anything.

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

385+startChannel :: MonadRunInIO m =>

386 Channel m a -> m ()

387startChannel chan = do

388 case yieldRateInfo chan of

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

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

17import Control.Concurrent (myThreadId)

18import Control.Concurrent.MVar (newEmptyMVar)

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

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

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

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

22import Data.IORef (newIORef, readIORef)

21import Data.IORef (newIORef, readIORef)

23import Streamly.Internal.Control.Concurrent

22import Streamly.Internal.Control.Concurrent

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

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

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

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

26

25

27import qualified Data.Set as Set

26import qualified Data.Set as Set

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

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

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

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

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

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

150-- 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 =>

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

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

154getFifoSVar mrun cfg = do

153getFifoSVar mrun cfg = do

155 outQ <- newIORef ([], 0)

154 outQ <- newIORef ([], 0)

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

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

30import Control.Monad (when)

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

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

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

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

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

35import Data.Maybe (isNothing)

34import Data.Maybe (isNothing)

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

35+import Streamly.Internal.Control.Concurrent

36+ (MonadAsync, MonadRunInIO, askRunInIO)

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

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

39import System.Mem (performMajorGC)

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

50

51#ifdef INSPECTION

52import Control.Exception (Exception)

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

53import Data.Typeable (Typeable)

54import Data.Typeable (Typeable)

54import Test.Inspection (inspect, hasNoTypeClassesExcept)

55import Test.Inspection (inspect, hasNoTypeClassesExcept)

55#endif

56#endif

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

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

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

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

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

90{-# INLINE toChannelK #-}

91{-# 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 ()

93toChannelK sv m = do

94 runIn <- askRunInIO

95 liftIO $ enqueue sv False (runIn, m)

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

98

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

100{-# 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 ()

103toChannel chan = toChannelK chan . Stream.toStreamK

102toChannel chan = toChannelK chan . Stream.toStreamK

104

103

105{-

104{-

streamly.cabal+13−2

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

196 manual: True

197 default: False

198

199+flag use-unliftio

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

201+ manual: True

202+ default: False

203+

199-------------------------------------------------------------------------------

204-------------------------------------------------------------------------------

200-- Common stanzas

205-- Common stanzas

201-------------------------------------------------------------------------------

206-------------------------------------------------------------------------------

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

248 if flag(limit-build-mem)

253 if flag(limit-build-mem)

249 ghc-options: +RTS -M1000M -RTS

254 ghc-options: +RTS -M1000M -RTS

250

255

256+ if flag(use-unliftio)

257+ cpp-options: -DUSE_UNLIFTIO

258+

251common default-extensions

259common default-extensions

252 default-extensions:

260 default-extensions:

253 BangPatterns

261 BangPatterns

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

473 , mtl >= 2.2 && < 2.4

481 , mtl >= 2.2 && < 2.4

474 , transformers >= 0.4 && < 0.7

482 , transformers >= 0.4 && < 0.7

475

483

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

477−

478 -- The core streamly package

484 -- The core streamly package

479 , streamly-core == 0.1.0

485 , streamly-core == 0.1.0

480

486

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

494 -- Network

500 -- Network

495 , network >= 2.6 && < 3.2

501 , network >= 2.6 && < 3.2

496

502

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+

497 if flag(dev)

508 if flag(dev)

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

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

499

510