Viewing/signaling conditions signaled in futures

[burchill]

One of the issues that I've thought about in furrr (and in future in general, actually), is that when the plan being used isn't sequential/transparent, the only condition that gets signaled to functions outside of the future call is the error condition. To the best of my knowledge, warnings and messages that arise within future_map/future are basically gone once the results are returned.
While I understand that the whole concept of futures precludes higher-up processes from dictating how the lower processes within the futures deal with these conditions, it seems to me inadvisable that potentially important warnings/messages are automatically thrown out. It seems like this design choice hampers "good code".

I don't know what your thoughts are about this, but if you're interested in implementing something like this, I've extended the future_map functions in my personal custom codebase so that I can collect all the messages, warnings, and errors signaled in these functions, and I can then signal/view everything that was collected, in case the functions calling future_map need to do anything about them.

I uploaded the relevant code as a gist that can be accessed here. Basically, I wrote a function that "pries open" the future_map functions, wraps the .f function in a function that collects conditions, and saves them as new functions. I went with this strategy for my code because I didn't see the point in hard-copying your code into new functions just to change something so small, but it has the added benefit of making things very flexible. I don't know if there are any theoretical problems with the condition collection in collect_all, but it's worked in all the scenarios I've been using it.

Just food for thought, basically! If you think such things are in the purview of this package, I could spruce it up and make a pull request, but I'd also be fine with you making your own code conceptually based off mine, if you just acknowledge me. At some level, I think stuff like this should probably be addressed in the future package itself, so I'll probably be asking Henrik about it regardless.

[yogat3ch]

@DavisVaughan
+1 for this.
It would really be useful for debugging purposes to have messages and warnings surface from the future_* functions either into the console, or at least into the outfile created by parallel::makeCluster as the foreach package allows.

Sometimes an error occurs in a specific sub-process because of the parameters passed to it, especially when doing parameter tuning for a model. It's very useful to message these parameters to the outfile at the start of each new task being passed to a worker such that the user can see which set of parameters caused the error and make changes accordingly. It sounds like @burchill has developed a method for catching these errors that I'm going to have a look at.
Right now, especially with long model build times, very long periods of time can pass without any indication that it is functioning as intended, even with .progress = T for jobs with few iterations but long internal sub-processes. It would definitely be helpful to surface internal progress markers via messages from the sub-processes, at least to the outfile created with the cluster plan and parallel::makeCluster(availableCores(), outfile = 'out.txt').

[burchill]

@yogat3ch Just as a heads up, this sort of condition-handling stuff ended up with me refining/generalizing the code there into the catchr package, which you might want to look at. It's not as focused on furrr as the gist I posted (e.g., it doesn't have the future_map_maker() function), but it's way more flexible, better written, and better documented. A simple example of what you want could be done like so:

library(future)
library(furrr)
library(catchr)
plan(sequential)

write_to_file <- function(cond) {
  cond_class <- class(cond)[1]
  msg <- paste(cond_class, ":", cond$message)
  write(msg, file="out.txt", append=TRUE)
}

cond_to_file <- make_catch_fn(
  warning   = c(write_to_file, muffle),
  message = c(write_to_file, muffle),
  error        = c(write_to_file, exit_with("Returned error!"))
)

# Example in action
fn <- function(x) {
  message("HEY")
  warning("UHOH")
  if (x==2) stop("It broke")
  paste("end","result!")
}

res <- future_map(1:2, ~cond_to_file(fn(.x)))
res

In the example above, the cond_to_file() wrapper writes all messages, warnings, and errors to out.txt, and if there's an error, returns the string "Returned error!" without stopping the rest of the map. I don't think the above code is necessarily thread-safe, but I think it gives you a start about how to do more. If you know what you're doing, it wouldn't be too hard to make it save separate output files for each item being mapped, etc.

[yogat3ch]

Hi @burchill,
Thanks for letting me know about catchr and writing out this example for me! It does exactly what I was hoping and works in parallel! 👏 👏

Many thanks!

If I specify the calls to catchr functions using :: will it require catchr to be loaded to function properly?
Update 2020-02-27 17:22 Just did a test and realized it doesn't require the package to be loaded, awesome! I'm implementing it throughout this long running parameter tuning script I'm building.

[DavisVaughan]

