@@ -1,7 +1,9 @@
1+{-# LANGUAGE ConstraintKinds #-}
1{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE MultiParamTypeClasses #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
4{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE RankNTypes #-}
6+{-# LANGUAGE UndecidableInstances #-} -- XXX
7-- Module : Asyncly.AsyncT
9-- Module : Asyncly.AsyncT
@@ -15,53 +17,101 @@
16module Asyncly.AsyncT
18module Asyncly.AsyncT
17 ( AsyncT (..)
19 ( AsyncT (..)
19-- , makeAsync
22-- , makeAsync
23import Control.Applicative (Alternative (..))
26import Control.Applicative (Alternative (..))
27+import Control.Concurrent (ThreadId, forkIO, myThreadId)
28+import Control.Concurrent.STM (TChan, atomically, newTChan,
29+ readTChan, writeTChan)
30+import Control.Exception (fromException, try,
32+ BlockedIndefinitelyOnSTM(..))
33+import qualified Control.Exception.Lifted as EL
24import Control.Monad (ap, liftM, MonadPlus(..), mzero)
34import Control.Monad (ap, liftM, MonadPlus(..), mzero)
25--import Control.Monad.Base (MonadBase (..), liftBaseDefault)
35--import Control.Monad.Base (MonadBase (..), liftBaseDefault)
26import Control.Monad.Catch (MonadThrow, throwM)
36import Control.Monad.Catch (MonadThrow, throwM)
27import Control.Monad.IO.Class (MonadIO(..))
37import Control.Monad.IO.Class (MonadIO(..))
28import Control.Monad.Trans.Class (MonadTrans (lift))
38import Control.Monad.Trans.Class (MonadTrans (lift))
39+import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
30import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
41import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
31 MonadTransControl (..),
42 MonadTransControl (..),
32 defaultLiftBaseWith,
43 defaultLiftBaseWith,
33 defaultRestoreM, liftBaseWith)
44 defaultRestoreM, liftBaseWith)
35−import Data.Maybe (maybe)
46+import Data.Maybe (fromJust, maybe)
37import Control.Monad.Trans.Recorder (MonadRecorder(..))
48import Control.Monad.Trans.Recorder (MonadRecorder(..))
38−import Asyncly.Threads (Context)
49+
51+------------------------------------------------------------------------------
52+-- Concurrency Semantics
53+------------------------------------------------------------------------------
55+-- Asyncly is essentially a concurrent list transformer. To concatenate the
56+-- lists it provides three different ways of composing them. The Monoid
57+-- instance concatenates lists in a non-concurrent in-order fashion. The
58+-- Alternative instance concatenates list in a concurrent manner potentially
59+-- running each action in parallel and therefore the order of the resulting
60+-- items is not deterministic. Computations in Alternative composition may or
61+-- may not run in parallel. Thirdly, the <||> operator provides a way to
62+-- compose computations that are guaranteed to run in parallel. This provides a
63+-- way to run infinite loops or long running computations without blocking
64+-- or starving other computations.
66+------------------------------------------------------------------------------
67+-- Parent child thread communication types
68+------------------------------------------------------------------------------
70+-- XXX for threads that yield a single value we can pack yield and done in
71+-- ChildDone (Either exception value) so that we do not use two messages over
72+-- the channel where we can use just one.
73+data ChildEvent a = ChildYield a | ChildDone ThreadId (Maybe SomeException)
75+------------------------------------------------------------------------------
76+-- State threaded around the monad for thread management
77+------------------------------------------------------------------------------
79+newtype Context a = Context { childChannel :: TChan (ChildEvent a) }
40-- The 'Maybe' is redundant as we can use 'stop' value in the Nothing case,
81-- The 'Maybe' is redundant as we can use 'stop' value in the Nothing case,
41−-- but it makes the fold using '<|>' 25% faster.
82+-- but it makes the fold using '<|>' 25% faster. Need to try again as the code
83+-- has gone through many changes after this was tested.
85+-- Currently the only state we need is the thread context, For generality we
86+-- can parameterize the type with a state type 's'.
42newtype AsyncT m a =
87newtype AsyncT m a =
44 runAsyncT :: forall r.
89 runAsyncT :: forall r.
45− Context -- state
90+ Maybe (Context a) -- state
46 -> m r -- stop
91 -> m r -- stop
47− -> (a -> Context -> Maybe (AsyncT m a) -> m r) -- yield
92+ -> (a -> Maybe (Context a) -> Maybe (AsyncT m a) -> m r) -- yield
96+type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
51------------------------------------------------------------------------------
98------------------------------------------------------------------------------
53------------------------------------------------------------------------------
100------------------------------------------------------------------------------
55-- We do not use bind for parallelism. That is, we do not start each iteration
102-- We do not use bind for parallelism. That is, we do not start each iteration
56-- of the list in parallel. That will introduce too much uncontrolled
103-- of the list in parallel. That will introduce too much uncontrolled
57−-- parallelism. Instead we control parallelsim using <|>, it is used for
104+-- parallelism. Instead we control parallelism using <|>, it is used for
58-- actions that 'may' be run in parallel. The library decides whether they will
105-- actions that 'may' be run in parallel. The library decides whether they will
59-- be started in parallel. This puts the parallel semantics in the hands of the
106-- be started in parallel. This puts the parallel semantics in the hands of the
60-- user and operational semantics in the hands of the library. Ideally we would
107-- user and operational semantics in the hands of the library. Ideally we would
61-- want parallelism to be completely hidden from the user, for example we can
108-- want parallelism to be completely hidden from the user, for example we can
62-- automatically decide where to start an action in parallel so that it is most
109-- automatically decide where to start an action in parallel so that it is most
110+-- efficient. We can decide whether a particular bind should fork a parallel
111+-- thread or not based on the level of the bind in the tree. Maybe at some
63−-- efficient. Maybe sometime we shall be able to do that?
112+-- point we shall be able to do that?
114+-- XXX make this (<>)
65serially :: AsyncT m a -> AsyncT m a -> AsyncT m a
115serially :: AsyncT m a -> AsyncT m a -> AsyncT m a
66serially (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
116serially (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
67 m1 ctx ((runAsyncT m2) ctx stp yld) $ \a c r ->
117 m1 ctx ((runAsyncT m2) ctx stp yld) $ \a c r ->
@@ -71,10 +121,13 @@ serially (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
71instance Monad m => Monad (AsyncT m) where
121instance Monad m => Monad (AsyncT m) where
72 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing
122 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing
124+ -- | On bind the new computation does not inherit the parent thread
125+ -- context, it starts a new node in the tree.
74− AsyncT m >>= f = AsyncT $ \ctx stp yld ->
126+ AsyncT m >>= f = AsyncT $ \_ stp yld ->
75− m ctx stp $ \a c r ->
127+ m Nothing stp $ \a _ r ->
76− let run x = (runAsyncT x) c stp yld
128+ let run x = (runAsyncT x) Nothing stp yld
77− in maybe (run $ f a) (\rx -> run $ serially (f a) (rx >>= f)) r
130+ in maybe (run next) (\rx -> run $ serially next (rx >>= f)) r
79------------------------------------------------------------------------------
132------------------------------------------------------------------------------
80-- Functor
133-- Functor
@@ -95,17 +148,136 @@ instance Monad m => Applicative (AsyncT m) where
95-- Alternative
148-- Alternative
96------------------------------------------------------------------------------
149------------------------------------------------------------------------------
151+newContext :: IO (Context a)
153+ childChan <- atomically newTChan
154+ return $ Context { childChannel = childChan }
156+doFork :: (MonadIO m, MonadBaseControl IO m)
158+ -> (Either SomeException () -> IO ())
160+doFork action preExit =
161+ EL.mask $ \restore ->
162+ liftBaseWith $ \runInIO -> forkIO $ do
163+ _ <- runInIO $ EL.try (restore action) >>= liftIO . preExit
164+ -- XXX restore state here
167+-- XXX make this a bounded channel so that we block if the previous value is
169+{-# INLINE channelDone #-}
170+channelDone :: MonadIO m => TChan (ChildEvent a) -> m ()
171+channelDone chan = do
172+ tid <- liftIO myThreadId
173+ liftIO $ atomically $ writeTChan chan (ChildDone tid Nothing)
175+{-# INLINE channelYield #-}
176+channelYield :: MonadIO m => TChan (ChildEvent a) -> a -> m ()
177+channelYield chan a = liftIO $ atomically $ writeTChan chan (ChildYield a)
179+push :: MonadIO m => Context a -> AsyncT m a -> m ()
180+push ctx m = go (Just ctx) m
184+ go c mx = (runAsyncT mx) c (channelDone pchan) yield
185+ pchan = childChannel ctx
188+ channelYield pchan a
191+ -- XXX If other threads are running, we should yield to the channel only
192+ -- when the channel becomes empty so that we do not start more threads
194+ {-# INLINE continue #-}
195+ continue a c mx = do
196+ channelYield pchan a
199+ {-# INLINE yield #-}
200+ yield a c r = maybe (done a) (\x -> continue a c x) r
202+-- If an exception occurs we push it to the channel so that it can handled by
203+-- the parent. 'Paused' exceptions are to be collected at the top level.
204+-- XXX Paused exceptions should only bubble up to the runRecorder computation
205+handlePushException :: TChan (ChildEvent a) -> Either SomeException () -> IO ()
206+handlePushException pchan res = do
212+ atomically $ writeTChan pchan (ChildDone tid r)
215+canFork = return True
217+-- XXX do not fork if the channel is never empty
218+-- We can fork if the channel goes empty sometimes
219+fork :: (MonadIO m, MonadBaseControl IO m) => AsyncT m a -> AsyncT m a
220+fork m = AsyncT $ \ctx stp yld -> do
221+ doAsync <- liftIO $ canFork
222+ let c = fromJust ctx -- XXX partial
224+ _ <- doFork (push c m) (handlePushException (childChannel c))
226+ else (runAsyncT m) ctx stp yld
228+-- We re-raise any exceptions received from the child threads, that way
229+-- exceptions get propagated to the top level computation and can be handled
231+pull :: (MonadIO m, MonadThrow m) => Context a -> AsyncT m a
232+pull ctx = AsyncT $ \_ stp yld -> do
233+ ethr <- liftIO $ try $ atomically $ readTChan (childChannel ctx)
235+ -- XXX for immediate cleanup and for killing any blocked threads we
236+ -- need to track the children
237+ Left e -> case fromException e of
238+ Just BlockedIndefinitelyOnSTM -> stp
242+ ChildYield a -> yld a Nothing (Just (pull ctx))
244+ let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
245+ maybe continue throwM e
247+pushpull :: MonadAsync m => AsyncT m a -> AsyncT m a
248+pushpull m = AsyncT $ \ctx stp yld ->
251+ c <- liftIO $ newContext
252+ -- XXX m must be a push type computation
253+ let comp = serially (fork m) (pull c)
254+ (runAsyncT comp) (Just c) stp yld
255+ Just _ -> (runAsyncT m) ctx stp yld
257+-- Child threads yield values to the parent's channel. The parent reads them
258+-- from there and yields to its consumer.
98−instance Monad m => Alternative (AsyncT m) where
259+instance MonadAsync m => Alternative (AsyncT m) where
99 empty = AsyncT $ \_ stp _ -> stp
260 empty = AsyncT $ \_ stp _ -> stp
101− -- XXX need to wait for the async threads in case of a stop is returned due
262+ -- Concurrency rate control. Our objective is to create more threads on
102− -- to async thread spawning.
263+ -- demand if the consumer is running faster than us. As soon as we
103− AsyncT m1 <|> m2 = AsyncT $ \ctx stp yld ->
264+ -- encounter an Alternative composition we create a push pull pair of
104− m1 ctx ((runAsyncT m2) ctx stp yld) $ \a c r ->
265+ -- threads. We use a channel for communication between the consumer that
105− let yield x = yld a c (Just x)
266+ -- pulls from the channel and the producer that pushes to the channel. The
106− in maybe (yield m2) (\rx -> yield $ rx <|> m2) r
267+ -- producer creates more threads if the channel becomes empty at times,
107−
268+ -- that is the consumer is running faster. However this mechanism can be
269+ -- problematic if the initial production latency is high, we may end up
270+ -- creating too many threads. So we need some way to monitor and use the
272+ -- TBD For quick response we may have to increase the rate in the middle of
273+ -- a serially running computation. For that we can use a state flag to fork
274+ -- the rest of the computation at any point of time inside the Monad bind
275+ -- operation if the consumer is running at a faster speed.
276+ m1 <|> m2 = AsyncT $ \ctx stp yld ->
277+ let comp = pushpull $ serially (fork m1) m2
278+ in (runAsyncT comp) ctx stp yld
108−instance Monad m => MonadPlus (AsyncT m) where
280+instance MonadAsync m => MonadPlus (AsyncT m) where
109 mzero = empty
281 mzero = empty
110 mplus = (<|>)
282 mplus = (<|>)