Can progressr handlers receive partial results from future_lapply?

[scottkosty]

First, thank you so much for all of your work on Futureverse. I'm really excited to use this framework for developing packages to get consistent (yet flexible) behavior on all OS's.

Background (feel free to skip): I'm working on a package to facilitate Monte Carlo simulations and am planning to rely on future.apply::future_lapply(). I would like to provide a feature where the user can add hooks that are run every M simulations. For example, the user could have a hook that would save partial results, which is nice if the simulations are expected to take a long time and there is a chance of error (e.g., too much RAM required for large number of observations, or electricity goes out). Another example of a hook would be to process and display the partial results. I would like to do this and at the same time still display a progress bar.

Below is a serial (and inefficient for several reasons) example to give an idea of what I hope to allow the user to do. I would like to achieve something like the example using future_lapply(), and I'm wondering if progressr can be used in combination to achieve it.

hook1 <- function(partial_results) {
  message("mean of (partial) results vector: ", mean(res))
}

hook2 <- function(partial_results) {
  # commenting out to keep things clean for the example
  # saveRDS(partial_results, file = "partial_results.Rds")
}

# hook1 and hook2 will be called every 'hook_freq' sims.
hook_freq <- 5000
res <- vector()
for (i in seq.int(50000)) {
  res <- c(res, rnorm(n = 1, mean = 0, sd = 100))
  if (i %% hook_freq == 0) {
    hook1(res)
    hook2(res)
  }
}

I wonder if hook_freq will need to be a multiple of future.chunk.size? After reading some of your explanations, I understand that this behavior will have a non-trivial cost in terms of efficiency, but hopefully if hook_freq/n is not too small there will not be too large of a hit?

After writing this up, it feels like my desired behavior could be an abuse of the progressr framework, even if it can be set up to get partial results. I post this discussion here since progressr is the only example I've seen where future_lapply() function can in some way signal/communicate with an outside function about progress/partial results. However, I now have the feeling that signaling where we are (i.e., the i in the above loop) might be a different beast from communicating the entire partial results vector. Please let me know if I should close here and post instead at https://github.com/HenrikBengtsson/future/discussions.

[scottkosty]

@HenrikBengtsson Do you know if this is currently possible?

[HenrikBengtsson]

Sorry for missing replying to this one.

I would like to achieve something like the example using future_lapply() ...

My approach would be to first rewrite your for loop as a lapply() call, verify that it does what's expected, and then replace that with future_lapply(). So, something like:

# hook1 and hook2 will be called every 'hook_freq' sims.
hook_freq <- 5000
res <- future_lapply(seq.int(50000), FUN = function(i) {
  res_i <- rnorm(n = 1, mean = 0, sd = 100)
  if (i %% hook_freq == 0) {
    hook1(res_i)
    hook2(res_i)
  }
  res_i
})

I would like to provide a feature where the user can add hooks that are run every M simulations.

The above rewrite should do exactly this, regardless of your parallel backend and how many workers you have. It will call the hook functions with the frequency you request. Note that, since it runs in parallel, you might of course see RDS files being saved out of order.

If you think about it, there's nothing with your need to cache results to the file system that is related to parallelization, or the future framework; your problem is exactly the same when you run sequentially with plan(sequential), or when you use lapply() or a for-loop.

... it feels like my desired behavior could be an abuse of the progressr framework ...

Nah, progressr does not support this, and, yes, it would be "abusing" it. The longer answer is that, although your gut feeling is that one could use a similar framework to progressr to send other type of information from parallel workers back to the parent R session, progressr can only signal a character string message. So, unless you encode your information into a character string, and then write your own progressr handler to decode that (and do what ever you want with that info), then progressr is not the way to go here.

[scottkosty]

Thanks for those details!

I would like to provide a feature where the user can add hooks that are run every M simulations.

The above rewrite should do exactly this, regardless of your parallel backend and how many workers you have. It will call the hook functions with the frequency you request. Note that, since it runs in parallel, you might of course see RDS files being saved out of order.

When you rewrote the for loop to an lapply() (and finally future_lapply()) call, I think you changed the functionality. The iterations of the for loop are not independent. In particular, hook1(res) in the for loop is acting on all of the completed results up to that point, not just the current iteration as your code hook1(res_i) does. The output of the for loop is the following (although the seed is not set so results will differ):

mean of (partial) results vector: -1.50017720401453
mean of (partial) results vector: -1.03122404077541
mean of (partial) results vector: -0.611001824461129
mean of (partial) results vector: -0.399810306867108
mean of (partial) results vector: -0.663713094401054
mean of (partial) results vector: -0.344964796916196
mean of (partial) results vector: -0.470787351448641
mean of (partial) results vector: -0.357298286601921
mean of (partial) results vector: -0.15412913375626
mean of (partial) results vector: -0.0940442478214597

The goal was to show, in real time, how the mean of the results (over all completed iterations so far) is converging.

I think I wrote something confusing in my original post. I wrote

I would like to provide a feature where the user can add hooks that are run every M simulations.

I should have specified:

I would like to provide a feature where the user can add hooks that accept the list of completed results so far and that are run every M simulations.

Another way of framing my dream functionality is that I would like a more efficient way of achieving the following implementation:

