@@ -18,7 +18,6 @@
1818module Asyncly.AsyncT
1919 ( AsyncT (..)
2020 , MonadAsync
21− , yield
2221-- , async
2322-- , makeAsync
2423 )
@@ -48,7 +47,6 @@ import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
4847import Data.Functor (void)
4948import Data.IORef (IORef, modifyIORef, newIORef,
5049 writeIORef, readIORef)
51−import Data.Maybe (maybe)
5250import Data.Monoid ((<>))
5351import Data.Set (Set)
5452import qualified Data.Set as S
@@ -119,9 +117,10 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
119117instance Monad m => Monoid (AsyncT m a) where
120118 mempty = AsyncT $ \_ stp _ -> stp
121119 mappend (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
122− m1 ctx ((runAsyncT m2) ctx stp yld) $ \a _ r ->
123− let yielder x = yld a ctx (Just x)
124− in maybe (yielder m2) (\rx -> yielder $ mappend rx m2) r
120+ let stop = (runAsyncT m2) ctx stp yld
121+ yield a c Nothing = yld a c (Just m2)
122+ yield a c (Just r) = yld a c (Just (mappend r m2))
123+ in m1 ctx stop yield
125124
126125-- We do not use bind for parallelism. That is, we do not start each iteration
127126-- of the list in parallel. That will introduce too much uncontrolled
@@ -141,13 +140,10 @@ instance Monad m => Monad (AsyncT m) where
141140 -- | A thread context is valid only until the next bind. Upon a bind we
142141 -- reset the context to Nothing.
143142 AsyncT m >>= f = AsyncT $ \_ stp yld ->
144− m Nothing stp $ \a _ r ->
145− let run x = (runAsyncT x) Nothing stp yld
146− next = f a
147− in maybe (run next) (\rx -> run $ next <> (rx >>= f)) r
148−
149−yield :: a -> AsyncT m a -> AsyncT m a
150−yield a m = AsyncT $ \ctx _ yld -> yld a ctx (Just m)
143+ let run x = (runAsyncT x) Nothing stp yld
144+ yield a _ Nothing = run $ f a
145+ yield a _ (Just r) = run $ f a <> (r >>= f)
146+ in m Nothing stp yield
151147
152148------------------------------------------------------------------------------
153149-- Functor
@@ -187,20 +183,21 @@ push context = run (Just context) (dequeueLoop context)
187183
188184 where
189185
190− run ctx m = (runAsyncT m) ctx channelStop yielder
186+ run ctx m = (runAsyncT m) ctx stop yield
191187
192− chan = childChannel context
193− channelYield a = liftIO $ atomically $ writeTBQueue chan (ChildYield a)
194− channelDone a = do
195− tid <- liftIO myThreadId
196− liftIO $ atomically $ writeTBQueue chan (ChildDone tid a)
197− channelStop = do
188+ chan = childChannel context
189+
190+ stop = do
198191 tid <- liftIO myThreadId
199192 liftIO $ atomically $ writeTBQueue chan (ChildStop tid Nothing)
200193
201− done a = channelDone a
202− continue a ctx m = channelYield a >> run ctx m
203− yielder a ctx r = maybe (done a) (\rx -> continue a ctx rx) r
194+ yield a _ Nothing = do
195+ tid <- liftIO myThreadId
196+ liftIO $ atomically $ writeTBQueue chan (ChildDone tid a)
197+
198+ yield a ctx (Just r) = do
199+ liftIO $ atomically $ writeTBQueue chan (ChildYield a)
200+ run ctx r
204201
205202-- Thread tracking has a significant performance overhead (~20% on empty
206203-- threads, it will be lower for heavy threads). It is needed for two reasons:
@@ -252,30 +249,24 @@ handleException e ctx tid = do
252249{-# NOINLINE pullWorker #-}
253250pullWorker :: (MonadIO m, MonadThrow m) => Context m a -> AsyncT m a
254251pullWorker ctx = AsyncT $ \_ stp yld -> do
255− ev <- liftIO $ atomically $ readTBQueue (childChannel ctx)
256− handleEvent ev stp yld
252+ let continue = (runAsyncT (pullWorker ctx)) Nothing stp yld
253+ yield a = yld a Nothing (Just (pullWorker ctx))
257254
258− where
259−
260− continue stp yld = (runAsyncT (pullWorker ctx)) Nothing stp yld
261−
262− {-# INLINE handleEvent #-}
263− handleEvent ev stp yld = do
264− let yielder a = yld a Nothing (Just (pullWorker ctx))
265− case ev of
266− ChildYield a -> yielder a
267− ChildDone tid a -> do
268− done <- delThread ctx tid
269− if done then (yld a Nothing Nothing) else (yielder a)
270− ChildStop tid e -> do
271− case e of
272− Nothing -> do
273− done <- delThread ctx tid
274− if done then stp else continue stp yld
275− Just x -> handleException x ctx tid
276− ChildCreate tid -> do
277− done <- addThread ctx tid
278− if done then stp else continue stp yld
255+ ev <- liftIO $ atomically $ readTBQueue (childChannel ctx)
256+ case ev of
257+ ChildYield a -> yield a
258+ ChildDone tid a -> do
259+ done <- delThread ctx tid
260+ if done then (yld a Nothing Nothing) else (yield a)
261+ ChildStop tid e -> do
262+ case e of
263+ Nothing -> do
264+ done <- delThread ctx tid
265+ if done then stp else continue
266+ Just x -> handleException x ctx tid
267+ ChildCreate tid -> do
268+ done <- addThread ctx tid
269+ if done then stp else continue
279270
280271-- If an exception occurs we push it to the channel so that it can handled by
281272-- the parent. 'Paused' exceptions are to be collected at the top level.
@@ -378,9 +369,9 @@ dequeueLoop ctx = AsyncT $ \_ stp yld -> do
378369 Nothing -> stp
379370 Just m -> do
380371 let loop = (runAsyncT (dequeueLoop ctx)) Nothing stp yld
381− yielder a c Nothing = yld a c (Just (dequeueLoop ctx))
382− yielder a c r = yld a c r
383− (runAsyncT m) (Just ctx) loop yielder
372+ yield a c Nothing = yld a c (Just (dequeueLoop ctx))
373+ yield a c r = yld a c r
374+ (runAsyncT m) (Just ctx) loop yield
384375
385376instance MonadAsync m => Alternative (AsyncT m) where
386377 empty = mempty