05ca066 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add GC based cleanup of SVars

05ca066Harendra Kumar8 years ago

src/Streamly/SVar.hs+16−4

compare with master

unifiedsplit

@@ -45,6 +45,8 @@ module Streamly.SVar

45 , getYieldLimit

46 , setYieldLimit

48+ , cleanupSVar

49+

48 -- SVar related

50 -- SVar related

49 , newAheadVar

51 , newAheadVar

50 , newParallelVar

52 , newParallelVar

@@ -95,7 +97,7 @@ module Streamly.SVar

95where

97where

97import Control.Concurrent

99import Control.Concurrent

98− (ThreadId, myThreadId, threadDelay, getNumCapabilities)

100+ (ThreadId, myThreadId, threadDelay, getNumCapabilities, throwTo)

99import Control.Concurrent.MVar

101import Control.Concurrent.MVar

100 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)

102 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)

101import Control.Exception (SomeException(..), catch, mask, assert, Exception)

103import Control.Exception (SomeException(..), catch, mask, assert, Exception)

@@ -318,9 +320,9 @@ data SVar t m a = SVar

318 , workerStopMVar :: MVar ()

320 , workerStopMVar :: MVar ()

319

321

320 , svarStats :: SVarStats

322 , svarStats :: SVarStats

321−#ifdef DIAGNOSTICS

322 -- to track garbage collection of SVar

323 -- to track garbage collection of SVar

323 , svarRef :: Maybe (IORef ())

324 , svarRef :: Maybe (IORef ())

325+#ifdef DIAGNOSTICS

324 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)

326 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)

325 -- Shared work queue (stream, seqNo)

327 -- Shared work queue (stream, seqNo)

326 , aheadWorkQueue :: IORef ([t m a], Int)

328 , aheadWorkQueue :: IORef ([t m a], Int)

@@ -460,6 +462,16 @@ setStreamLatency n st =

460getStreamLatency :: State t m a -> Maybe NanoSecs

462getStreamLatency :: State t m a -> Maybe NanoSecs

461getStreamLatency = _streamLatency

463getStreamLatency = _streamLatency

462

464

465+-------------------------------------------------------------------------------

466+-- Cleanup

467+-------------------------------------------------------------------------------

468+

469+cleanupSVar :: SVar t m a -> IO ()

470+cleanupSVar sv = do

471+ workers <- readIORef (workerThreads sv)

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

473+ (S.toList workers)

474+

463-------------------------------------------------------------------------------

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

464-- Dumping the SVar for debug/diag

476-- Dumping the SVar for debug/diag

465-------------------------------------------------------------------------------

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

@@ -1648,8 +1660,8 @@ getAheadSVar st f = do

1648 , workerCount = active

1660 , workerCount = active

1649 , accountThread = delThread sv

1661 , accountThread = delThread sv

1650 , workerStopMVar = stopMVar

1662 , workerStopMVar = stopMVar

1651−#ifdef DIAGNOSTICS

1652 , svarRef = Nothing

1663 , svarRef = Nothing

1664+#ifdef DIAGNOSTICS

1653 , aheadWorkQueue = q

1665 , aheadWorkQueue = q

1654 , outputHeap = outH

1666 , outputHeap = outH

1655#endif

1667#endif

@@ -1733,8 +1745,8 @@ getParallelSVar st = do

1733 , workerCount = active

1745 , workerCount = active

1734 , accountThread = modifyThread sv

1746 , accountThread = modifyThread sv

1735 , workerStopMVar = undefined

1747 , workerStopMVar = undefined

1736−#ifdef DIAGNOSTICS

1737 , svarRef = Nothing

1748 , svarRef = Nothing

1749+#ifdef DIAGNOSTICS

1738 , aheadWorkQueue = undefined

1750 , aheadWorkQueue = undefined

1739 , outputHeap = undefined

1751 , outputHeap = undefined

1740#endif

1752#endif

src/Streamly/Streams/Async.hs+2−2

