ce79079 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add a thread limit for async ahead types

ce79079Harendra Kumar8 years ago

src/Streamly/Core.hs+7−1

compare with master

unifiedsplit

@@ -486,7 +486,13 @@ sendWorkerWait sv = do

486 when (n <= 0) $ do

487 done <- queueEmpty sv

488 if not done

489+ then do

490+ cnt <- liftIO $ readIORef $ activeWorkers sv

491+ if (cnt < 1500)

492+ then do

493+ pushWorker sv

489− then pushWorker sv >> sendWorkerWait sv

494+ sendWorkerWait sv

495+ else liftIO $ takeMVar (doorBell sv)

490 else liftIO $ takeMVar (doorBell sv)

496 else liftIO $ takeMVar (doorBell sv)

491

497

492-- | Pull a stream from an SVar.

498-- | Pull a stream from an SVar.

test/Main.hs+6−1

@@ -9,7 +9,7 @@ import Control.Exception (Exception, try)

9import Control.Monad.Catch (throwM, MonadThrow)

10import Control.Monad.Error.Class (throwError, MonadError)

11import Control.Monad.Trans.Except (runExceptT, ExceptT)

12−import Data.Foldable (forM_)

12+import Data.Foldable (forM_, fold)

13import Data.List (sort)

14import Test.Hspec

@@ -334,6 +334,11 @@ main = hspec $ do

334 describe "Composed MonadThrow asCoparAhead" $ composeWithMonadThrow asCoparAhead

335 describe "Composed MonadThrow asParallel" $ composeWithMonadThrow asParallel

336

337+ it "Crosses thread limit (2000 threads)" $

338+ runStream (asParAhead $ fold $

339+ replicate 2000 $ A.once $ threadDelay 1000000)

340+ `shouldReturn` ()

341+

337-- XXX need to test that we have promptly cleaned up everything after the error

342-- XXX need to test that we have promptly cleaned up everything after the error

338-- XXX We can also check the output that we are expected to get before the

343-- XXX We can also check the output that we are expected to get before the

339-- error occurs.

344-- error occurs.