adab08d · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

cleanup parallel yield limit code

adab08dHarendra Kumar8 years ago

src/Streamly/SVar.hs+9

compare with master

unifiedsplit

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

46 , setYieldLimit

48 , cleanupSVar

49+ , cleanupSVarFromWorker

50 -- SVar related

51 -- SVar related

51 , newAheadVar

52 , newAheadVar

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

114import Data.Int (Int64)

115import Data.Int (Int64)

115import Data.IORef

116import Data.IORef

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

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

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

117import Data.Maybe (fromJust)

119import Data.Maybe (fromJust)

118import Data.Set (Set)

120import Data.Set (Set)

119import Data.Word (Word64)

121import Data.Word (Word64)

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

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

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

473 (S.toList workers)

475 (S.toList workers)

474

476

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+

475-------------------------------------------------------------------------------

484-------------------------------------------------------------------------------

476-- Dumping the SVar for debug/diag

485-- Dumping the SVar for debug/diag

477-------------------------------------------------------------------------------

486-------------------------------------------------------------------------------

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

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

34 )

35where

37−import Control.Concurrent (throwTo, myThreadId)

38import Control.Monad (ap)

37import Control.Monad (ap)

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

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

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

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

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

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

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

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

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

46import Data.Functor (void)

45import Data.Functor (void)

47−import Data.IORef (newIORef, readIORef)

46+import Data.IORef (newIORef)

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

49import Data.Maybe (fromJust)

47import Data.Maybe (fromJust)

50import Data.Semigroup (Semigroup(..))

48import Data.Semigroup (Semigroup(..))

51import Prelude hiding (map)

49import Prelude hiding (map)

52import System.Clock

50import System.Clock

54−import qualified Data.Set as S

55−

56import Streamly.Streams.SVar (fromSVar)

52import Streamly.Streams.SVar (fromSVar)

57import Streamly.Streams.Serial (map)

53import Streamly.Streams.Serial (map)

58import Streamly.SVar

54import Streamly.SVar

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

82 yieldLimitOk <- liftIO $ decrementYieldLimitPost sv

78 yieldLimitOk <- liftIO $ decrementYieldLimitPost sv

83 if yieldLimitOk

79 if yieldLimitOk

84 then action

80 then action

85− else liftIO $ do

81+ else liftIO $ cleanupSVarFromWorker sv

86− workers <- readIORef (workerThreads sv)

87− self <- myThreadId

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

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

91 stop = liftIO $ winfo >>= sendStop sv

83 stop = liftIO $ winfo >>= sendStop sv

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

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