@@ -303,8 +303,8 @@ getLifoSVar st = do

303 , workerCount = active

304 , accountThread = delThread sv

305 , workerStopMVar = undefined

306−#ifdef DIAGNOSTICS

307 , svarRef = Nothing

306 , svarRef = Nothing

307+#ifdef DIAGNOSTICS

308 , aheadWorkQueue = undefined

309 , outputHeap = undefined

310#endif

@@ -397,8 +397,8 @@ getFifoSVar st = do

397 , workerCount = active

398 , accountThread = delThread sv

399 , workerStopMVar = undefined

400−#ifdef DIAGNOSTICS

401 , svarRef = Nothing

400 , svarRef = Nothing

401+#ifdef DIAGNOSTICS

402 , aheadWorkQueue = undefined

403 , outputHeap = undefined

404#endif

src/Streamly/Streams/SVar.hs+10−6

@@ -39,9 +39,10 @@ where

39import Control.Exception (fromException)

40import Control.Monad.Catch (throwM)

41import Data.Int (Int64)

42−#ifdef DIAGNOSTICS

43import Control.Monad.IO.Class (liftIO)

42import Control.Monad.IO.Class (liftIO)

44−import Data.IORef (newIORef, writeIORef, mkWeakIORef)

43+import Data.IORef (newIORef, mkWeakIORef)

44+#ifdef DIAGNOSTICS

45+import Data.IORef (writeIORef)

45import System.IO (hPutStrLn, stderr)

46import System.IO (hPutStrLn, stderr)

46import System.Clock (Clock(Monotonic), getTime)

47import System.Clock (Clock(Monotonic), getTime)

47#endif

48#endif

@@ -111,13 +112,16 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

111fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a

112fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a

112fromSVar sv = do

113fromSVar sv = do

113 fromStream $ Stream $ \st stp sng yld -> do

114 fromStream $ Stream $ \st stp sng yld -> do

114−#ifdef DIAGNOSTICS_VERBOSE

115 ref <- liftIO $ newIORef ()

116− _ <- liftIO $ mkWeakIORef ref (printSVar sv "SVar Garbage Collected")

116+ _ <- liftIO $ mkWeakIORef ref hook

117 unStream (fromStreamVar sv{svarRef = Just ref}) st stp sng yld

118−#else

118+ where

119− unStream (fromStreamVar sv) st stp sng yld

119+

120+ hook = do

121+#ifdef DIAGNOSTICS_VERBOSE

122+ printSVar sv "SVar Garbage Collected"

120#endif

123#endif

124+ cleanupSVar sv

121

125

122-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

126-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

123-- be read back from the SVar using 'fromSVar'.

127-- be read back from the SVar using 'fromSVar'.

test/Main.hs+26−10

@@ -12,6 +12,7 @@ import Control.Monad.Error.Class (throwError, MonadError)

12import Control.Monad.Trans.Except (runExceptT, ExceptT)

13import Data.Foldable (forM_, fold)

14import Data.List (sort)

15+import System.Mem (performMajorGC)

16import Data.IORef

17import Data.IORef

17import Test.Hspec as H

18import Test.Hspec as H

@@ -88,17 +89,24 @@ main = hspec $ do

88 it "take 1 aheadly" $ checkCleanup aheadly (S.take 1)

89 it "take 1 aheadly" $ checkCleanup aheadly (S.take 1)

89 it "take 1 parallely" $ checkCleanup parallely (S.take 1)

90 it "take 1 parallely" $ checkCleanup parallely (S.take 1)

