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 $ sendYield maxBuf sv (ChildYield a)
76 if res then stop else liftIO $ sendStop sv
78 res <- liftIO $ sendYield 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 $ sendYield maxBuf sv (ChildYield a)
101 if res then stop else liftIO $ sendStop sv
103 res <- liftIO $ sendYield 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
124 yl <- case yieldLimit st of
125 Nothing -> return Nothing
126 Just x -> Just <$> newIORef x
134 let checkEmpty = null <$> readIORef q
136 SVar { outputQueue = outQ
138 , outputDoorBell = outQMv
139 , readOutputQ = readOutputQBounded (threadsHigh st) sv
140 , postProcess = postProcessBounded sv
141 , workerThreads = running
142 , workLoop = workLoopLIFO runStreamLIFO
143 st{streamVar = Just sv} q
144 , enqueue = enqueueLIFO sv q
145 , isWorkDone = checkEmpty
147 , svarStyle = AsyncVar
148 , workerCount = active
149 , accountThread = delThread sv
151 , aheadWorkQueue = undefined
152 , outputHeap = undefined
153 , maxWorkers = maxWrk
154 , totalDispatches = disp
155 , maxOutQSize = maxOq
156 , maxHeapSize = maxHs
157 , maxWorkQSize = maxWq
162getFifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)
164 outQ <- newIORef ([], 0)
165 outQMv <- newEmptyMVar
167 wfw <- newIORef False
168 running <- newIORef S.empty
170 yl <- case yieldLimit st of
171 Nothing -> return Nothing
172 Just x -> Just <$> newIORef x
181 SVar { outputQueue = outQ
183 , outputDoorBell = outQMv
184 , readOutputQ = readOutputQBounded (threadsHigh st) sv
185 , postProcess = postProcessBounded sv
186 , workerThreads = running
187 , workLoop = workLoopFIFO runStreamFIFO
188 st{streamVar = Just sv} q
189 , enqueue = enqueueFIFO sv q
190 , isWorkDone = nullQ q
192 , svarStyle = WAsyncVar
193 , workerCount = active
194 , accountThread = delThread sv
196 , aheadWorkQueue = undefined
197 , outputHeap = undefined
198 , totalDispatches = disp
199 , maxWorkers = maxWrk
200 , maxOutQSize = maxOq
201 , maxHeapSize = maxHs
202 , maxWorkQSize = maxWq
207{-# INLINABLE newAsyncVar #-}
208newAsyncVar :: MonadAsync m
209 => State Stream m a -> Stream m a -> m (SVar Stream m a)
211 sv <- liftIO $ getLifoSVar st
214-- XXX Get rid of this?
215-- | Make a stream asynchronous, triggers the computation and returns a stream
216-- in the underlying monad representing the output generated by the original
217-- computation. The returned action is exhaustible and must be drained once. If
218-- not drained fully we may have a thread blocked forever and once exhausted it
219-- will always return 'empty'.
222{-# INLINABLE mkAsync #-}
223mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
224mkAsync m = newAsyncVar defState (toStream m) >>= return . fromSVar
226{-# INLINABLE mkAsync' #-}
227mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
228mkAsync' st m = newAsyncVar st (toStream m) >>= return . fromSVar
230-- | Create a new SVar and enqueue one stream computation on it.
231{-# INLINABLE newWAsyncVar #-}
232newWAsyncVar :: MonadAsync m
233 => State Stream m a -> Stream m a -> m (SVar Stream m a)
234newWAsyncVar st m = do
235 sv <- liftIO $ getFifoSVar st
238------------------------------------------------------------------------------
239-- Running streams concurrently
240------------------------------------------------------------------------------
242-- Concurrency rate control.
244-- Our objective is to create more threads on demand if the consumer is running
245-- faster than us. As soon as we encounter a concurrent composition we create a
246-- push pull pair of threads. We use an SVar for communication between the
247-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
248-- The producer creates more threads if the SVar drains and becomes empty, that
249-- is the consumer is running faster.
251-- XXX Note 1: This mechanism can be problematic if the initial production
252-- latency is high, we may end up creating too many threads. So we need some
253-- way to monitor and use the latency as well. Having a limit on the dispatches
254-- (programmer controlled) may also help.
256-- TBD Note 2: We may want to run computations at the lower level of the
257-- composition tree serially even when they are composed using a parallel
258-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
259-- place of 'wAsync'. If we find that an SVar immediately above a computation
260-- gets drained empty we can switch to parallelizing the computation. For that
261-- we can use a state flag to fork the rest of the computation at any point of
262-- time inside the Monad bind operation if the consumer is running at a faster
265-- TBD Note 3: the binary operation ('parallel') composition allows us to
266-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
267-- chunksizes we will need to compose the parallel actions using a data
268-- constructor (A Free container) instead so that we can divide it in chunks of
269-- arbitrary size before dispatching. If the stream is composed of
270-- hierarchically composed grains of different sizes then we can always switch
271-- to a desired granularity depending on the consumer speed.
273-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
274-- into just the number of CPUs.
276-- | Join two computations on the currently running 'SVar' queue for concurrent
277-- execution. When we are using parallel composition, an SVar is passed around
278-- as a state variable. We try to schedule a new parallel computation on the
279-- SVar passed to us. The first time, when no SVar exists, a new SVar is
280-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
281-- already scheduled on the SVar is further evaluated. For example, when (a
282-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
283-- the current scheduler queue.
285-- The 'SVarStyle' required by the current composition context is passed as one
286-- of the parameters. If the scheduling and composition style of the new
287-- computation being scheduled is different than the style of the current SVar,
288-- then we create a new SVar and schedule it on that. The newly created SVar
289-- joins as one of the computations on the current SVar queue.
291-- Cases when we need to switch to a new SVar:
293-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
294-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
295-- because of the scheduling policy change.
296-- * if we 'adapt' a stream of type 'async' to a stream of type
297-- 'Parallel', we create a new SVar at the transitioning bind.
298-- * When the stream is switching from disjunctive composition to conjunctive
299-- composition and vice-versa we create a new SVar to isolate the scheduling
302forkSVarAsync :: MonadAsync m
303 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
304forkSVarAsync style m1 m2 = Stream $ \st stp sng yld -> do
306 AsyncVar -> newAsyncVar st (concurrently m1 m2)
307 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
308 _ -> error "illegal svar type"
309 unStream (fromSVar sv) (rstState st) stp sng yld
311 concurrently ma mb = Stream $ \st stp sng yld -> do
312 liftIO $ enqueue (fromJust $ streamVar st) mb
313 unStream ma st stp sng yld
315{-# INLINE joinStreamVarAsync #-}
316joinStreamVarAsync :: MonadAsync m
317 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
318joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld -> do
320 Just sv | svarStyle sv == style ->
321 liftIO (enqueue sv m2) >> unStream m1 st stp sng yld
322 _ -> unStream (forkSVarAsync style m1 m2) st stp sng yld
324------------------------------------------------------------------------------
325-- Semigroup and Monoid style compositions for parallel actions
326------------------------------------------------------------------------------
329asyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
330asyncS = joinStreamVarAsync AsyncVar
332-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
333-- Merges two streams possibly concurrently, preferring the
334-- elements from the left one when available.
338async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
339async m1 m2 = fromStream $
340 joinStreamVarAsync AsyncVar (toStream m1) (toStream m2)
345{-# DEPRECATED (<|) "Please use 'async' instead." #-}
347(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
350-- | XXX we can implement it more efficienty by directly implementing instead
351-- of combining streams using async.
352{-# INLINE consMAsync #-}
353consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
354consMAsync m r = K.yieldM m `asyncS` r
356------------------------------------------------------------------------------
358------------------------------------------------------------------------------
360-- | Deep async composition or async composition with depth first traversal. In
361-- a left to right 'Semigroup' composition it tries to yield elements from the
362-- left stream as long as it can, but it can run the right stream in parallel
363-- if it needs to, based on demand. The right stream can be run if the left
364-- stream blocks on IO or cannot produce elements fast enough for the consumer.
367-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
373-- Any exceptions generated by a constituent stream are propagated to the
374-- output stream. The output and exceptions from a single stream are guaranteed
375-- to arrive in the same order in the resulting stream as they were generated
376-- in the input stream. However, the relative ordering of elements from
377-- different streams in the resulting stream can vary depending on scheduling
378-- and generation delays.
380-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
381-- concurrently based on demand. More concurrent iterations are started only
382-- if the previous iterations are not able to produce enough output for the
387-- import qualified "Streamly.Prelude" as S
388-- import Control.Concurrent
390-- main = 'runStream' . 'asyncly' $ do
391-- n <- return 3 \<\> return 2 \<\> return 1
393-- threadDelay (n * 1000000)
394-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
397-- ThreadId 40: Delay 1
398-- ThreadId 39: Delay 2
399-- ThreadId 38: Delay 3
402-- All iterations may run in the same thread if they do not block.
404-- Note that async composition with depth first traversal can be used to
405-- combine infinite number of streams as it explores only a bounded number of
409newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
410 deriving (MonadTrans)
412-- | A demand driven left biased parallely composing IO stream of elements of
413-- type @a@. See 'AsyncT' documentation for more details.
416type Async a = AsyncT IO a
418-- | Fix the type of a polymorphic stream as 'AsyncT'.
421asyncly :: IsStream t => AsyncT m a -> t m a
424instance IsStream AsyncT where
429 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
430 consM m r = fromStream $ consMAsync m (toStream r)
433 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
436------------------------------------------------------------------------------
438------------------------------------------------------------------------------
440instance MonadAsync m => Semigroup (AsyncT m a) where
443------------------------------------------------------------------------------
445------------------------------------------------------------------------------
447instance MonadAsync m => Monoid (AsyncT m a) where
451------------------------------------------------------------------------------
453------------------------------------------------------------------------------
455instance MonadAsync m => Monad (AsyncT m) where
457 (AsyncT m) >>= f = AsyncT $ K.bindWith asyncS m (getAsyncT . f)
459------------------------------------------------------------------------------
461------------------------------------------------------------------------------
463MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
464MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
466------------------------------------------------------------------------------
468------------------------------------------------------------------------------
470{-# INLINE wAsyncS #-}
471wAsyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
472wAsyncS = joinStreamVarAsync WAsyncVar
474-- | XXX we can implement it more efficienty by directly implementing instead
475-- of combining streams using wAsync.
476{-# INLINE consMWAsync #-}
477consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
478consMWAsync m r = K.yieldM m `wAsyncS` r
480-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
481-- Merges two streams concurrently choosing elements from both fairly.
485wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
486wAsync m1 m2 = fromStream $ wAsyncS (toStream m1) (toStream m2)
488-- | Wide async composition or async composition with breadth first traversal.
489-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
490-- streams using a depth first travesal or in a round robin fashion, yielding
491-- elements from both streams alternately.
494-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
500-- Any exceptions generated by a constituent stream are propagated to the
501-- output stream. The output and exceptions from a single stream are guaranteed
502-- to arrive in the same order in the resulting stream as they were generated
503-- in the input stream. However, the relative ordering of elements from
504-- different streams in the resulting stream can vary depending on scheduling
505-- and generation delays.
507-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
508-- concurrently using a round robin scheduling.
512-- import qualified "Streamly.Prelude" as S
513-- import Control.Concurrent
515-- main = 'runStream' . 'wAsyncly' $ do
516-- n <- return 3 \<\> return 2 \<\> return 1
518-- threadDelay (n * 1000000)
519-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
522-- ThreadId 40: Delay 1
523-- ThreadId 39: Delay 2
524-- ThreadId 38: Delay 3
527-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
528-- concurrently, unconditionally.
530-- Note that async composition with breadth first traversal can only combine a
531-- finite number of streams as it needs to retain state for each unfinished
535newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
536 deriving (MonadTrans)
538-- | A round robin parallely composing IO stream of elements of type @a@.
539-- See 'WAsyncT' documentation for more details.
542type WAsync a = WAsyncT IO a
544-- | Fix the type of a polymorphic stream as 'WAsyncT'.
547wAsyncly :: IsStream t => WAsyncT m a -> t m a
550instance IsStream WAsyncT where
551 toStream = getWAsyncT
555 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
556 consM m r = fromStream $ consMWAsync m (toStream r)
559 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
562------------------------------------------------------------------------------
564------------------------------------------------------------------------------
566instance MonadAsync m => Semigroup (WAsyncT m a) where
569------------------------------------------------------------------------------
571------------------------------------------------------------------------------
573instance MonadAsync m => Monoid (WAsyncT m a) where
577------------------------------------------------------------------------------
579------------------------------------------------------------------------------
581instance MonadAsync m => Monad (WAsyncT m) where
584 WAsyncT $ K.bindWith wAsyncS m (getWAsyncT . f)
586------------------------------------------------------------------------------
588------------------------------------------------------------------------------
590MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
591MONAD_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