gc policy

R/class_active.R

@@ -189,6 +189,8 @@ active_class <- R6::R6Class(
         counter_del_name(self$scheduler$progress$queued, name)
       } else if (target_should_run(target, self$meta)) {
         self$flush_upload_meta_file(target)
+        runtime_increment_targets_run(tar_runtime)
+        target_gc(target)
         self$run_target(name)
       } else {
         self$skip_target(target)

R/class_builder.R

@@ -125,7 +125,6 @@ target_run.tar_builder <- function(target, envir, path_store) {
     builder_unset_tar_runtime()
     target$subpipeline <- NULL
   })
-  target_gc(target)
   builder_ensure_deps(target, target$subpipeline, "worker")
   frames <- frames_produce(envir, target, target$subpipeline)
   builder_set_tar_runtime(target, frames)
@@ -146,12 +145,14 @@ target_run_worker.tar_builder <- function(
   options,
   envvars
 ) {
+  set_envvars(envvars)
+  tar_options$import(options)
   envir <- if_any(identical(envir, "globalenv"), globalenv(), envir)
   tar_option_set(envir = envir)
   tar_runtime$store <- path_store
   tar_runtime$fun <- fun
-  tar_options$import(options)
-  set_envvars(envvars)
+  runtime_increment_targets_run(tar_runtime)
+  target_gc(target)
   builder_unmarshal_subpipeline(target)
   target_run(target, envir, path_store)
   builder_marshal_value(target)

R/class_runtime.R

@@ -15,7 +15,8 @@ runtime_new <- function(
   traceback = NULL,
   pid_parent = NULL,
   file_systems = NULL,
-  trust_timestamps_store = NULL
+  trust_timestamps_store = NULL,
+  number_targets_run = NULL
 ) {
   force(target)
   force(frames)
@@ -34,6 +35,7 @@ runtime_new <- function(
   force(pid_parent)
   force(file_systems)
   force(trust_timestamps_store)
+  force(number_targets_run)
   environment()
 }
 
@@ -75,6 +77,12 @@ runtime_validate_basics <- function(x) {
     tar_assert_chr(x$fun)
     tar_assert_nzchar(x$fun)
   }
+  if (!is.null(x$number_targets_run)) {
+    tar_assert_scalar(x$number_targets_run)
+    tar_assert_int(x$number_targets_run)
+    tar_assert_none_na(x$number_targets_run)
+    tar_assert_ge(x$number_targets_run, 1L)
+  }
 }
 
 runtime_validate_extras <- function(x) {
@@ -146,6 +154,15 @@ runtime_file_systems <- function() {
   out
 }
 
+runtime_increment_targets_run <- function(x) {
+  count <- .subset2(x, "number_targets_run")
+  if (is.null(count)) {
+    count <- 0L
+  }
+  count <- count + 1L
+  x$number_targets_run <- count
+}
+
 runtime_reset <- function(x) {
   for (field in names(x)) {
     x[[field]] <- NULL

R/class_target.R

@@ -293,6 +293,12 @@ target_run_worker <- function(
 target_gc <- function(target) {
   if (target$settings$garbage_collection) {
     gc()
+  } else {
+    count <- .subset2(tar_runtime, "number_targets_run") %|||% 0L
+    interval <- .subset2(tar_options, "get_garbage_collection")()
+    if (interval > 0L && (count %% interval) == 0L) {
+      gc()
+    }
   }
 }
 

tests/testthat/test-class_runtime.R

@@ -227,3 +227,12 @@ tar_test("runtime inventories", {
   x$inventories <- ""
   expect_error(runtime_validate(x), class = "tar_condition_validate")
 })
+
+tar_test("runtime_increment_targets_run()", {
+  x <- runtime_new()
+  expect_null(x$number_targets_run)
+  runtime_increment_targets_run(x)
+  expect_equal(x$number_targets_run, 1L)
+  runtime_increment_targets_run(x)
+  expect_equal(x$number_targets_run, 2L)
+})