@@ -48,6 +48,11 @@ import GHC.Prim (Any)
4848import Asyncly.Threads
4949import Asyncly.AsyncT
5050
51+-- This transformer runs AsyncT under a state to manage the threads.
52+-- Separating the state from the pure ListT transformer is cleaner but it
53+-- results in 2x performance degradation. At some point if that performance is
54+-- really needed we can combine the two.
55+
5156newtype AsynclyT m a = AsynclyT { runAsynclyT :: AsyncT (StateT Context m) a }
5257
5358deriving instance MonadAsync m => Functor (AsynclyT m)
@@ -58,6 +63,10 @@ deriving instance MonadAsync m => MonadIO (AsynclyT m)
5863instance MonadTrans (AsynclyT) where
5964 lift mx = AsynclyT $ AsyncT $ \s k -> lift mx >>= (\a -> (k a Nothing))
6065
66+------------------------------------------------------------------------------
67+-- Running the monad
68+------------------------------------------------------------------------------
69+
6170getContext :: Maybe (IORef [Recording]) -> IO Context
6271getContext lref = do
6372 childChan <- atomically newTChan
@@ -65,9 +74,9 @@ getContext lref = do
6574 credit <- newIORef maxBound
6675 return $ initContext childChan pendingRef credit lref
6776
68−-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
77+-- | Run an 'AsynclyT m' computation, wait for it to finish and discard the
6978-- results.
70−{-# SPECIALIZE wait_ :: AsynclyT IO a -> IO () #-}
79+{-# INLINABLE wait_ #-}
7180wait_ :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m ()
7281wait_ (AsynclyT m) = do
7382 ctx <- liftIO $ getContext Nothing
@@ -76,31 +85,35 @@ wait_ (AsynclyT m) = do
7685
7786 where
7887
79− run m = (runAsyncT m) (return ())
80− (\a r -> maybe (return ()) (\x -> run x) r)
88+ stop = return ()
89+ run m = (runAsyncT m) stop (\_ r -> maybe stop run r)
8190
8291data Step a r = Stop | Done a | Yield a r
8392
84−{-# SPECIALIZE wait :: AsynclyT IO a -> IO [a] #-}
85−{-# INLINE wait #-}
93+{-# INLINABLE wait #-}
94+-- | Run an 'AsynclyT m' computation and collect the results generated by each
95+-- thread of the computation in a list.
8696wait :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m [a]
8797wait (AsynclyT m) = liftIO (getContext Nothing) >>= run m
8898
8999 where
90100
101+ stop = return Stop
102+ done a = return (Done a)
103+ yield a x = return $ Yield a x
104+
91105 run ma ctx = do
92− (res, ctx') <- runStateT ((runAsyncT ma) (return Stop)
93− (\a r -> maybe (return (Done a)) (\x -> return $ Yield a x) r)) ctx
106+ (res, ctx') <- runStateT
107+ ((runAsyncT ma) stop (\a r -> maybe (done a) (\x -> yield a x) r))
108+ ctx
94109 case res of
95110 Yield x mb -> liftM (x :) (run mb ctx')
96− -- XXX pull from children here
97111 Stop -> return []
98112 Done x -> return (x : [])
99113
100114-- scatter
101−{-# SPECIALIZE each :: [a] -> AsynclyT IO a #-}
102−each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
103− => [a] -> AsynclyT m a
115+{-# INLINABLE each #-}
116+each :: MonadAsync m => [a] -> AsynclyT m a
104117each xs = foldr (<|>) empty $ map return xs
105118
106119------------------------------------------------------------------------------
@@ -114,95 +127,6 @@ each xs = foldr (<|>) empty $ map return xs
114127-- threads n action = AsynclyT $ AsyncT $ threadCtl n (runAsyncT $ runAsynclyT action)
115128
116129{-
117−wait_ :: AsyncT IO a -> IO ()
118−wait_ m = do
119− ctx <- liftIO $ getContext (return :: a -> AsyncT IO a) Nothing
120− _ <- runStateT (run m) ctx
121− return ()
122−
123− where
124−
125− run m = runAsyncT m >>= \x ->
126− case x of
127− Yield _ r -> run r
128− -- XXX pull from children here
129− _ -> return ()
130−
131−wait :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]
132−wait m = liftIO (getContext (return :: a -> AsyncT m a) Nothing) >>= run m
133−
134− where
135−
136− run ma ctx = do
137− (a, ctx') <- runStateT (runAsyncT ma) ctx
138− case a of
139− Yield x mb -> liftM (x :) (run mb ctx')
140− -- XXX pull from children here
141− Stop -> return []
142−-}
143−
144−{-
145−------------------------------------------------------------------------------
146−-- Running the monad
147−------------------------------------------------------------------------------
148−
149−-- | Run an 'AsyncT m' computation and collect the results generated by each
150−-- thread of the computation in a list.
151−{-# SPECIALIZE waitAsync :: (a -> StateT Context IO (Maybe a))
152− -> Maybe (IORef [Recording]) -> AsyncT IO a -> IO () #-}
153−waitAsync :: (MonadAsync m, MonadCatch m)
154− => (a -> StateT Context m (Maybe a))
155− -> Maybe (IORef [Recording])
156− -> AsyncT m a -> m ()
157−waitAsync finalizer lref m = do
158− childChan <- liftIO $ atomically newTChan
159− pendingRef <- liftIO $ newIORef []
160− credit <- liftIO $ newIORef maxBound
161−
162− let ctx = initContext childChan pendingRef credit finalizer lref
163−
164− r <- try $ runStateT comp ctx
165− e <- handleResult ctx r
166− maybe (return ()) throwM e
167−
168− where
169− comp = do
170− x <- runAsyncT m
171− maybe (return Nothing) finalizer x
172−
173−-- TBD throttling of producer based on consumption rate.
174−
175−-- | Invoked to store the result of the computation in the context and finish
176−-- the computation when the computation is done
177−{-# SPECIALIZE gatherResult :: IORef [a] -> a -> StateT Context IO (Maybe a) #-}
178−gatherResult :: MonadAsync m => IORef [a] -> a -> StateT Context m (Maybe a)
179−gatherResult ref r = do
180− liftIO $ atomicModifyIORefCAS ref $ \rs -> (r : rs, ())
181− return Nothing
182−
183−gather :: (MonadAsync m, MonadCatch m)
184− => AsyncT m a
185− -> StateT Context m (Maybe [a])
186−gather m = do
187− resultsRef <- liftIO $ newIORef []
188− lift $ waitAsync (gatherResult resultsRef) Nothing m
189− r <- liftIO $ readIORef resultsRef
190− return $ Just r
191−
192−-- | Run an 'AsyncT m' computation and collect the results generated by each
193−-- thread of the computation in a list.
194−{-# SPECIALIZE wait :: AsyncT IO a -> IO [a] #-}
195−wait :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]
196−wait m = do
197− resultsRef <- liftIO $ newIORef []
198− waitAsync (gatherResult resultsRef) Nothing m
199− liftIO $ readIORef resultsRef
200−
201−-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
202−-- results.
203−wait_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m ()
204−wait_ m = waitAsync (const (return Nothing)) Nothing m
205−
206130------------------------------------------------------------------------------
207131-- Logging
208132------------------------------------------------------------------------------