@@ -1,7 +1,9 @@
1+{-# LANGUAGE ConstraintKinds #-}
12{-# LANGUAGE FlexibleContexts #-}
23{-# LANGUAGE FlexibleInstances #-}
34{-# LANGUAGE MultiParamTypeClasses #-}
45{-# LANGUAGE RankNTypes #-}
6+{-# LANGUAGE UndecidableInstances #-} -- XXX
57
68-- |
79-- Module : Asyncly.AsyncT
@@ -15,53 +17,101 @@
1517--
1618module Asyncly.AsyncT
1719 ( AsyncT (..)
20+ , MonadAsync
1821-- , async
1922-- , makeAsync
2023 )
2124where
2225
2326import 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,
31+ SomeException (..),
32+ BlockedIndefinitelyOnSTM(..))
33+import qualified Control.Exception.Lifted as EL
2434import Control.Monad (ap, liftM, MonadPlus(..), mzero)
2535--import Control.Monad.Base (MonadBase (..), liftBaseDefault)
2636import Control.Monad.Catch (MonadThrow, throwM)
2737import Control.Monad.IO.Class (MonadIO(..))
2838import Control.Monad.Trans.Class (MonadTrans (lift))
39+import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
2940{-
3041import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
3142 MonadTransControl (..),
3243 defaultLiftBaseWith,
3344 defaultRestoreM, liftBaseWith)
3445 -}
35−import Data.Maybe (maybe)
46+import Data.Maybe (fromJust, maybe)
3647
3748import Control.Monad.Trans.Recorder (MonadRecorder(..))
38−import Asyncly.Threads (Context)
49+
50+
51+------------------------------------------------------------------------------
52+-- Concurrency Semantics
53+------------------------------------------------------------------------------
54+--
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.
65+
66+------------------------------------------------------------------------------
67+-- Parent child thread communication types
68+------------------------------------------------------------------------------
69+
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)
74+
75+------------------------------------------------------------------------------
76+-- State threaded around the monad for thread management
77+------------------------------------------------------------------------------
78+
79+newtype Context a = Context { childChannel :: TChan (ChildEvent a) }
3980
4081-- 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.
84+--
85+-- Currently the only state we need is the thread context, For generality we
86+-- can parameterize the type with a state type 's'.
4287newtype AsyncT m a =
4388 AsyncT {
4489 runAsyncT :: forall r.
45− Context -- state
90+ Maybe (Context a) -- state
4691 -> 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
4893 -> m r
4994 }
5095
96+type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
97+
5198------------------------------------------------------------------------------
5299-- Monad
53100------------------------------------------------------------------------------
54101
55102-- We do not use bind for parallelism. That is, we do not start each iteration
56103-- 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
58105-- actions that 'may' be run in parallel. The library decides whether they will
59106-- be started in parallel. This puts the parallel semantics in the hands of the
60107-- user and operational semantics in the hands of the library. Ideally we would
61108-- want parallelism to be completely hidden from the user, for example we can
62109-- automatically decide where to start an action in parallel so that it is most
63−-- efficient. Maybe sometime we shall be able to do that?
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
112+-- point we shall be able to do that?
64113
114+-- XXX make this (<>)
65115serially :: AsyncT m a -> AsyncT m a -> AsyncT m a
66116serially (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
67117 m1 ctx ((runAsyncT m2) ctx stp yld) $ \a c r ->
@@ -71,10 +121,13 @@ serially (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
71121instance Monad m => Monad (AsyncT m) where
72122 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing
73123
74− AsyncT m >>= f = AsyncT $ \ctx stp yld ->
75− m ctx stp $ \a c r ->
76− let run x = (runAsyncT x) c stp yld
77− in maybe (run $ f a) (\rx -> run $ serially (f a) (rx >>= f)) r
124+ -- | On bind the new computation does not inherit the parent thread
125+ -- context, it starts a new node in the tree.
126+ AsyncT m >>= f = AsyncT $ \_ stp yld ->
127+ m Nothing stp $ \a _ r ->
128+ let run x = (runAsyncT x) Nothing stp yld
129+ next = f a
130+ in maybe (run next) (\rx -> run $ serially next (rx >>= f)) r
78131
79132------------------------------------------------------------------------------
80133-- Functor
@@ -95,17 +148,136 @@ instance Monad m => Applicative (AsyncT m) where
95148-- Alternative
96149------------------------------------------------------------------------------
97150
98−instance Monad m => Alternative (AsyncT m) where
151+newContext :: IO (Context a)
152+newContext = do
153+ childChan <- atomically newTChan
154+ return $ Context { childChannel = childChan }
155+
156+doFork :: (MonadIO m, MonadBaseControl IO m)
157+ => m ()
158+ -> (Either SomeException () -> IO ())
159+ -> m ThreadId
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
165+ return ()
166+
167+-- XXX make this a bounded channel so that we block if the previous value is
168+-- not consumed yet.
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)
174+
175+{-# INLINE channelYield #-}
176+channelYield :: MonadIO m => TChan (ChildEvent a) -> a -> m ()
177+channelYield chan a = liftIO $ atomically $ writeTChan chan (ChildYield a)
178+
179+push :: MonadIO m => Context a -> AsyncT m a -> m ()
180+push ctx m = go (Just ctx) m
181+
182+ where
183+
184+ go c mx = (runAsyncT mx) c (channelDone pchan) yield
185+ pchan = childChannel ctx
186+
187+ done a = do
188+ channelYield pchan a
189+ channelDone pchan
190+
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
193+ -- unnecessarily.
194+ {-# INLINE continue #-}
195+ continue a c mx = do
196+ channelYield pchan a
197+ go c mx
198+
199+ {-# INLINE yield #-}
200+ yield a c r = maybe (done a) (\x -> continue a c x) r
201+
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
207+ let r = case res of
208+ Left e -> Just e
209+ Right _ -> Nothing
210+
211+ tid <- myThreadId
212+ atomically $ writeTChan pchan (ChildDone tid r)
213+
214+canFork :: IO Bool
215+canFork = return True
216+
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
223+ if doAsync then do
224+ _ <- doFork (push c m) (handlePushException (childChannel c))
225+ stp
226+ else (runAsyncT m) ctx stp yld
227+
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
230+-- there.
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)
234+ case ethr of
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
239+ Nothing -> throwM e
240+ Right ev ->
241+ case ev of
242+ ChildYield a -> yld a Nothing (Just (pull ctx))
243+ ChildDone _ e -> do
244+ let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
245+ maybe continue throwM e
246+
247+pushpull :: MonadAsync m => AsyncT m a -> AsyncT m a
248+pushpull m = AsyncT $ \ctx stp yld ->
249+ case ctx of
250+ Nothing -> do
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
256+
257+-- Child threads yield values to the parent's channel. The parent reads them
258+-- from there and yields to its consumer.
259+instance MonadAsync m => Alternative (AsyncT m) where
99260 empty = AsyncT $ \_ stp _ -> stp
100261
101− -- XXX need to wait for the async threads in case of a stop is returned due
102− -- to async thread spawning.
103− AsyncT m1 <|> m2 = AsyncT $ \ctx stp yld ->
104− m1 ctx ((runAsyncT m2) ctx stp yld) $ \a c r ->
105− let yield x = yld a c (Just x)
106− in maybe (yield m2) (\rx -> yield $ rx <|> m2) r
107−
108−instance Monad m => MonadPlus (AsyncT m) where
262+ -- Concurrency rate control. Our objective is to create more threads on
263+ -- demand if the consumer is running faster than us. As soon as we
264+ -- encounter an Alternative composition we create a push pull pair of
265+ -- threads. We use a channel for communication between the consumer that
266+ -- pulls from the channel and the producer that pushes to the channel. The
267+ -- producer creates more threads if the channel becomes empty at times,
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
271+ -- latency as well.
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
279+
280+instance MonadAsync m => MonadPlus (AsyncT m) where
109281 mzero = empty
110282 mplus = (<|>)
111283