Resolve the job's cgroup subpath for memory/thread detection; make detectors testable

aboyounp-gne · aboyounp-gne · commit cd601e712b7f · 2026-07-23T22:02:06.000-07:00
The cgroup memory_limit / threads detectors read fixed root paths
(/sys/fs/cgroup/memory.max, .../cpu.max, ...), which only match when the
process runs in the ROOT cgroup. A cgroup-constrained non-SLURM job was
therefore undetected and fell back to physical RAM / hardware concurrency
-- the over-commit the defaults exist to prevent (F14). The helpers also
read /proc and /sys directly, so they could not be unit-tested (F15).

Add .cgroupDirs(), which resolves the job's own cgroup subpath from
/proc/self/cgroup (cgroup v2 unified "0::&lt;path&gt;" and v1 per-controller
"&lt;n&gt;:&lt;ctrls&gt;:&lt;path&gt;", including combined mounts like cpu,cpuacct) and
searches it before the mount root. The mount root stays in the candidate
list, so detection never regresses where root paths already worked. All
detectors (.cgroupDirs, .cgroupMemoryBytes, .cgroupCpus,
.physicalMemoryBytes) take injectable path parameters and are covered by
fixture-based tests.

- R/DuckDBConnection.R: .cgroupDirs, .readCgroupBytes, .cgroupQuotaCpus;
  subpath-aware .cgroupMemoryBytes/.cgroupCpus; path-param .physicalMemoryBytes; docs
- tests/testthat/test-DuckDBConnection.R: fixture tests (v2/v1 subpath, root fallback, meminfo)
- DESCRIPTION, NEWS.md: 0.99.13
diff --git a/DESCRIPTION b/DESCRIPTION
@@ -1,5 +1,5 @@
 Package: DuckDBDataFrame
-Version: 0.99.12
+Version: 0.99.13
 Date: 2026-07-23
 Title: DuckDB-Backed DataFrame and Table Structures
 Description:
diff --git a/NEWS.md b/NEWS.md
@@ -1,3 +1,20 @@
+# DuckDBDataFrame 0.99.13
+
+## Bug fixes
+
+- The cgroup `memory_limit` / `threads` detectors used by `configureOutOfCore()`
+  now resolve the job's **own** cgroup subpath from `/proc/self/cgroup` (both
+  cgroup v2 unified and v1 per-controller, including combined mounts like
+  `cpu,cpuacct`) before falling back to the mount root. A cgroup-constrained
+  **non-SLURM** job was previously undetected — the fixed root paths
+  (`/sys/fs/cgroup/memory.max`, …) only match when the process runs in the root
+  cgroup — so it silently fell back to physical RAM / hardware concurrency, the
+  over-commit the defaults are meant to prevent. The mount root stays in the
+  search list, so detection never regresses where it already worked. The
+  detectors (`.cgroupDirs()`, `.cgroupMemoryBytes()`, `.cgroupCpus()`,
+  `.physicalMemoryBytes()`) take injectable path parameters and are now covered
+  by fixture-based unit tests.
+
 # DuckDBDataFrame 0.99.12
 
 ## Bug fixes
