Background execution and R process pool

Makefile

@@ -47,15 +47,11 @@ requirements-rs:
 .PHONY: build
 build: ## Compile polars R package with all features and generate Rd files
 	export RPOLARS_ALL_FEATURES=true \
-	&& export NOT_CRAN=true \
-	&& export RPOLARS_PROFILE=release-optimized \
 	&& Rscript -e 'if (!(require(arrow)&&require(nanoarrow))) warning("could not load arrow/nanoarrow, igonore changes to nanoarrow.Rd"); rextendr::document()'
 
 .PHONY: install
 install:
 	export RPOLARS_ALL_FEATURES=true \
-	&& export NOT_CRAN=true \
-	&& export RPOLARS_PROFILE=release-optimized \
 	&& R CMD INSTALL --no-multiarch --with-keep.source .
 
 .PHONY: all

R/rbackground.R

@@ -9,35 +9,110 @@
 }
 
 
+#' @title RThreadHandle to string
+#' @description get description of RThreadHandle as string
+#' @param x RThreadHandle
+#' @param ... not used
 #' @export
-#' @noRd
+#' @keywords internal
 as.character.RThreadHandle = function(x, ...) {
   .pr$RThreadHandle$thread_description(x) |>
-  unwrap_or("An exhausted thread handle")
+  unwrap_or("An exhausted RThreadHandle")
 }
 
-
+#' s3 method print RThreadHandle
+#'
+#' @keywords internal
+#' @param x RThreadHandle
+#' @param ... not used
+#'
+#' @return self
 #' @export
-#' @noRd
+#'
+#' @examples
+#' handle = pl$LazyFrame()$select(pl$lit(2) + 2)$collect_in_background()
+#' print(handle)
+#' handle$join()
+#' print(handle)
 print.RThreadHandle = function(x, ...) as.character(x) |> cat("\n")
 
 
-#' @title Wait for the thread to complete its job
+#' @title The RThreadHandle class
+#' @name RThreadHandle_RThreadHandle_class
+#' @description A handle to some polars query running in a background thread.
+#' @details
+#' `<LazyFrame>$collect_in_background()` will execute a polars query detached from the R session
+#' and return a `RThreadHandle` immediately. The RThreadHandle has the methods `is_finished()` and
+#' `$join()`.
+#'
+#' The background thread may access the pool of extra R sessions to process R code
+#' embedded in polars query via `$map(...,background = TRUE)` or `$apply(background=TRUE)`. Use
+#' `pl$set_global_rpool_cap()` to limit number of parallel R sessions. Extra R sessions are spawned
+#' and used if `background` arg is set to TRUE.
+#'
+#' Starting polars `<LazyFrame>$collect_in_background()` with e.g. some
+#' `$map(...,background = FALSE)` will raise an Error as the main R session is not available to
+#' process the query.
+#' @return see methods
+#' @keywords RThreadHandle
+RThreadHandle
+
+#' Join a RThreadHandle
 #' @keywords RThreadHandle
-#' @param ... a RThreadHandle
-#' @return a DataFrame, which is the result of the job
+#' @details method `<RThreadHandle>$join()`: will block until job is done and then return some value
+#'  or raise an error from the thread.
+#' Calling `<RThreadHandle>$join()` a second time will raise an error because handle is already
+#' exhausted.
 #' @export
