Suggestion: mirai package

[shikokuchuo]

Hi Sebastian, following feedback from useR!2024 I'd like to propose my own package.

I've seen your reasoning for not including parallel processing to date, and have hence held back thus far.

However I want to highlight that there are now 'fast' and 'slow' ways to do parallel processing / distributed computing itself. The gain in speed is easily over 100x from existing solutions (see https://shikokuchuo.net/user2024-conference/#/mirai---100x-faster).

This is achieved by the minimalist design of mirai, and the fact it has just one (recursive) dependency on nanonext, which is itself a high-performance binding written almost entirely in C (including interfacing R serialization at the C API level etc.). Also to be noted is the (optional) integration with the later event loop (developed in collaboration with Joe Cheng), which enables event-driven, zero-latency promise resolution for Shiny / Plumber.

We are talking about taking round-trip task resolution from the millisecond to the microsecond scale - enabling R usage in contexts where it wasn't feasible before such as high frequency real-time data, or just for more responsive Shiny apps!

If you are receptive, am happy to draft or review a PR. Thanks.

[SebKrantz]

Thanks @shikokuchuo. I'm not principally opposed if the framework is, as you say, very fast and easy to use. However, looking at the main vignette, I'm not seeing an example that I find extremely compelling that would prompt me to use it. Take the first example:

library(mirai)

input <- list(x = 2, y = 5, z = double(1e8))

m <- mirai(
  {
    res <- rnorm(1e8, mean = mean, sd = sd)
    max(res) - min(res)
  },
  mean = input$x,
  sd = input$y
)

I understand this expression gets evaluated in a different R session. But where is the parallelism? Any why do I need z? Can I run this in parallel using input <- list(x = 1:2, y = 4:5) (e.g. like parallel mapply). I was wondering if

Map(function(x, y) mirai({
    res <- rnorm(1e8, mean = mean, sd = sd)
    max(res) - min(res)
  },
  mean = x,
  sd = y
), 1:4, 4:7) |> lapply(`[`)

gives parallel execution? In any case, it should be very obvious how the package can be used to generate greater performance. A major performance bottlenecks of course is duplication of objects in memory. For example

z <- rnorm(1e7)
Map(function(x, y) mirai({
  res <- rnorm(1e8, mean = mean, sd = sd)
  res <- res + z
  max(res) - min(res)
},
mean = x,
sd = y
), 1:4, 4:7) |> lapply(`[`)

Does not work it seems. So, bottom line, I need to see how this is useful to optimize common operations. Ideally with shared memory. That would be well worth featuring if its a kind of real fire and forget solution.

[shikokuchuo]

Hi @SebKrantz, mirai() is a general function for evaluating arbitrary code in another R process, locally or across the network, hence the readme example probably did not accord with what you had in mind.

In many async use cases it suffices that the computation is done in (offloaded to) a different process so as not to block the main R session. For more convenient use in data manipulation, I believe mirai_map() is the function you are searching for.

Having said that, your example above does seem to work and give parallel execution:

Map(function(x, y) mirai({
    res <- rnorm(1e8, mean = mean, sd = sd)
    max(res) - min(res)
  },
  mean = x,
  sd = y
), 1:4, 4:7) |> lapply(`[`)

An even better example would be to set persistent daemons and use mirai_map(), for example re-writing your last example:

library(mirai)
daemons(4)
m <- mirai_map(
  1:4,
  function(x) {res <- rnorm(1e8, mean = x, sd = x + 3); res <- res + z; max(res) - min(res)},
  z = rnorm(1e7)
)
m[]

Just so there is no confusion, we are still talking about message-passing here and not shared memory.

[SebKrantz]

Ok, got it. Feel free to send a PR with a bullet point for your package under the section "Parallelization, High-Performance Computing and Out-Of-Memory Data". Please ensure that someone who knows very little about parallel computing and is just looking for performance can understand it, i.e., if you use terms like "asynchronously", "resolving automatically upon completion" either explain them or put them in simple terms, e.g. "the mirai package allows executing arbitrary R expressions in parallel within different R processes (asynchonously) on either local or network computing resources and automatically terminates these processes and returns the result upon completion". Also, please ensure your docs / README etc. includes examples like the above. Most people want to just parallelize simple data/processing or model fitting code, and often there are some inputs that change and some others that are immutable / ideally shared between the parallel tasks. Somebody looking at your repo/docs should need less than 1 minute to get the idead how this can be done using your package.

[shikokuchuo]

Excellent, thanks for the suggestions! I'll have a re-draft of the package readme before I open up a PR with you.

[SebKrantz]

Thanks!

[SebKrantz]

Hi @shikokuchuo, I'm just trying to build something rather complex using your package and encountered some issues, so I wanted to get your opinion. Basically, my code looks like this:

  # Code above... defines graph and results lists
  if(!is.finite(nthreads) || nthreads <= 1L) {
    res$final_flows <- run_assignment_core(seq_len(N), verbose)
  } else {
    # Split OD matrix in equal parts
    ind_list <- gsplit(g = sample.int(as.integer(nthreads), N, replace = TRUE))
    daemons(n = nthreads - 1L)
    # Pass current environment dynamically
    everywhere({}, environment())
    # Now run the map in the background
    final_flows_other <- mirai_map(ind_list[-1L], function(x) run_assignment_core(x, FALSE))
    # Runs the first instance in the current session
    final_flows <- run_assignment_core(ind_list[[1L]], verbose)
    # Collect other mirai's results
    final_flows_other <- collect_mirai(final_flows_other) # [.stop, .progress]
    # Deactivate Daemons
    daemons(0)
    # Combine Results
    for (fflow in final_flows_other) final_flows %+=% fflow
    res$final_flows <- final_flows
    rm(final_flows_other, fflow, final_flows)
  }
# Code below, precesses and returns result object, 

Where run_assignment_core() assigns traffic flows to the edges of a network, and takes as input indices of rows in the Origin-Destination Matrix to interate over. The matrix has N rows and is melted to a long format with columns from, to, and flow. If verbose = TRUE, it prints a fancy progress bar, which was the intention here of using mirai to have the first set of indices run in the active session and the rest asynchonously in the background. The function relies on rather large global objects, most notable a graph of the network (created using igraph), and a full distance matrix between all nodes. I would like those objects to be shared by all daemons (not deep-copied), and thought everywhere({}, environment()) would accomplish this (it seems to be working).

The other two particularities are that I have defined global results lists above the parallel section, and let run_assignment_core() internally assign results to the right indices using result1[[i]] <<- object_to_save. There is also some C-code called inside run_assignment_core(). Now it seems I have some issues with this setup. First, it seems global assignment from inside the deamons is not working (expectably so). But also Other parts of the code, notably the final flows vectors in final_flows_other, which are not globally assigned but returned by the function, are not computed properly (I have many NA's).

So I just wanted to hear your input on which parts of this pipeline are feasible? I guess having large objects shared between multiple daemons is feasible? If so, is my setup ok? And is there a scope to do global assignment (potentially also using C/by reference)?

[SebKrantz]

Anyway thanks if you have time to respond.