git.iko.soy/streamly
on f580793 help sign in
streamly/src/Asyncly/Core.hs
617 lines 25 kB history raw
1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE UndecidableInstances #-} -- XXX
8-- |
9-- Module : Asyncly.Core
10-- Copyright : (c) 2017 Harendra Kumar
11--
12-- License : BSD3
13-- Maintainer : harendra.kumar@gmail.com
14-- Stability : experimental
15-- Portability : GHC
16--
17--
18module Asyncly.Core
19 (
20 MonadAsync
22 -- * Streams
23 , Stream (..)
24 , yields
25 , interleave
27 -- * Concurrent Stream Vars (SVars)
28 , SVar
29 , SVarSched (..)
30 , SVarTag (..)
31 , SVarStyle (..)
32 , EndOfStream (..)
33 , newSVar1
34 , newSVar2
35 , joinSVar2
36 , streamSVar
38 -- * Concurrent Streams
39 , parAlt
40 , parLeft
41 )
42where
44import Control.Applicative (Alternative (..))
45import Control.Concurrent (ThreadId, forkIO,
46 myThreadId, threadDelay)
47import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
48 tryPutMVar, takeMVar)
49import Control.Exception (SomeException (..), Exception)
50import qualified Control.Exception.Lifted as EL
51import Control.Monad (MonadPlus(..), mzero, when)
52import Control.Monad.Base (MonadBase (..), liftBaseDefault)
53import Control.Monad.Catch (MonadThrow, throwM)
54import Control.Monad.Error.Class (MonadError(..))
55import Control.Monad.IO.Class (MonadIO(..))
56import Control.Monad.Reader.Class (MonadReader(..))
57import Control.Monad.State.Class (MonadState(..))
58import Control.Monad.Trans.Class (MonadTrans (lift))
59import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
60import Data.Atomics (atomicModifyIORefCAS,
61 atomicModifyIORefCAS_)
62import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
63 tryPopR, nullQ)
64import Data.Functor (void)
65import Data.IORef (IORef, modifyIORef, newIORef,
66 readIORef)
67import Data.Maybe (isNothing)
68import Data.Semigroup (Semigroup(..))
69import Data.Set (Set)
70import qualified Data.Set as S
72------------------------------------------------------------------------------
73-- Parent child thread communication type
74------------------------------------------------------------------------------
76-- | Events that a child thread may send to a parent thread.
77data ChildEvent a =
78 ChildYield a
79 | ChildStop ThreadId (Maybe SomeException)
81------------------------------------------------------------------------------
82-- State threaded around the monad for thread management
83------------------------------------------------------------------------------
85-- | Conjunction is used for monadic/product style composition. Disjunction is
86-- used for fold/sum style composition. We need to distiguish the two types of
87-- SVars so that the scheduling of the two is independent.
88data SVarTag = Conjunction | Disjunction deriving Eq
90-- | For fairly interleaved parallel composition the sched policy is FIFO
91-- whereas for left biased parallel composition it is LIFO.
92data SVarSched = LIFO | FIFO deriving Eq
94-- | Identify the type of the SVar. Two computations using the same style can
95-- be scheduled on the same SVar.
96data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
98-- | An SVar (A Stream Var or an Sched Var) is a conduit to multiple streams
99-- running concurrently. It has an associated runqueue that holds the streams
100-- to be picked and run by a pool of worker threads. It has an associated
101-- output queue where the output stream elements are placed by the worker
102-- threads. A doorBell is used by the worker threads to intimate the consumer
103-- thread about availability of new results in the output queue. More workers
104-- are added to the SVar by 'streamSVar' on demand if the output produced is
105-- not keeping pace with the consumer. On bounded SVars, workers block on the
106-- output queue to provide throttling when the consumer is not pulling fast
107-- enough. The number of workers may even get reduced depending on the
108-- consuming pace.
109--
110-- New work is enqueued either at the time of creation of the SVar or as a
111-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
112-- already enqueued computations get evaluated. See 'joinSVar2'.
113data SVar m a =
114 SVar { outputQueue :: IORef [ChildEvent a]
115 , doorBell :: MVar Bool -- wakeup mechanism for outQ
116 , enqueue :: Stream m a -> IO ()
117 , runqueue :: m ()
118 , runningThreads :: IORef (Set ThreadId)
119 , queueEmpty :: m Bool
120 , svarStyle :: SVarStyle
121 }
123------------------------------------------------------------------------------
124-- The stream type
125------------------------------------------------------------------------------
127-- TBD use a functor instead of the bare type a?
129-- | Represents a monadic stream of values of type 'a' constructed using
130-- actions in monad 'm'. Streams can be composed sequentially or in parallel;
131-- in product style compositions (monadic bind multiplies streams in a ListT
132-- fashion) or in sum style compositions like 'Semigroup', 'Monoid',
133-- 'Alternative' or variants of these.
134newtype Stream m a =
135 Stream {
136 runStream :: forall r.
137 Maybe (SVar m a) -- local state
138 -> m r -- stop
139 -> (a -> Maybe (Stream m a) -> m r) -- yield
140 -> m r
141 }
143-- | A monad that can perform asynchronous/concurrent IO operations.
144type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
146-- | Yield a singleton value as a stream.
147yields :: a -> Stream m a
148yields a = Stream $ \_ _ yld -> yld a Nothing
150------------------------------------------------------------------------------
151-- Semigroup
152------------------------------------------------------------------------------
154-- | '<>' concatenates two streams sequentially i.e. the first stream is
155-- exhausted completely before yielding any element from the second stream.
156instance Semigroup (Stream m a) where
157 m1 <> m2 = go m1
158 where
159 go (Stream m) = Stream $ \_ stp yld ->
160 let stop = (runStream m2) Nothing stp yld
161 yield a Nothing = yld a (Just m2)
162 yield a (Just r) = yld a (Just (go r))
163 in m Nothing stop yield
165------------------------------------------------------------------------------
166-- Monoid
167------------------------------------------------------------------------------
169instance Monoid (Stream m a) where
170 mempty = Stream $ \_ stp _ -> stp
171 mappend = (<>)
173------------------------------------------------------------------------------
174-- Interleave
175------------------------------------------------------------------------------
177-- | Same as '<=>'.
178interleave :: Stream m a -> Stream m a -> Stream m a
179interleave m1 m2 = Stream $ \_ stp yld -> do
180 let stop = (runStream m2) Nothing stp yld
181 yield a Nothing = yld a (Just m2)
182 yield a (Just r) = yld a (Just (interleave m2 r))
183 (runStream m1) Nothing stop yield
185------------------------------------------------------------------------------
186-- Spawning threads and collecting result in streamed fashion
187------------------------------------------------------------------------------
189{-# INLINE doFork #-}
190doFork :: MonadBaseControl IO m
191 => m ()
192 -> (SomeException -> m ())
193 -> m ThreadId
194doFork action exHandler =
195 EL.mask $ \restore ->
196 liftBaseWith $ \runInIO -> forkIO $ do
197 -- XXX test the exception handling
198 _ <- runInIO $ EL.catch (restore action) exHandler
199 -- XXX restore state here?
200 return ()
202-- XXX exception safety of all atomic/MVar operations
204{-# INLINE send #-}
205send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
206send sv msg = liftIO $ do
207 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
208 -- XXX need a memory barrier? The wake up must happen only after the
209 -- store has finished otherwise we can have lost wakeup problems.
210 void $ tryPutMVar (doorBell sv) True
212{-# INLINE sendStop #-}
213sendStop :: MonadIO m => SVar m a -> m ()
214sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
216-- Note: Left associated compositions can grow this queue to a large size
217{-# INLINE enqueueLIFO #-}
218enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
219enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
221runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
222runqueueLIFO sv q = run
224 where
226 run = do
227 work <- dequeue
228 case work of
229 Nothing -> sendStop sv
230 Just m -> (runStream m) (Just sv) run yield
232 sendit a = send sv (ChildYield a)
233 yield a Nothing = sendit a >> run
234 yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
236 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
237 case ms of
238 [] -> ([], Nothing)
239 x : xs -> (xs, Just x)
241{-# INLINE enqueueFIFO #-}
242enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
243enqueueFIFO = pushL
245runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
246runqueueFIFO sv q = run
248 where
250 run = do
251 work <- dequeue
252 case work of
253 Nothing -> sendStop sv
254 Just m -> (runStream m) (Just sv) run yield
256 dequeue = liftIO $ tryPopR q
257 sendit a = send sv (ChildYield a)
258 yield a Nothing = sendit a >> run
259 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
261-- Thread tracking is needed for two reasons:
262--
263-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
264-- themselves because they may run for significant times before going away or
265-- worse they may be stuck in IO and never go away.
266--
267-- 2) To know when all threads are done.
269{-# NOINLINE addThread #-}
270addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
271addThread sv tid =
272 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.insert tid s)
274{-# INLINE delThread #-}
275delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
276delThread sv tid =
277 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
279{-# INLINE allThreadsDone #-}
280allThreadsDone :: MonadIO m => SVar m a -> m Bool
281allThreadsDone sv = liftIO $ do
282 readIORef (runningThreads sv) >>= return . S.null
284{-# NOINLINE handleChildException #-}
285handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
286handleChildException sv e = do
287 tid <- liftIO myThreadId
288 send sv (ChildStop tid (Just e))
290{-# NOINLINE pushWorker #-}
291pushWorker :: MonadAsync m => SVar m a -> m ()
292pushWorker sv =
293 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
295-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
296-- Also, if a worker blocks on the output queue we can decide if we want to
297-- block or make it go away entirely, depending on the number of workers and
298-- the type of the queue.
299{-# INLINE sendWorkerWait #-}
300sendWorkerWait :: MonadAsync m => SVar m a -> m ()
301sendWorkerWait sv = do
302 case svarStyle sv of
303 SVarStyle _ LIFO -> liftIO $ threadDelay 200
304 SVarStyle _ FIFO -> liftIO $ threadDelay 0
306 output <- liftIO $ readIORef (outputQueue sv)
307 when (null output) $ do
308 done <- queueEmpty sv
309 if (not done)
310 then (pushWorker sv) >> sendWorkerWait sv
311 else void (liftIO $ takeMVar (doorBell sv))
313-- | A stream being pulled from 'SVar' has ended.
314data EndOfStream = EndOfStream deriving Show
315instance Exception EndOfStream
317-- | Pull a stream from an SVar.
318{-# NOINLINE streamSVar #-}
319streamSVar :: MonadAsync m => SVar m a -> Stream m a
320streamSVar sv = Stream $ \_ stp yld -> do
321 -- XXX if reading the IORef is costly we can use a flag in the SVar to
322 -- indicate we are done.
323 done <- allThreadsDone sv
324 when done $ throwM EndOfStream
326 res <- liftIO $ tryTakeMVar (doorBell sv)
327 when (isNothing res) $ sendWorkerWait sv
328 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
329 -- To avoid lock overhead we read all events at once instead of reading one
330 -- at a time. We just reverse the list to process the events in the order
331 -- they arrived. Maybe we can use a queue instead?
332 (runStream $ processEvents (reverse list)) Nothing stp yld
334 where
336 handleException e tid = do
337 delThread sv tid
338 -- XXX implement kill async exception handling
339 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
340 throwM e
342 {-# INLINE processEvents #-}
343 processEvents [] = Stream $ \_ stp yld -> do
344 done <- allThreadsDone sv
345 if not done
346 then (runStream (streamSVar sv)) Nothing stp yld
347 else stp
349 processEvents (ev : es) = Stream $ \_ stp yld -> do
350 let continue = (runStream (processEvents es)) Nothing stp yld
351 yield a = yld a (Just (processEvents es))
353 case ev of
354 ChildYield a -> yield a
355 ChildStop tid e ->
356 case e of
357 Nothing -> delThread sv tid >> continue
358 Just ex -> handleException ex tid
360getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
361getFifoSVar ctype = do
362 outQ <- newIORef []
363 outQMv <- newEmptyMVar
364 running <- newIORef S.empty
365 q <- newQ
366 let sv =
367 SVar { outputQueue = outQ
368 , doorBell = outQMv
369 , runningThreads = running
370 , runqueue = runqueueFIFO sv q
371 , enqueue = pushL q
372 , queueEmpty = liftIO $ nullQ q
373 , svarStyle = ctype
374 }
375 in return sv
377getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
378getLifoSVar ctype = do
379 outQ <- newIORef []
380 outQMv <- newEmptyMVar
381 running <- newIORef S.empty
382 q <- newIORef []
383 let checkEmpty = liftIO (readIORef q) >>= return . null
384 let sv =
385 SVar { outputQueue = outQ
386 , doorBell = outQMv
387 , runningThreads = running
388 , runqueue = runqueueLIFO sv q
389 , enqueue = enqueueLIFO q
390 , queueEmpty = checkEmpty
391 , svarStyle = ctype
392 }
393 in return sv
395-- | Create a new SVar and enqueue one stream computation on it.
396newSVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
397newSVar1 style m = do
398 sv <- liftIO $
399 case style of
400 SVarStyle _ FIFO -> do
401 c <- getFifoSVar style
402 return c
403 SVarStyle _ LIFO -> do
404 c <- getLifoSVar style
405 return c
406 -- Note: We must have all the work on the queue before sending the
407 -- pushworker, otherwise the pushworker may exit before we even get a
408 -- chance to push.
409 liftIO $ (enqueue sv) m
410 pushWorker sv
411 return sv
413-- | Create a new SVar and enqueue two stream computations on it.
414newSVar2 :: MonadAsync m
415 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
416newSVar2 style m1 m2 = do
417 -- Note: We must have all the work on the queue before sending the
418 -- pushworker, otherwise the pushworker may exit before we even get a
419 -- chance to push.
420 sv <- liftIO $
421 case style of
422 SVarStyle _ FIFO -> do
423 c <- getFifoSVar style
424 (enqueue c) m1 >> (enqueue c) m2
425 return c
426 SVarStyle _ LIFO -> do
427 c <- getLifoSVar style
428 (enqueue c) m2 >> (enqueue c) m1
429 return c
430 pushWorker sv
431 return sv
433------------------------------------------------------------------------------
434-- Running streams concurrently
435------------------------------------------------------------------------------
437-- Concurrency rate control. Our objective is to create more threads on demand
438-- if the consumer is running faster than us. As soon as we encounter an
439-- Alternative composition we create a push pull pair of threads. We use a
440-- channel for communication between the consumer pulling from the channel and
441-- the producer who pushing to the channel. The producer creates more threads
442-- if no output is seen on the channel, that is the consumer is running faster.
443-- However this mechanism can be problematic if the initial production latency
444-- is high, we may end up creating too many threads. So we need some way to
445-- monitor and use the latency as well.
446--
447-- TBD We may run computations at the lower level of the composition tree
448-- serially even if they are composed using a parallel combinator. We can use
449-- <> in place of <| and <=> in place of <|>. If we find that a parallel
450-- channel immediately above a computation becomes empty we can switch to
451-- parallelizing the computation. For that we can use a state flag to fork the
452-- rest of the computation at any point of time inside the Monad bind operation
453-- if the consumer is running at a faster speed.
454--
455-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
456-- If we have to dispatch in arbitrary chunksizes we will need to compose the
457-- parallel actions using a data constructor (Free Alternative) instead so that
458-- we can divide it in chunks of arbitrary size before dispatch. If the stream
459-- is composed of hierarchically composed grains of different sizes then we can
460-- always switch to a desired granularity depending on the consumer speed.
461--
462-- TBD for pure work (when we are not in the IO monad) we can divide it into
463-- just the number of CPUs.
465{-# NOINLINE withNewSVar2 #-}
466withNewSVar2 :: MonadAsync m
467 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
468withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
469 sv <- newSVar2 style m1 m2
470 (runStream (streamSVar sv)) Nothing stp yld
472-- | Join two computations on the currently running 'SVar' queue for concurrent
473-- execution. The 'SVarStyle' required by the current composition context is
474-- passed as one of the parameters. If the style does not match with the style
475-- of the current 'SVar' we create a new 'SVar' and schedule the computations
476-- on that. The newly created SVar joins as one of the computations on the
477-- current SVar queue.
478--
479-- When we are using parallel composition, an SVar is passed around as a state
480-- variable. We try to schedule a new parallel computation on the SVar passed
481-- to us. The first time, when no SVar exists, a new SVar is created.
482-- Subsequently, 'joinSVar2' may get called when a computation already
483-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
484-- evaluated it calls a 'joinSVar2' to put 'a' and 'b' on the current scheduler
485-- queue. However, if the scheduling and composition style of the new
486-- computation being scheduled is different than the style of the current SVar,
487-- then we create a new SVar and schedule it on that.
488--
489-- For example:
490--
491-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
492-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
493-- because of the scheduling policy change.
494-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
495-- 'ParallelT', we create a new SVar at the transitioning bind.
496-- * When the stream is switching from disjunctive composition to conjunctive
497-- composition and vice-versa we create a new SVar to isolate the scheduling
498-- of the two.
499--
500{-# INLINE joinSVar2 #-}
501joinSVar2 :: MonadAsync m
502 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
503joinSVar2 style m1 m2 = Stream $ \st stp yld -> do
504 case st of
505 Just sv | svarStyle sv == style ->
506 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
507 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
509------------------------------------------------------------------------------
510-- Semigroup and Monoid style compositions for parallel actions
511------------------------------------------------------------------------------
513{-
514-- | Same as '<>|'.
515parAhead :: Stream m a -> Stream m a -> Stream m a
516parAhead = undefined
518-- | Sequential composition similar to '<>' except that it can execute the
519-- action on the right in parallel ahead of time. Returns the results in
520-- sequential order like '<>' from left to right.
521(<>|) :: Stream m a -> Stream m a -> Stream m a
522(<>|) = parAhead
523-}
525-- | Same as '<|>'. Since this schedules all the composed streams fairly you
526-- cannot fold infinite number of streams using this operation.
527{-# INLINE parAlt #-}
528parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
529parAlt = joinSVar2 (SVarStyle Disjunction FIFO)
531-- | Same as '<|'. Since this schedules the left side computation first you can
532-- right fold an infinite container using this operator. However a left fold
533-- will not work well as it first unpeels the whole structure before scheduling
534-- a computation requiring an amount of memory proportional to the size of the
535-- structure.
536{-# INLINE parLeft #-}
537parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
538parLeft = joinSVar2 (SVarStyle Disjunction LIFO)
540-------------------------------------------------------------------------------
541-- Instances (only used for deriving newtype instances)
542-------------------------------------------------------------------------------
544-- Stream type is not exposed, these instances are only for deriving instances
545-- for the newtype wrappers based on Stream.
547-- Dummy Instances, defined to enable the definition of other instances that
548-- require a Monad constraint. Must be defined by the newtypes.
550instance Monad m => Functor (Stream m) where
551 fmap = undefined
553instance Monad m => Applicative (Stream m) where
554 pure = undefined
555 (<*>) = undefined
557instance Monad m => Monad (Stream m) where
558 return = pure
559 (>>=) = undefined
561------------------------------------------------------------------------------
562-- Alternative & MonadPlus
563------------------------------------------------------------------------------
565-- | `empty` represents an action that takes non-zero time to complete. Since
566-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
567-- monoidal composition executing all actions in parallel, it is similar to
568-- '<>' except that it runs all the actions in parallel and interleaves their
569-- results fairly.
570instance MonadAsync m => Alternative (Stream m) where
571 empty = mempty
572 (<|>) = parAlt
574instance MonadAsync m => MonadPlus (Stream m) where
575 mzero = empty
576 mplus = (<|>)
578-------------------------------------------------------------------------------
579-- Transformer
580-------------------------------------------------------------------------------
582instance MonadTrans Stream where
583 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
585instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
586 liftBase = liftBaseDefault
588------------------------------------------------------------------------------
589-- Standard transformer instances
590------------------------------------------------------------------------------
592instance MonadIO m => MonadIO (Stream m) where
593 liftIO = lift . liftIO
595instance MonadThrow m => MonadThrow (Stream m) where
596 throwM = lift . throwM
598-- XXX handle and test cross thread state transfer
599instance MonadError e m => MonadError e (Stream m) where
600 throwError = lift . throwError
601 catchError m h = Stream $ \st stp yld ->
602 let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
603 yield a Nothing = yld a Nothing
604 yield a (Just r) = yld a (Just (catchError r h))
605 in handle $ (runStream m) st stp yield
607instance MonadReader r m => MonadReader r (Stream m) where
608 ask = lift ask
609 local f m = Stream $ \st stp yld ->
610 let yield a Nothing = local f $ yld a Nothing
611 yield a (Just r) = local f $ yld a (Just (local f r))
612 in (runStream m) st (local f stp) yield
614instance MonadState s m => MonadState s (Stream m) where
615 get = lift get
616 put x = lift (put x)
617 state k = lift (state k)