@@ -154,6 +154,7 @@ data SVar m a =
154154 , runqueue :: m ()
155155 , runningThreads :: IORef (Set ThreadId)
156156 , queueEmpty :: m Bool
157+ , activeWorkers :: IORef Int
157158 , svarStyle :: SVarStyle
158159 }
159160
@@ -401,7 +402,8 @@ handleChildException sv e = do
401402
402403{-# NOINLINE pushWorker #-}
403404pushWorker :: MonadParallel m => SVar m a -> m ()
404−pushWorker sv =
405+pushWorker sv = do
406+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
405407 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
406408
407409-- | In contrast to pushWorker which always happens only from the consumer
@@ -455,6 +457,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
455457 where
456458
457459 handleException e tid = do
460+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
458461 modifyThread sv tid
459462 -- XXX implement kill async exception handling
460463 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
@@ -475,13 +478,17 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
475478 ChildYield a -> yield a
476479 ChildStop tid e ->
477480 case e of
478− Nothing -> modifyThread sv tid >> continue
481+ Nothing -> do
482+ let active = activeWorkers sv
483+ liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
484+ modifyThread sv tid >> continue
479485 Just ex -> handleException ex tid
480486
481487getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
482488getFifoSVar ctype = do
483489 outQ <- newIORef []
484490 outQMv <- newEmptyMVar
491+ active <- newIORef 0
485492 running <- newIORef S.empty
486493 q <- newQ
487494 let sv =
@@ -492,6 +499,7 @@ getFifoSVar ctype = do
492499 , enqueue = pushL q
493500 , queueEmpty = liftIO $ nullQ q
494501 , svarStyle = ctype
502+ , activeWorkers = active
495503 }
496504 in return sv
497505
@@ -499,6 +507,7 @@ getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
499507getLifoSVar ctype = do
500508 outQ <- newIORef []
501509 outQMv <- newEmptyMVar
510+ active <- newIORef 0
502511 running <- newIORef S.empty
503512 q <- newIORef []
504513 let checkEmpty = null <$> liftIO (readIORef q)
@@ -510,6 +519,7 @@ getLifoSVar ctype = do
510519 , enqueue = enqueueLIFO q
511520 , queueEmpty = checkEmpty
512521 , svarStyle = ctype
522+ , activeWorkers = active
513523 }
514524 in return sv
515525
@@ -517,6 +527,7 @@ getParSVar :: SVarStyle -> IO (SVar m a)
517527getParSVar style = do
518528 outQ <- newIORef []
519529 outQMv <- newEmptyMVar
530+ active <- newIORef 0
520531 running <- newIORef S.empty
521532 let sv =
522533 SVar { outputQueue = outQ
@@ -526,6 +537,7 @@ getParSVar style = do
526537 , enqueue = undefined
527538 , queueEmpty = undefined
528539 , svarStyle = style
540+ , activeWorkers = active
529541 }
530542 in return sv
531543