-RThreadHandle_join = function(...) {
+RThreadHandle_join = function() {
   .pr$RThreadHandle$join(self) |> unwrap()
 }
 
 
-#' @title Check if the thread completes its job
+#' Ask if RThreadHandle is finished?
 #' @keywords RThreadHandle
-#' @param ... a RThreadHandle
-#' @return a boolean indicating the whether the job has finished
-#'  or NULL if the handle has been joined
+#' @details method `<RThreadHandle>$is_finished()`: Calling `<RThreadHandle>$is_finished()` returns
+#' trinary value: `TRUE` if finished, `FALSE` if not, and `NULL` if the handle was exhausted
+#' (already joined).
 #' @export
-RThreadHandle_is_finished = function(...) {
+RThreadHandle_is_finished = function() {
   .pr$RThreadHandle$is_finished(self) |> unwrap_or(NULL)
 }
+
+
+
+
+#' get/set global R session pool cap
+#' @name global_rpool_cap
+#' @param n integer, the capacity limit R sessions to process R code.
+#' @return for `pl$get_global_rpool_cap()` a list(available = ? , capacity = ?)
+#' where available is how many R session already spawned in pool. Capacity is the limit of
+#' how many new R sessions to spawn. Anytime a polars thread worker needs a background R session
+#' specifically to run R code embedded in a query via `$map(..., in_background = TRUE)` or
+#' `$apply(..., in_background = TRUE)` it will obtain any R session idling in rpool, otherwise spawn
+#' a new R session (process) and add it to pool if not `capacity` has been reached. If capacity has
+#' been reached already the thread worker will sleep until an R session is idling.
+#'
+#' Background R sessions communicate via polars arrow IPC (series/vectors) or R serialize +
+#' shared memory buffers via the rust crate `ipc-channel`. Multi-process communication has overhead
+#' because all data must be serialized/de-serialized and sent via buffers. Using multiple R sessions
+#' will likely only give a speed-up in a `low io - high cpu` scenario. Native polars query syntax
+#' runs in threads and have no overhead.
+#'
+#' @examples
+#' default = pl$get_global_rpool_cap()
+#' print(default)
+#' pl$set_global_rpool_cap(8)
+#' pl$get_global_rpool_cap()
+#' pl$set_global_rpool_cap(default$capacity)
+pl$get_global_rpool_cap = function() {
+  get_global_rpool_cap() |> unwrap()
+}
+
+#' @rdname global_rpool_cap
+#' @name set_global_rpool_cap
+pl$set_global_rpool_cap = function(n) {
+  set_global_rpool_cap(n) |> unwrap() |> invisible()
+}
+
+

R/zzz.R

@@ -96,9 +96,7 @@ replace_private_with_pub_methods(Series, "^Series_")
 # RThreadHandle
 replace_private_with_pub_methods(RThreadHandle, "^RThreadHandle_")
 
-# Global R process pool configuration
-pl$get_global_rpool_cap = function() get_global_rpool_cap() |> unwrap()
-pl$set_global_rpool_cap = function(c) set_global_rpool_cap(c) |> unwrap() |> invisible()
+
 
 
 # expression constructors

docs/make-docs.R

@@ -125,7 +125,8 @@ make_doc_hierarchy = function() {
   out = list()
   # order determines order in sidebar
   classes = c("pl", "Series", "DataFrame", "LazyFrame", "GroupBy",
-              "LazyGroupBy", "arr", "ExprBin", "ExprDT", "ExprMeta", "ExprStr", "ExprStruct", "Expr")
+              "LazyGroupBy", "arr", "ExprBin", "ExprDT", "ExprMeta", "ExprStr", "ExprStruct",
+              "Expr", "RThreadHandle")
   for (cl in classes) {
     files = grep(paste0("^", cl, "_"), other, value = TRUE)
     tmp = sprintf("%s: reference/%s", sub("\\.md", "", sub("[^_]*_", "", files)), files)

inst/misc/benchmark_rbackground.R

@@ -52,15 +52,16 @@ large_compute_bg$collect_in_background()$join() |> system.time()
 
 ### 3a -----------  Use R processes in parallel, low io, low cpu
 lf <- pl$LazyFrame(lapply(1:100,\(i) rep(i,5)))
-f_sum_all_cols <-  \(lf,...) lf$select(pl$all()$map(\(x) {x$to_r() |> sum()},...))
+f_sum_all_cols <-  \(lf,...) lf$select(pl$all()$map(\(x) {print("hey");x$to_r() |> sum()},...))
 
 f_sum_all_cols(lf)$collect() |> system.time()
 
 pl$set_global_rpool_cap(1)
 f_sum_all_cols(lf, in_background = TRUE)$collect() |> system.time() #burn-in start processes
 f_sum_all_cols(lf, in_background = TRUE)$collect() |> system.time()
 
-pl$set_global_rpool_cap(4)
+
+pl$set_global_rpool_cap(8)
 f_sum_all_cols(lf, in_background = TRUE)$collect() |> system.time() #burn-in start processes
 f_sum_all_cols(lf, in_background = TRUE)$collect() |> system.time()