91−{-

92 it "takeWhile (< 0) asyncly" $ checkCleanup asyncly (S.takeWhile (< 0))

93 it "takeWhile (< 0) wAsyncly" $ checkCleanup wAsyncly (S.takeWhile (< 0))

94 it "takeWhile (< 0) aheadly" $ checkCleanup aheadly (S.takeWhile (< 0))

95 it "takeWhile (< 0) parallely" $ checkCleanup parallely (S.takeWhile (< 0))

97− it "head asyncly" $ checkCleanupFold asyncly (S.head)

97+ testFoldOpsCleanup "head" S.head

98− it "head wAsyncly" $ checkCleanupFold wAsyncly (S.head)

98+ testFoldOpsCleanup "null" S.null

99− it "head aheadly" $ checkCleanupFold aheadly (S.head)

99+ testFoldOpsCleanup "elem" (S.elem 0)

100− it "head parallely" $ checkCleanupFold parallely (S.head)

100+ testFoldOpsCleanup "notElem" (S.notElem 0)

101−-}

101+ testFoldOpsCleanup "elemIndex" (S.elemIndex 0)

102+ -- S.lookup

103+ testFoldOpsCleanup "notElem" (S.notElem 0)

104+ testFoldOpsCleanup "find" (S.find (==0))

105+ testFoldOpsCleanup "findIndex" (S.findIndex (==0))

106+ testFoldOpsCleanup "all" (S.all (==1))

107+ testFoldOpsCleanup "any" (S.any (==0))

108+ testFoldOpsCleanup "and" (S.and . S.map (==1))

109+ testFoldOpsCleanup "or" (S.or . S.map (==0))

102

110

103 ---------------------------------------------------------------------------

111 ---------------------------------------------------------------------------

104 -- Semigroup/Monoidal Composition strict ordering checks

112 -- Semigroup/Monoidal Composition strict ordering checks

@@ -133,13 +141,13 @@ checkCleanup t op = do

133 runStream . serially $ do

141 runStream . serially $ do

134 _ <- t $ op $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil

142 _ <- t $ op $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil

135 return ()

143 return ()

144+ performMajorGC

136 threadDelay 500000

145 threadDelay 500000

137 res <- readIORef r

146 res <- readIORef r

138 res `shouldBe` 0

147 res `shouldBe` 0

139 where

148 where

140− delay ref i = threadDelay (i*100000) >> writeIORef ref i >> return i

149+ delay ref i = threadDelay (i*200000) >> writeIORef ref i >> return i

141

150

142−{-

143checkCleanupFold :: IsStream t

151checkCleanupFold :: IsStream t

144 => (t IO Int -> SerialT IO Int)

152 => (t IO Int -> SerialT IO Int)

145 -> (SerialT IO Int -> IO (Maybe Int))

153 -> (SerialT IO Int -> IO (Maybe Int))

@@ -147,12 +155,20 @@ checkCleanupFold :: IsStream t

147checkCleanupFold t op = do

155checkCleanupFold t op = do

148 r <- newIORef (-1 :: Int)

156 r <- newIORef (-1 :: Int)

149 _ <- op $ t $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil

157 _ <- op $ t $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil

158+ performMajorGC

150 threadDelay 500000

159 threadDelay 500000

151 res <- readIORef r

160 res <- readIORef r

152 res `shouldBe` 0

161 res `shouldBe` 0

153 where

162 where

154− delay ref i = threadDelay (i*100000) >> writeIORef ref i >> return i

163+ delay ref i = threadDelay (i*200000) >> writeIORef ref i >> return i

155−-}

164+

165+testFoldOpsCleanup :: String -> (SerialT IO Int -> IO a) -> Spec

166+testFoldOpsCleanup name f = do

167+ let testOp op x = op x >> return Nothing

168+ it (name ++ " asyncly") $ checkCleanupFold asyncly (testOp f)

169+ it (name ++ " wAsyncly") $ checkCleanupFold wAsyncly (testOp f)

170+ it (name ++ " aheadly") $ checkCleanupFold aheadly (testOp f)

171+ it (name ++ " parallely") $ checkCleanupFold parallely (testOp f)

156

172

157parallelTests :: SpecWith ()

173parallelTests :: SpecWith ()

158parallelTests = H.parallel $ do

174parallelTests = H.parallel $ do