@@ -22,19 +22,21 @@
2222--
2323module Asyncly.AsyncT
2424 ( AsyncT (..)
25− , Step (..)
26− , stop
25+ -- , Step (..)
26+ -- , stop
2727 , waitA_
2828 , eachA
2929-- , async
3030-- , makeAsync
3131
32+{-
3233 , before
3334 , (*>>)
3435 , thereafter
3536 , (>>*)
3637 , afterfirst
3738 , (>>|)
39+ -}
3840
3941 -- internal
4042 , dbg
@@ -42,6 +44,7 @@ module Asyncly.AsyncT
4244where
4345
4446import Control.Applicative (Alternative (..))
47+import Data.Maybe (maybe)
4548import Control.Monad (ap, mzero, when)
4649import Control.Monad.Base (MonadBase (..), liftBaseDefault)
4750import Control.Monad.Catch (MonadThrow, throwM)
@@ -61,8 +64,15 @@ import Control.Monad.Trans.Recorder (MonadRecorder(..))
6164
6265import Asyncly.Threads (MonadAsync, Context)
6366
64−data Step a m = Stop | Yield a (AsyncT m a)
65−newtype AsyncT m a = AsyncT { runAsyncT :: m (Step a m) }
67+-- data Step a m = Stop | Yield a (AsyncT m a)
68+-- newtype AsyncT m a = AsyncT { runAsyncT :: m (Step a m) }
69+-- stop value is needed for Alternative instance
70+newtype AsyncT m a = AsyncT {
71+ runAsyncT :: forall r.
72+ m r
73+ -> (a -> Maybe (AsyncT m a) -> m r)
74+ -> m r
75+ }
6676
6777------------------------------------------------------------------------------
6878-- Utilities
@@ -78,36 +88,33 @@ dbg _ = return ()
7888-- Monad
7989------------------------------------------------------------------------------
8090
81−stop :: Monad m => AsyncT m a
82−stop = AsyncT . return $ Stop
83−
84−-- TBD real parallel
85−{-# SPECIALIZE parallel
86− :: AsyncT (StateT Context IO) a
87− -> AsyncT (StateT Context IO) a
88− -> StateT Context IO (Step a (StateT Context IO)) #-}
91+-- XXX need to wait for the async threads in case of a stop is returned due to
92+-- async thread spawning.
8993{-# SPECIALIZE parallel
9094 :: AsyncT IO a
9195 -> AsyncT IO a
92− -> IO (Step a IO) #-}
93−parallel :: Monad m => AsyncT m a -> AsyncT m a -> m (Step a m)
94−parallel ma mb = do
95− runAsyncT ma >>= \step ->
96− case step of
97− Stop -> runAsyncT mb
98− Yield a mx -> return (Yield a (AsyncT (parallel mx mb)))
96+ -> AsyncT IO a #-}
97+{-# INLINE parallel #-}
98+parallel :: Monad m => AsyncT m a -> AsyncT m a -> AsyncT m a
99+parallel (AsyncT m1) m2 = AsyncT $ \stp yld ->
100+ m1 ((runAsyncT m2) stp yld) (\a ax ->
101+ case ax of
102+ Just mx -> yld a (Just (parallel mx m2))
103+ Nothing -> yld a (Just m2)
104+ )
99105
100106instance MonadAsync m => Monad (AsyncT m) where
101− {-# SPECIALIZE instance Monad (AsyncT (StateT Context IO)) #-}
107+ -- {-# SPECIALIZE instance Monad (AsyncT (StateT Context IO)) #-}
102108 {-# SPECIALIZE instance Monad (AsyncT IO) #-}
103− return a = AsyncT . return $ Yield a stop
109+ return a = AsyncT $ \_ yld -> yld a Nothing
104110
105− -- {-# INLINE (>>=) #-}
106− m >>= f = AsyncT $ do
107− runAsyncT m >>= \step ->
108− case step of
109− Stop -> return Stop -- XXX need to wait for the child threads
110− Yield a ma -> runAsyncT $ (f a) <|> (ma >>= f)
111+ {-# INLINE (>>=) #-}
112+ AsyncT m >>= f = AsyncT $ \stp yld ->
113+ m stp (\a ma ->
114+ case ma of
115+ Just mx -> (runAsyncT (parallel (f a) (mx >>= f))) stp yld
116+ Nothing -> (runAsyncT (f a)) stp yld
117+ )
111118
112119------------------------------------------------------------------------------
113120-- Functor
@@ -146,15 +153,16 @@ instance MonadAsync m => Applicative (AsyncT m) where
146153-- they can go away rather than waiting indefinitely.
147154
148155instance MonadAsync m => Alternative (AsyncT m) where
149− {-# SPECIALIZE instance Alternative (AsyncT (StateT Context IO)) #-}
156+ -- {-# SPECIALIZE instance Alternative (AsyncT (StateT Context IO)) #-}
150157 {-# SPECIALIZE instance Alternative (AsyncT IO) #-}
151− empty = stop
152− (<|>) m1 m2 = AsyncT $ parallel m1 m2
158+ empty = AsyncT $ \stp _ -> stp
159+ (<|>) m1 m2 = parallel m1 m2
153160
154161instance MonadAsync m => MonadPlus (AsyncT m) where
155162 mzero = empty
156163 mplus = (<|>)
157164
165+{-
158166instance (Monoid a, MonadAsync m) => Monoid (AsyncT m a) where
159167 mappend x y = mappend <$> x <*> y
160168 mempty = return mempty
@@ -171,12 +179,13 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
171179 abs f = f >>= return . abs
172180 signum f = f >>= return . signum
173181
182+-}
174183-------------------------------------------------------------------------------
175184-- AsyncT transformer
176185-------------------------------------------------------------------------------
177186
178187instance MonadTrans AsyncT where
179− lift mx = AsyncT $ mx >>= return . (\a -> (Yield a stop))
188+ lift mx = AsyncT $ \_ yld -> mx >>= (\a -> (yld a Nothing))
180189
181190instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
182191 liftBase = liftBaseDefault
@@ -226,6 +235,7 @@ instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where
226235-- Special compositions
227236------------------------------------------------------------------------------
228237
238+{-
229239-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.
230240before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
231241before ma mb = AsyncT $ do
@@ -264,6 +274,7 @@ infixr 1 >>|
264274-- | Same as 'afterfirst'.
265275(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
266276(>>|) = afterfirst
277+-}
267278
268279------------------------------------------------------------------------------
269280-- Async primitives
@@ -295,17 +306,9 @@ makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a
295306makeAsync = AsyncT . makeCont
296307-}
297308
298−{-# SPECIALIZE waitA_ :: AsyncT IO a -> IO () #-}
309+-- {-# SPECIALIZE waitA_ :: AsyncT IO a -> IO () #-}
299310waitA_ :: (MonadAsync m) => AsyncT m a -> m ()
300−waitA_ (AsyncT m) = run m
301−
302− where
303−
304− run m = m >>= \x ->
305− case x of
306− Yield _ r -> run (runAsyncT r)
307− -- XXX pull from children here
308− _ -> return ()
311+waitA_ (AsyncT m) = m (return ()) (\_ r -> maybe (return ()) waitA_ r)
309312
310313{-
311314waitA :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]
@@ -321,6 +324,8 @@ waitA (AsyncT m) = run m
321324-}
322325
323326-- scatter
327+-- XXX unthreaded case can be more efficient by yielding continuously rather
328+-- than folding a yield followed by Nothing in the parallel case.
324329{-# SPECIALIZE eachA :: [a] -> AsyncT IO a #-}
325330eachA :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
326331 => [a] -> AsyncT m a