Fix #931: Observer memory leak (#1256)

* Fix #931: Observer memory leak

Observers were being prevented from being garbage collected by
their own onReactiveDomainEnded() event handlers. This commit
fixes that by making sure that those event handlers are only
registered when autoDestroy=TRUE, and that they are unregistered
both on destruction and when autoDestroy is changed.

* Remove extraneous self$ prefixes

* Add comment explaining autoDestroyHandle

R/reactive-domains.R

@@ -42,11 +42,11 @@ NULL
 #
 ## ------------------------------------------------------------------------
 createMockDomain <- function() {
-  callbacks <- list()
+  callbacks <- Callbacks$new()
   ended <- FALSE
   domain <- new.env(parent = emptyenv())
   domain$onEnded <- function(callback) {
-    callbacks <<- c(callbacks, callback)
+    return(callbacks$register(callback))
   }
   domain$isEnded <- function() {
     ended
@@ -55,7 +55,7 @@ createMockDomain <- function() {
   domain$end <- function() {
     if (!ended) {
       ended <<- TRUE
-      lapply(callbacks, do.call, list())
+      callbacks$invoke()
     }
     invisible()
   }

R/reactives.R

@@ -591,6 +591,11 @@ Observer <- R6Class(
     .domain = 'ANY',
     .priority = numeric(0),
     .autoDestroy = logical(0),
+    # A function that, when invoked, unsubscribes the autoDestroy
+    # listener (or NULL if autodestroy is disabled for this observer).
+    # We must unsubscribe when this observer is destroyed, or else
+    # the observer cannot be garbage collected until the session ends.
+    .autoDestroyHandle = 'ANY',
     .invalidateCallbacks = list(),
     .execCount = integer(0),
     .onResume = 'function',
@@ -620,7 +625,6 @@ registerDebugHook("observerFunc", environment(), label)
       }
       .label <<- label
       .domain <<- domain
-      .autoDestroy <<- autoDestroy
       .priority <<- normalizePriority(priority)
       .execCount <<- 0L
       .suspended <<- suspended
@@ -628,7 +632,9 @@ registerDebugHook("observerFunc", environment(), label)
       .destroyed <<- FALSE
       .prevId <<- ''
 
-      onReactiveDomainEnded(.domain, self$.onDomainEnded)
+      .autoDestroy <<- FALSE
+      .autoDestroyHandle <<- NULL
+      setAutoDestroy(autoDestroy)
 
       # Defer the first running of this until flushReact is called
       .createContext()$invalidate()
@@ -706,11 +712,28 @@ registerDebugHook("observerFunc", environment(), label)
       "Sets whether this observer should be automatically destroyed when its
       domain (if any) ends. If autoDestroy is TRUE and the domain already
       ended, then destroy() is called immediately."
+
+      if (.autoDestroy == autoDestroy) {
+        return(.autoDestroy)
+      }
+
       oldValue <- .autoDestroy
       .autoDestroy <<- autoDestroy
-      if (!is.null(.domain) && .domain$isEnded()) {
-        destroy()
+
+      if (autoDestroy) {
+        if (!.destroyed && !is.null(.domain)) { # Make sure to not try to destroy twice.
+          if (.domain$isEnded()) {
+            destroy()
+          } else {
+            .autoDestroyHandle <<- onReactiveDomainEnded(.domain, .onDomainEnded)
+          }
+        }
+      } else {
+        if (!is.null(.autoDestroyHandle))
+          .autoDestroyHandle()
+        .autoDestroyHandle <<- NULL
       }
+
       invisible(oldValue)
     },
     suspend = function() {
@@ -743,6 +766,11 @@ registerDebugHook("observerFunc", environment(), label)
       suspend()
       .destroyed <<- TRUE
 
+      if (!is.null(.autoDestroyHandle)) {
+        .autoDestroyHandle()
+      }
+      .autoDestroyHandle <<- NULL
+
       if (!is.null(.ctx)) {
         .ctx$invalidate()
       }

tests/testthat/test-reactivity.r

@@ -819,6 +819,66 @@ test_that("observers autodestroy (or not)", {
   })
 })
 
+test_that("observers are garbage collected when destroyed", {
+  domain <- createMockDomain()
+  rv <- reactiveValues(x = 1)
+
+  # Auto-destroy. GC on domain end.
+  a <- observe(rv$x, domain = domain)
+  # No auto-destroy. GC with rv.
+  b <- observe(rv$x, domain = domain, autoDestroy = FALSE)
+  # No auto-destroy and no reactive dependencies. GC immediately.
+  c <- observe({}, domain = domain)
+  c$setAutoDestroy(FALSE)
+  # Similar to b, but we'll set it to autoDestroy later.
+  d <- observe(rv$x, domain = domain, autoDestroy = FALSE)
+  # Like a, but we'll destroy it immediately.
+  e <- observe(rx$x, domain = domain)
+  e$destroy()
+
+  collected <- new.env(parent = emptyenv())
+
+  reg.finalizer(a, function(o) collected$a <- TRUE)
+  reg.finalizer(b, function(o) collected$b <- TRUE)
+  reg.finalizer(c, function(o) collected$c <- TRUE)
+  reg.finalizer(d, function(o) collected$d <- TRUE)
+  reg.finalizer(e, function(o) collected$e <- TRUE)
+
+  rm(list = c("a", "b", "c", "e")) # Not "d"
+
+  gc()
+  # Nothing can be GC'd yet, because all of the observers are
+  # pending execution (i.e. waiting for flushReact).
+  expect_equal(ls(collected), character())
+
+  flushReact()
+  # Now "c" can be garbage collected, because it ran and took
+  # no dependencies (and isn't tied to the session in any way).
+  # And "e" can also be garbage collected, it's been destroyed.
+  gc()
+  expect_equal(ls(collected), c("c", "e"))
+
+  domain$end()
+  # We can GC "a" as well; even though it references rv, it is
+  # destroyed when the session ends.
+  gc()
+  expect_equal(sort(ls(collected)), c("a", "c", "e"))
+
+  # It's OK to turn on auto-destroy even after the session was
+  # destroyed.
+  d$setAutoDestroy(TRUE)
+  # This should no-op.
+  d$setAutoDestroy(FALSE)
+  rm(d)
+  gc()
+  expect_equal(sort(ls(collected)), c("a", "c", "d", "e"))
+
+  rm(rv)
+  # Both rv and "b" can now be collected.
+  gc()
+  expect_equal(sort(ls(collected)), c("a", "b", "c", "d", "e"))
+})
+
 test_that("maskReactiveContext blocks use of reactives", {
   vals <- reactiveValues(x = 123)