Understanding memory usage and performance of `furrr::future_apply`

[alessandro-peron-sdg]

I am facing some issues parallelizing processes with furrr::future_apply.

This is the setting I am having issues with:

    rm(list=ls(all=TRUE))
    
    require(future)
    require(furrr)
    require(dplyr)
    require(readr)
    require(parallel)
    set.seed(123)
    
    # fake data
    my_list <-   replicate(1000000, rnorm(1000), simplify = FALSE)
    
    # function to parallelize
    f_to_parallelize <- function(x){
      
      y <- sum(x)
      
      return(y)
      
    }
    
    # plans to test
    plan(sequential)
    #plan(multisession, workers=2)
    #plan(multisession, workers=6)
    #plan(multisession, workers=15)
    
    l <- future_walk(my_list, f_to_parallelize)
    

When I profile memory and time for these 4 plans this is what I get:

I have launched 4 different jobs from R studio server, while I was profiling all memory used for processes with my user in a separate job to get data for the graph.

This is the outpu of my sessionInfo()) of the parallelization jobs:

R version 4.2.2 Patched (2022-11-10 r83330)
Platform: x86_64-pc-linux-gnu (64-bit)
Running under: Ubuntu 20.04.6 LTS
Matrix products: default
BLAS: /usr/lib/x86_64-linux-gnu/blas/libblas.so.3.9.0
LAPACK: /usr/lib/x86_64-linux-gnu/lapack/liblapack.so.3.9.0
locale:
[1] LC_CTYPE=C.UTF-8 LC_NUMERIC=C LC_TIME=C.UTF-8
[4] LC_COLLATE=C.UTF-8 LC_MONETARY=C.UTF-8 LC_MESSAGES=C.UTF-8
[7] LC_PAPER=C.UTF-8 LC_NAME=C LC_ADDRESS=C
[10] LC_TELEPHONE=C LC_MEASUREMENT=C.UTF-8 LC_IDENTIFICATION=C
attached base packages:
[1] parallel stats graphics grDevices utils datasets methods
[8] base
other attached packages:
[1] readr_2.1.2 dplyr_1.1.0 furrr_0.2.3 future_1.24.0
loaded via a namespace (and not attached):
[1] rstudioapi_0.13 parallelly_1.30.0 magrittr_2.0.2 hms_1.1.1
[5] tidyselect_1.2.0 R6_2.5.1 rlang_1.1.1 fansi_1.0.2
[9] globals_0.14.0 tools_4.2.2 utf8_1.2.2 cli_3.6.0
[13] ellipsis_0.3.2 digest_0.6.29 tibble_3.1.6 lifecycle_1.0.3
[17] crayon_1.5.0 tzdb_0.2.0 purrr_1.0.1 vctrs_0.5.2
[21] codetools_0.2-18 glue_1.6.2 compiler_4.2.2 pillar_1.7.0
[25] generics_0.1.2 listenv_0.8.0 pkgconfig_2.0.3

Is this behavior normal? I did not expected the steep increase in memory for all the plans, other than the increase in time when I increase the number of workers.

I also tested the sys.sleep(1) function in parallel, and I got the result I expected, time decreases as I increase workers.

What I am trying to parallelize is far more complex than this, i.e. a series of nested wrapped functions that do some training for some time series models and inference writing a csv and not returning anything.

I fill like I am losing something very simple but yet I cannot wrap my head around it, what concerns me the most is the memory increase, as it would be a very memory intensive function.

[DavisVaughan]

Can you also post the code you used to do the memory profiling?

[alessandro-peron-sdg]

Sure, here it is:

rm(list=ls(all=TRUE))

library(tibble)
library(glue)
library(lubridate)

run_command <- function(filename) {
  while (TRUE) {
    output <- system("ps -u username--no-headers -o rss | awk '{sum+=$1} END {print (sum/1024/1024)}'", intern = TRUE)
    output_df <- tibble(time = format(Sys.time(), "%Y-%m-%d %H:%M:%S"), output = output)
    
    # Check if the file exists
    if (file.exists(filename)) {
      # Append the output to the existing CSV file
      write.table(output_df, file = filename, append = TRUE, sep = ",", row.names = FALSE, col.names = FALSE)
    } else {
      # Create a new CSV file and write the output
      write.table(output_df, file = filename, append = FALSE, sep = ",", row.names = FALSE, col.names = TRUE)
    }
    
    Sys.sleep(1)
  }
}

run_command("output_file.csv")

[alessandro-peron-sdg]

@DavisVaughan any news on this?

[rasmusrhl]

I'm having the same issue.

[D3SL]

Same for me as well. I also recently several gigabytes of temp files had been created and never cleaned up and parallelized functions do not complete as quickly as they used to. I've been using furrr for years with excellent performance, this is unusual. It feels like something else changed in the R ecosystem that's impacting furrr.

I may have to switch to Crew (powered by mirai). It's a shame because nothing comes close to furrr in terms of syntactic sugar and ease of use.

[DavisVaughan]

I've run some tests with this example locally and can't reproduce anything worrysome. Here's a reproducible script below.

require(furrr)
library(ggplot2)
library(dplyr)
set.seed(123)

# fake data
my_list <- replicate(1000000, seq_len(1000), simplify = FALSE)

# function to parallelize
f_to_parallelize <- function(x) {
  y <- sum(x)

  return(y)
}

get_pids <- function(plan) {
  workers <- attributes(plan)$backend$workers
  purrr::map_int(unclass(workers), function(worker) {
    worker$session_info$process$pid
  })
}

test_one <- function(workers) {
  plan(multisession, workers = workers)

  pids <- get_pids(plan())

  bg <- callr::r_bg(args = list(pids = pids), func = function(pids) {
    out <- list()
    row <- 1L
    stop <- Sys.time() + 20

    while (Sys.time() < stop) {
      for (pid in pids) {
        handle <- ps::ps_handle(pid)
        info <- ps::ps_memory_info(handle)
        out[[row]] <- vctrs::new_data_frame(list(
          pid = pid,
          time = unclass(Sys.time()),
          rss = info[[1]],
          vms = info[[2]]
        ))
        row <- row + 1L
      }
      Sys.sleep(.1)
    }

    out <- vctrs::vec_rbind(!!!out)
    out
  })

  l <- future_walk(my_list, f_to_parallelize)

  while (bg$is_alive()) {
    Sys.sleep(.5)
  }

  tibble::as_tibble(bg$get_result())
}

workers <- c(2, 4, 6, 8, 10)
results <- lapply(workers, test_one)

out <- tibble::tibble(
  workers = workers,
  results = results
) |>
  tidyr::unnest(results)

out |>
  mutate(time = floor(time)) |>
  summarise(rss = mean(rss), .by = c(workers, pid, time)) |>
  summarise(rss = sum(rss), .by = c(workers, time)) |>
  mutate(workers = as.factor(workers)) |>
  ggplot(aes(x = time, y = rss, color = workers)) +
  geom_point() +
  scale_y_continuous(labels = prettyunits::pretty_bytes) +
  labs(y = "Total memory used across all workers")

As you add more workers the baseline memory used by the set of workers increases, because a static R instance requires some memory. And as they start doing some work you see the amount of memory increase, but then it levels off. I don't see any runaway here.

I'm going to close. If you can provide a reproducible script like the one above that shows some kind of issue, feel free to open another issue. Thanks!