9e55dd7 · streamly · git.iko.soy

git.iko.soy/streamly

Change yield limit type to Int64 instead of Word64

We need a signed type because multiple threads can pre-decrement it to below zero and then increment it to undo that. If it never goes below zero the increment can actually increment it without decrementing.

commit9e55dd781decfaa52755889e8d49c355f512e1ed authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 15 Aug 2018, 18:59 (8 years ago) parentadab08d

2 files changed+8−13

compare with master

unifiedsplit

src/Streamly/SVar.hs+8−10 src/Streamly/Streams/Async.hs−3

src/Streamly/SVar.hs+8−10

@@ -153,7 +153,7 @@ newtype NanoSecs = NanoSecs Word64

153153 , Ord

154154 )

155155

156−newtype Count = Count Word64

156+newtype Count = Count Int64

157157 deriving ( Eq

158158 , Read

159159 , Show

@@ -672,14 +672,16 @@ doFork action exHandler =

672672-- througput is high or when the cost of synchronization is high. For example

673673-- if the application is distributed then inc/dec of a shared variable may be

674674-- very costly.

675+--

676+-- Note that we need it to be an Int type so that we have the ability to undo a

677+-- decrement that takes below zero.

675678{-# INLINE decrementYieldLimit #-}

676679decrementYieldLimit :: SVar t m a -> IO Bool

677680decrementYieldLimit sv =

678681 case remainingYields sv of

679682 Nothing -> return True

680683 Just ref -> do

681− r <- atomicModifyIORefCAS ref $ \x ->

682− (if x >= 1 then x - 1 else 0, x)

684+ r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

683685 return $ r >= 1

684686

685687-- decrementYieldLimit returns False when the old limit is 0. This one returns

@@ -690,13 +692,9 @@ decrementYieldLimitPost sv =

690692 case remainingYields sv of

691693 Nothing -> return True

692694 Just ref -> do

693− r <- atomicModifyIORefCAS ref $ \x ->

694− (if x >= 1 then x - 1 else 0, x)

695+ r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

695696 return $ r > 1

696697

697−-- XXX increment is not safe when the yield limit was already 0

698−-- and therefore was not decremented. We need yield limit to be

699−-- int and negative if we want this to work.

700698{-# INLINE incrementYieldLimit #-}

701699incrementYieldLimit :: SVar t m a -> IO ()

702700incrementYieldLimit sv =

@@ -1143,8 +1141,8 @@ dispatchWorker yieldCount sv = do

11431141 -- style.

11441142 limit <- case remainingYields sv of

11451143 Nothing -> return workerLimit

1146− Just x -> do

1147− n <- liftIO $ readIORef x

1144+ Just ref -> do

1145+ n <- liftIO $ readIORef ref

11481146 return $

11491147 case workerLimit of

11501148 Unlimited -> Limited (fromIntegral n)

src/Streamly/Streams/Async.hs−3

@@ -134,9 +134,6 @@ workLoopLIFOLimited q st sv winfo = run

134134 yieldLimitOk <- liftIO $ decrementYieldLimit sv

135135 if yieldLimitOk

136136 then do

137− -- XXX increment is not safe when the yield limit was already 0

138− -- and therefore was not decremented. We need yield limit to be

139− -- int and negative if we want this to work.

140137 let stop = liftIO (incrementYieldLimit sv) >> run

141138 unStream m st stop single yieldk

142139 -- Avoid any side effects, undo the yield limit decrement if we