@@ -22,11 +22,12 @@
2222--
2323module Asyncly.AsyncT
2424 ( AsyncT (..)
25−
26− , threads
25+ , Step (..)
26+ , stop
27+ , waitA_
28+ , eachA
2729-- , async
2830-- , makeAsync
29− , each
3031
3132 , before
3233 , (*>>)
@@ -58,10 +59,10 @@ import Unsafe.Coerce (unsafeCoerce)
5859import Control.Monad.Trans.Recorder (MonadRecorder(..))
5960--import Debug.Trace (traceM)
6061
61−import Asyncly.Threads
62+import Asyncly.Threads (MonadAsync, Context)
6263
63−--data Step a m = Stop | Yield a (AsyncT m a)
64−--newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Step a m) }
64+data Step a m = Stop | Yield a (AsyncT m a)
65+newtype AsyncT m a = AsyncT { runAsyncT :: m (Step a m) }
6566
6667------------------------------------------------------------------------------
6768-- Utilities
@@ -80,35 +81,33 @@ dbg _ = return ()
8081stop :: Monad m => AsyncT m a
8182stop = AsyncT . return $ Stop
8283
83−{-# SPECIALIZE runAction :: AsyncT IO a -> StateT Context IO (Step a IO) #-}
84−runAction :: Monad m => AsyncT m a -> StateT Context m (Step a m)
85−runAction m = do
86− ctx <- get
87− (step, ctx') <- lift $ runStateT (runAsyncT m) ctx
88− put ctx'
89− return step
90−
9184-- TBD real parallel
92−{-# SPECIALIZE parallel :: AsyncT IO a -> AsyncT IO a -> StateT Context IO (Step a IO) #-}
93−parallel :: Monad m
94− => AsyncT m a
95− -> AsyncT m a
96− -> StateT Context m (Step a m)
85+{-# SPECIALIZE parallel
86+ :: AsyncT (StateT Context IO) a
87+ -> AsyncT (StateT Context IO) a
88+ -> StateT Context IO (Step a (StateT Context IO)) #-}
89+{-# SPECIALIZE parallel
90+ :: AsyncT IO a
91+ -> AsyncT IO a
92+ -> IO (Step a IO) #-}
93+parallel :: Monad m => AsyncT m a -> AsyncT m a -> m (Step a m)
9794parallel ma mb = do
98− step <- runAction ma
99− case step of
100− Stop -> runAsyncT mb
101− Yield a mx -> return (Yield a (AsyncT (parallel mx mb)))
95+ runAsyncT ma >>= \step ->
96+ case step of
97+ Stop -> runAsyncT mb
98+ Yield a mx -> return (Yield a (AsyncT (parallel mx mb)))
10299
103100instance MonadAsync m => Monad (AsyncT m) where
101+ {-# SPECIALIZE instance Monad (AsyncT (StateT Context IO)) #-}
104102 {-# SPECIALIZE instance Monad (AsyncT IO) #-}
105103 return a = AsyncT . return $ Yield a stop
106104
105+ -- {-# INLINE (>>=) #-}
107106 m >>= f = AsyncT $ do
108− step <- runAction m
109− case step of
110− Stop -> return Stop -- XXX need to wait for the child threads
111− Yield a ma -> runAsyncT $ (f a) <|> (ma >>= f)
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)
112111
113112------------------------------------------------------------------------------
114113-- Functor
@@ -147,6 +146,7 @@ instance MonadAsync m => Applicative (AsyncT m) where
147146-- they can go away rather than waiting indefinitely.
148147
149148instance MonadAsync m => Alternative (AsyncT m) where
149+ {-# SPECIALIZE instance Alternative (AsyncT (StateT Context IO)) #-}
150150 {-# SPECIALIZE instance Alternative (AsyncT IO) #-}
151151 empty = stop
152152 (<|>) m1 m2 = AsyncT $ parallel m1 m2
@@ -176,7 +176,7 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
176176-------------------------------------------------------------------------------
177177
178178instance MonadTrans AsyncT where
179− lift mx = AsyncT $ lift mx >>= return . (\a -> (Yield a stop))
179+ lift mx = AsyncT $ mx >>= return . (\a -> (Yield a stop))
180180
181181instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
182182 liftBase = liftBaseDefault
@@ -295,18 +295,33 @@ makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a
295295makeAsync = AsyncT . makeCont
296296-}
297297
298−-- scatter
299−{-# SPECIALIZE each :: [a] -> AsyncT IO a #-}
300−each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
301− => [a] -> AsyncT m a
302−each xs = foldr (<|>) empty $ map return xs
298+{-# SPECIALIZE waitA_ :: AsyncT IO a -> IO () #-}
299+waitA_ :: (MonadAsync m) => AsyncT m a -> m ()
300+waitA_ (AsyncT m) = run m
303301
304−------------------------------------------------------------------------------
305−-- Controlling thread quota
306−------------------------------------------------------------------------------
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 ()
307309
308−-- | Runs a computation under a given thread limit. A limit of 0 means all new
309−-- tasks start synchronously in the current thread unless overridden by
310−-- 'async'.
311−threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
312−threads n action = AsyncT $ threadCtl n (runAsyncT action)
310+{-
311+waitA :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]
312+waitA (AsyncT m) = run m
313+
314+ where
315+
316+ run ma = do
317+ ma >>= \a -> case a of
318+ Yield x mb -> liftM (x :) (run mb ctx')
319+ -- XXX pull from children here
320+ Stop -> return []
321+-}
322+
323+-- scatter
324+{-# SPECIALIZE eachA :: [a] -> AsyncT IO a #-}
325+eachA :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
326+ => [a] -> AsyncT m a
327+eachA xs = foldr (<|>) empty $ map return xs