@@ -36,7 +36,7 @@ import Data.IORef (IORef, atomicModifyIORef,
3636 modifyIORef, newIORef, readIORef,
3737 writeIORef)
3838import Data.List (delete)
39−import Data.Maybe (fromJust)
39+import Data.Maybe (fromJust, isJust)
4040import Unsafe.Coerce (unsafeCoerce)
4141
4242import Strands.AsyncT
@@ -88,52 +88,35 @@ runContext ctx = do
8888-- on consumption rate.
8989processOneEvent :: MonadIO m
9090 => ChildEvent a
91− -> Either (TChan (ChildEvent a)) (IORef [a])
9291 -> [ThreadId]
9392 -> m ([ThreadId], Maybe SomeException)
94−processOneEvent ev dest pending = do
95− -- Collect results unless we have already encountered an exception.
96− case ev of
97− ChildDone _ (Just e) -> handleException e
98− ChildResult (Left e) -> handleException e
99− ChildDone tid Nothing -> return (delete tid pending, Nothing)
100− ChildResult (Right r) -> do
101− case dest of
102− Left chan -> liftIO $ atomically $ writeTChan chan ev
103− Right ref ->
104− liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs
105− return (pending, Nothing)
106− where
107−
108− handleException e = do
109− liftIO $ mapM_ killThread pending
110− return (pending, Just e)
93+processOneEvent (ChildDone tid e) pending = do
94+ when (isJust e) $ liftIO $ mapM_ killThread pending
95+ return (delete tid pending, Nothing)
11196
11297drainChildren :: MonadIO m
113− => Either (TChan (ChildEvent a)) (IORef [a])
114− -> TChan (ChildEvent a)
98+ => TChan (ChildEvent a)
11599 -> [ThreadId]
116100 -> m ([ThreadId], Maybe SomeException)
117−drainChildren dest cchan pending =
101+drainChildren cchan pending =
118102 case pending of
119103 [] -> return (pending, Nothing)
120104 _ -> do
121105 ev <- liftIO $ atomically $ readTChan cchan
122− (p, e) <- processOneEvent ev dest pending
123− maybe (drainChildren dest cchan p) (const $ return (p, e)) e
106+ (p, e) <- processOneEvent ev pending
107+ maybe (drainChildren cchan p) (const $ return (p, e)) e
124108
125109waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)
126110waitForChildren ctx = do
127111 let pendingRef = pendingThreads ctx
128112 pending <- liftIO $ readIORef pendingRef
129− (p, e) <- drainChildren (getCtxResultDest ctx) (childChannel ctx) pending
113+ (p, e) <- drainChildren (childChannel ctx) pending
130114 liftIO $ writeIORef pendingRef p
131115 return e
132116
133117tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()
134118tryReclaimZombies ctx = do
135− let dest = getCtxResultDest ctx
136− cchan = childChannel ctx
119+ let cchan = childChannel ctx
137120 pendingRef = pendingThreads ctx
138121
139122 pending <- liftIO $ readIORef pendingRef
@@ -144,7 +127,7 @@ tryReclaimZombies ctx = do
144127 case mev of
145128 Nothing -> return ()
146129 Just ev -> do
147− (p, e) <- processOneEvent ev dest pending
130+ (p, e) <- processOneEvent ev pending
148131 liftIO $ writeIORef pendingRef p
149132 maybe (return ()) throwM e
150133 tryReclaimZombies ctx
@@ -153,13 +136,12 @@ waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()
153136waitForOneEvent ctx = do
154137 -- XXX assert pending must have at least one element
155138 -- assert that the tid is found in our list
156− let dest = getCtxResultDest ctx
157− cchan = childChannel ctx
139+ let cchan = childChannel ctx
158140 pendingRef = pendingThreads ctx
159141
160142 ev <- liftIO $ atomically $ readTChan cchan
161143 pending <- liftIO $ readIORef pendingRef
162− (p, e) <- processOneEvent ev dest pending
144+ (p, e) <- processOneEvent ev pending
163145 liftIO $ writeIORef pendingRef p
164146 maybe (return ()) throwM e
165147
@@ -226,7 +208,6 @@ forkContext context = do
226208 { parentChannel = Just (childChannel ctx)
227209 , pendingThreads = pendingRef
228210 , childChannel = chan
229− , accumResults = Nothing
230211 }
231212
232213 beforeExit ctx res = do
@@ -361,7 +342,6 @@ loopContextWith ioaction context = AsyncT $ do
361342 { parentChannel = parentChannel curCtx
362343 , pendingThreads = pendingThreads curCtx
363344 , childChannel = childChannel curCtx
364− , accumResults = accumResults curCtx
365345 }
366346
367347 where
@@ -518,36 +498,17 @@ noTrans x = AsyncT $ x >>= return . Just
518498-- Running the monad
519499------------------------------------------------------------------------------
520500
521−collectResult :: MonadIO m => a -> StateT Context m ()
522−collectResult r = do
523− ctx <- get
524− case parentChannel ctx of
525− Nothing -> do
526− let ref = fromJust $ accumResults ctx
527− liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs
528− Just chan -> do
529− -- XXX can we pass the result directly to the root thread
530− -- instead of passing through all the parents? We can let the
531− -- parent go away and handle the ChildDone events as well in
532− -- the root thread.
533− liftIO $ atomically $ writeTChan chan
534− (ChildResult (Right (unsafeCoerce r)))
535−
536−-- XXX pass a collector function and return a Traversable?
537−-- XXX Ideally it should be a non-empty list.
538−
539501-- | Run an 'AsyncT m' computation and collect the results generated by each
540502-- thread of the computation in a list.
541503waitAsync :: forall m a b. (MonadIO m, MonadCatch m)
542− => (a -> AsyncT m a) -> AsyncT m a -> m [a]
504+ => (a -> AsyncT m a) -> AsyncT m a -> m ()
543505waitAsync finalizer m = do
544506 childChan <- liftIO $ atomically newTChan
545507 pendingRef <- liftIO $ newIORef []
546− resultsRef <- liftIO $ newIORef []
547508 credit <- liftIO $ newIORef maxBound
548509
549510 let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit
550− finalizer resultsRef
511+ finalizer
551512
552513 r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx
553514
@@ -559,25 +520,28 @@ waitAsync finalizer m = do
559520 e <- waitForChildren ctx
560521 case e of
561522 Just (exc :: SomeException) -> throwM exc
562− Nothing -> liftIO $ readIORef resultsRef
523+ Nothing -> return ()
524+
525+-- TBD throttling of producer based on conumption rate.
563526
564527-- | Invoked to store the result of the computation in the context and finish
565528-- the computation when the computation is done
566−gatherResult :: MonadIO m => a -> AsyncT m a
567−gatherResult x = AsyncT $ do
568− collectResult x
569− return Nothing
529+gatherResult :: MonadIO m => IORef [a] -> a -> AsyncT m a
530+gatherResult ref r = do
531+ liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, rs)
532+ mzero
570533
571534-- | Run an 'AsyncT m' computation and collect the results generated by each
572535-- thread of the computation in a list.
573536gather :: forall m a. (MonadIO m, MonadCatch m)
574537 => AsyncT m a -> m [a]
575−gather m = waitAsync gatherResult m
538+gather m = do
539+ resultsRef <- liftIO $ newIORef []
540+ waitAsync (gatherResult resultsRef) m
541+ liftIO $ readIORef resultsRef
576542
577543-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
578544-- results.
579545wait :: forall m a. (MonadIO m, MonadCatch m)
580546 => AsyncT m a -> m ()
581−wait m = do
582− _ <- waitAsync (const mzero) m
583− return ()
547+wait m = waitAsync (const mzero) m