Allowing future's to timeout

[rcannood]

First off, thanks for this great package. It's insanely useful.

I've been trying to get a future expression to get killed after a given timeout, but I'm not sure whether the code shown below is the best way of going about this. Do you have any suggestions?

Thanks in advance!

Example:

future_timeout(
  expr = {
    Sys.sleep(100) # a really boring function
    data.frame(result = "finished", time = sleeper_time)
  },
  wait_time = 10,
  cancel_output_fun = function(t) data.frame(result = "killed", time = t),
  check_interval = 1,
  verbose = TRUE
)

Attempt:

#' Kill a future after a certain period of time
#'
#' @param expr An R \code{\link[base]{expression}} to be evaluated.
#' @param wait_time How long to wait before killing it.
#' @param cancel_output_fun A function \code{function(time_elapsed){...}} to call when the total time
#'   has exceeded the \code{wait_time}.
#' @param check_interval The time waited between checking whether the future has already finished.
#' @param verbose Whether or not to print waiting times.
#' @param ... extra arguments to be passed to the \code{\link[future]{future}} call.
#'
#' @importFrom future future plan resolved value
#' @export
#'
#' @examples
#' future_timeout(
#'   expr = {
#'     Sys.sleep(100) # a really boring function
#'     data.frame(result = "finished", time = sleeper_time)
#'   },
#'   wait_time = 10,
#'   cancel_output_fun = function(t) data.frame(result = "killed", time = t),
#'   check_interval = 1,
#'   verbose = TRUE
#' )
future_timeout <- function(expr, wait_time, cancel_output_fun, check_interval = 1, verbose = FALSE, ...) {
  expr <- substitute(expr)
  envir <- parent.frame()

  requireNamespace("future")
  start_time <- Sys.time()

  future_handle <- future(
    { expr },
    evaluator = plan("multisession"),
    substitute = FALSE,
    envir = envir,
    ...
  )

  time_waited <- 0
  while (!resolved(future_handle) && time_waited < wait_time) {
    Sys.sleep(check_interval)
    time_waited <- as.numeric(difftime(Sys.time(), start_time, units = "secs"))
    if (verbose) cat("Time waited: ", time_waited, "\n", sep = "")
  }

  if (resolved(future_handle)) {
    value(future_handle)
  } else {
    future:::ClusterRegistry("stop")
    cancel_output_fun(time_waited)
  }
}

Test:

context("Future helper")

test_that("Future helper works correctly", {
  test_fun <- function(sleeper_time, wait_time) {
    dry_run <- future_timeout({1}, wait_time = 5, function(x) x, 1)
    future_timeout(
      expr = {
        Sys.sleep(sleeper_time)
        dplyr::data_frame(result = "finished", time = sleeper_time)
      },
      wait_time = wait_time,
      cancel_output_fun = function(t) dplyr::data_frame(result = "killed", time = t),
      check_interval = 1
    )
  }

  out1 <- test_fun(10, 5)
  expect_equal(out1$result, "killed")
  expect_gte(out1$time, 5)

  out2 <- test_fun(5, 10)
  expect_equal(out2$result, "finished")
  expect_gte(out2$time, 5)

  # Also in parallel settings
  sleeper_times <- sample(c(runif(10, .5, 3), runif(10, 7, 10)))
  wait_time <- 4

  parallelMap::parallelStartMulticore(cpus = 2, show.info = TRUE)
  outp <- dplyr::bind_rows(parallelMap::parallelMap(test_fun, sleeper_times, more.args = list(wait_time = wait_time)))
  parallelMap::parallelStop()

  expected_result <- ifelse(sleeper_times < wait_time, "finished", "killed")
  expect_lte( mean(outp$result == expected_result), .9 ) # allow some tests to fail
  expect_true( all(outp$time >= pmin(sleeper_times, wait_time)) )
})

[HenrikBengtsson]

Thank you - I'm glad you find it useful.

Before anything else, maybe

f <- future(R.utils::withTimeout({ expr }, timeout = 10))

will achieve what you're after? This will cause value(f) throw a TimeoutException (catchable) if timed out.

(from my phone)

[rcannood]

Unfortunately, this does not work when R calls C/C++ code or even a system call. Examples are:

R.utils::withTimeout({ z <- rnorm(1000000000) }, timeout = 1) # requires about ~8GB free memory
R.utils::withTimeout({ system("sleep 10") }, timeout = 1)

[HenrikBengtsson]

I see - this clarifies your initial attempts. Rather swamped but here's some quick feedback:

About limitations of withTimeout() - I've created HenrikBengtsson/R.utils#74 and HenrikBengtsson/R.utils#75 (thxs). PS/FYI, in R (>= 3.5.0), system() calls will (to some extent) support timeouts.

Generic support for evaluation timeout is tricky and as far as I understand it, the most likely solution for this is to terminate a subprocess (and we certaintly don't want to terminate the main R process). The termination of subprocesses can/should probably be done in different ways, e.g. signalling an interrupt (SIGINT) and if that's not enough bump of the signal (e.g. to SIGTERM or SIGKILL). The Future API as it stands now does not provide a "kill" mechanism, but it is on the todo/wish list, cf. Issue #93. As you see there, not all future backends support temination of the evaluation. Because of this, it needs to be figured out how to extend the Future API with an option to terminate, a way for the developer to specify that termination is optional or required, etc. The way I'm starting to think about the Future API, is that there is a Future Minimal API that all backends have to conform to and then there is a (to be developed/defined) Future Extended API that some backends may supported partially or in full. Your use case helps to identify what should go in to this extended/optional API. Other examples are support for progress updates and forwarding of stdout/stderr (e.g. #141 and #67).

BTW, your idea of using a per-future timer is related to Issue #59. Before this discussion, I though about the timer as an internal / non-exported property, but I can now see it could be an exported feature. Since it's something that is likely to be supported by all types of futures / backends ("Future Minimal API").

For your needs, I would recommend to try to tackled (in some way) the concept of being able to terminate a future. There are some ideas in #93, which are limited to locally running futures/workers, i.e. multicore and multisession (and cluster). Killing individual futures seems a bit less abrupt that calling all workers (e.g. future:::ClusterRegistry("stop")).

It's not obvious to me how to restart workers that have been killed, but it should be doable, i.e. launching a new one and "hotswap" a stopped cluster "node" with a new one. That probably warrants another issue by itself - I'm happy to discuss / consider a solution for that. BTW, this is partially related to Issue https://github.com/HenrikBengtsson/future/issues/154 (on what to do with crashed cluster workers). I also have some todo notes (not yet in an issue) on being able to shrink and grow the set of cluster workers dynamically which is related to this.

Hope this helps a little bit.

[rcannood]

Thanks for having added this feature to the wishlist.

In the short term, it might make more sense for me to convince withTimeout to terminate a subprocess if it's taking too long. In the meantime, future_timeout works, though it might not be the most elegant solution. I'll keep you posted on further developments.

Since there is already HenrikBengtsson/R.utils#74 and #93, I will be closing this issue.