As a workaround to Issues #39 and #42, encode variable names before exporting to BatchJobs and modify the future expression to decode (=rename) them first thing.

Add support for inspect() on listenv:s as well. Either all, a subset or a single element.

findGlobals() does not consider the order of subexpressions. So, although

> async:::findGlobals(substitute({ a <- b }))
[1] "{"  "<-" "b"

is correct, the following is not:

async:::findGlobals(substitute({ a <- b; b <- 1 }))
[1] "{" "<-"

WORKAROUND: Currently one must be careful not overriding globals by local variables.

Update debug option async::debug to future.debug, where the latter is defined by the future package. This is simplify the API.

Make it possible to specify a BatchJobs backend by specifying the pathname to it's configuration script, e.g.

backend('~/.BatchJobs.R')
backend('~/.BatchJobs,cluster.R')
backend('configs/.BatchJobs,cluster.R')

and so on. Aliases should work automagically, e.g.

backend(cluster='configs/.BatchJobs,cluster.R')
backend('cluster')

Add class attibute to batchjobs(), cf. future 0.12.0.

Add LazyFuture using delayedAssign() - just for demonstration. This is basically already in place by the old code, but lets write it up such that there could be a very bare bone Future API package, e.g. future with LazyFuture as the built-in "future", which can be used in practice but also for other developers to look at.

Is it possible to pass asynchronous evaluations to other asynchronous expressions? Currently async values are passed such that it pauses and waits for the value before continuing.

CLEANUP: Drop async() - use batchjobs() instead

%<=% (or delayedAsyncAssign()) assigns AsyncTask to the wrong environment, causing them to be overwritten. For example:

> x <- listenv() ## same with new.env()
> y <- listenv() ## same with new.env()
> x$a %<=% 1
> y$a %<=% 2  ## Same task name!

> stopifnot(!identical(inspect(x$a), inspect(y$a)))
Error: !identical(inspect(x$a), inspect(y$a)) is not TRUE

> inspect(x$a)
AsyncTask:
Expression:
  [1] 2
Status: 'done', 'started', 'submitted'
Backend:
Job registry:  async13548092
  Number of jobs:  1
  Files dir: x:/async/.async/async13548092-files
  Work dir: x:/async
  Multiple result files: FALSE
  Seed: 250509008
  Required packages: BatchJobs

> inspect(y$a)
AsyncTask:
Expression:
  [1] 2
Status: 'done', 'started', 'submitted'
Backend:
Job registry:  async13548092
  Number of jobs:  1
  Files dir: x:/async/.async/async13548092-files
  Work dir: x:/async
  Multiple result files: FALSE
  Seed: 250509008
  Required packages: BatchJobs

but the asynchronous values are correct in the end:

> x$a
[1] 1
> y$a
[1] 2

Add more details to help("backend")

Package passes R CMD check, but...

• tests/AsyncListEnv.R
• tests/inspect.R

only works with options(future=async), while they should work with any Future including options(future=lazyfuture).

Update batchjobs() to handle ... the same way lazy() does it, cf. HenrikBengtsson/future#13.

> x %<=% { a <<- 1; 2 * a }
Error in getGlobals(exprs, envir = envir, tweak = tweakExpression, primitive = FALSE,  :
  Identified a global by static code inspection, but failed to locate the corresponding object in the relevant environments: 'a'

This is because a <<- 1 technically creates a non-local (=global) variable.

findGlobals() incorrectly detects field as a global in subset(df, x < 3), e.g.

> library(async)
> df <- data.frame(x=1:5, y=(1:5)^2)
> small %<=% { subset(df, x < 3) }
Error in FUN("x"[[1L]], ...) : object 'x' not found
}

