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