library("future.apply")

hook1 <- function(partial_results) {
  message("mean of (all completed) results vector: ", mean(simplify2array(partial_results)))
}

hook2 <- function(partial_results) {
  # commenting out to keep things clean for the example
  # saveRDS(partial_results, file = "partial_results.Rds")
}


# hook1 and hook2 will be called every 'hook_freq' sims.
hook_freq <- 5000
nits <- 50000
n_chunks <- nits/hook_freq
# could use an outer future_lapply here, but let's keep things simple (?)
res_all_chunks <- vector()
for (j in seq.int(n_chunks)) {
  res_chunk_j <- future_lapply(seq.int(hook_freq), FUN = function(i) {
    res_i <- rnorm(n = 1, mean = 0, sd = 100)
  }, future.seed = TRUE)
  res_all_chunks <- c(res_all_chunks, res_chunk_j)
  hook1(res_all_chunks)
  hook2(res_all_chunks)
}

By "more efficient", I mean in my dream functionality I would not have to wait until all works finishes for a specific chunk, then call the hooks, and then have to restart the workers. Even with my dream functionality, I understand there will be non-trivial overhead, but hopefully not as much as the above implementation.

I see that future_lapply() has an argument future.chunk.size, but I think that's purely about the division of labor, where I'm looking for a way to access the completed results so far.

Now that I've hopefully clarified my initial post, is this dream functionality currently possible or should I give up?

[HenrikBengtsson]

I see.

There are two things here:

1. Writing partial results to the file system at regular intervals (here, in separate, parallel processes)
2. Once in a while, read any partial results on file and produce a summary (in the main R process)

The first "step" can be done as already done. Since you're doing it in parallel, the different parallel processes must not write to the same file, because then it is likely that they overwrite each other either completely or partially. The best is probably to write to separate files and then merge them in the second "step". I also recommend writing these files atomically, e.g. use saveRDS(..., file = "foo.rds.tmp") and then file.rename("foo.rds.tmp", "foo.rds"). Then, signal a progress, e.g. call p().

The second "step" can be triggered by a custom progression handler, which can be created by make_progression_handler(). This custom handler should, on an update, search for available *.rds files, read them, merge the data, and produce the summary.

[scottkosty]

The first "step" can be done as already done.

Do you mean that I should keep the code as is in #121 (reply in thread) where I manually split the chunks, and have an outer for loop (that is not parallelized) ? Or are you referring to your code, in the part if (i %% hook_freq == 0) { ?

Since you're doing it in parallel, the different parallel processes must not write to the same file, because then it is likely that they overwrite each other either completely or partially. The best is probably to write to separate files and then merge them in the second "step".

This makes me think you're talking about your code that uses if (i %% hook_freq == 0) { since I indeed understand that would cause parallel writes. But that code could save data only from the current iteration. That is, the body of the iteration doesn't have access to the partial results that are done so it can't save all the partial results, unless I am missing a way? For example, if I change "resi" to "res" in the code you wrote, in the following I get the expected error object 'res' not found:

library("future.apply")

hook1 <- function(partial_results) {
  message("mean of (partial) results vector: ", mean(partial_results))
}

hook2 <- function(partial_results) {
  # commenting out to keep things clean for the example
  # saveRDS(partial_results, file = "partial_results.Rds")
}

hook_freq <- 5000
res <- future_lapply(seq.int(50000), FUN = function(i) {
  res_i <- rnorm(n = 1, mean = 0, sd = 100)
  if (i %% hook_freq == 0) {
    # I changed "resi" to "res" in the following calls
    hook1(res)
    hook2(res)
  }
  res_i
}, future.seed = TRUE)

I also recommend writing these files atomically, e.g. use saveRDS(..., file = "foo.rds.tmp") and then file.rename("foo.rds.tmp", "foo.rds").

Makes sense.

The second "step" can be triggered by a custom progression handler, which can be created by make_progression_handler(). This custom handler should, on an update, search for available *.rds files, read them, merge the data, and produce the summary.

Sounds good, I'll give this a shot once I understand how to do the first step.

[HenrikBengtsson]

It might be easier if you roll your own future-based for-loop, e.g.

library(future)

my_fcn <- function(n = 100L, summary_freq = 10L) {
  p <- progressr::progressor(n)
  fs <- vector("list", length = n)
  for (ii in seq_len(n)) {
    fs[[ii]] <- future({
      Sys.sleep(0.1)
      rnorm(n = 1L, mean = 0, sd = 100)
    }, future.seed = TRUE)
    p()

    ## For every n:th iteration, calculate a
    ## summary of currently available results
    if (ii %% summary_freq == 0) {
      rs <- resolved(fs)
      vs_resolved <- value(fs[rs])
      vs_resolved <- unlist(vs_resolved)
      summary <- mean(vs_resolved)
      p(sprintf("Average: %g", summary), amount = 0)
    }
  }
  vs <- value(fs)
  vs <- unlist(vs)
  vs
}


plan(multisession)

progressr::handlers(global = TRUE)
progressr::handlers("cli")

y <- my_fcn(100L)

[scottkosty]

I think you're right this is the approach (I didn't know) I was looking for. Thanks for taking the time to understand my goal and to help me towards it. Much appreciated!