ce79079 · streamly · git.iko.soy

git.iko.soy/streamly

Add a thread limit for async ahead types

currently hardcoded to 1500 threads

commitce790790023843c7876205c135467ac8c81fc76b authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 11 May 2018, 01:49 (8 years ago) parent5c975de

2 files changed+13−2

unifiedsplit

src/Streamly/Core.hs+7−1

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

486486 when (n <= 0) $ do

487487 done <- queueEmpty sv

488488 if not done

489− then pushWorker sv >> sendWorkerWait sv

489+ then do

490+ cnt <- liftIO $ readIORef $ activeWorkers sv

491+ if (cnt < 1500)

492+ then do

493+ pushWorker sv

494+ sendWorkerWait sv

495+ else liftIO $ takeMVar (doorBell sv)

490496 else liftIO $ takeMVar (doorBell sv)

491497

492498-- | Pull a stream from an SVar.

test/Main.hs+6−1

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

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

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

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

12−import Data.Foldable (forM_)

12+import Data.Foldable (forM_, fold)

1313import Data.List (sort)

1414import Test.Hspec

1515

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

334334 describe "Composed MonadThrow asCoparAhead" $ composeWithMonadThrow asCoparAhead

335335 describe "Composed MonadThrow asParallel" $ composeWithMonadThrow asParallel

336336

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

338+ runStream (asParAhead $ fold $

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

340+ `shouldReturn` ()

341+

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

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

339344-- error occurs.