diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,11 @@ # Revision history for reflex +## 0.9.4.1 + +* Bug fix for `CausalityLoopException`s caused by incorrect coincidence and merge height invalidation by @parenthetical in [#536](https://github.com/reflex-frp/reflex/pull/536). +* Space leak fixed: Applicative combination of Behaviors, with at least one never changing, no longer causes unbounded memory growth by @parenthetical in [#535](https://github.com/reflex-frp/reflex/pull/535). +* Stray debug print removed from non-debug builds. + ## 0.9.4.0 * Add note about `requesting` semantics by @LightAndLight in https://github.com/reflex-frp/reflex/pull/508 diff --git a/bench/RunAll.hs b/bench/RunAll.hs --- a/bench/RunAll.hs +++ b/bench/RunAll.hs @@ -121,6 +121,7 @@ firing n = group ("firing " <> show n) $ Focused.firing n merging n = group ("merging " <> show n) $ Focused.merging n dynamics n = group ("dynamics " <> show n) $ Focused.dynamics n + shared w = group ("sharedInvalidators " <> show w) $ Focused.sharedInvalidators w cases = concat [ sub 100 40 , dynamics 100 @@ -131,6 +132,10 @@ , merging 50 , merging 100 , merging 200 + , shared 30 + , shared 100 + , shared 300 + , shared 1000 ] pattern RunTestCaseFlag = "--run-test-case" diff --git a/reflex.cabal b/reflex.cabal --- a/reflex.cabal +++ b/reflex.cabal @@ -1,6 +1,6 @@ cabal-version: 1.22 name: reflex -version: 0.9.4.0 +version: 0.9.4.1 license: BSD3 license-file: LICENSE maintainer: ryan.trinkle@gmail.com @@ -446,6 +446,19 @@ , reflex , ref-tf , these + +test-suite behaviorLeak + type: exitcode-stdio-1.0 + main-is: behaviorLeak.hs + hs-source-dirs: test + default-language: Haskell2010 + ghc-options: -Wall -O2 -rtsopts "-with-rtsopts=-T" + build-depends: + base >= 4.11 && < 4.22 + , dependent-sum + , reflex + , ref-tf + , primitive benchmark spider-bench type: exitcode-stdio-1.0 diff --git a/src/Reflex/Spider/Internal.hs b/src/Reflex/Spider/Internal.hs --- a/src/Reflex/Spider/Internal.hs +++ b/src/Reflex/Spider/Internal.hs @@ -709,7 +709,7 @@ case val of Just subscribed -> do askParentsRef >>= mapM_ (\r -> liftIO $ modifyIORef' r (SomeBehaviorSubscribed (Some (BehaviorSubscribedPull subscribed)) :)) - askInvalidator >>= mapM_ (\wi -> liftIO $ modifyIORef' (pullSubscribedInvalidators subscribed) (wi:)) + askInvalidator >>= mapM_ (liftIO . addInvalidatorAmortized (pullSubscribedInvalidators subscribed)) liftIO $ touch $ pullSubscribedOwnInvalidator subscribed return $ pullSubscribedValue subscribed Nothing -> do @@ -718,7 +718,8 @@ parentsRef <- liftIO $ newIORef [] holdInits <- askBehaviorHoldInits a <- liftIO $ runReaderIO (unBehaviorM $ pullCompute p) (Just (wi, parentsRef), holdInits) - invsRef <- liftIO . newIORef . maybeToList =<< askInvalidator + inv0 <- maybeToList <$> askInvalidator + invsRef <- liftIO $ newIORef $ InvalidatorList (length inv0) invalidatorPruneThreshold inv0 parents <- liftIO $ readIORef parentsRef let subscribed = PullSubscribed { pullSubscribedValue = a @@ -737,11 +738,49 @@ readHoldTracked :: Hold x p -> BehaviorM x (PatchTarget p) readHoldTracked h = do result <- liftIO $ readIORef $ holdValue h - askInvalidator >>= mapM_ (\wi -> liftIO $ modifyIORef' (holdInvalidators h) (wi:)) + askInvalidator >>= mapM_ (liftIO . addInvalidatorAmortized (holdInvalidators h)) askParentsRef >>= mapM_ (\r -> liftIO $ modifyIORef' r (SomeBehaviorSubscribed (Some (BehaviorSubscribedHold h)) :)) liftIO $ touch h -- Otherwise, if this gets inlined enough, the hold's parent reference may get collected return result +data InvalidatorList x = InvalidatorList + { invalidatorListSize :: !Int + , invalidatorListPruneAt :: !Int + , invalidatorListElems :: ![Weak (Invalidator x)] + } + +emptyInvalidatorList :: InvalidatorList x +emptyInvalidatorList = InvalidatorList + { invalidatorListSize = 0 + , invalidatorListPruneAt = invalidatorPruneThreshold + , invalidatorListElems = [] + } + +-- | Add invalidator weak ref to invalidator list, pruning finalized entries +-- once the list grows to its "prune at" size. +{-# INLINE addInvalidatorAmortized #-} +addInvalidatorAmortized :: IORef (InvalidatorList x) -> Weak (Invalidator x) -> IO () +addInvalidatorAmortized ref weakInvalidator = do + InvalidatorList listSize pruneAt weakInvalidators <- readIORef ref + if listSize < pruneAt + then writeIORef ref $! InvalidatorList (listSize + 1) pruneAt (weakInvalidator : weakInvalidators) + else do + (liveCount, liveInvalidators) <- + foldrM + (\weakInvalidator' (!liveCount, liveInvalidators) -> + (\case Just _ -> (liveCount + 1, weakInvalidator' : liveInvalidators) + Nothing -> (liveCount, liveInvalidators)) + <$> deRefWeak weakInvalidator') + (0, []) + weakInvalidators + writeIORef ref $! InvalidatorList + (liveCount + 1) + (max invalidatorPruneThreshold (2 * liveCount)) + (weakInvalidator : liveInvalidators) + +invalidatorPruneThreshold :: Int +invalidatorPruneThreshold = 100 + {-# INLINABLE readBehaviorUntracked #-} readBehaviorUntracked :: Defer (SomeHoldInit x) m => Behavior x a -> m a readBehaviorUntracked b = do @@ -795,7 +834,7 @@ --type role Hold representational data Hold x p = Hold { holdValue :: !(IORef (PatchTarget p)) - , holdInvalidators :: !(IORef [Weak (Invalidator x)]) + , holdInvalidators :: !(IORef (InvalidatorList x)) , holdEvent :: Event x p -- This must be lazy, or holds cannot be defined before their input Events , holdParent :: !(IORef (Maybe (EventSubscription x))) -- Keeps its parent alive (will be undefined until the hold is initialized) --TODO: Probably shouldn't be an IORef #ifdef DEBUG_NODEIDS @@ -845,6 +884,7 @@ , eventEnvRootClears :: !(IORef [Some RootClear]) , eventEnvCurrentHeight :: !(IORef Height) -- Needed for Subscribe , eventEnvResetCoincidences :: !(IORef [SomeResetCoincidence x]) -- Needed for Subscribe + , eventEnvInvalidatedCoincidences :: !(IORef [SomeCoincidenceSubscribed x]) -- Coincidences whose height was set to 'invalidHeight' this frame and must be recalculated; populated by 'invalidateCoincidenceHeight' , eventEnvDelayedMerges :: !(IORef (IntMap [EventM x ()])) } @@ -947,7 +987,7 @@ hold :: (Patch p, Defer (SomeHoldInit x) m) => PatchTarget p -> Event x p -> m (Hold x p) hold v0 e = do valRef <- liftIO $ newIORef v0 - invsRef <- liftIO $ newIORef [] + invsRef <- liftIO $ newIORef emptyInvalidatorList parentRef <- liftIO $ newIORef Nothing #ifdef DEBUG_NODEIDS nodeId <- liftIO newNodeId @@ -1012,7 +1052,7 @@ --type role PullSubscribed representational data PullSubscribed x a = PullSubscribed { pullSubscribedValue :: !a - , pullSubscribedInvalidators :: !(IORef [Weak (Invalidator x)]) + , pullSubscribedInvalidators :: !(IORef (InvalidatorList x)) , pullSubscribedOwnInvalidator :: !(Invalidator x) , pullSubscribedParents :: ![SomeBehaviorSubscribed x] -- Need to keep parent behaviors alive, or they won't let us know when they're invalidated } @@ -1339,7 +1379,7 @@ newtype RootClear k = RootClear (IORef (DMap k Identity)) -data SomeAssignment x = forall a. SomeAssignment {-# UNPACK #-} !(IORef a) {-# UNPACK #-} !(IORef [Weak (Invalidator x)]) a +data SomeAssignment x = forall a. SomeAssignment {-# UNPACK #-} !(IORef a) {-# UNPACK #-} !(IORef (InvalidatorList x)) a debugFinalize :: Bool debugFinalize = False @@ -1352,8 +1392,6 @@ then Just $ debugStrLn $ "finalizing: " ++ debugNote else Nothing -type WeakList a = [Weak a] - type CanTrace x m = (HasSpiderTimeline x, MonadIO m) @@ -1502,8 +1540,8 @@ propagateSubscriberHold :: forall x p. (HasSpiderTimeline x, Patch p) => Hold x p -> p -> EventM x () propagateSubscriberHold h a = do {-# SCC "trace" #-} when debugPropagate $ traceM (Proxy :: Proxy x) $ liftIO $ do - invalidators <- liftIO $ readIORef $ holdInvalidators h - return $ "SubscriberHold" <> showNodeId h <> ": " ++ show (length invalidators) + InvalidatorList n _ _ <- liftIO $ readIORef $ holdInvalidators h + return $ "SubscriberHold" <> showNodeId h <> ": " ++ show n v <- {-# SCC "read" #-} liftIO $ readIORef $ holdValue h case {-# SCC "apply" #-} apply a v of @@ -1515,7 +1553,9 @@ iRef <- {-# SCC "iRef" #-} liftIO $ evaluate $ holdInvalidators h defer $ {-# SCC "assignment" #-} SomeAssignment vRef iRef v' -data SomeResetCoincidence x = forall a. SomeResetCoincidence !(EventSubscription x) !(Maybe (CoincidenceSubscribed x a)) -- The CoincidenceSubscriber will be present only if heights need to be reset +-- | 'CoincidenceSubscribed' is present only when the coincidence raised its height while subscribing to the inner event. +data SomeResetCoincidence x = forall a. SomeResetCoincidence !(EventSubscription x) !(Maybe (CoincidenceSubscribed x a)) +data SomeCoincidenceSubscribed x = forall a. SomeCoincidenceSubscribed !(CoincidenceSubscribed x a) runBehaviorM :: BehaviorM x a -> Maybe (Weak (Invalidator x), IORef [SomeBehaviorSubscribed x]) -> IORef [SomeHoldInit x] -> IO a runBehaviorM a mwi holdInits = runReaderIO (unBehaviorM a) (mwi, holdInits) @@ -2257,8 +2297,6 @@ oldParents <- liftIO $ FastMutableIntMap.applyPatch parents newSubscriptions liftIO $ for_ oldParents $ \oldParent -> do oldParentHeight <- getEventSubscribedHeight $ _eventSubscription_subscribed oldParent - - print ("updateMe", oldParentHeight) modifyIORef' heightBagRef $ heightBagRemove oldParentHeight return $ IntMap.elems oldParents let changeSubscriber = Subscriber @@ -2274,7 +2312,7 @@ -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain -- its subscription. liftIO $ writeIORef changeSubdRef (changeSubscriber, changeSubscription) - let unsubscribeAll = traverse_ unsubscribe =<< FastMutableIntMap.getFrozenAndClear parents + let unsubscribeAll = traverse_ (unsubscribe . snd) =<< FastMutableIntMap.toList parents return (EventSubscription unsubscribeAll subscribed, occ) @@ -2323,11 +2361,12 @@ toClearIntRef <- newIORef [] toClearRootRef <- newIORef [] coincidenceInfosRef <- newIORef [] + invalidatedCoincidencesRef <- newIORef [] delayedRef <- newIORef IntMap.empty - return $ EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef delayedRef + return $ EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef invalidatedCoincidencesRef delayedRef clearEventEnv :: EventEnv x -> IO () -clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef delayedRef) = do +clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef invalidatedCoincidencesRef delayedRef) = do writeIORef toAssignRef [] writeIORef holdInitRef [] writeIORef dynInitRef [] @@ -2338,6 +2377,7 @@ writeIORef toClearIntRef [] writeIORef toClearRootRef [] writeIORef coincidenceInfosRef [] + writeIORef invalidatedCoincidencesRef [] writeIORef delayedRef IntMap.empty -- | Run an event action outside of a frame @@ -2403,10 +2443,12 @@ writeIORef (switchSubscribedHeight subscribed) $! invalidHeight WeakBag.traverse_ (switchSubscribedSubscribers subscribed) $ invalidateSubscriberHeight myHeight mapM_ _someMergeUpdate_invalidateHeight mergeUpdates --TODO: In addition to when the patch is completely empty, we should also not run this if it has some Nothing values, but none of them have actually had any effect; potentially, we could even check for Just values with no effect (e.g. by comparing their IORefs and ignoring them if they are unchanged); actually, we could just check if the new height is different - forM_ coincidenceInfos $ \(SomeResetCoincidence subscription mcs) -> do + forM_ coincidenceInfos $ \(SomeResetCoincidence subscription mInvalidate) -> do unsubscribe subscription - mapM_ invalidateCoincidenceHeight mcs - forM_ coincidenceInfos $ \(SomeResetCoincidence _ mcs) -> mapM_ recalculateCoincidenceHeight mcs + mapM_ invalidateCoincidenceHeight mInvalidate + invalidatedCoincidences <- readIORef $ eventEnvInvalidatedCoincidences env + writeIORef (eventEnvInvalidatedCoincidences env) [] + forM_ invalidatedCoincidences $ \(SomeCoincidenceSubscribed subscribed) -> recalculateCoincidenceHeight subscribed mapM_ _someMergeUpdate_recalculateHeight mergeUpdates forM_ toReconnect $ \(SomeSwitchSubscribed subscribed) -> do height <- calculateSwitchHeight subscribed @@ -2437,11 +2479,13 @@ then invalidHeight else Height $ succ a -invalidateCoincidenceHeight :: CoincidenceSubscribed x a -> IO () +invalidateCoincidenceHeight :: forall x a. HasSpiderTimeline x => CoincidenceSubscribed x a -> IO () invalidateCoincidenceHeight subscribed = do oldHeight <- readIORef $ coincidenceSubscribedHeight subscribed when (oldHeight /= invalidHeight) $ do writeIORef (coincidenceSubscribedHeight subscribed) $! invalidHeight + let env = _spiderTimeline_eventEnv (unSTE (spiderTimeline :: SpiderTimelineEnv x)) + modifyIORef' (eventEnvInvalidatedCoincidences env) (SomeCoincidenceSubscribed subscribed :) WeakBag.traverse_ (coincidenceSubscribedSubscribers subscribed) $ invalidateSubscriberHeight oldHeight updateSwitchHeight :: Height -> SwitchSubscribed x a -> IO () @@ -2472,8 +2516,8 @@ data SomeSwitchSubscribed x = forall a. SomeSwitchSubscribed {-# NOUNPACK #-} (SwitchSubscribed x a) -invalidate :: IORef [SomeSwitchSubscribed x] -> WeakList (Invalidator x) -> IO (WeakList (Invalidator x)) -invalidate toReconnectRef wis = do +invalidate :: IORef [SomeSwitchSubscribed x] -> InvalidatorList x -> IO (InvalidatorList x) +invalidate toReconnectRef (InvalidatorList _ _ wis) = do forM_ wis $ \wi -> do mi <- deRefWeak wi case mi of @@ -2492,7 +2536,7 @@ InvalidatorSwitch subscribed -> do traceInvalidate $ "invalidate: Switch" <> showNodeId subscribed modifyIORef' toReconnectRef (SomeSwitchSubscribed subscribed :) - return [] -- Since we always finalize everything, always return an empty list --TODO: There are some things that will need to be re-subscribed every time; we should try to avoid finalizing them + return emptyInvalidatorList -- Since we always finalize everything, always return an empty list --TODO: There are some things that will need to be re-subscribed every time; we should try to avoid finalizing them -------------------------------------------------------------------------------- -- Reflex integration diff --git a/test/Reflex/Bench/Focused.hs b/test/Reflex/Bench/Focused.hs --- a/test/Reflex/Bench/Focused.hs +++ b/test/Reflex/Bench/Focused.hs @@ -8,6 +8,7 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeSynonymInstances #-} +{-# LANGUAGE TypeApplications #-} module Reflex.Bench.Focused where @@ -376,6 +377,12 @@ counters = countMany =<< sparse - - - +-- Stress-test sampling a single behavior through a large, shared set of pulls, +-- exercising the invalidator-list pruning path. +sharedInvalidators :: Word -> [(String, TestCase)] +sharedInvalidators width = + [ testB "sumPulls" $ do + b <- current <$> (count @_ @_ @Int =<< events 10) + let ps = [ (+ fromIntegral i) <$> b | i <- [1 .. width] ] + return $ pull $ sum <$> traverse sample ps + ] diff --git a/test/Reflex/Test/Micro.hs b/test/Reflex/Test/Micro.hs --- a/test/Reflex/Test/Micro.hs +++ b/test/Reflex/Test/Micro.hs @@ -20,7 +20,9 @@ import Data.Foldable import Data.Functor.Misc import qualified Data.Map as Map +import qualified Data.IntMap as IntMap import Data.Monoid +import Reflex.Patch.MapWithMove (moveMapKey, insertMapKey, deleteMapKey) import Prelude @@ -214,6 +216,41 @@ e <- events1 return $ coincidence (deep e <$ e) + , testE "coincidence-incremental-height" $ do + tick <- ticks + e2 <- mergeMapIncremental + <$> holdIncremental mempty ((mempty :: PatchMap Int (Event t ())) <$ tick) + pure (leftmost [ void $ coincidence (e2 <$ leftmost [tick, tick]) + , tick]) + + , testE "coincidence-int-incremental-height" $ do + tick <- ticks + h <- mergeIntIncremental + <$> holdIncremental + (IntMap.singleton 1 (void tick)) + (PatchIntMap (IntMap.singleton 1 (Just (void tick))) <$ tick) + pure (leftmost [ void $ coincidence (void h <$ leftmost [tick, tick]) + , tick]) + + , testE "coincidence-switch-reconnect-height" $ do + tick <- ticks + flag <- foldDyn (\_ b -> not b) False tick + let hi = leftmost [tick, tick] + pure (leftmost [void (coincidence (switch (current $ (\b -> if b then hi else tick) <$> flag) <$ hi)), tick]) + + , testE "mergeWithMove-height" $ do + tick <- ticks + let lo = tick + hi = leftmost [tick, tick] + patches <- plan + [ (1, insertMapKey 1 hi) + , (2, deleteMapKey 0) + , (3, insertMapKey 0 lo) + , (4, moveMapKey 1 2) + , (5, deleteMapKey 2) ] + mergeMapIncrementalWithMove + <$> holdIncremental (Map.singleton (0 :: Int) lo) patches + , testB "holdWhileFiring" $ do e <- events1 eo <- headE e @@ -336,6 +373,11 @@ events1 = plan [(1, "a"), (2, "b"), (5, "c"), (7, "d"), (8, "e")] events2 = plan [(1, "e"), (3, "d"), (4, "c"), (6, "b"), (7, "a")] events3 = liftA2 mappend events1 events2 + + -- Unit-valued event occurrences on consecutive frames, for the height/teardown + -- regression tests below (which need occurrences but not their values). + ticks :: TestPlan t m => m (Event t ()) + ticks = plan [(1, ()), (2, ()), (3, ()), (4, ()), (5, ()), (6, ())] eithers :: TestPlan t m => m (Event t (Either String String)) eithers = plan [(1, Left "e"), (3, Left "d"), (4, Right "c"), (6, Right "b"), (7, Left "a")] diff --git a/test/behaviorLeak.hs b/test/behaviorLeak.hs new file mode 100644 --- /dev/null +++ b/test/behaviorLeak.hs @@ -0,0 +1,79 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} + +-- | Test whether Applicative combination of Behaviors leaks memory when one of +-- the Behaviors' source events never occurs. +-- (https://github.com/reflex-frp/reflex/issues/490) + +import Control.Monad +import Control.Monad.IO.Class +import Control.Monad.Primitive (touch) +import Control.Monad.Ref +import Data.Dependent.Sum +import Data.Functor.Identity +import Data.Maybe (fromJust) +import GHC.Stats +import Reflex +import Reflex.Host.Class +import System.Exit +import System.Mem (performMajorGC) +import Text.Printf + +warmupTicks :: Int +warmupTicks = 200000 +windowTicks :: Int +windowTicks = 200000 +samplesPerWindow :: Int +samplesPerWindow = 20 +ticksBetweenWindows :: Int +ticksBetweenWindows = 9 * windowTicks +allowedGrowthBytes :: Integer +allowedGrowthBytes = 512 * 1024 + +testCase :: (Reflex t, MonadHold t m) => Event t b -> m (Behavior t (Int, Int)) +testCase tickE = do + a <- hold (0 :: Int) (0 <$ tickE) + c <- hold (0 :: Int) never + pure $ (,) <$> a <*> c + +main :: IO () +main = do + enabled <- getRTSStatsEnabled + unless enabled $ do + putStrLn "Failed: RTS stats not enabled (run with +RTS -T)" + exitFailure + + runSpiderHost $ do + (tickE, tickTriggerRef) <- newEventWithTriggerRef + b <- runHostFrame $ testCase tickE + trigger <- fromJust <$> readRef tickTriggerRef + let fireOnce = void $ fireEventsAndRead [trigger :=> Identity ()] $ do + v <- sample b + v `seq` pure () + liveBytes = liftIO $ do + performMajorGC + fromIntegral . gcdetails_live_bytes . gc <$> getRTSStats :: IO Integer + measureWindow = do + let chunk = windowTicks `div` samplesPerWindow + samples <- replicateM samplesPerWindow $ do + replicateM_ chunk fireOnce + liveBytes + pure $ sum samples `div` fromIntegral samplesPerWindow + + replicateM_ warmupTicks fireOnce + firstWindowAvg <- measureWindow + replicateM_ ticksBetweenWindows fireOnce + secondWindowAvg <- measureWindow + liftIO $ touch b + liftIO $ do + let growth = secondWindowAvg - firstWindowAvg + if growth <= allowedGrowthBytes + then putStrLn "Succeeded" + else do + printf "Failed: Behavior Applicative space leak\n" + printf " first-window avg live bytes: %d\n" firstWindowAvg + printf " second-window avg live bytes: %d\n" secondWindowAvg + printf " growth over %d intervening ticks: %d bytes (allowed %d)\n" + ticksBetweenWindows growth allowedGrowthBytes + exitFailure