adab08d · streamly · git.iko.soy

git.iko.soy/streamly

cleanup parallel yield limit code

commitadab08de440ab7ea6cd987cba0ff9717fa2d9da5 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 15 Aug 2018, 06:19 (8 years ago) parent05ca066

2 files changed+11−10

compare with master

unifiedsplit

src/Streamly/SVar.hs+9 src/Streamly/Streams/Parallel.hs+2−10

src/Streamly/SVar.hs+9

@@ -46,6 +46,7 @@ module Streamly.SVar

4646 , setYieldLimit

4747

4848 , cleanupSVar

49+ , cleanupSVarFromWorker

4950

5051 -- SVar related

5152 , newAheadVar

@@ -114,6 +115,7 @@ import Data.Heap (Heap, Entry(..))

114115import Data.Int (Int64)

115116import Data.IORef

116117 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)

118+import Data.List ((\\))

117119import Data.Maybe (fromJust)

118120import Data.Set (Set)

119121import Data.Word (Word64)

@@ -472,6 +474,13 @@ cleanupSVar sv = do

472474 Prelude.mapM_ (\tid -> throwTo tid ThreadAbort)

473475 (S.toList workers)

474476

477+cleanupSVarFromWorker :: SVar t m a -> IO ()

478+cleanupSVarFromWorker sv = do

479+ workers <- readIORef (workerThreads sv)

480+ self <- myThreadId

481+ mapM_ (\tid -> throwTo tid ThreadAbort)

482+ (S.toList workers \\ [self])

483+

475484-------------------------------------------------------------------------------

476485-- Dumping the SVar for debug/diag

477486-------------------------------------------------------------------------------

src/Streamly/Streams/Parallel.hs+2−10

@@ -34,7 +34,6 @@ module Streamly.Streams.Parallel

3434 )

3535where

3636

37−import Control.Concurrent (throwTo, myThreadId)

3837import Control.Monad (ap)

3938import Control.Monad.Base (MonadBase(..), liftBaseDefault)

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

@@ -44,15 +43,12 @@ import Control.Monad.Reader.Class (MonadReader(..))

4443import Control.Monad.State.Class (MonadState(..))

4544import Control.Monad.Trans.Class (MonadTrans(lift))

4645import Data.Functor (void)

47−import Data.IORef (newIORef, readIORef)

48−import Data.List ((\\))

46+import Data.IORef (newIORef)

4947import Data.Maybe (fromJust)

5048import Data.Semigroup (Semigroup(..))

5149import Prelude hiding (map)

5250import System.Clock

5351

54−import qualified Data.Set as S

55−

5652import Streamly.Streams.SVar (fromSVar)

5753import Streamly.Streams.Serial (map)

5854import Streamly.SVar

@@ -82,11 +78,7 @@ runOne st m = unStream m st stop single yieldk

8278 yieldLimitOk <- liftIO $ decrementYieldLimitPost sv

8379 if yieldLimitOk

8480 then action

85− else liftIO $ do

86− workers <- readIORef (workerThreads sv)

87− self <- myThreadId

88− mapM_ (\tid -> throwTo tid ThreadAbort)

89− (S.toList workers \\ [self])

81+ else liftIO $ cleanupSVarFromWorker sv

9082

9183 stop = liftIO $ winfo >>= sendStop sv

9284 sendit a = liftIO $ winfo >>= \x -> sendYield sv x (ChildYield a)