9e55dd7 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Change yield limit type to Int64 instead of Word64

9e55dd7Harendra Kumar8 years ago

src/Streamly/SVar.hs+8−10

compare with master

unifiedsplit

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

153 , Ord

154 )

155

156−newtype Count = Count Word64

156+newtype Count = Count Int64

157 deriving ( Eq

158 , Read

159 , Show

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

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

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

674-- 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.

675{-# INLINE decrementYieldLimit #-}

678{-# INLINE decrementYieldLimit #-}

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

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

677decrementYieldLimit sv =

680decrementYieldLimit sv =

678 case remainingYields sv of

681 case remainingYields sv of

679 Nothing -> return True

682 Nothing -> return True

680 Just ref -> do

683 Just ref -> do

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

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

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

683 return $ r >= 1

685 return $ r >= 1

684

686

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

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

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

690 case remainingYields sv of

692 case remainingYields sv of

691 Nothing -> return True

693 Nothing -> return True

692 Just ref -> do

694 Just ref -> do

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

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

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

695 return $ r > 1

696 return $ r > 1

696

697

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.

700{-# INLINE incrementYieldLimit #-}

698{-# INLINE incrementYieldLimit #-}

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

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

702incrementYieldLimit sv =

700incrementYieldLimit sv =

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

1143 -- style.

1141 -- style.

1144 limit <- case remainingYields sv of

1142 limit <- case remainingYields sv of

1145 Nothing -> return workerLimit

1143 Nothing -> return workerLimit

1146− Just x -> do

1144+ Just ref -> do

1147− n <- liftIO $ readIORef x

1145+ n <- liftIO $ readIORef ref

1148 return $

1146 return $

1149 case workerLimit of

1147 case workerLimit of

1150 Unlimited -> Limited (fromIntegral n)

1148 Unlimited -> Limited (fromIntegral n)

src/Streamly/Streams/Async.hs−3

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

134 yieldLimitOk <- liftIO $ decrementYieldLimit sv

135 if yieldLimitOk

136 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.

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

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

141 unStream m st stop single yieldk

138 unStream m st stop single yieldk

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

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