Running out of memory when using future_lapply within another software

[blenback]

Hi,
I have a problem with a very specific use case of the future package that maybe someone could shed some light on even though I can't easily provide a reprex.

I am creating a land use change model that links R with a dedicated environmental modelling software called Dinamica EGO. Dinamica EGO has the capability to run either R or Python code from inside it's models launching it's own local sessions. I have a section of R code that optionally involves a parallel process using a future_lapply loop (see code chunk below) or can perform the same task using sequential processing.

I have obviously tested the code in isolation in R and the parallel processing seems to work fine in terms of clearing the memory of the workers (I include gc() calls and deliberately call plan(sequential) after the loop). This is important because there is one large global that needs to be exported to the workers.

However when I run this code as part of my model in Dinamica, it seems like the memory of the workers is not being cleared because I receive an error along the lines of: "No process exists with this PID, i.e. the localhost worker is no longer alive" which I believe can indicate that a memory limit has been reached?

This problem could be related to how the Dinamica EGO software is creating it's local R sessions and I have reached out to the software team to aks. However another important detail is that the Dinamica model itself is being run using a system2 command from R so in reality what is happening is: R -> Dinamica EGO -> R which could also be causing the problem.

As I said, a hyper-specific use case but still I thought I would create a topic incase some may have some advice for me?
Thanks,
Ben.

The code chunk used in the R session started by Dinamica EGO:

future::plan(multisession(workers = availableCores()-2))

size = 5000*1024^2 
options(future.globals.maxSize= size)

#loop over transitions (currently takes 4 mins in parallel)
results <- future.apply::future_lapply(1:nrow(Model_lookup),
                                       future.packages = packs,
                                       function(i){

gc()
                                         
#vector trans_name
Trans_name <- Model_lookup[i, "Trans_name"]
Trans_ID <- Model_lookup[i, "Trans_ID"]
Region <- Model_lookup[i,"Region"]
Final_LULC <- Model_lookup[i, "Final_LULC"]
Initial_LULC <- Model_lookup[i, "Initial_LULC"]

#load model
Fitted_model <- readRDS(Model_lookup[i, "File_path"])

#subset data for prediction
pred_data <- Trans_dataset_complete[Trans_dataset_complete[Initial_LULC] == 1 &
                                      Trans_dataset_complete$Bioregion_Class_Names == Region,] 

#predict using fitted model
prob_predicts <- as.data.frame(predict(Fitted_model, pred_data, type="prob"))
names(prob_predicts)[[2]] <- paste0("Prob_", Final_LULC)

#bind to ID
predict_ID <- cbind(ID = pred_data[, c("ID")], prob_predicts[paste0("Prob_", Final_LULC)])

gc()
#return the prediction results to be bound together
return(predict_ID)
}) #close loop over Models
plan(sequential)

[scottkosty]

I am a novice in futureverse, but I have a few questions that might help others help you:

1. Are you on Windows? Is that why you are using multisession? Do you have access to a Linux machine by chance, so you can see if you can reproduce there? First try to reproduce with multisession and then try to reproduce with multicore.

2. On your current system, if you replace future::plan(multisession(workers = availableCores()-2)) with plan(sequential), can you reproduce the issue? I'm guessing the issue is "no", but that's still helpful to clarify. Do you observe any memory leak at all? i.e., more memory used in the system than if you run the code directly in R?

3. It would be helpful to have some information on your system. e.g., can you give the output to the following? parallelly::availableCores(which = "all"), parallelly::freeConnections(), Sys.info().

[blenback]

Hi thanks for the suggestions,

1. Yes, sorry I should have made it clear that I am indeed using Windows hence using multisession rather than multicore so I assume no point trying to test multicore as Windows is not capable of forking.

2. I just ran a test replacing plan(multisession) with plan(sequential and was not able to reproduce the error. I will have to do some more detailed testing regarding memory leakage.

3. parallelly::availableCores(which = "all")
   system 12
   parallelly::freeConnections: [1] 125
   Sys.info() returns:
   sysname release version nodename machine
   "Windows" "10 x64" "build 19045" "PLUS-BBLACK" "x86-64"

After speaking to the Dinamica software developers they have told me that when sessions are created inside Dinamica there are no restrictions placed on memory access so the session should function like a normal R session.

[scottkosty]

Thanks for that quick reply and extra information. One more command that would be good to run is: futureSessionInfo(). Other than that, I don't have much advice. If you are really stuck and no one else has any idea, you might consider running Dinamica in Linux (it does have a Linux version). I know that would probably take a lot of time to set up, but if everywhere else is a dead end, maybe it's worth a shot.

[scottkosty]

By the way, you might want to run those commands inside the script that Dinamica calls, and have them save to a file. Then compare the output to the output you get when running the commands directly in R. If there is a difference, that could pinpoint a problem. For example, in particular future::futureSessionInfo() might be interesting because your error is about PIDs, and that command gives info related to PIDs.

[HenrikBengtsson]

Looking at your code snippet, each future_lapply() iteration uses only one row of Model_lookup, e.g. Model_lookup[i, "Trans_name"]. This means you can avoid exporting the full Model_lookup object. I recommend using future_by() instead, which has the capability of iterating over rows, e.g.

results <- future.apply::future_by(Model_lookup, INDICES = seq_len(nrow(Model_lookup)), future.packages = packs, FUN = function(row) {
  #vector trans_name
  Trans_name <- row[1, "Trans_name"]
  Trans_ID <- row[1, "Trans_ID"]
  ...
})

Because future.apply chunks up the iterations, this construct will export only the chunk of rows that are needed, instead of the full Model_lookup object. This scales much better. For example, if you have 12 parallel workers, each chunk will contain nrow(Model_lookup)/12 rows, i.e. a 12:th of the full data.frame.

By using future.schedule, or the bijective future.chunk.size, you can tell future.apply to chunk it up more aggressively. If you pass future.schedule = 2.0 (the default is 1.0), you can use twice as many chunks, i.e. each chunk will contain nrow(Model_lookup)/24 rows. future.schedule = 4.0, nrow(Model_lookup)/48 rows, and so on. Alternatively, you can future.chunk.size = 20L to use 20 rows in each chunk, or go to the very extreme by using future.chunk.size = 1L.

---

You could also try to use future.callr::callr instead of multisession; that'll use a fresh parallel worker for each chunk, and then shut it down immediately when done. It comes with more overhead, but it could be more efficient that garbage collecting parallel workers that live longer.

BTW, instead of availableCores() - 2, use availableCores(omit = 2). See https://www.jottr.org/2021/04/30/parallelly-1.25.0/ for why this is better.

---

PS. @blenback, please see https://docs.github.com/en/get-started/writing-on-github/working-with-advanced-formatting/creating-and-highlighting-code-blocks for how to format your code blocks using Markdown. You can edit your existing comments to fix this.

[blenback]

Thanks for the tips Henrik.

To start I swapped to future.callr::callr and availableCores(omit = 2) which appears to have solved the memory problem however the time taken to complete the parallel process is actually longer than the sequential alternative perhaps because the of the additional overhead.

More generally I think I need to re-visit how I approach the loop as Model_lookup is a small global compared to Trans_data_complete which is a large dataframe of ~4million rows and max 100 columns and is currently exported to all workers.

Model_lookup contains statistical models for several spatial regions so perhaps instead I use an upper parallelized loop over the regions and then export only the portion of Trans_dataset_complete that corresponds to a given region, then do an inner loop over Trans_name for each region.

Anyway thanks for the help!