b9c1c92 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add the wait primitive to discard the results

b9c1c92Harendra Kumar9 years ago

src/Strands.hs+2−1

compare with master

unifiedsplit

@@ -9,7 +9,8 @@

9--

11module Strands

12+ ( wait

12− ( gather

13+ , gather

13 , async

14 , async

14 , sample

15 , sample

15 , threads

16 , threads

src/Strands/Context.hs+2

@@ -88,6 +88,8 @@ data Context = Context

88 -- the channel until all its children are cleaned up.

89 ---------------------------------------------------------------------------

91+ -- XXX setup a cleanup computation to run rather than passing all these

92+ -- params.

91 -- XXX use Either parentChannel accumResults

93 -- XXX use Either parentChannel accumResults

92 , accumResults :: Maybe (IORef [Any])

94 , accumResults :: Maybe (IORef [Any])

93 -- ^ Accumulated results, only the top level thread context accumulates.

95 -- ^ Accumulated results, only the top level thread context accumulates.

src/Strands/Threads.hs+29−13

@@ -9,6 +9,8 @@ module Strands.Threads

9 , sync

10 --, react

11 , threads

12+

13+ , wait

12 , gather

14 , gather

13 )

15 )

14where

16where

@@ -25,7 +27,7 @@ import qualified Control.Exception.Lifted as EL

25import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,

27import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,

26 try)

28 try)

27import Control.Monad.IO.Class (MonadIO (..))

29import Control.Monad.IO.Class (MonadIO (..))

28−import Control.Monad.State (get, gets, modify,

30+import Control.Monad.State (get, gets, modify, mzero,

29 runStateT, when, StateT)

31 runStateT, when, StateT)

30import Control.Monad.Trans.Class (MonadTrans (lift))

32import Control.Monad.Trans.Class (MonadTrans (lift))

31import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

33import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

@@ -531,30 +533,23 @@ collectResult r = do

531 liftIO $ atomically $ writeTChan chan

533 liftIO $ atomically $ writeTChan chan

532 (ChildResult (Right (unsafeCoerce r)))

534 (ChildResult (Right (unsafeCoerce r)))

533

535

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−

541-- XXX pass a collector function and return a Traversable?

536-- XXX pass a collector function and return a Traversable?

542-- XXX Ideally it should be a non-empty list.

537-- XXX Ideally it should be a non-empty list.

543

538

544-- | Run an 'AsyncT m' computation and collect the results generated by each

539-- | Run an 'AsyncT m' computation and collect the results generated by each

545-- thread of the computation in a list.

540-- thread of the computation in a list.

546−gather :: forall m a. (MonadIO m, MonadCatch m)

541+waitAsync :: forall m a b. (MonadIO m, MonadCatch m)

547− => AsyncT m a -> m [a]

542+ => (a -> AsyncT m a) -> AsyncT m a -> m [a]

548−gather m = do

543+waitAsync finalizer m = do

549 childChan <- liftIO $ atomically newTChan

544 childChan <- liftIO $ atomically newTChan

550 pendingRef <- liftIO $ newIORef []

545 pendingRef <- liftIO $ newIORef []

551 resultsRef <- liftIO $ newIORef []

546 resultsRef <- liftIO $ newIORef []

552 credit <- liftIO $ newIORef maxBound

547 credit <- liftIO $ newIORef maxBound

553

548

554 let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit

549 let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit

555− finishComputation resultsRef

550+ finalizer resultsRef

556

551

557− r <- try $ runStateT (runAsyncT $ m >>= finishComputation) ctx

552+ r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx

558

553

559 case r of

554 case r of

560 Left (exc :: SomeException) -> do

555 Left (exc :: SomeException) -> do

@@ -565,3 +560,24 @@ gather m = do

565 case e of

560 case e of

566 Just (exc :: SomeException) -> throwM exc

561 Just (exc :: SomeException) -> throwM exc

567 Nothing -> liftIO $ readIORef resultsRef

562 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 ()

test/3-sample.hs+1−1

@@ -5,7 +5,7 @@ import System.IO

6import Strands

8−main = waitAsync $ threads 3 $ do

8+main = wait $ threads 3 $ do

9 liftIO $ hSetBuffering stdout LineBuffering

10 mainThread <- liftIO myThreadId

11 liftIO $ putStrLn $ "Main thread: " ++ show mainThread