@@ -11,11 +11,6 @@
1111
1212module Strands.AsyncT
1313 ( AsyncT (..)
14− , waitAsync
15− , processOneEvent
16− , drainChildren
17− , waitForChildren
18− , getCtxResultDest
1914 , (<**)
2015 , onNothing
2116 , dbg
@@ -23,12 +18,8 @@ module Strands.AsyncT
2318where
2419
2520import Control.Applicative (Alternative (..))
26−import Control.Concurrent (ThreadId, killThread)
27−import Control.Concurrent.STM (TChan, atomically, newTChan,
28− readTChan, writeTChan)
2921import Control.Monad.Base (MonadBase (..), liftBaseDefault)
30−import Control.Monad.Catch (MonadCatch, MonadThrow, try,
31− throwM, SomeException)
22+import Control.Monad.Catch (MonadThrow, throwM)
3223import Control.Monad.State (MonadIO (..), MonadPlus (..),
3324 MonadState (..), StateT (..),
3425 liftM, modify, runStateT, when)
@@ -38,11 +29,6 @@ import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
3829 defaultLiftBaseWith,
3930 defaultRestoreM)
4031import Data.Dynamic (Typeable)
41−import Data.IORef (IORef, modifyIORef, newIORef, readIORef,
42− writeIORef)
43−import Data.List (delete)
44−import Data.Maybe (fromJust, isJust, isNothing)
45−import Unsafe.Coerce (unsafeCoerce)
4632
4733import Strands.Context
4834
@@ -157,115 +143,6 @@ instance (MonadBaseControl b m, MonadIO m) => MonadBaseControl b (AsyncT m) wher
157143instance MonadThrow m => MonadThrow (AsyncT m) where
158144 throwM e = lift $ throwM e
159145
160−------------------------------------------------------------------------------
161−-- Thread management
162−------------------------------------------------------------------------------
163−
164−-- XXX We are using unbounded channels so this will not block on writing to
165−-- pchan. We can use bounded channels to throttle the creation of threads based
166−-- on consumption rate.
167−processOneEvent :: MonadIO m
168− => ChildEvent a
169− -> Either (TChan (ChildEvent a)) (IORef [a])
170− -> [ThreadId]
171− -> m ([ThreadId], Maybe SomeException)
172−processOneEvent ev dest pending = do
173− -- Collect results unless we have already encountered an exception.
174− case ev of
175− ChildDone _ (Just e) -> handleException e
176− ChildResult (Left e) -> handleException e
177− ChildDone tid Nothing -> return (delete tid pending, Nothing)
178− ChildResult (Right r) -> do
179− case dest of
180− Left chan -> liftIO $ atomically $ writeTChan chan ev
181− Right ref ->
182− liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs
183− return (pending, Nothing)
184− where
185−
186− handleException e = do
187− liftIO $ mapM_ killThread pending
188− return (pending, Just e)
189−
190−drainChildren :: MonadIO m
191− => Either (TChan (ChildEvent a)) (IORef [a])
192− -> TChan (ChildEvent a)
193− -> [ThreadId]
194− -> m ([ThreadId], Maybe SomeException)
195−drainChildren dest cchan pending =
196− case pending of
197− [] -> return (pending, Nothing)
198− _ -> do
199− ev <- liftIO $ atomically $ readTChan cchan
200− (p, e) <- processOneEvent ev dest pending
201− maybe (drainChildren dest cchan p) (const $ return (p, e)) e
202−
203−waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)
204−waitForChildren ctx = do
205− let pendingRef = pendingThreads ctx
206− pending <- liftIO $ readIORef pendingRef
207− (p, e) <- drainChildren (getCtxResultDest ctx) (childChannel ctx) pending
208− liftIO $ writeIORef pendingRef p
209− return e
210−
211−getCtxResultDest :: Context -> Either (TChan (ChildEvent a)) (IORef [a])
212−getCtxResultDest ctx =
213− maybe (Right $ unsafeCoerce $ fromJust $ accumResults ctx)
214− (Left . unsafeCoerce) (parentChannel ctx)
215−
216−------------------------------------------------------------------------------
217−-- Running the monad
218−------------------------------------------------------------------------------
219−
220−collectResult :: MonadIO m => a -> StateT Context m ()
221−collectResult r = do
222− ctx <- get
223− case parentChannel ctx of
224− Nothing -> do
225− let ref = fromJust $ accumResults ctx
226− liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs
227− Just chan -> do
228− -- XXX can we pass the result directly to the root thread
229− -- instead of passing through all the parents? We can let the
230− -- parent go away and handle the ChildDone events as well in
231− -- the root thread.
232− liftIO $ atomically $ writeTChan chan
233− (ChildResult (Right (unsafeCoerce r)))
234−
235−-- | Invoked to store the result of the computation in the context and finish
236−-- the computation when the computation is done
237−finishComputation :: MonadIO m => a -> AsyncT m b
238−finishComputation x = AsyncT $ do
239− collectResult x
240− return Nothing
241−
242−-- XXX pass a collector function and return a Traversable.
243−-- XXX Ideally it should be a non-empty list instead.
244−-- | Run an 'AsyncT m' computation. Returns a list of results of the
245−-- computation or may throw an exception.
246−waitAsync :: forall m a. (MonadIO m, MonadCatch m)
247− => AsyncT m a -> m [a]
248−waitAsync m = do
249− childChan <- liftIO $ atomically newTChan
250− pendingRef <- liftIO $ newIORef []
251− resultsRef <- liftIO $ newIORef []
252− credit <- liftIO $ newIORef maxBound
253−
254− let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit
255− finishComputation resultsRef
256−
257− r <- try $ runStateT (runAsyncT $ m >>= finishComputation) ctx
258−
259− case r of
260− Left (exc :: SomeException) -> do
261− liftIO $ readIORef pendingRef >>= mapM_ killThread
262− throwM exc
263− Right _ -> do
264− e <- waitForChildren ctx
265− case e of
266− Just (exc :: SomeException) -> throwM exc
267− Nothing -> liftIO $ readIORef resultsRef
268−
269146------------------------------------------------------------------------------
270147-- * Extensible State: Session Data Management
271148------------------------------------------------------------------------------