120d0ed · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add API to gather results of multiple computations

120d0edHarendra Kumar9 years ago

src/Strands.hs+1

compare with master

unifiedsplit

@@ -16,6 +16,7 @@ module Strands

16 , async

17 , waitEvents

18 , each

19+ , gather

19 , sample

20 , sample

20 , threads

21 , threads

21 , logged

22 , logged

src/Strands/Threads.hs+8

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

8 , each

9 , sample

10 , sync

11+ , gather

11 --, react

12 --, react

12 , threads

13 , threads

@@ -537,6 +538,13 @@ gatherResult ref r = do

537 liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, ())

538 liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, ())

538 mzero

539 mzero

539

540

541+gather :: (MonadIO m, MonadCatch m) => AsyncT m a -> AsyncT m [a]

542+gather m = AsyncT $ do

543+ resultsRef <- liftIO $ newIORef []

544+ lift $ waitAsync (gatherResult resultsRef) Nothing m

545+ r <- liftIO $ readIORef resultsRef

546+ return $ Just r

547+

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

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

541-- thread of the computation in a list.

549-- thread of the computation in a list.

542wait :: (MonadIO m, MonadCatch m) => AsyncT m a -> m [a]

550wait :: (MonadIO m, MonadCatch m) => AsyncT m a -> m [a]

test/gather.hsnew file+14

@@ -0,0 +1,14 @@

1+import Control.Concurrent (threadDelay, myThreadId)

2+import Control.Monad.IO.Class (liftIO)

3+import System.Random (randomIO)

4+import System.IO

6+import Strands

8+--- Check space leaks

10+main = do

11+ x <- wait $ do

12+ y <- gather $ each [1..10]

13+ liftIO $ print y

14+ print x