future now relays all conditions as of early 2019
https://github.com/HenrikBengtsson/future/blob/9a5da2f906c8677c9bc4d738428f1ec3a662642a/NEWS#L496-L500

library(furrr)
#> Loading required package: future
#> Warning: package 'future' was built under R version 4.0.2

plan(multisession, workers = 2)

future_map(1:5, ~{
  if (.x == 3L || .x == 5L) {
    warning("oh no!")
  }
  .x
})
#> Warning in ...furrr_fn(...): oh no!
#> Warning in ...furrr_fn(...): oh no!
#> [[1]]
#> [1] 1
#> 
#> [[2]]
#> [1] 2
#> 
#> [[3]]
#> [1] 3
#> 
#> [[4]]
#> [1] 4
#> 
#> [[5]]
#> [1] 5

plan(sequential)

^{Created on 2020-08-06 by the reprex package (v0.3.0)}

[yogat3ch]

I noticed this! Thanks for reminded me about this. However, I'm still using catchr for surfacing progress/status messages in the code when running in parallel from an RStudio background process. Is it possible to surface messages from future/furrr directly or is catchr still the best way to do this?

[DavisVaughan]

can you try and provide a full reproducible example of what you'd like to do?

[yogat3ch]

I was not seeing messages surface when running a cluster plan in an Rstudio background processes in August 2019 for some reason, but it definitely appears that messages are now surfacing! I thought I had the dev version installed but perhaps I didn't.
Anyway, case closed! Thank you all for the effort on this amazingly useful package!

[yogat3ch]

@DavisVaughan
Ok, I tricked myself testing a deceptively simple reprex and didn't realize that the messages weren't actually getting printed while the cluster was running, but rather printed afterwards - so the issue still remains.
I'm trying to surface messages for long running background threads while they're running in a background job. Currently the only way I'm aware of to do this is to print them to a logfile with catchr. The background job actually isn't necessary for the reprex because the behavior is the same regardless of whether it's running in a background job or in the main R session. The messages indicate where the BG process is at so it would be helpful to see them in the console as they execute, alas the messages only surface at the completion of all the processes in lieu of an outfile to surface them as soon as they execute.

Here's the reprex:

cl <- parallel::makeCluster(2)
future::plan(future::cluster, workers = cl)

furrr::future_map(1:5, ~{
  purrr::map(1:10, ~{
    message(paste("Progress message:", .x))
    Sys.sleep(1)
  })
  .x
})

parallel::stopCluster(cl)

[DavisVaughan]

Ah, currently messages are captured in the future on the individual workers, and are relayed back to the user all at once after that future finishes.

For the example above, furrr will make 2 future objects, one with the elements 1:2 and one with elements 3:5. So as soon as the map over 1:2 finishes, you get those messages, then 3:5 finishes a little bit later and you get those messages.

This example makes things a little clearer:

cl <- future::makeClusterPSOCK(2)
future::plan(future::cluster, workers = cl)

furrr::future_map(1:5, ~{
  idx <- .x
  
  purrr::map(1:3, ~{
    message(paste("Progress message:", .x, "from idx", idx))
    Sys.sleep(1)
  })
  .x
})

# These messages are relayed after indices 1:2 finish
Progress message: 1 from idx 1
Progress message: 2 from idx 1
Progress message: 3 from idx 1
Progress message: 1 from idx 2
Progress message: 2 from idx 2
Progress message: 3 from idx 2

# These messages are relayed a little bit later after 3:5 finish
# (They are a little slower because they have one more element to process (5))
Progress message: 1 from idx 3
Progress message: 2 from idx 3
Progress message: 3 from idx 3
Progress message: 1 from idx 4
Progress message: 2 from idx 4
Progress message: 3 from idx 4
Progress message: 1 from idx 5
Progress message: 2 from idx 5
Progress message: 3 from idx 5

What you are asking for is "near real time updates". These have recently been made possible in future so that it can support near real time progress updates. It is supported for multisession, sequential, and cluster futures, but not multicore as of right now.

It is not publicly advertised, but it is possible to hook in to this feature for your own "near real time" messages. The idea is to subclass your message with "immediateCondition", and then future will relay them back as soon as possible. For example, I get the following chain of messages with the following setup. Notice how idx 1 comes back, then idx 3 (these are on two different workers!). It should also be happening in real time.

