git.iko.soy/streamly
on 02a745f help sign in
streamly/src/Asyncly/AsyncT.hs
492 lines 21 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.AsyncT
10-- Copyright : (c) 2017 Harendra Kumar
11--
12-- License : MIT-style
13-- Maintainer : harendra.kumar@gmail.com
14-- Stability : experimental
15-- Portability : GHC
16--
17--
18module Asyncly.AsyncT
19 ( AsyncT (..)
20 , MonadAsync
21 , yield
22-- , async
23-- , makeAsync
24 )
25where
27import Control.Applicative (Alternative (..))
28import Control.Concurrent (ThreadId, forkIO, killThread,
29 myThreadId)
30import Control.Concurrent.STM (TChan, atomically, newTChan,
31 readTChan, tryReadTChan,
32 writeTChan)
33import Control.Exception (SomeException (..))
34import qualified Control.Exception.Lifted as EL
35import Control.Monad (ap, liftM, MonadPlus(..), mzero)
36--import Control.Monad.Base (MonadBase (..), liftBaseDefault)
37import Control.Monad.Catch (MonadThrow, throwM)
38import Control.Monad.IO.Class (MonadIO(..))
39import Control.Monad.Trans.Class (MonadTrans (lift))
40import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
41{-
42import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
43 MonadTransControl (..),
44 defaultLiftBaseWith,
45 defaultRestoreM, liftBaseWith)
46 -}
47import Data.Functor (void)
48import Data.IORef (IORef, modifyIORef, newIORef,
49 writeIORef, readIORef)
50import Data.Maybe (maybe)
51import Data.Monoid ((<>))
52import Data.Set (Set)
53import qualified Data.Set as S
55import Control.Monad.Trans.Recorder (MonadRecorder(..))
58------------------------------------------------------------------------------
59-- Concurrency Semantics
60------------------------------------------------------------------------------
61--
62-- Asyncly is essentially a concurrent list transformer. To concatenate the
63-- lists it provides three different ways of composing them. The Monoid
64-- instance concatenates lists in a non-concurrent in-order fashion. The
65-- Alternative instance concatenates list in a concurrent manner potentially
66-- running each action in parallel and therefore the order of the resulting
67-- items is not deterministic. Computations in Alternative composition may or
68-- may not run in parallel. Thirdly, the <||> operator provides a way to
69-- compose computations that are guaranteed to run in parallel. This provides a
70-- way to run infinite loops or long running computations without blocking
71-- or starving other computations.
73------------------------------------------------------------------------------
74-- Parent child thread communication types
75------------------------------------------------------------------------------
77data ChildEvent a =
78 ChildYield a
79 | ChildDone ThreadId a
80 | ChildStop ThreadId (Maybe SomeException)
81 | ChildCreate ThreadId
83------------------------------------------------------------------------------
84-- State threaded around the monad for thread management
85------------------------------------------------------------------------------
87data Context a =
88 Context { childChannel :: TChan (ChildEvent a)
89 , pullSide :: Bool
90 , runningThreads :: IORef (Set ThreadId)
91 , doneThreads :: IORef (Set ThreadId)
92 }
94-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the
95-- Nothing case, but it makes the fold using '<|>' 25% faster. Need to try
96-- again as the code has gone through many changes after this was tested.
97-- With the Maybe, 'stop' is required only to represent 'empty' in an
98-- Alternative composition.
99--
100-- Currently the only state we need is the thread context, For generality we
101-- can parameterize the type with a state type 's'.
102newtype AsyncT m a =
103 AsyncT {
104 runAsyncT :: forall r.
105 Maybe (Context a) -- state
106 -> m r -- stop
107 -> (a -> Maybe (Context a) -> Maybe (AsyncT m a) -> m r) -- yield
108 -> m r
109 }
111type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
113------------------------------------------------------------------------------
114-- Monad
115------------------------------------------------------------------------------
117-- | Appends the results of two AsyncT computations in order.
118instance Monad m => Monoid (AsyncT m a) where
119 mempty = AsyncT $ \_ stp _ -> stp
120 mappend (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
121 m1 ctx ((runAsyncT m2) ctx stp yld) $ \a _ r ->
122 let yielder x = yld a ctx (Just x)
123 in maybe (yielder m2) (\rx -> yielder $ mappend rx m2) r
125-- We do not use bind for parallelism. That is, we do not start each iteration
126-- of the list in parallel. That will introduce too much uncontrolled
127-- parallelism. Instead we control parallelism using <|>, it is used for
128-- actions that 'may' be run in parallel. The library decides whether they will
129-- be started in parallel. This puts the parallel semantics in the hands of the
130-- user and operational semantics in the hands of the library. Ideally we would
131-- want parallelism to be completely hidden from the user, for example we can
132-- automatically decide where to start an action in parallel so that it is most
133-- efficient. We can decide whether a particular bind should fork a parallel
134-- thread or not based on the level of the bind in the tree. Maybe at some
135-- point we shall be able to do that?
137instance Monad m => Monad (AsyncT m) where
138 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing
140 -- | A thread context is valid only until the next bind. Upon a bind we
141 -- reset the context to Nothing.
142 AsyncT m >>= f = AsyncT $ \_ stp yld ->
143 m Nothing stp $ \a _ r ->
144 let run x = (runAsyncT x) Nothing stp yld
145 next = f a
146 in maybe (run next) (\rx -> run $ next <> (rx >>= f)) r
148yield :: a -> AsyncT m a -> AsyncT m a
149yield a m = AsyncT $ \ctx _ yld -> yld a ctx (Just m)
151------------------------------------------------------------------------------
152-- Functor
153------------------------------------------------------------------------------
155instance Monad m => Functor (AsyncT m) where
156 fmap = liftM
158------------------------------------------------------------------------------
159-- Applicative
160------------------------------------------------------------------------------
162instance Monad m => Applicative (AsyncT m) where
163 pure = return
164 (<*>) = ap
166------------------------------------------------------------------------------
167-- Alternative
168------------------------------------------------------------------------------
170{-# INLINE doFork #-}
171doFork :: (MonadIO m, MonadBaseControl IO m)
172 => m ()
173 -> (SomeException -> IO ())
174 -> m ThreadId
175doFork action exHandler =
176 EL.mask $ \restore ->
177 liftBaseWith $ \runInIO -> forkIO $ do
178 -- XXX test the exception handling
179 _ <- runInIO $ EL.catch (restore action) (liftIO . exHandler)
180 -- XXX restore state here?
181 return ()
183{-# NOINLINE push #-}
184push :: MonadIO m => Context a -> AsyncT m a -> m ()
185push context action = run (Just context) action
187 where
189 run ctx m = (runAsyncT m) ctx channelStop yielder
191 -- XXX make this a bounded channel so that we block if the previous value
192 -- is not consumed yet.
193 chan = childChannel context
194 channelYield a = liftIO $ atomically $ writeTChan chan (ChildYield a)
195 channelDone a = do
196 tid <- liftIO myThreadId
197 liftIO $ atomically $ writeTChan chan (ChildDone tid a)
198 channelStop = do
199 tid <- liftIO myThreadId
200 liftIO $ atomically $ writeTChan chan (ChildStop tid Nothing)
202 done a = channelDone a
203 continue a ctx m = channelYield a >> run ctx m
204 yielder a ctx r = maybe (done a) (\rx -> continue a ctx rx) r
206-- If an exception occurs we push it to the channel so that it can handled by
207-- the parent. 'Paused' exceptions are to be collected at the top level.
208-- XXX Paused exceptions should only bubble up to the runRecorder computation
209{-# NOINLINE handleChildException #-}
210handleChildException :: TChan (ChildEvent a) -> SomeException -> IO ()
211handleChildException pchan e = do
212 tid <- myThreadId
213 atomically $ writeTChan pchan (ChildStop tid (Just e))
215{-# NOINLINE pushSideDispatch #-}
216pushSideDispatch :: MonadAsync m
217 => Context a -> AsyncT m a -> AsyncT m a -> AsyncT m a
218pushSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do
219 let chan = childChannel ctx
220 tid <- doFork (push ctx m1) (handleChildException chan)
221 liftIO $ atomically $ writeTChan chan (ChildCreate tid)
222 (runAsyncT m2) (Just ctx) stp yld
224-- This is a bit messy because ChildCreate and ChildDone events can arrive out
225-- of order in case of pushSideDispatch. Returns whether we are done draining
226-- threads.
227{-# INLINE accountThread #-}
228accountThread :: MonadIO m
229 => ThreadId -> IORef (Set ThreadId) -> IORef (Set ThreadId) -> m Bool
230accountThread tid ref1 ref2 = liftIO $ do
231 s1 <- readIORef ref1
232 s2 <- readIORef ref2
234 if (S.member tid s1) then do
235 let r = S.delete tid s1
236 writeIORef ref1 r
237 return $ S.null r && S.null s2
238 else do
239 liftIO $ writeIORef ref2 (S.insert tid s2)
240 return False
242{-# NOINLINE addThread #-}
243addThread :: MonadIO m => Context a -> ThreadId -> m Bool
244addThread ctx tid = accountThread tid (doneThreads ctx) (runningThreads ctx)
246{-# INLINE delThread #-}
247delThread :: MonadIO m => Context a -> ThreadId -> m Bool
248delThread ctx tid = accountThread tid (runningThreads ctx) (doneThreads ctx)
250handleException :: (MonadIO m, MonadThrow m)
251 => SomeException -> Context a -> ThreadId -> m r
252handleException e ctx tid = do
253 void (delThread ctx tid)
254 liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
255 throwM e
257-- XXX the TBQueue size should be proportional to the pending threads.
258-- We re-raise any exceptions received from the child threads, that way
259-- exceptions get propagated to the top level computation and can be handled
260-- there.
261{-# NOINLINE pullDispatch #-}
262pullDispatch :: (MonadIO m, MonadThrow m)
263 => Context a -> AsyncT m a -> Bool -> AsyncT m a
264pullDispatch ctx m dispatch = AsyncT $ \_ stp yld -> do
265 if dispatch then do
266 (runAsyncT m) (Just ctx) stp yld
267 else do
268 res <- liftIO $ atomically $ tryReadTChan (childChannel ctx)
269 maybe (continue stp yld) (\ev -> handleEvent ev stp yld) res
271 where
273 {-# INLINE continue #-}
274 continue stp yld = (runAsyncT (pullDispatch ctx m True)) (Just ctx) stp yld
276 {-# INLINE handleEvent #-}
277 handleEvent ev stp yld =
278 case ev of
279 ChildYield a -> yld a Nothing (Just (pullDispatch ctx m False))
280 ChildDone tid a -> do
281 void $ delThread ctx tid
282 yld a Nothing (Just (pullDispatch ctx m True))
283 ChildStop tid e -> do
284 case e of
285 Nothing -> void (delThread ctx tid) >> continue stp yld
286 Just x -> handleException x ctx tid
287 ChildCreate tid -> void (addThread ctx tid) >> continue stp yld
289-- | run m1 in a new thread, pushing its results to a pull channel and then run
290-- m2 in the parent thread. Any exceptions are also pushed to the channel.
291{-# INLINE pullSideDispatch #-}
292pullSideDispatch :: MonadAsync m
293 => Context a -> AsyncT m a -> AsyncT m a -> AsyncT m a
294pullSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do
295 let chan = childChannel ctx
296 tid <- doFork (push (ctx {pullSide = False}) m1)
297 (handleChildException chan)
298 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
299 (runAsyncT (pullDispatch ctx m2 False)) Nothing stp yld
301{-# NOINLINE pullDrain #-}
302pullDrain :: (MonadIO m, MonadThrow m) => Context a -> AsyncT m a
303pullDrain ctx = AsyncT $ \_ stp yld -> do
304 let yielder a = yld a Nothing (Just (pullDrain ctx))
305 continue = (runAsyncT (pullDrain ctx)) Nothing stp yld
307 res <- liftIO $ atomically $ readTChan (childChannel ctx)
308 case res of
309 ChildYield a -> yielder a
310 ChildDone tid a -> do
311 done <- delThread ctx tid
312 if done then (yld a Nothing Nothing) else (yielder a)
313 ChildStop tid e -> do
314 case e of
315 Nothing -> do
316 done <- delThread ctx tid
317 if done then stp else continue
318 Just x -> handleException x ctx tid
319 ChildCreate tid -> do
320 done <- addThread ctx tid
321 if done then stp else continue
323-- | Split the original computation in a pull-push pair. The original
324-- computation pulls from a Channel while m1 and m2 push to the channel.
325{-# NOINLINE pullFork #-}
326pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
327pullFork m1 m2 = AsyncT $ \_ stp yld -> do
328 ctx <- liftIO $ newContext
329 let m = pullDispatch ctx ((m1 <|> m2) <> finalizer ctx) True
330 (runAsyncT m) Nothing stp yld
332 where
334 finalizer ctx = AsyncT $ \_ stp yld -> do
335 running <- liftIO $ readIORef (runningThreads ctx)
336 done <- liftIO $ readIORef (runningThreads ctx)
337 if (S.null running && S.null done) then stp
338 else (runAsyncT (pullDrain ctx)) Nothing stp yld
340 newContext = do
341 channel <- atomically newTChan
342 running <- liftIO $ newIORef S.empty
343 done <- liftIO $ newIORef S.empty
344 return $ Context { childChannel = channel
345 , pullSide = True
346 , runningThreads = running
347 , doneThreads = done
348 }
350-- Concurrency rate control. Our objective is to create more threads on
351-- demand if the consumer is running faster than us. As soon as we
352-- encounter an Alternative composition we create a push pull pair of
353-- threads. We use a channel for communication between the consumer that
354-- pulls from the channel and the producer that pushes to the channel. The
355-- producer creates more threads if the channel becomes empty at times,
356-- that is the consumer is running faster. However this mechanism can be
357-- problematic if the initial production latency is high, we may end up
358-- creating too many threads. So we need some way to monitor and use the
359-- latency as well.
360--
361-- TBD For quick response we may have to increase the rate in the middle of
362-- a serially running computation. For that we can use a state flag to fork
363-- the rest of the computation at any point of time inside the Monad bind
364-- operation if the consumer is running at a faster speed.
365--
366-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
367-- If we have to dispatch in arbitrary chunksizes we will need to compose the
368-- parallel actions using a data constructor instead so that we can divide it
369-- in chunks of arbitrary size before dispatch. When batching we can convert
370-- the structure into Alternative batches of Monoid composition. That will also
371-- allow us to dispatch more work to existing threads rather than creating new
372-- threads always.
373--
374-- TBD for pure work (when we are not in the IO monad) we can divide it into
375-- just the number of CPUs.
376--
377-- XXX to rate control left folded structrues we will have to return the
378-- residual work back to the dispatcher. It will also consume a lot of
379-- memory due to queueing of all the work before execution starts.
380--
381instance MonadAsync m => Alternative (AsyncT m) where
382 empty = mempty
384 -- Note: This is designed to scale for right associated compositions,
385 -- therefore always use a right fold for folding bigger structures.
386 {-# INLINE (<|>) #-}
387 m1 <|> m2 = AsyncT $ \ctx stp yld -> do
388 case ctx of
389 Nothing -> (runAsyncT (pullFork m1 m2)) ctx stp yld
390 Just c -> do
391 if pullSide c then
392 (runAsyncT (pullSideDispatch c m1 m2)) ctx stp yld
393 else
394 -- for left associated compositions
395 (runAsyncT (pushSideDispatch c m1 m2)) ctx stp yld
397instance MonadAsync m => MonadPlus (AsyncT m) where
398 mzero = empty
399 mplus = (<|>)
401------------------------------------------------------------------------------
402-- Num
403------------------------------------------------------------------------------
405instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
406 fromInteger = return . fromInteger
407 mf + mg = (+) <$> mf <*> mg
408 mf * mg = (*) <$> mf <*> mg
409 negate f = f >>= return . negate
410 abs f = f >>= return . abs
411 signum f = f >>= return . signum
413-------------------------------------------------------------------------------
414-- AsyncT transformer
415-------------------------------------------------------------------------------
417instance MonadTrans AsyncT where
418 lift mx = AsyncT $ \ctx _ yld -> mx >>= (\a -> (yld a ctx Nothing))
420{-
421instance (MonadBase b m, Monad m) => MonadBase b (AsyncT m) where
422 liftBase = liftBaseDefault
424-------------------------------------------------------------------------------
425-- monad-control
426-------------------------------------------------------------------------------
428instance MonadTransControl AsyncT where
429 type StT AsyncT a = (Maybe a, Context)
430 liftWith f = AsyncT $ StateT $ \s ->
431 liftM (\x -> (Just x, s))
432 (f $ \t -> runStateT (runAsyncT t) s)
433 restoreT = AsyncT . StateT . const
434 {-# INLINABLE liftWith #-}
435 {-# INLINABLE restoreT #-}
437instance (MonadBaseControl b m, Monad m) => MonadBaseControl b (AsyncT m) where
438 type StM (AsyncT m) a = ComposeSt AsyncT m a
439 liftBaseWith = defaultLiftBaseWith
440 restoreM = defaultRestoreM
441 {-# INLINABLE liftBaseWith #-}
442 {-# INLINABLE restoreM #-}
443-}
445------------------------------------------------------------------------------
446-- Standard transformer instances
447------------------------------------------------------------------------------
449instance MonadIO m => MonadIO (AsyncT m) where
450 liftIO = lift . liftIO
452instance MonadThrow m => MonadThrow (AsyncT m) where
453 throwM = lift . throwM
455------------------------------------------------------------------------------
456-- MonadRecorder
457------------------------------------------------------------------------------
459instance MonadRecorder m => MonadRecorder (AsyncT m) where
460 getJournal = lift getJournal
461 putJournal = lift . putJournal
462 play = lift . play
464------------------------------------------------------------------------------
465-- Async primitives
466------------------------------------------------------------------------------
467--
468-- Only those actions that are marked with 'async' are guaranteed to be
469-- asynchronous. Asyncly is free to run other actions synchronously or
470-- asynchronously and it should not matter to the semantics of the program, if
471-- it does then use async to force.
472--
473-- Why not make async as default and ask the programmer to use a 'sync'
474-- primitive to force an action to run synchronously? But then we would not
475-- have the freedom to convert the async actions to sync dynamically. Note that
476-- it is safe to convert a sync action to async but vice-versa is not true.
477-- Converting an async to sync can cause change in semantics if the async
478-- action was an infinite loop for example.
479--
480-- | In an 'Alternative' composition, force the action to run asynchronously.
481-- The @\<|\>@ operator implies "can be parallel", whereas 'async' implies
482-- "must be parallel". Note that outside an 'Alternative' composition 'async'
483-- is not useful and should not be used. Even in an 'Alternative' composition
484-- 'async' is not useful as the last action as the last action always runs in
485-- the current thread.
486{-
487async :: Monad m => AsyncT m a -> AsyncT m a
488async action = AsyncT $ runAsyncTask True (runAsyncT action)
490makeAsync :: Monad m => ((a -> m ()) -> m ()) -> AsyncT m a
491makeAsync = AsyncT . makeCont
492-}