2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE StandaloneDeriving #-}
9{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- Module : Streamly.Streams.Async
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
21module Streamly.Streams.Async
38import Control.Monad (ap)
39import Control.Monad.Base (MonadBase(..), liftBaseDefault)
40import Control.Monad.Catch (MonadThrow, throwM)
41import Control.Concurrent.MVar (newEmptyMVar)
42-- import Control.Monad.Error.Class (MonadError(..))
43import Control.Monad.IO.Class (MonadIO(..))
44import Control.Monad.Reader.Class (MonadReader(..))
45import Control.Monad.State.Class (MonadState(..))
46import Control.Monad.Trans.Class (MonadTrans(lift))
47import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ)
48import Data.IORef (IORef, newIORef, readIORef)
49import Data.Maybe (fromJust)
50import Data.Semigroup (Semigroup(..))
52import Prelude hiding (map)
53import qualified Data.Set as S
55import Streamly.Streams.SVar (fromSVar)
56import Streamly.Streams.Serial (map)
58import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)
59import qualified Streamly.Streams.StreamK as K
61#include "Instances.hs"
63-------------------------------------------------------------------------------
65-------------------------------------------------------------------------------
67{-# INLINE runStreamLIFO #-}
68runStreamLIFO :: MonadIO m
69 => State Stream m a -> IORef [Stream m a] -> Stream m a -> m () -> m ()
70runStreamLIFO st q m stop = unStream m st stop single yieldk
72 sv = fromJust $ streamVar st
73 maxBuf = bufferHigh st
75 res <- liftIO $ send maxBuf sv (ChildYield a)
76 if res then stop else liftIO $ sendStop sv
78 res <- liftIO $ send maxBuf sv (ChildYield a)
80 then (unStream r) st stop single yieldk
81 else liftIO $ enqueueLIFO sv q r >> sendStop sv
83-------------------------------------------------------------------------------
85-------------------------------------------------------------------------------
87{-# INLINE runStreamFIFO #-}
91 -> LinkedQueue (Stream m a)
95runStreamFIFO st q m stop = unStream m st stop single yieldk
97 sv = fromJust $ streamVar st
98 maxBuf = bufferHigh st
100 res <- liftIO $ send maxBuf sv (ChildYield a)
101 if res then stop else liftIO $ sendStop sv
103 res <- liftIO $ send maxBuf sv (ChildYield a)
104 liftIO (enqueueFIFO sv q r)
105 if res then stop else liftIO $ sendStop sv
107-------------------------------------------------------------------------------
109-- This code belongs in SVar.hs but is kept here for perf reasons
110-------------------------------------------------------------------------------
112-- XXX we have this function in this file because passing runStreamLIFO as a
113-- function argument to this function results in a perf degradation of more
114-- than 10%. Need to investigate what the root cause is.
115-- Interestingly, the same thing does not make any difference for Ahead.
116getLifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)
118 outQ <- newIORef ([], 0)
119 outQMv <- newEmptyMVar
121 wfw <- newIORef False
122 running <- newIORef S.empty
131 let checkEmpty = null <$> readIORef q
133 SVar { outputQueue = outQ
134 , outputDoorBell = outQMv
135 , readOutputQ = readOutputQBounded (threadsHigh st) sv
136 , postProcess = postProcessBounded sv
137 , workerThreads = running
138 , workLoop = workLoopLIFO runStreamLIFO
139 st{streamVar = Just sv} q
140 , enqueue = enqueueLIFO sv q
141 , isWorkDone = checkEmpty
143 , svarStyle = AsyncVar
144 , workerCount = active
145 , accountThread = delThread sv
147 , aheadWorkQueue = undefined
148 , outputHeap = undefined
149 , maxWorkers = maxWrk
150 , totalDispatches = disp
151 , maxOutQSize = maxOq
152 , maxHeapSize = maxHs
153 , maxWorkQSize = maxWq
158getFifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)
160 outQ <- newIORef ([], 0)
161 outQMv <- newEmptyMVar
163 wfw <- newIORef False
164 running <- newIORef S.empty
174 SVar { outputQueue = outQ
175 , outputDoorBell = outQMv
176 , readOutputQ = readOutputQBounded (threadsHigh st) sv
177 , postProcess = postProcessBounded sv
178 , workerThreads = running
179 , workLoop = workLoopFIFO runStreamFIFO
180 st{streamVar = Just sv} q
181 , enqueue = enqueueFIFO sv q
182 , isWorkDone = nullQ q
184 , svarStyle = WAsyncVar
185 , workerCount = active
186 , accountThread = delThread sv
188 , aheadWorkQueue = undefined
189 , outputHeap = undefined
190 , totalDispatches = disp
191 , maxWorkers = maxWrk
192 , maxOutQSize = maxOq
193 , maxHeapSize = maxHs
194 , maxWorkQSize = maxWq
199{-# INLINABLE newAsyncVar #-}
200newAsyncVar :: MonadAsync m
201 => State Stream m a -> Stream m a -> m (SVar Stream m a)
203 sv <- liftIO $ getLifoSVar st
206-- XXX Get rid of this?
207-- | Make a stream asynchronous, triggers the computation and returns a stream
208-- in the underlying monad representing the output generated by the original
209-- computation. The returned action is exhaustible and must be drained once. If
210-- not drained fully we may have a thread blocked forever and once exhausted it
211-- will always return 'empty'.
214{-# INLINABLE mkAsync #-}
215mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
216mkAsync m = newAsyncVar defState (toStream m) >>= return . fromSVar
218{-# INLINABLE mkAsync' #-}
219mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
220mkAsync' st m = newAsyncVar st (toStream m) >>= return . fromSVar
222-- | Create a new SVar and enqueue one stream computation on it.
223{-# INLINABLE newWAsyncVar #-}
224newWAsyncVar :: MonadAsync m
225 => State Stream m a -> Stream m a -> m (SVar Stream m a)
226newWAsyncVar st m = do
227 sv <- liftIO $ getFifoSVar st
230------------------------------------------------------------------------------
231-- Running streams concurrently
232------------------------------------------------------------------------------
234-- Concurrency rate control.
236-- Our objective is to create more threads on demand if the consumer is running
237-- faster than us. As soon as we encounter a concurrent composition we create a
238-- push pull pair of threads. We use an SVar for communication between the
239-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
240-- The producer creates more threads if the SVar drains and becomes empty, that
241-- is the consumer is running faster.
243-- XXX Note 1: This mechanism can be problematic if the initial production
244-- latency is high, we may end up creating too many threads. So we need some
245-- way to monitor and use the latency as well. Having a limit on the dispatches
246-- (programmer controlled) may also help.
248-- TBD Note 2: We may want to run computations at the lower level of the
249-- composition tree serially even when they are composed using a parallel
250-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
251-- place of 'wAsync'. If we find that an SVar immediately above a computation
252-- gets drained empty we can switch to parallelizing the computation. For that
253-- we can use a state flag to fork the rest of the computation at any point of
254-- time inside the Monad bind operation if the consumer is running at a faster
257-- TBD Note 3: the binary operation ('parallel') composition allows us to
258-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
259-- chunksizes we will need to compose the parallel actions using a data
260-- constructor (A Free container) instead so that we can divide it in chunks of
261-- arbitrary size before dispatching. If the stream is composed of
262-- hierarchically composed grains of different sizes then we can always switch
263-- to a desired granularity depending on the consumer speed.
265-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
266-- into just the number of CPUs.
268-- | Join two computations on the currently running 'SVar' queue for concurrent
269-- execution. When we are using parallel composition, an SVar is passed around
270-- as a state variable. We try to schedule a new parallel computation on the
271-- SVar passed to us. The first time, when no SVar exists, a new SVar is
272-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
273-- already scheduled on the SVar is further evaluated. For example, when (a
274-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
275-- the current scheduler queue.
277-- The 'SVarStyle' required by the current composition context is passed as one
278-- of the parameters. If the scheduling and composition style of the new
279-- computation being scheduled is different than the style of the current SVar,
280-- then we create a new SVar and schedule it on that. The newly created SVar
281-- joins as one of the computations on the current SVar queue.
283-- Cases when we need to switch to a new SVar:
285-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
286-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
287-- because of the scheduling policy change.
288-- * if we 'adapt' a stream of type 'async' to a stream of type
289-- 'Parallel', we create a new SVar at the transitioning bind.
290-- * When the stream is switching from disjunctive composition to conjunctive
291-- composition and vice-versa we create a new SVar to isolate the scheduling
294forkSVarAsync :: MonadAsync m
295 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
296forkSVarAsync style m1 m2 = Stream $ \st stp sng yld -> do
298 AsyncVar -> newAsyncVar st (concurrently m1 m2)
299 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
300 _ -> error "illegal svar type"
301 unStream (fromSVar sv) (rstState st) stp sng yld
303 concurrently ma mb = Stream $ \st stp sng yld -> do
304 liftIO $ enqueue (fromJust $ streamVar st) mb
305 unStream ma st stp sng yld
307{-# INLINE joinStreamVarAsync #-}
308joinStreamVarAsync :: MonadAsync m
309 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
310joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld ->
312 Just sv | svarStyle sv == style ->
313 liftIO (enqueue sv m2) >> unStream m1 st stp sng yld
314 _ -> unStream (forkSVarAsync style m1 m2) (rstState st) stp sng yld
316------------------------------------------------------------------------------
317-- Semigroup and Monoid style compositions for parallel actions
318------------------------------------------------------------------------------
321asyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
322asyncS = joinStreamVarAsync AsyncVar
324-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
325-- Merges two streams possibly concurrently, preferring the
326-- elements from the left one when available.
330async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
331async m1 m2 = fromStream $
332 joinStreamVarAsync AsyncVar (toStream m1) (toStream m2)
337{-# DEPRECATED (<|) "Please use 'async' instead." #-}
339(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
342-- | XXX we can implement it more efficienty by directly implementing instead
343-- of combining streams using async.
344{-# INLINE consMAsync #-}
345consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
346consMAsync m r = K.yieldM m `asyncS` r
348------------------------------------------------------------------------------
350------------------------------------------------------------------------------
352-- | Deep async composition or async composition with depth first traversal. In
353-- a left to right 'Semigroup' composition it tries to yield elements from the
354-- left stream as long as it can, but it can run the right stream in parallel
355-- if it needs to, based on demand. The right stream can be run if the left
356-- stream blocks on IO or cannot produce elements fast enough for the consumer.
359-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
365-- Any exceptions generated by a constituent stream are propagated to the
366-- output stream. The output and exceptions from a single stream are guaranteed
367-- to arrive in the same order in the resulting stream as they were generated
368-- in the input stream. However, the relative ordering of elements from
369-- different streams in the resulting stream can vary depending on scheduling
370-- and generation delays.
372-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
373-- concurrently based on demand. More concurrent iterations are started only
374-- if the previous iterations are not able to produce enough output for the
379-- import qualified "Streamly.Prelude" as S
380-- import Control.Concurrent
382-- main = 'runStream' . 'asyncly' $ do
383-- n <- return 3 \<\> return 2 \<\> return 1
385-- threadDelay (n * 1000000)
386-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
389-- ThreadId 40: Delay 1
390-- ThreadId 39: Delay 2
391-- ThreadId 38: Delay 3
394-- All iterations may run in the same thread if they do not block.
396-- Note that async composition with depth first traversal can be used to
397-- combine infinite number of streams as it explores only a bounded number of
401newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
402 deriving (MonadTrans)
404-- | A demand driven left biased parallely composing IO stream of elements of
405-- type @a@. See 'AsyncT' documentation for more details.
408type Async a = AsyncT IO a
410-- | Fix the type of a polymorphic stream as 'AsyncT'.
413asyncly :: IsStream t => AsyncT m a -> t m a
416instance IsStream AsyncT where
421 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
422 consM m r = fromStream $ consMAsync m (toStream r)
425 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
428------------------------------------------------------------------------------
430------------------------------------------------------------------------------
432instance MonadAsync m => Semigroup (AsyncT m a) where
435------------------------------------------------------------------------------
437------------------------------------------------------------------------------
439instance MonadAsync m => Monoid (AsyncT m a) where
443------------------------------------------------------------------------------
445------------------------------------------------------------------------------
447instance MonadAsync m => Monad (AsyncT m) where
449 (AsyncT m) >>= f = AsyncT $ K.bindWith asyncS m (getAsyncT . f)
451------------------------------------------------------------------------------
453------------------------------------------------------------------------------
455MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
456MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
458------------------------------------------------------------------------------
460------------------------------------------------------------------------------
462{-# INLINE wAsyncS #-}
463wAsyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
464wAsyncS = joinStreamVarAsync WAsyncVar
466-- | XXX we can implement it more efficienty by directly implementing instead
467-- of combining streams using wAsync.
468{-# INLINE consMWAsync #-}
469consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
470consMWAsync m r = K.yieldM m `wAsyncS` r
472-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
473-- Merges two streams concurrently choosing elements from both fairly.
477wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
478wAsync m1 m2 = fromStream $ wAsyncS (toStream m1) (toStream m2)
480-- | Wide async composition or async composition with breadth first traversal.
481-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
482-- streams using a depth first travesal or in a round robin fashion, yielding
483-- elements from both streams alternately.
486-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
492-- Any exceptions generated by a constituent stream are propagated to the
493-- output stream. The output and exceptions from a single stream are guaranteed
494-- to arrive in the same order in the resulting stream as they were generated
495-- in the input stream. However, the relative ordering of elements from
496-- different streams in the resulting stream can vary depending on scheduling
497-- and generation delays.
499-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
500-- concurrently using a round robin scheduling.
504-- import qualified "Streamly.Prelude" as S
505-- import Control.Concurrent
507-- main = 'runStream' . 'wAsyncly' $ do
508-- n <- return 3 \<\> return 2 \<\> return 1
510-- threadDelay (n * 1000000)
511-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
514-- ThreadId 40: Delay 1
515-- ThreadId 39: Delay 2
516-- ThreadId 38: Delay 3
519-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
520-- concurrently, unconditionally.
522-- Note that async composition with breadth first traversal can only combine a
523-- finite number of streams as it needs to retain state for each unfinished
527newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
528 deriving (MonadTrans)
530-- | A round robin parallely composing IO stream of elements of type @a@.
531-- See 'WAsyncT' documentation for more details.
534type WAsync a = WAsyncT IO a
536-- | Fix the type of a polymorphic stream as 'WAsyncT'.
539wAsyncly :: IsStream t => WAsyncT m a -> t m a
542instance IsStream WAsyncT where
543 toStream = getWAsyncT
547 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
548 consM m r = fromStream $ consMWAsync m (toStream r)
551 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
554------------------------------------------------------------------------------
556------------------------------------------------------------------------------
558instance MonadAsync m => Semigroup (WAsyncT m a) where
561------------------------------------------------------------------------------
563------------------------------------------------------------------------------
565instance MonadAsync m => Monoid (WAsyncT m a) where
569------------------------------------------------------------------------------
571------------------------------------------------------------------------------
573instance MonadAsync m => Monad (WAsyncT m) where
576 WAsyncT $ K.bindWith wAsyncS m (getWAsyncT . f)
578------------------------------------------------------------------------------
580------------------------------------------------------------------------------
582MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
583MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list