test_that("nested event loops", { ## Create a function that finishes while its event loop is inactive sleeper <- function(x) { Sys.sleep(x); Sys.getpid() } afun1 <- async(function(x) { x; call_function(sleeper, args = list(x)) }) afun2 <- async(function(x1, x2) { x1; x2 p1 <- afun1(x1) p2 <- delay(0)$then(function() { synchronise(afun1(x2)) }) when_all(p1, p2) }) res <- synchronise(afun2(1, 2)) expect_equal(length(res), 2) expect_true(res[[1]]$result %in% async_env$worker_pool$list_workers()$pid) expect_true(res[[2]]$result %in% async_env$worker_pool$list_workers()$pid) }) test_that("successful call", { afun <- async(function(x) { call_function(function() 100)$ then(function(x) x$result) }) res <- synchronise(afun()) expect_identical(res, 100) }) test_that("successful calls", { afun <- async(function(x) { when_all( call_function(function() Sys.getpid()), call_function(function() Sys.getpid()), call_function(function() Sys.getpid()), call_function(function() Sys.getpid()) ) }) res <- synchronise(afun()) expect_true(is.integer(viapply(res, "[[", "result"))) }) test_that("calls that error", { skip_without_package("processx", "3.4.1.9001") afun <- async(function(x) { when_all( call_function(function() Sys.getpid()), call_function(function() Sys.getpid()), call_function(function() Sys.getpid()), call_function(function() stop("nope")) ) }) expect_error(synchronise(afun()), "nope", class = "error") }) test_that("calls that crash", { skip_without_package("processx", "3.4.1.9001") skip_without_package("callr", "3.7.6.9000") afun <- async(function(x) { when_all( call_function(function() Sys.getpid()), call_function(function() Sys.getpid()), call_function(function() Sys.getpid()), call_function(function() asNamespace("callr")$crash()) ) }) ## Either might happen here err <- tryCatch(synchronise(afun()), error = function(x) x) expect_true( grepl("R session crashed with exit code", err$message) || grepl("R session closed the process connection", err$message)) afun <- async(function(x) { when_all( call_function(function() asNamespace("callr")$crash()), call_function(function() asNamespace("callr")$crash()), call_function(function() asNamespace("callr")$crash()), call_function(function() asNamespace("callr")$crash()) ) }) ## Either might happen here err <- tryCatch(synchronise(afun()), error = function(x) x) expect_true( grepl("R session crashed with exit code", err$message) || grepl("R session closed the process connection", err$message)) }) test_that("handling call errors", { skip_without_package("processx", "3.4.1.9001") skip_without_package("callr", "3.7.1") worker_pid <- async(function() { call_function(function() Sys.getpid())$then(function(x) x$result) }) afun <- async(function(x) { when_all( worker_pid(), worker_pid(), worker_pid(), call_function(function() stop("nope"))$ catch(error = function(e) e) ) }) res <- synchronise(afun()) expect_true(is_count(res[[1]])) expect_true(is_count(res[[2]])) expect_true(is_count(res[[3]])) expect_s3_class(res[[4]], "async_rejected") expect_match(res[[4]]$message, "in callr subprocess") }) test_that("mix calls with others", { skip_on_cran() px <- asNamespace("processx")$get_tool("px") afun <- async(function() { when_all( delay = delay(1/1000)$ then(function() 1), http = http_get(http$url("/status/418"))$ then(function(x) x$status_code), process = run_process(px, c("outln", "foobar"))$ then(function(x) str_trim(x$stdout)), r_process = run_r_process(function() 2)$ then(function(x) x$result), call = call_function(function() 3)$ then(function(x) x$result) ) }) res <- synchronise(afun()) expect_equal( res, list(delay = 1, http = 418, process = "foobar", r_process = 2, call = 3) ) })