immediateMessage <- function(..., domain = NULL, appendLF = TRUE) {
  msg <- .makeMessage(..., domain = domain, appendLF = appendLF)
  call <- sys.call()
  m <- simpleMessage(msg, call)
  
  cls <- class(m)
  cls <- setdiff(cls, "condition")
  cls <- c(cls, "immediateCondition", "condition")
  class(m) <- cls
  
  message(m)
  invisible(m)
}

cl <- future::makeClusterPSOCK(2)
future::plan(future::cluster, workers = cl)

furrr::future_map(1:5, ~{
  idx <- .x
  
  purrr::map(1:3, ~{
    immediateMessage(paste("Progress message:", .x, "from idx", idx))
    Sys.sleep(1)
  })
  .x
})

# These are relayed as soon as possible, essentially in real time
# Notice how they are not sequential
Progress message: 1 from idx 1
Progress message: 1 from idx 3
Progress message: 2 from idx 1
Progress message: 2 from idx 3
Progress message: 3 from idx 1
Progress message: 3 from idx 3
Progress message: 1 from idx 2
Progress message: 1 from idx 4
Progress message: 2 from idx 2
Progress message: 2 from idx 4
Progress message: 3 from idx 2
Progress message: 3 from idx 4
Progress message: 1 from idx 5
Progress message: 2 from idx 5
Progress message: 3 from idx 5

Now, I'm not sure that this feature is 100% stable. @HenrikBengtsson would be able to tell you if this is a good idea or not.

[HenrikBengtsson]

Now, I'm not sure that this feature is 100% stable. @HenrikBengtsson would be able to tell you if this is a good idea or not.

Yes, the idea is to be able to use immediateCondition:s for other purposes as well - not just progress updates. There are some simple wrappers for immediateMessage() and immediateWarning() in the future package tests:

https://github.com/HenrikBengtsson/future/blob/425d2ac75a6202f49be8be448941b3c4505538f3/tests/immediateCondition.R#L7-L23

Having said this, I don't want to over-promise anything right now so stride carefully. For example, there is a risk that immediateCondition:s might be signaled twice - once in a near-live fashion and once when the future value is queried. I cannot remember where it stands on that right now. I now that progressr's 'progression' conditions, which inherits from 'immediateCondition', has built-in protect against this.

BTW, note that you can "sticky" progressr messages, e.g p(msg, class = "sticky"). The traditional terminal-based progression handler will push such messages up above the progress bar. See https://cran.r-project.org/web/packages/progressr/vignettes/progressr-intro.html for an example.

[yogat3ch]

Hi @DavisVaughan,

This example makes things a little clearer:

It does indeed!

What you are asking for is "near real time updates"

Yes exactly.

It is not publicly advertised, but it is possible to hook in to this feature for your own "near real time" messages. The idea is to subclass your message with "immediateCondition"

This is exactly what I need! Thank you for the reprex as I would have stumbled around for a while trying to figure out how to implement that on my own! Much gratitude for your expert and expedited assistance!

There are some simple wrappers for immediateMessage() and immediateWarning() in the future package tests:
https://github.com/HenrikBengtsson/future/blob/425d2ac75a6202f49be8be448941b3c4505538f3/tests/immediateCondition.R#L7-L23

@HenrikBengtsson Excellent, I'll have a look. Thanks for hopping in here to help out!

For example, there is a risk that immediateCondition:s might be signaled twice - once in a near-live fashion and once when the future value is queried

Noted. Thank you for letting me know. In the way I'm intending to use this to monitor a long-running background ML model build, duplicate signaling should not be an issue.

I now that progressr's 'progression' conditions, which inherits from 'immediateCondition', has built-in protect against this.

I'll check out progressr and see what features it offers. Most of the messages I'm intending to surface are to provide evidence when steps of the process complete along with completion times such that I can estimate the total build time necessary for multiple models.

BTW, note that you can "sticky" progressr messages, e.g p(msg, class = "sticky"). The traditional terminal-based progression handler will push such messages up above the progress bar.

That's a neat feature, I can think of a use for it already. Thanks for mentioning!

Thank you both so much for the excellent advice! 🤞 that I'm able to implement it without any snags.
This assistance couldn't have been better timed as I just ran into a bizarre issue using the bang bang !! with the catchr implementation that @burchill so kindly provided. I'll open an issue on catchr with a reprex for this shortly.

Grateful for everyone's detailed advice and efforts on these wonderfully useful packages!