# Workaround
The workaround is not to use `subset()`, e.g.
```r
> library(async)
> df <- data.frame(x=1:5, y=(1:5)^2)
> small %<=% { keep <- (df$x < 3); df[keep,] }
> small
  x y
1 1 1
2 2 4
}

Add argument 'backend' to tempRegistry() to specify what type of BatchJobs 'backend to use. Also add it to BatchJobsAsyncTask() and pass it down to tempRegistry(). Finally add ... to async() and pass down.

This should allow for

plan(async, backend="multicore")
...

Add argument resources for BatchJobsAsyncTask futures. These resource parameters should be passed along in the submitJobs() call.

Add listenv() - an environment with ordered elements, i.e. that can be iterated over, e.g.

x <- listenv()
x[[1]] <- "a"
x[[4]] <- 3.14
names(x) <- c("A", "B", "C", "D")
as.list(x)

y <- listenv(length=4L)
names(y) <- c("A", "B", "C", "D")
for (ii in seq_along(y)) {
  print(y[[ii]])
}
for (name in names(y)) {
  print(y[[name]])
}

Drop asyncEvalQ().

Can async support magrittr style piping to async()? Example:

y <- f(x) %>% async

and

res <- listenv()
for (kk in 1:10) y[[kk]] <- f(kk) %>% async

The problem of finding global / unknown variables is independent of 'async' package and a solution would be useful elsewhere / for other packages.

To help debugging/troubleshooting, record BatchJobs cluster functions in task object. It looks like these are not recorded in the Registry, but only come in play when you submitJobs() is called.

So, these should be recorded by run(), e.g.

conf <- BatchJobs%:::%getBatchJobsConf()
cf <- BatchJobs%:::%getClusterFunctions(conf)

Add protection against exporting too large objects by mistake. Make it possible to specify warn and error limits via options, e.g.

maxSizeOfGlobals <- getOption("async::maxSizeOfGlobals"=100e6)  # 100 MB
if (totalExportSize > maxSizeOfGlobals) {
  throw("The total size of all objects needed to be exported by the asynchronous
         expression exceeds the maximum allowed size (option 'maxSizeOfGlobals'):
         %g bytes > %g bytes", totalExportSize, maxSizeOfGlobals)
}

Drop makeClusterFunctionsRscript().

BiocParallel::bpvalidate() can be used to identify "unknowns" and "required" packages. Maybe find/getGlobals() can utilize that functions, because it does a better job handling when expressions also attaches new packages.

The backend already set is overridden iff BatchJobs is loaded "on-the-fly" later. This explains for instance why tests/AsyncListEnv.R uses .BatchJobs.R config file when ran through R CMD check async although the script itself clearly specified backend("local").

Add multicore future based on the 'parallel' package, e.g.

multicore <- function (expr, envir=parent.frame(), substitute=TRUE, ...) {
    if (substitute) expr <- substitute(expr)
    future <- MulticoreFuture(expr=expr, envir=envir, substitute=FALSE)
    future <- run(future)
    future
}

where run() for MulticoreFuture uses p <- parallel::mcparallel() and value() uses parallel::mccollect(p, wait=TRUE).

From example(mcparallel, package="parallel"):

p <- mcparallel(1:10)
q <- mcparallel(1:20)
# wait for both jobs to finish and collect all results
res <- mccollect(list(p, q))

Maybe this will so simple that it could be part of the future package?

Extend existing listenv to AsyncTaskList to hold AsyncTask elements. Implement methods status(), finished(), value(), error() etc. that return a named vector of the same length as the list.

For example:

x <- AsyncListEnv()
for (ii in 1:3) {
  name <- letters[ii]
  x[[name]] %<=% { ii^2 }
}

such that for instance:

> length(x)
[1] 3

> names(x)
[1] "a" "b" "c"

> finished(x)
    a     b     c
FALSE  TRUE FALSE

> error()
    a     b     c
FALSE FALSE FALSE

> value(x[-1])
b c
4 9

Rename class BatchJobsAsyncTask to BatchJobsFuture to harmonize with how classes in the future package are labelled.

Due to BatchJobs limitations, it is currently not possible to export operators, replacement functions etc. For example:

batchExport(reg, li=list("names<-"=NA))
Error in batchExport(reg, li=list("names<-"=NA))
  Assertion on 'li' failed: Vector must be named according to R's variable naming rules

and

batchExport(reg, li=list("%>%"=NA))
Error in batchExport(reg, li = list(`%>%` = NA)) :
  Assertion on 'li' failed: Vector must be named according to R's variable naming rules

This will be solved as soon as BatchJobs Issue #93 is resolved.

Update asyncBatchEvalQ() to handle globals just like future package. Export that functionality in the future package to be used by async?

backend("multicore") should acknowledge assigned number of cores, if possible. For instance, this can be done by also looking at Torque/PBS variables, e.g. PBS_NUM_NODES (or PBS_NUM_PPN).

Currently the async package puts all BatchJobs directories directly under ./async/. If one run several R sessions each utilizing async, then it is very hard to know which BatchJobs directory belongs to which R session. If one of the R session is aborted, it will fail to garbage collect meaning all its BatchJobs directories will also remain.

Suggestion: Have each R session use a unique ./async/async_<unique_id>/ directory.

CLEANUP: Drop delayedAssign() operator %<-% and tests.

Now the async BatchJobs work directories are in the current directory, e.g. ./async983970872-files. This tend to clutter up the working directory quite a bit. Instead, move them to a subdirectory, e.g. ./.async/983970872-files/.

Problem

The backend() functions needs to be called explicitly to override the default setup when package is loaded. It is not possible to override this using plan(batchjobs, backend=...) function.

Example

> library("async")

## Explicitly set backend
> backend("multicore=4")
[1] "multicore=4"
> plan(batchjobs)

> Sys.getpid()
[1] 406326

> pid %<=% { Sys.getpid() }
> pid
[1] 407866
> pid %<=% { Sys.getpid() }
> pid
[1] 408592

## Try to change backend via plan()
> plan(batchjobs, backend="interactive")
> pid %<=% { Sys.getpid() }
> pid
[1] 410531
## Hmm... a different one than the main R process
> stopifnot(pid == Sys.getpid())
Error: pid == Sys.getpid() is not TRUE

Troubleshooting

This is because internally run() for BatchJobsAsyncTask calls BatchJobs::submitJobs() which in turn uses conf <- BatchJobs:::getBatchJobsConf(). The latter is set by our backend(), which is

> backend()
[1] "multicore=4"

We are currently temporarily setting backend(backend) when a BatchJobsAsyncTask future is created, or more precisely when the BatchJobs register is created using async:::tempRegistry(). We need to do this in run() instead.

Example:

library("R.utils")
library("async")
backend("multicore=2")
## [1] "multicore=2"

env <- listenv()
t0 <- Sys.time()
for (ii in 1:10) {
  t1 <- Sys.time()
  printf("Task %d @ %.2fs - ", ii, t1-t0)
  env[[ii]] %<=% { Sys.sleep(5); 1000+ii }
  printf("%.2fs\n", Sys.time()-t1)
}
printf("%.2fs: complete!\n", Sys.time() - t0)

## Task 1 @ 0.00s - 0.95s
## Task 2 @ 0.95s - 0.74s
## Task 3 @ 1.70s - 1 temporary submit errors logged to file ## '/home/henrik/.async/async1610630674-files/submit.log'.
## First message: Multicore busy: R
## 11.02s
## Task 4 @ 12.71s - 0.69s
## Task 5 @ 13.40s - 1 temporary submit errors logged to file ## '/home/henrik/.async/async1330253475-files/submit.log'.
## First message: Multicore busy: R
## 11.13s
## Task 6 @ 24.54s - 0.75s
## Task 7 @ 25.29s - 1 temporary submit errors logged to file ## '/home/henrik/.async/async1662451945-files/submit.log'.
## First message: Multicore busy: R
## 10.95s
## Task 8 @ 36.24s - 0.75s
## Task 9 @ 37.00s - 0.83s
## Task 10 @ 37.82s - 1 temporary submit errors logged to file ## '/home/henrik/.async/async1856376469-files/submit.log'.
## First message: Multicore busy: R
## 11.02s
## 48.84s: complete!

str(as.list(env))
## List of 10
##  $ : num 1001
##  $ : num 1002
##  $ : num 1003
##  $ : num 1004
##  $ : num 1005
##  $ : num 1006
##  $ : num 1007
##  $ : num 1008
##  $ : num 1009
##  $ : num 1010

QUESTION: Instead of stalling the main/interactive R process, can we have another cheap background process launch each of them and wait?

This package is solely based on the BatchJobs package/framework. We can now use it with the 'future' package (by specifying future::plan()). Other future mechanisms will be implemented elsewhere/in other packages. Thus, is 'async' too much of a generic name? It's certainly catchy.

Background / current status

## A global variable
a <- 0
f <- lazy({
  42 * a
})

## Since 'a' is a global variable in _lazy_ future 'f',
## which still hasn't been resolved, any changes to
## 'a' until 'f' is resolved, will affect its value.
a <- 1
v <- value(f)
print(v)
stopifnot(v == 42) 

Proposal

Should there be an option to specify when globals in lazy futures should be resolved, e.g.

a <- 0
f <- lazy({
  42 * a
}, globals="eager")
a <- 1
v <- value(f)
print(v)
stopifnot(v == 0) 

versus (current):

a <- 0
f <- lazy({
  42 * a
}, globals="lazy")
a <- 1
v <- value(f)
print(v)
stopifnot(v == 42) 

CLEANUP: Turn BatchJobsAsyncTask() into a simple constructor. Use batchjobs() every where else.

It is not possible to export globals starting with a period, e.g.

> library(async)
> plan(batchjobs, backend="local")
> a <- 1
> x %<=% { 2 * a }
> x
[1] 2
> .b <- 2
> x %<=% { 2 * a * .b }
Error in asyncBatchEvalQ(reg, exprs = list(task$expr), envir = task$envir,  :
  BatchJobs does not support exported variables that start with a period
  (see https://github.com/tudo-r/BatchJobs/issues/103): '.b'

See tudo-r/BatchJobs#103 for troubleshooting and details.

An operator for memoized evaluation, e.g.

y %<-$% f(x)

where $ is for "cache", or

y %<-% f(x) %>% memoize
y %<-% f(x) %>% cache

Should this be part of R.cache?

ANSWER: Yes, eventually. But not before the globals package is on CRAN.

Background

findGlobals() on locally defined S3 methods identifies the generic functions but not the corresponding S3 methods. This is not surprising.

As long as the S3 methods are implemented in packages and those packages are loaded/attached, calling the generic function - as found by findGlobals() - will work. The S3 method to be dispatched on will be found at run time (since the class of the object needs to be known).

Problem

However, if an S3 method is defined "locally" (e.g. in the global environment as part of prototyping/development), that S3 method will not be part of the set of globals that are exported to the job/task.

Workaround

The workaround is to:

1. Check whether an identified global is a generic function or not.
2. If it is a generic function, use methods(<generic>) to search for all matching S3 methods.
3. For each S3 method, check if it is in a package name space or in another environment (e.g. global environment).
4. If in a package namespace, add the package to set of required packages to be loaded (conservative approach).
5. If in a non-package environment, then add the S3 method mth to the list of globals and run findGlobals(mth) on it to make sure it also gets everything it needs.

Problem

There should also be a way to export options(), e.g.

options(width=60)
a %<=% {   getOption("width") 
 }

will return the (default) 80, and not the set 60, because options are currently not exported. Same with par() options etc.

Workaround

Currently one needs to explicitly set the options as in.

options(width=60)
a %<=% {
  options(width=60)
  getOption("width") 
}

Action

This is a trick one, because it could be that we do not want to export all possible options, e.g. BatchJobs options(?!?).

This works:

backend("multicore")
> a <- async({ Sys.sleep(60); 1 })
> status(a)
[1] "running"   "started"   "submitted"

but this doesn't:

> b %<=% { Sys.sleep(60); 2 }
> status(b) ## This stalls!
^C
> status(b)
^C
Warning message:
In status(b) : restarting interrupted promise evaluation
> inspect(b)
AsyncTask:
Expression:
  {
      Sys.sleep(60)
      2
  }
Status: 'running', 'started', 'submitted'
Backend:
Job registry:  async1840699759
  Number of jobs:  1
  Files dir: /home/henrik/projects/exome-copy-number/.async/async1840699759-files
  Work dir: /home/henrik/projects/exome-copy-number
  Multiple result files: FALSE
  Seed: 953806506
  Required packages: BatchJobs

For each async expression/jobs, generate a checksum based on the expression and its global objects. Use this checksum for the async jobs id. This way it should be possible to avoid launching the same asynchroneous job if already running/done/available.

Move asAssignTarget() to listenv and make it a public function. Some more restructuring of that function is possible as will.

Observation

> library(future)
> plan(eager)

> backend("local")
[1] "local"

> x %<=% { 1 } %plan% batchjobs(backend="interactive")
> x
[1] 1

> backend()
[1] "interactive"

Should it?

Troubleshooting

This is because batchjobs() does not undo the backend set, e.g.

> backend()
[1] "local"
> f <- batchjobs({ 1 }, backend="interactive")
> backend()
[1] "interactive"
> value(f)
[1] 1
> backend()
[1] "interactive"

This is in turn because BatchJobsAsyncTask({ 1 }, backend="interactive") does not undo it. Continuing, the problem occurs because of tempRegistry():

> backend("local")
[1] "local"
> reg <- async:::tempRegistry(backend="interactive")
Creating dir: /cbc/henrik/repositories/GitHub/async/.async/async28374250-files
Loading required package: BatchJobs
Loading required package: BBmisc
Saving registry: /cbc/henrik/repositories/GitHub/async/.async/async28374250-files/registry.RData
> backend()
[1] "interactive"

Add a AsyncError class (extending the error class) such that it's possible to register also the AsyncTask objects, BatchJob registry objects and so on.

Currently we just report simple errors, e.g.

msg <- sprintf("BatchJobError: %s", error(obj))
stop(msg, call.=FALSE)

msg <- sprintf("BatchJobExpiration: Job of registry '%s' expired: %s", reg$id, reg$file.dir)
stop(msg, call.=FALSE)

msg <- sprintf("AsyncNotReadyError: Polled for results %d times every %g seconds, but asynchroneous evaluation is still running: BatchJobs registry '%s' (%s)", tries-1L, interval, reg$id, reg$file.dir)
stop(msg, call.=FALSE)

asyncEvalQ() no longer dispatches to jobs workers