120d0ed · streamly · git.iko.soy

git.iko.soy/streamly

Add API to gather results of multiple computations

commit120d0ed5f0bfab0253ae0d0e74dd94d741a9d220 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 29 Jun 2017, 05:40 (9 years ago) parentafbd289

3 files changed+23

compare with master

unifiedsplit

src/Strands.hs+1 src/Strands/Threads.hs+8 test/gather.hs+14

src/Strands.hs+1

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

1616 , async

1717 , waitEvents

1818 , each

19+ , gather

1920 , sample

2021 , threads

2122 , logged

src/Strands/Threads.hs+8

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

88 , each

99 , sample

1010 , sync

11+ , gather

1112 --, react

1213 , threads

1314

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

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

538539 mzero

539540

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+

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

541549-- thread of the computation in a list.

542550wait :: (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