@@ -9,6 +9,8 @@ module Strands.Threads
99 , sync
1010 --, react
1111 , threads
12+
13+ , wait
1214 , gather
1315 )
1416where
@@ -25,7 +27,7 @@ import qualified Control.Exception.Lifted as EL
2527import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,
2628 try)
2729import Control.Monad.IO.Class (MonadIO (..))
28−import Control.Monad.State (get, gets, modify,
30+import Control.Monad.State (get, gets, modify, mzero,
2931 runStateT, when, StateT)
3032import Control.Monad.Trans.Class (MonadTrans (lift))
3133import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
@@ -531,30 +533,23 @@ collectResult r = do
531533 liftIO $ atomically $ writeTChan chan
532534 (ChildResult (Right (unsafeCoerce r)))
533535
534−-- | Invoked to store the result of the computation in the context and finish
535−-- the computation when the computation is done
536−finishComputation :: MonadIO m => a -> AsyncT m b
537−finishComputation x = AsyncT $ do
538− collectResult x
539− return Nothing
540−
541536-- XXX pass a collector function and return a Traversable?
542537-- XXX Ideally it should be a non-empty list.
543538
544539-- | Run an 'AsyncT m' computation and collect the results generated by each
545540-- thread of the computation in a list.
546−gather :: forall m a. (MonadIO m, MonadCatch m)
547− => AsyncT m a -> m [a]
548−gather m = do
541+waitAsync :: forall m a b. (MonadIO m, MonadCatch m)
542+ => (a -> AsyncT m a) -> AsyncT m a -> m [a]
543+waitAsync finalizer m = do
549544 childChan <- liftIO $ atomically newTChan
550545 pendingRef <- liftIO $ newIORef []
551546 resultsRef <- liftIO $ newIORef []
552547 credit <- liftIO $ newIORef maxBound
553548
554549 let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit
555− finishComputation resultsRef
550+ finalizer resultsRef
556551
557− r <- try $ runStateT (runAsyncT $ m >>= finishComputation) ctx
552+ r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx
558553
559554 case r of
560555 Left (exc :: SomeException) -> do
@@ -565,3 +560,24 @@ gather m = do
565560 case e of
566561 Just (exc :: SomeException) -> throwM exc
567562 Nothing -> liftIO $ readIORef resultsRef
563+
564+-- | Invoked to store the result of the computation in the context and finish
565+-- 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
570+
571+-- | Run an 'AsyncT m' computation and collect the results generated by each
572+-- thread of the computation in a list.
573+gather :: forall m a. (MonadIO m, MonadCatch m)
574+ => AsyncT m a -> m [a]
575+gather m = waitAsync gatherResult m
576+
577+-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
578+-- results.
579+wait :: forall m a. (MonadIO m, MonadCatch m)
580+ => AsyncT m a -> m ()
581+wait m = do
582+ _ <- waitAsync (const mzero) m
583+ return ()