diff --git a/R/DuckDBConnection.R b/R/DuckDBConnection.R
@@ -72,7 +72,8 @@ reg.finalizer(.duckdb, function(env) {
 #' the most-restrictive detected ceiling -- an explicit SLURM allocation
 #' (\code{SLURM_MEM_PER_NODE}, or \code{SLURM_MEM_PER_CPU} times
 #' \code{SLURM_CPUS_ON_NODE}), the cgroup limit (v2 \code{memory.max} then v1
-#' \code{memory.limit_in_bytes}), then physical RAM. DuckDB's own default is 80\%
+#' \code{memory.limit_in_bytes}, read from the job's own cgroup resolved via
+#' \code{/proc/self/cgroup} before the mount root), then physical RAM. DuckDB's own default is 80\%
 #' of *physical* RAM, which ignores a SLURM / cgroup cap and can over-commit
 #' (near-OOM on the first large scan); the detected default keeps a big
 #' aggregation spilling within the job's allocation. When nothing can be detected
@@ -81,7 +82,8 @@ reg.finalizer(.duckdb, function(env) {
 #' \code{threads} is likewise defaulted when unset: the most-restrictive detected
 #' CPU allocation -- a SLURM allocation (\code{SLURM_CPUS_PER_TASK}, or
 #' \code{SLURM_CPUS_ON_NODE}) then the cgroup CFS quota (v2 \code{cpu.max}, v1
-#' \code{cpu.cfs_quota_us}/\code{cpu.cfs_period_us}). DuckDB otherwise defaults to
+#' \code{cpu.cfs_quota_us}/\code{cpu.cfs_period_us}, read from the job's own
+#' cgroup subpath before the mount root). DuckDB otherwise defaults to
 #' hardware concurrency, which ignores a SLURM/cgroup cpuset and over-subscribes
 #' on a shared node (each extra thread also carries working memory against the
 #' memory cap). When no allocation is detected, DuckDB's default is left in place.
@@ -228,37 +230,73 @@ setExtensionDirectory <- function(conn) {
     NULL
 }
 
-# Memory ceiling (bytes) from the cgroup limit (v2 then v1). NULL when
-# unconstrained ("max", the v1 huge sentinel, or the files are absent).
-.cgroupMemoryBytes <- function() {
-    read_limit <- function(path) {
-        if (!file.exists(path)) {
-            return(NULL)
+# Ordered candidate cgroup directories for a controller, most specific first
+.cgroupDirs <- function(controller,
+                        proc_cgroup = "/proc/self/cgroup",
+                        root = "/sys/fs/cgroup") {
+    dirs <- character(0)
+    lines <- if (file.exists(proc_cgroup)) {
+        tryCatch(readLines(proc_cgroup, warn = FALSE), error = function(e) character())
+    } else {
+        character(0)
+    }
+    for (ln in lines) {
+        parts <- strsplit(ln, ":", fixed = TRUE)[[1L]]
+        if (length(parts) < 3L) {
+            next
         }
-        val <- tryCatch(readLines(path, n = 1L, warn = FALSE),
-                        error = function(e) character())
-        if (!length(val) || !nzchar(val[1L]) || val[1L] == "max") {
-            return(NULL)
+        ctrls <- parts[2L]
+        sub <- sub("^/", "", paste(parts[-(1:2)], collapse = ":"))  # path may hold ':'
+        if (!nzchar(ctrls)) {
+            # v2 unified: files live directly under root/<path>
+            dirs <- c(dirs, if (nzchar(sub)) file.path(root, sub) else root)
+        } else if (controller %in% strsplit(ctrls, ",", fixed = TRUE)[[1L]]) {
+            # v1: files under root/<mount>/<path>; the mount may be the combined
+            # controller string ("cpu,cpuacct") or a per-controller symlink.
+            for (mnt in unique(c(ctrls, controller))) {
+                dirs <- c(dirs,
+                          if (nzchar(sub)) file.path(root, mnt, sub) else file.path(root, mnt))
+            }
         }
-        b <- suppressWarnings(as.numeric(val[1L]))
-        # cgroup v1 "unlimited" is a huge sentinel (~PAGE_COUNTER_MAX); treat an
-        # implausibly large value (>= 1 EiB) as unset.
-        if (is.na(b) || b <= 0 || b >= 2^60) {
-            return(NULL)
+    }
+    dirs <- c(dirs, root, file.path(root, controller))  # root fallbacks
+    unique(dirs[dir.exists(dirs)])
+}
+
+# Parse a cgroup byte-limit file (v2 memory.max / v1 memory.limit_in_bytes)
+.readCgroupBytes <- function(path) {
+    if (!file.exists(path)) {
+        return(NULL)
+    }
+    val <- tryCatch(readLines(path, n = 1L, warn = FALSE), error = function(e) character())
+    if (!length(val) || !nzchar(val[1L]) || val[1L] == "max") {
+        return(NULL)
+    }
+    b <- suppressWarnings(as.numeric(val[1L]))
+    if (is.na(b) || b <= 0 || b >= 2^60) {
+        return(NULL)
+    }
+    b
+}
+
+# Memory ceiling (bytes) from the job's cgroup (v2 then v1)
+.cgroupMemoryBytes <- function(dirs = .cgroupDirs("memory")) {
+    for (d in dirs) {
+        b <- .readCgroupBytes(file.path(d, "memory.max")) %||%          # v2
+            .readCgroupBytes(file.path(d, "memory.limit_in_bytes"))     # v1
+        if (!is.null(b)) {
+            return(b)
         }
-        b
     }
-    read_limit("/sys/fs/cgroup/memory.max") %||%                    # cgroup v2
-        read_limit("/sys/fs/cgroup/memory/memory.limit_in_bytes")  # cgroup v1
+    NULL
 }
 
 # Physical RAM (bytes) from /proc/meminfo (Linux only). NULL elsewhere.
-.physicalMemoryBytes <- function() {
-    if (!file.exists("/proc/meminfo")) {
+.physicalMemoryBytes <- function(path = "/proc/meminfo") {
+    if (!file.exists(path)) {
         return(NULL)
     }
-    ln <- tryCatch(readLines("/proc/meminfo", warn = FALSE),
-                   error = function(e) character())
+    ln <- tryCatch(readLines(path, warn = FALSE), error = function(e) character())
     kb <- grep("^MemTotal:", ln, value = TRUE)
     if (!length(kb)) {
         return(NULL)
@@ -301,39 +339,41 @@ setExtensionDirectory <- function(conn) {
     NULL
 }
 
-# CPU count from the cgroup CFS quota (v2 `cpu.max`, then v1
-# `cpu.cfs_quota_us`/`cpu.cfs_period_us`) = floor(quota / period). NULL when
-# unconstrained ("max", quota <= 0, or the files are absent).
-.cgroupCpus <- function() {
-    quota_cpus <- function(q, p) {
-        if (is.na(q) || is.na(p) || q <= 0 || p <= 0) {
-            return(NULL)
+# floor(quota / period) as a CPU count, or NULL when either is non-positive/NA.
+.cgroupQuotaCpus <- function(quota, period) {
+    if (is.na(quota) || is.na(period) || quota <= 0 || period <= 0) {
+        return(NULL)
+    }
+    max(1L, as.integer(floor(quota / period)))
+}
+
+# CPU count from the job's cgroup CFS quota
+.cgroupCpus <- function(dirs = .cgroupDirs("cpu")) {
+    for (d in dirs) {
+        v2 <- file.path(d, "cpu.max")
+        if (file.exists(v2)) {
+            val <- tryCatch(strsplit(trimws(readLines(v2, n = 1L, warn = FALSE)),
+                                     "[[:space:]]+")[[1L]],
+                            error = function(e) character())
+            if (length(val) >= 2L && val[1L] != "max") {
+                n <- .cgroupQuotaCpus(suppressWarnings(as.numeric(val[1L])),
+                                      suppressWarnings(as.numeric(val[2L])))
+                if (!is.null(n)) {
+                    return(n)
+                }
+            }
         }
-        max(1L, as.integer(floor(q / p)))
-    }
-    v2 <- "/sys/fs/cgroup/cpu.max"
-    if (file.exists(v2)) {
-        val <- tryCatch(strsplit(trimws(readLines(v2, n = 1L, warn = FALSE)),
-                                 "[[:space:]]+")[[1L]],
-                        error = function(e) character())
-        if (length(val) >= 2L && val[1L] != "max") {
-            n <- quota_cpus(suppressWarnings(as.numeric(val[1L])),
-                            suppressWarnings(as.numeric(val[2L])))
+        qf <- file.path(d, "cpu.cfs_quota_us")
+        pf <- file.path(d, "cpu.cfs_period_us")
+        if (file.exists(qf) && file.exists(pf)) {
+            n <- .cgroupQuotaCpus(
+                suppressWarnings(as.numeric(readLines(qf, n = 1L, warn = FALSE))),
+                suppressWarnings(as.numeric(readLines(pf, n = 1L, warn = FALSE))))
             if (!is.null(n)) {
                 return(n)
             }
         }
     }
-    qf <- "/sys/fs/cgroup/cpu/cpu.cfs_quota_us"
-    pf <- "/sys/fs/cgroup/cpu/cpu.cfs_period_us"
-    if (file.exists(qf) && file.exists(pf)) {
-        n <- quota_cpus(
-            suppressWarnings(as.numeric(readLines(qf, n = 1L, warn = FALSE))),
-            suppressWarnings(as.numeric(readLines(pf, n = 1L, warn = FALSE))))
-        if (!is.null(n)) {
-            return(n)
-        }
-    }
     NULL
 }
 
diff --git a/man/DuckDBConnection.Rd b/man/DuckDBConnection.Rd
diff --git a/tests/testthat/test-DuckDBConnection.R b/tests/testthat/test-DuckDBConnection.R
@@ -109,3 +109,63 @@ test_that("configureOutOfCore defaults threads to the SLURM allocation when unse
         DBI::dbGetQuery(con, "SELECT current_setting('threads') AS v")$v)
     expect_identical(thr, 1L)
 })
+
+test_that(".cgroupDirs / cgroup detectors resolve the job subpath then fall back", {
+    mk <- function(p, txt) {
+        dir.create(dirname(p), recursive = TRUE, showWarnings = FALSE)
+        writeLines(txt, p)
+    }
+
+    # cgroup v2: the job's own subpath must win over the mount root
+    b <- tempfile("v2_"); root <- file.path(b, "cg"); proc <- file.path(b, "proc")
+    on.exit(unlink(b, recursive = TRUE), add = TRUE)
+    mk(file.path(root, "myjob", "memory.max"), "8589934592")   # 8 GiB (job)
+    mk(file.path(root, "memory.max"), "68719476736")           # 64 GiB (root)
+    mk(file.path(root, "myjob", "cpu.max"), "200000 100000")   # 2 cpus (job)
+    mk(file.path(root, "cpu.max"), "800000 100000")            # 8 cpus (root)
+    writeLines("0::/myjob", proc)
+    expect_identical(
+        DuckDBDataFrame:::.cgroupMemoryBytes(DuckDBDataFrame:::.cgroupDirs("memory", proc, root)),
+        8 * 2^30)
+    expect_identical(
+        DuckDBDataFrame:::.cgroupCpus(DuckDBDataFrame:::.cgroupDirs("cpu", proc, root)),
+        2L)
+
+    # cgroup v1 with a combined controller mount (cpu,cpuacct) + subpath
+    b <- tempfile("v1_"); root <- file.path(b, "cg"); proc <- file.path(b, "proc")
+    on.exit(unlink(b, recursive = TRUE), add = TRUE)
+    mk(file.path(root, "memory", "mygrp", "memory.limit_in_bytes"), "4294967296") # 4 GiB
+    mk(file.path(root, "cpu,cpuacct", "mygrp", "cpu.cfs_quota_us"), "300000")
+    mk(file.path(root, "cpu,cpuacct", "mygrp", "cpu.cfs_period_us"), "100000")    # 3 cpus
+    writeLines(c("7:memory:/mygrp", "5:cpu,cpuacct:/mygrp"), proc)
+    expect_identical(
+        DuckDBDataFrame:::.cgroupMemoryBytes(DuckDBDataFrame:::.cgroupDirs("memory", proc, root)),
+        4 * 2^30)
+    expect_identical(
+        DuckDBDataFrame:::.cgroupCpus(DuckDBDataFrame:::.cgroupDirs("cpu", proc, root)),
+        3L)
+
+    # root fallback: no /proc/self/cgroup -> the mount root is still searched
+    b <- tempfile("rt_"); root <- file.path(b, "cg"); proc <- file.path(b, "absent")
+    on.exit(unlink(b, recursive = TRUE), add = TRUE)
+    mk(file.path(root, "memory.max"), "1073741824")            # 1 GiB at root
+    expect_identical(
+        DuckDBDataFrame:::.cgroupMemoryBytes(DuckDBDataFrame:::.cgroupDirs("memory", proc, root)),
+        2^30)
+
+    # unconstrained ("max") -> NULL
+    b <- tempfile("mx_"); root <- file.path(b, "cg"); proc <- file.path(b, "proc")
+    on.exit(unlink(b, recursive = TRUE), add = TRUE)
+    mk(file.path(root, "memory.max"), "max")
+    writeLines("0::/", proc)
+    expect_null(
+        DuckDBDataFrame:::.cgroupMemoryBytes(DuckDBDataFrame:::.cgroupDirs("memory", proc, root)))
+})
+
+test_that(".physicalMemoryBytes parses an injected /proc/meminfo", {
+    mi <- tempfile()
+    on.exit(unlink(mi), add = TRUE)
+    writeLines(c("MemTotal:       16307188 kB", "SwapTotal:  0 kB"), mi)
+    expect_identical(DuckDBDataFrame:::.physicalMemoryBytes(mi), 16307188 * 1024)
+    expect_null(DuckDBDataFrame:::.physicalMemoryBytes(tempfile()))
+})
