From 3f951c5934dc758519cadde1cba6ee8db4ce83b0 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Fri, 2 Oct 2026 13:46:37 +0300 Subject: [PATCH 1/5] docs: running several instances on one machine A guide to CPU sets, memory limits, ports, journals and spill files per instance, and to what is safe when instances share a database: any number of readers, one writing process per database root, no in-place edits of table files, one rayforce version. The storage pages said symbol-file writes were safe from several processes and that symbol files carry checksums; neither holds. The spill-file naming now lists direct blocks too. The REPL and pipeline pages describe the pool size and core pinning. --- docs/docs/architecture/memory.md | 2 +- docs/docs/architecture/pipeline.md | 2 + docs/docs/guides/instances.md | 74 ++++++++++++++++++++++++++++++ docs/docs/guides/storage.md | 2 +- docs/docs/language/repl.md | 4 +- docs/docs/storage/index.md | 6 +-- mkdocs.yml | 1 + 7 files changed, 85 insertions(+), 6 deletions(-) create mode 100644 docs/docs/guides/instances.md diff --git a/docs/docs/architecture/memory.md b/docs/docs/architecture/memory.md index 57d72c3ae..a7a3bdf89 100644 --- a/docs/docs/architecture/memory.md +++ b/docs/docs/architecture/memory.md @@ -89,7 +89,7 @@ Pools are normally backed by anonymous `mmap` — fast, lazy-committed by the ke The fallback is transparent to every `ray_alloc` caller. Pages never written stay as holes in the sparse backing file, so the over-allocate-and-trim alignment trick costs zero real disk space; only the pages you actually touch consume disk pages. On heap teardown the fd is closed and the tempfile is unlinked, so the swap directory doesn't accumulate orphans. - **Configurable swap directory.** The directory is resolved as `RAY_HEAP_SWAP` → `TMPDIR` → `/tmp`. `/tmp` is the default rather than the working directory, which may be read-only or shared. Trailing slash is added automatically. The chosen directory must exist and be writable by the running process. -- **Tempfile naming.** Format is `rayheap___.dat`. Files are opened `O_EXCL` so no clashes between concurrent processes; the counter is per-process atomic so no clashes within a process. +- **Tempfile naming.** Pools use `rayheap___.dat`, direct blocks `raydirect___.dat`. The pid keeps processes of one PID namespace apart, and the counter is a per-process atomic. Files are opened `O_EXCL`, so a name that already exists fails that allocation rather than sharing the file. - **POSIX only today.** The fallback path uses `open` / `ftruncate` / `mmap MAP_SHARED` / `munmap` / `unlink`. Windows pools currently take only the anonymous `VirtualAlloc` path. - **Distinct from block offloading.** Block offloading (see [Block Offloading](offloading.md)) streams pre-existing parted-table data through queries one segment at a time. The file-backed pool fallback handles fresh anonymous allocations that exceed RAM. Both let workloads exceed RAM, but at different layers and for different shapes of work. diff --git a/docs/docs/architecture/pipeline.md b/docs/docs/architecture/pipeline.md index 1d3d5f927..14d64d9bf 100644 --- a/docs/docs/architecture/pipeline.md +++ b/docs/docs/architecture/pipeline.md @@ -207,6 +207,8 @@ Rayforce uses a global thread pool (`ray_pool_t`) for parallel execution. Querie Each dispatch splits its tasks into one contiguous range per worker (the main thread included); a worker runs its range in order and, once it is empty, steals half of another worker's remaining range from the tail, so an idle thread never waits on a busy one and the rows a worker touches are sequential. Tasks of one dispatch may therefore run in any order and on any worker: operators place results by task index, and any early stop (a `take:` cutoff, a top-N bound) is derived from row positions or values, never from "the earlier tasks have run". The scheduler is selected once at pool creation (`RAY_POOL_STEAL`, default `1`; `0` restores the shared claim cursor for A/B comparison) and `RAY_POOL_TRACE=1` prints the own-versus-stolen task histogram per worker to stderr at pool destroy; see [REPL Reference](../language/repl.md#environment-variables). +The default pool has one thread per CPU of the process affinity mask. On Linux such a pool is pinned: each spawned worker pins itself to one CPU before it creates its heap, so the heap's pages are first touched on the worker's own NUMA node and stay local to it. Workers take one hardware thread of every physical core first, round-robin across nodes starting from the node with most of the set's cores, then the hyperthread siblings. The `rayforce` binary also pins the main thread (worker 0) to the first CPU of that order; a pool created inside an embedding application leaves the caller's thread alone. Children started with `.sys.exec` get the whole affinity mask back. `RAY_POOL_PIN=0` disables pinning and `RAY_POOL_PIN=1` also pins a pool smaller than the mask. + ### Radix-Partitioned Hash Join Hash joins use adaptive radix partitioning to ensure each partition's hash table fits in L2 cache. The number of radix bits adapts between `RAY_JOIN_MIN_RADIX` (2, producing 4 partitions) and `RAY_JOIN_MAX_RADIX` (14, producing 16K partitions) based on input size, targeting a per-partition working set of `RAY_JOIN_L2_TARGET` (256 KB). diff --git a/docs/docs/guides/instances.md b/docs/docs/guides/instances.md new file mode 100644 index 000000000..46f810db9 --- /dev/null +++ b/docs/docs/guides/instances.md @@ -0,0 +1,74 @@ +# Running Several Instances + +Several rayforce processes can run on one machine. Each process sizes its thread pool and its memory for what it can see, and it does not know about the others. Give every instance its own share of CPUs and memory, its own port and journal, and keep each database written by one process at a time. + +## CPUs + +Choose the CPUs of each instance when you start it: `taskset -c LIST`, `docker run --cpuset-cpus=LIST`, or `CPUAffinity=` in a systemd unit. Every thread of the instance, and every program it starts with `.sys.exec`, stays within that list. + +On Linux an instance uses one thread per CPU of its list. Such a pool is pinned: the main thread runs on the first CPU and each worker on one of the others, taking one hardware thread of every physical core first, alternating NUMA nodes, and the hyperthread siblings after. Pinning keeps each worker's memory on its own node. + +```bash +taskset -c 0,2,4,6,24,26,28,30 rayforce -p 5001 & +taskset -c 8,10,12,14,32,34,36,38 rayforce -p 5002 & +``` + +**Choosing the lists.** Read the machine's layout from `lscpu -e` (columns `NODE`, `SOCKET`, `CORE`), or from `/sys/devices/system/cpu/cpuN/topology/thread_siblings_list` and `/sys/devices/system/node/nodeN/cpulist`. + +- Give each instance whole cores: both hardware threads of a core go to the same instance. Two instances on the two threads of one core slow each other down. +- Keep an instance on one NUMA node when its CPU count fits in one. +- CPU numbering does not follow the layout on every machine. On a two-socket machine that numbers sockets alternately (even CPUs on socket 0, odd on socket 1) with CPU `k` and `k+24` on one core, the ranges `0-7`, `8-15`, … `40-47` put every instance on both sockets, and the last three ranges on the hyperthreads of the first three. For six instances of 8 threads on that machine use `0,2,4,6,24,26,28,30`, `8,10,12,14,32,34,36,38`, `16,18,20,22,40,42,44,46`, `1,3,5,7,25,27,29,31`, `9,11,13,15,33,35,37,39` and `17,19,21,23,41,43,45,47`. + +**Do not:** + +- start two instances on overlapping lists: their pinned threads share CPUs and neither can move away; +- start instances without a list and leave each at its default size: every instance takes every CPU; +- pass `-c` larger than the list: the threads share CPUs and are not pinned (rayforce prints a warning). + +`RAY_POOL_PIN=0` turns pinning off; then size the instances with `-c` so that their pools add up to the CPU count. `RAY_POOL_PIN=1` also pins a pool smaller than its list, always to the same first CPUs of it, so use it only with lists that do not overlap. + +To see where an instance's threads run, read every thread: `grep Cpus_allowed_list /proc/PID/task/*/status`. `taskset -p PID` shows only the main thread. + +## Memory + +Each instance treats the machine's whole physical RAM as its own and does not read container memory limits. It moves its working set to disk only when its own memory would pass that amount, so several busy instances can together exhaust RAM before any of them spills. Start every instance with `-m SIZE` and keep the sizes, plus room for the page cache, within RAM: + +```bash +taskset -c 0-11 rayforce -m 40G -p 5001 & +taskset -c 12-23 rayforce -m 40G -p 5002 & +``` + +`-m` also bounds the freed memory an idle instance keeps for reuse (a sixteenth of SIZE). + +Instances that read the same tables share the operating system's page cache, so the sum of their resident sizes overstates the memory in use. + +## Ports, journals and spill files + +- **Port:** each instance listens on its own `-p` port. A second instance on a busy port exits with an error. +- **Journal:** each instance needs its own journal base (`-l /path/instance1`). Two processes appending to one journal break its framing, and recovery stops at the first broken entry. Never run `.log.roll`, `.log.snapshot` or `.log.purge` on a journal another process writes. +- **Spill files:** names carry the process id and are removed as soon as they are mapped, so instances never share one. Point `RAY_HEAP_SWAP` at a disk with room for every instance that may spill at once. + +## Sharing a database + +**Readers.** Any number of instances can read the same splayed or partitioned tables. A reader sees either the whole previous version of a table or the whole new one, and a table it has already loaded stays valid when another process rewrites it. + +**One writer per database.** Only one process at a time may write a database: `.db.splayed.set`, a `.csv` or `.parquet` import into an existing root, or a live `insert` into a partitioned table followed by its rollover. All tables and partitions of a database share the root's symbol file. + +- When two processes write the same database, a write that races another one fails with a `corrupt` error. Nothing on disk is damaged. If the failing process holds tables it loaded from that database, its later reads and writes of the database fail until it drops them. +- Two processes that update one table by reading it, changing it and writing it back lose each other's updates; the last write wins. +- A partitioned reader fails while a new partition is being written, because the partition directory exists before its schema does. Write a new partition under a dot directory of the root, passing the root's symbol file, and rename it into place; dot directories are skipped when the partitions are listed: + +```lisp +(.db.splayed.set "/db/.stage/2026.10.02/t/" t "/db/.sym") +(.sys.exec "mv /db/.stage/2026.10.02 /db/2026.10.02") +``` + +- If a first write of a table fails, its directory is left without a schema and every `.db.parted.get` of that database fails with `cannot read .d schema`. Remove that directory. + +**Files.** Never change table files in place: no `cp` over them, no `rsync --inplace`, no `truncate`, no editors. Instances that have the table open see the change at once, its stored statistics no longer match the data, and a shortened file crashes them with `SIGBUS`. Replace tables with `.db.splayed.set`, or write a new file and rename it over the old one. The same applies to files written with `write`, `write-bytes` and `hnsw-save`, which rewrite their target in place. + +**Versions.** All instances that share a database must run the same rayforce version; upgrade them together. + +## Interactive sessions + +The REPL rewrites `~/.rayforce_history` when it exits, so with several interactive sessions only the history of the last one to exit is kept. diff --git a/docs/docs/guides/storage.md b/docs/docs/guides/storage.md index 3f185eeb5..5eb259eef 100644 --- a/docs/docs/guides/storage.md +++ b/docs/docs/guides/storage.md @@ -69,7 +69,7 @@ ray_err_t ray_sym_save(const char* path); ray_err_t ray_sym_load(const char* path); ``` -The symbol file uses an append-only format: new symbols are appended on save, and loading merges them into the running table. File locking (`flock` on POSIX, `LockFileEx` on Windows) ensures safe concurrent access. +`ray_sym_save` writes the whole runtime table to a temporary file and renames it over the target, so a reader never sees a partial file; there is no merge with another writer's file, so one process writes a given file. Loading merges the file into the running table. In typical usage, you do not call these directly — the splayed and partitioned table functions handle symbol persistence automatically. diff --git a/docs/docs/language/repl.md b/docs/docs/language/repl.md index 34cd12946..f8672f8e1 100644 --- a/docs/docs/language/repl.md +++ b/docs/docs/language/repl.md @@ -17,7 +17,9 @@ The `rayforce` binary operates in three modes depending on how it is invoked: echo '(+ 1 2)' | ./rayforce ``` -Interactive mode is detected automatically when stdin is a terminal. On startup the REPL prints a banner with the version, CPU, memory, logical CPU count, and worker count, then shows the `‣` prompt. By default the pool includes every online logical CPU, including SMT threads. Use `-c N` to choose a total worker count explicitly: N must contain decimal digits only; `0` selects the automatic size, and `1..1023` sets the total count, including the main thread. Values outside this range are rejected before the pool starts. Individual operations may use fewer tasks when their workload or memory budget requires it. Use `-m SIZE` (`--mem`; a byte count or a `K`/`M`/`G`/`T` suffix) to set the RAM budget: anonymous memory past it spills to disk instead of risking the OOM killer. `(.sys.mem)` reports the effective value as [`anon-watermark`](../namespaces/sys.md#sys-mem). +Interactive mode is detected automatically when stdin is a terminal. On startup the REPL prints a banner with the version, CPU, memory, logical CPU count, and worker count, then shows the `‣` prompt. By default the pool includes every logical CPU the process may run on (all online CPUs, unless `taskset` or a cpuset narrowed them), including SMT threads. Use `-c N` to choose a total worker count explicitly: N must contain decimal digits only; `0` selects the automatic size, and `1..1023` sets the total count, including the main thread. Values outside this range are rejected before the pool starts. Individual operations may use fewer tasks when their workload or memory budget requires it. Use `-m SIZE` (`--mem`; a byte count or a `K`/`M`/`G`/`T` suffix) to set the RAM budget: anonymous memory past it spills to disk instead of risking the OOM killer. `(.sys.mem)` reports the effective value as [`anon-watermark`](../namespaces/sys.md#sys-mem). + +On Linux a pool with one thread per allowed CPU is pinned: the main thread runs on the first CPU and each worker on one of the others, one hardware thread of every physical core first, alternating NUMA nodes, then the hyperthread siblings. Programs started with `.sys.exec` still get every allowed CPU. `RAY_POOL_PIN=0` turns pinning off; `RAY_POOL_PIN=1` also pins a smaller pool. See [Running Several Instances](../guides/instances.md) for choosing CPUs per instance. ### Environment Variables diff --git a/docs/docs/storage/index.md b/docs/docs/storage/index.md index c4e1390e6..d228a3ac5 100644 --- a/docs/docs/storage/index.md +++ b/docs/docs/storage/index.md @@ -222,7 +222,7 @@ The global symbol intern table maps strings to integer IDs. When saving data to ### Append-Only .sym Files -Symbol files use an append-only format. New symbols are appended to the end of the file without rewriting existing entries. This makes concurrent writes safe and enables incremental updates. +Symbol files use an append-only format. New symbols are appended to the end of the file without rewriting existing entries, so a symbol's position never changes and updates are incremental. One process writes a database at a time: a writer whose view of the file is out of date fails with `RAY_ERR_CORRUPT` instead of overwriting another writer's symbols (see [Running Several Instances](../guides/instances.md)). | Function | Description | |---|---| @@ -243,7 +243,7 @@ ray_sym_load("db/.sym"); ### Concurrency and Integrity - **File locking** — `flock()` on POSIX, `LockFileEx()` on Windows. Multiple processes can safely read the symbol file; writes acquire an exclusive lock. -- **Corruption detection** — symbol files include checksums. If a file is truncated or corrupted, `ray_sym_load()` returns `RAY_ERR_CORRUPT` and leaves the in-memory table unchanged. +- **Corruption detection** — symbol files carry no checksum. The loader checks the magic, the entry count and every entry's length; a truncated or malformed file makes `ray_sym_load()` return `RAY_ERR_CORRUPT`. - **Arena backing** — interned strings are allocated from a dedicated arena (`ray_arena_t`), making bulk allocation fast and ensuring all strings are freed together when the symbol table is destroyed. ## CSV Import and Export @@ -412,7 +412,7 @@ For every table present in **any** partition, ensures **every** partition has it Symbol tables are persisted automatically when you use `.db.splayed.set`. A `.sym` file is written into the table directory containing all interned symbol strings. When loading with `.db.splayed.get` or `.db.parted.get`, the symbol table is loaded first so that symbol columns decode correctly. -At the C API level, `ray_sym_save(path)` and `ray_sym_load(path)` handle persistence directly. The format is append-only — new symbols are appended without rewriting existing entries, making concurrent access safe. +At the C API level, `ray_sym_save(path)` and `ray_sym_load(path)` handle persistence directly. `ray_sym_save` writes the whole runtime table to a temporary file and renames it over `path`; one process writes a given file. ## Working Example: Round-Trip diff --git a/mkdocs.yml b/mkdocs.yml index 9a9e05296..df85bf0da 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -167,6 +167,7 @@ nav: - IPC Guide: docs/guides/ipc.md - Error Handling: docs/guides/errors.md - Memory & Monitoring: docs/guides/memory.md + - Running Several Instances: docs/guides/instances.md - Type Casting: docs/guides/types.md - Timespan: docs/guides/timespan.md - Architecture: From 9cd146f39676f892f25cd25f3f0716ae75d3d616 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Thu, 22 Oct 2026 17:50:21 +0300 Subject: [PATCH 2/5] perf(pool): size the default pool from the CPUs the process may run on The default pool had one thread per online CPU also under taskset or a cgroup cpuset, so an instance confined to 8 CPUs ran a thread per machine CPU on them, and under a CPU quota (docker --cpus, systemd CPUQuota=) more threads than the quota lets run, throttled in turn. The default size is now the process affinity mask on Linux (the online count elsewhere), or the CPU quota rounded up when that is smaller: the smallest v2 cpu.max or v1 cfs quota over its period on the process's cgroup and its ancestors. Threads are not tied to CPUs; the scheduler places them within the mask. A -c larger than that is obeyed with a warning on stderr. .sys.info reported the online CPU count; cores is now the CPUs the process may run on, and the new key workers is the pool size, main included. The banner prints the same CPU count. The RFL runner takes ";; @requires: linux" for files that check Linux-only facilities (taskset, cgroups). --- docs/docs/architecture/pipeline.md | 2 +- docs/docs/guides/instances.md | 16 ++-- docs/docs/language/repl.md | 2 +- docs/docs/namespaces/sys.md | 9 +- docs/docs/storage/index.md | 4 +- src/app/main.c | 7 ++ src/app/repl.c | 6 +- src/core/platform.c | 116 +++++++++++++++++++++++++ src/core/platform.h | 7 ++ src/core/pool.c | 16 +++- src/core/pool.h | 5 ++ src/ops/system.c | 11 ++- test/main.c | 7 ++ test/rfl/system/pool_cpuset.rfl | 28 ++++++ test/rfl/system/reserved_namespace.rfl | 2 +- test/rfl/system/system_branch_cov2.rfl | 6 +- test/test_pool.c | 49 ++++++++++- test/test_repl.c | 28 ++++-- 18 files changed, 285 insertions(+), 36 deletions(-) create mode 100644 test/rfl/system/pool_cpuset.rfl diff --git a/docs/docs/architecture/pipeline.md b/docs/docs/architecture/pipeline.md index 14d64d9bf..7e68d4b8c 100644 --- a/docs/docs/architecture/pipeline.md +++ b/docs/docs/architecture/pipeline.md @@ -207,7 +207,7 @@ Rayforce uses a global thread pool (`ray_pool_t`) for parallel execution. Querie Each dispatch splits its tasks into one contiguous range per worker (the main thread included); a worker runs its range in order and, once it is empty, steals half of another worker's remaining range from the tail, so an idle thread never waits on a busy one and the rows a worker touches are sequential. Tasks of one dispatch may therefore run in any order and on any worker: operators place results by task index, and any early stop (a `take:` cutoff, a top-N bound) is derived from row positions or values, never from "the earlier tasks have run". The scheduler is selected once at pool creation (`RAY_POOL_STEAL`, default `1`; `0` restores the shared claim cursor for A/B comparison) and `RAY_POOL_TRACE=1` prints the own-versus-stolen task histogram per worker to stderr at pool destroy; see [REPL Reference](../language/repl.md#environment-variables). -The default pool has one thread per CPU of the process affinity mask. On Linux such a pool is pinned: each spawned worker pins itself to one CPU before it creates its heap, so the heap's pages are first touched on the worker's own NUMA node and stay local to it. Workers take one hardware thread of every physical core first, round-robin across nodes starting from the node with most of the set's cores, then the hyperthread siblings. The `rayforce` binary also pins the main thread (worker 0) to the first CPU of that order; a pool created inside an embedding application leaves the caller's thread alone. Children started with `.sys.exec` get the whole affinity mask back. `RAY_POOL_PIN=0` disables pinning and `RAY_POOL_PIN=1` also pins a pool smaller than the mask. +The default pool has one thread per CPU of the process affinity mask (Linux; the online CPU count elsewhere), or per CPU of a cgroup CPU quota, rounded up, when that is smaller. Threads are not tied to CPUs: the scheduler places them within the mask. ### Radix-Partitioned Hash Join diff --git a/docs/docs/guides/instances.md b/docs/docs/guides/instances.md index 46f810db9..eafdb74b5 100644 --- a/docs/docs/guides/instances.md +++ b/docs/docs/guides/instances.md @@ -6,7 +6,7 @@ Several rayforce processes can run on one machine. Each process sizes its thread Choose the CPUs of each instance when you start it: `taskset -c LIST`, `docker run --cpuset-cpus=LIST`, or `CPUAffinity=` in a systemd unit. Every thread of the instance, and every program it starts with `.sys.exec`, stays within that list. -On Linux an instance uses one thread per CPU of its list. Such a pool is pinned: the main thread runs on the first CPU and each worker on one of the others, taking one hardware thread of every physical core first, alternating NUMA nodes, and the hyperthread siblings after. Pinning keeps each worker's memory on its own node. +On Linux an instance sizes its thread pool to its list: one thread per CPU of it, the main thread included. The operating system keeps the threads within the list and moves them between its CPUs as it sees fit; rayforce does not tie a thread to one CPU. On other platforms the pool has one thread per online CPU. ```bash taskset -c 0,2,4,6,24,26,28,30 rayforce -p 5001 & @@ -19,19 +19,19 @@ taskset -c 8,10,12,14,32,34,36,38 rayforce -p 5002 & - Keep an instance on one NUMA node when its CPU count fits in one. - CPU numbering does not follow the layout on every machine. On a two-socket machine that numbers sockets alternately (even CPUs on socket 0, odd on socket 1) with CPU `k` and `k+24` on one core, the ranges `0-7`, `8-15`, … `40-47` put every instance on both sockets, and the last three ranges on the hyperthreads of the first three. For six instances of 8 threads on that machine use `0,2,4,6,24,26,28,30`, `8,10,12,14,32,34,36,38`, `16,18,20,22,40,42,44,46`, `1,3,5,7,25,27,29,31`, `9,11,13,15,33,35,37,39` and `17,19,21,23,41,43,45,47`. +**CPU quota.** `docker run --cpus=N` and systemd `CPUQuota=` limit CPU time, not CPUs: the instance may still run on every CPU. Its default pool then has one thread per CPU of the quota, rounded up. Give it `--cpuset-cpus` or `CPUAffinity=` as well to choose its CPUs. + **Do not:** -- start two instances on overlapping lists: their pinned threads share CPUs and neither can move away; +- start two instances on overlapping lists: they share those CPUs; - start instances without a list and leave each at its default size: every instance takes every CPU; -- pass `-c` larger than the list: the threads share CPUs and are not pinned (rayforce prints a warning). - -`RAY_POOL_PIN=0` turns pinning off; then size the instances with `-c` so that their pools add up to the CPU count. `RAY_POOL_PIN=1` also pins a pool smaller than its list, always to the same first CPUs of it, so use it only with lists that do not overlap. +- pass `-c` larger than the list: the threads take turns on its CPUs (rayforce prints a warning). -To see where an instance's threads run, read every thread: `grep Cpus_allowed_list /proc/PID/task/*/status`. `taskset -p PID` shows only the main thread. +`(.sys.info)` reports the CPUs of the instance (`cores`) and its pool size (`workers`); `taskset -p PID` shows its list. ## Memory -Each instance treats the machine's whole physical RAM as its own and does not read container memory limits. It moves its working set to disk only when its own memory would pass that amount, so several busy instances can together exhaust RAM before any of them spills. Start every instance with `-m SIZE` and keep the sizes, plus room for the page cache, within RAM: +Each instance treats the machine's whole physical RAM, or the memory limit of its container or cgroup when one is set, as its own. It moves its working set to disk only when its own memory would pass that amount, so several busy instances without limits of their own can together exhaust RAM before any of them spills. Start every instance with `-m SIZE` and keep the sizes, plus room for the page cache, within RAM: ```bash taskset -c 0-11 rayforce -m 40G -p 5001 & @@ -63,8 +63,6 @@ Instances that read the same tables share the operating system's page cache, so (.sys.exec "mv /db/.stage/2026.10.02 /db/2026.10.02") ``` -- If a first write of a table fails, its directory is left without a schema and every `.db.parted.get` of that database fails with `cannot read .d schema`. Remove that directory. - **Files.** Never change table files in place: no `cp` over them, no `rsync --inplace`, no `truncate`, no editors. Instances that have the table open see the change at once, its stored statistics no longer match the data, and a shortened file crashes them with `SIGBUS`. Replace tables with `.db.splayed.set`, or write a new file and rename it over the old one. The same applies to files written with `write`, `write-bytes` and `hnsw-save`, which rewrite their target in place. **Versions.** All instances that share a database must run the same rayforce version; upgrade them together. diff --git a/docs/docs/language/repl.md b/docs/docs/language/repl.md index f8672f8e1..ed5809764 100644 --- a/docs/docs/language/repl.md +++ b/docs/docs/language/repl.md @@ -19,7 +19,7 @@ echo '(+ 1 2)' | ./rayforce Interactive mode is detected automatically when stdin is a terminal. On startup the REPL prints a banner with the version, CPU, memory, logical CPU count, and worker count, then shows the `‣` prompt. By default the pool includes every logical CPU the process may run on (all online CPUs, unless `taskset` or a cpuset narrowed them), including SMT threads. Use `-c N` to choose a total worker count explicitly: N must contain decimal digits only; `0` selects the automatic size, and `1..1023` sets the total count, including the main thread. Values outside this range are rejected before the pool starts. Individual operations may use fewer tasks when their workload or memory budget requires it. Use `-m SIZE` (`--mem`; a byte count or a `K`/`M`/`G`/`T` suffix) to set the RAM budget: anonymous memory past it spills to disk instead of risking the OOM killer. `(.sys.mem)` reports the effective value as [`anon-watermark`](../namespaces/sys.md#sys-mem). -On Linux a pool with one thread per allowed CPU is pinned: the main thread runs on the first CPU and each worker on one of the others, one hardware thread of every physical core first, alternating NUMA nodes, then the hyperthread siblings. Programs started with `.sys.exec` still get every allowed CPU. `RAY_POOL_PIN=0` turns pinning off; `RAY_POOL_PIN=1` also pins a smaller pool. See [Running Several Instances](../guides/instances.md) for choosing CPUs per instance. +Under a cgroup CPU quota (`docker --cpus`, systemd `CPUQuota=`) the default pool has one thread per CPU of the quota, rounded up. A `-c` larger than the allowed CPUs prints a warning. See [Running Several Instances](../guides/instances.md) for choosing CPUs per instance. ### Environment Variables diff --git a/docs/docs/namespaces/sys.md b/docs/docs/namespaces/sys.md index a86eb9bcc..bca88a9b7 100644 --- a/docs/docs/namespaces/sys.md +++ b/docs/docs/namespaces/sys.md @@ -11,7 +11,7 @@ Process-level introspection (build, memory, host info) and command-style operati |---|---|---|---| | [`.sys.args`](#sys-args) | variadic | — | Command-line arguments as a typed dict. | | [`.sys.build`](#sys-build) | nullary | — | Version + build date as a dict. | -| [`.sys.info`](#sys-info) | nullary | — | Host and process facts: cores, page size, total memory, pid, hostname. | +| [`.sys.info`](#sys-info) | nullary | — | Host and process facts: cores, page size, total memory, pool size, pid, hostname. | | [`.sys.mem`](#sys-mem) | nullary | — | Allocator statistics. | | [`.sys.prof`](#sys-prof) | nullary | — | Last profiled query's per-step statistics as a table. | | [`.sys.querylog`](#sys-querylog) | variadic | — | Ambient per-query statistics ring as a table. | @@ -75,12 +75,15 @@ Signature: `(.sys.build)`. Returns a dict with `version` (string) and `build-dat ## `.sys.info` { #sys-info } -Signature: `(.sys.info)`. Returns `{cores: i64, page-size: i64, total-mem: i64, pid: i64, hostname: str}` on every platform (on Windows from `GetSystemInfo` / `GlobalMemoryStatusEx`). +Signature: `(.sys.info)`. Returns `{cores: i64, page-size: i64, total-mem: i64, workers: i64, pid: i64, hostname: str}` on every platform (on Windows from `GetSystemInfo` / `GlobalMemoryStatusEx`). + +- `cores`: the CPUs this process may run on. On Linux that is its affinity mask, which `taskset`, a cpuset or `docker --cpuset-cpus` may narrow below the machine's CPU count. +- `workers`: the threads of the execution pool, main included (see [Running Several Instances](../guides/instances.md)). ```lisp (.sys.info) ;; => {cores: 10, page-size: 16384, total-mem: 68719476736, -;; pid: 48213, hostname: "tp-prod-01"} +;; workers: 10, pid: 48213, hostname: "tp-prod-01"} ``` `pid` and `hostname` describe *this process*, not the machine, and exist so an diff --git a/docs/docs/storage/index.md b/docs/docs/storage/index.md index d228a3ac5..4e16dce3a 100644 --- a/docs/docs/storage/index.md +++ b/docs/docs/storage/index.md @@ -222,7 +222,9 @@ The global symbol intern table maps strings to integer IDs. When saving data to ### Append-Only .sym Files -Symbol files use an append-only format. New symbols are appended to the end of the file without rewriting existing entries, so a symbol's position never changes and updates are incremental. One process writes a database at a time: a writer whose view of the file is out of date fails with `RAY_ERR_CORRUPT` instead of overwriting another writer's symbols (see [Running Several Instances](../guides/instances.md)). +A table's symbol file, which `.db.splayed.set` and the importers write, is append-only: new symbols are appended to the end of the file without rewriting existing entries, so a symbol's position never changes and updates are incremental. One process writes a database at a time: a writer whose view of the file is out of date fails with `RAY_ERR_CORRUPT` instead of overwriting another writer's symbols (see [Running Several Instances](../guides/instances.md)). + +The C functions below work on the runtime's global symbol table instead. `ray_sym_save` writes the whole table to a temporary file and renames it over the target, and `ray_sym_load` merges a file into the running table. | Function | Description | |---|---| diff --git a/src/app/main.c b/src/app/main.c index e14d39068..464f031d1 100644 --- a/src/app/main.c +++ b/src/app/main.c @@ -313,6 +313,13 @@ int main(int argc, char** argv) { * (file load, REPL eval, builtins). If -c wasn't given, leave the * pool to its lazy default on first use. */ if (n_cores >= 0) { + /* the CPUs this process may run on, or its CPU quota when smaller */ + uint32_t ncpu = ray_pool_default_total(); + if ((uint32_t)n_cores > ncpu) + fprintf(stderr, "warning: -c %d gives %d threads, more than the %u CPU%s this process " + "may use, and they take turns; use -c %u, or drop -c to size the pool " + "to the CPUs\n", + n_cores, n_cores, ncpu, ncpu == 1 ? "" : "s", ncpu); ray_err_t err = ray_pool_init_total((uint32_t)n_cores); if (err != RAY_OK) { fprintf(stderr, "error: ray_pool_init_total(%d) failed (%d)\n", diff --git a/src/app/repl.c b/src/app/repl.c index 0a8646862..4e8e1cb8d 100644 --- a/src/app/repl.c +++ b/src/app/repl.c @@ -396,11 +396,7 @@ static void print_banner(void) { char cpu[256]; get_cpu_name(cpu, sizeof(cpu)); int64_t mem_mb = get_total_mem_mb(); -#if defined(RAY_OS_WINDOWS) - int ncores = (int)ray_thread_count(); -#else - int ncores = (int)sysconf(_SC_NPROCESSORS_ONLN); -#endif + int ncores = (int)ray_cpu_set_count(); /* the CPUs this process may run on */ /* "Using" count reflects the actual worker-pool size, not ncores. * ray_pool_get() is a lazy initializer — callers might not have diff --git a/src/core/platform.c b/src/core/platform.c index 3bd433a8b..1e6052a68 100644 --- a/src/core/platform.c +++ b/src/core/platform.c @@ -62,6 +62,8 @@ uint64_t ray_vm_willneed_bytes(void) { #include /* EINTR — ray_os_cgroup_stat_flush */ #include #include /* getrusage: ray_io_counters */ +#include /* sched_getaffinity: ray_cpu_set_count */ +#include /* INT64_MAX: the cgroup quota */ #if !defined(RUSAGE_THREAD) #define RUSAGE_THREAD 1 #endif @@ -415,6 +417,23 @@ uint32_t ray_thread_count(void) { return (n > 0) ? (uint32_t)n : 1; } +/* The CPUs this process may run on: its affinity mask on Linux, which + * taskset, a cgroup cpuset or docker --cpuset-cpus may narrow below the + * online count; the online count elsewhere. The mask read is the calling + * thread's: the process's, unless an embedding application narrowed the + * thread that first uses the pool (whose workers then inherit it). */ +uint32_t ray_cpu_set_count(void) { +#if defined(RAY_OS_LINUX) + cpu_set_t set; + CPU_ZERO(&set); + if (sched_getaffinity(0, sizeof(set), &set) == 0) { + int n = CPU_COUNT(&set); + if (n > 0) return (uint32_t)n; + } +#endif + return ray_thread_count(); +} + /* Physical cores (SMT siblings collapsed). The worker pool's kernels are * memory-bound; two hyperthreads sharing one core's load/store machinery * only add contention (measured: the full ClickBench suite runs ~11% @@ -634,6 +653,98 @@ int64_t ray_os_cgroup_mem_limit(void) { return cgroup_mem_limit(NULL, 0); } +/* CPU quota (docker --cpus, systemd CPUQuota=): the CPU time a cgroup may + * use per period, as v2 cpu.max "QUOTA PERIOD" ("max" = none) or v1 + * cpu.cfs_quota_us (-1 = none) over cpu.cfs_period_us, in millicpus. A + * quota does not narrow the affinity mask, so a pool sized to the mask + * would run more threads than the quota lets run and be throttled. */ +static int64_t cgroup_cpu_read(const char* dir, bool v2) { + char path[640], buf[64]; + long long q = 0, per = 0; + if (v2) { + snprintf(path, sizeof(path), "%s/cpu.max", dir); + FILE* f = fopen(path, "r"); + if (!f) return 0; + buf[0] = 0; + if (!fgets(buf, sizeof(buf), f)) buf[0] = 0; + fclose(f); + if (strncmp(buf, "max", 3) == 0 || sscanf(buf, "%lld %lld", &q, &per) != 2) return 0; + } else { + snprintf(path, sizeof(path), "%s/cpu.cfs_quota_us", dir); + FILE* f = fopen(path, "r"); + if (!f) return 0; + if (fscanf(f, "%lld", &q) != 1) q = 0; + fclose(f); + snprintf(path, sizeof(path), "%s/cpu.cfs_period_us", dir); + f = fopen(path, "r"); + if (!f) return 0; + if (fscanf(f, "%lld", &per) != 1) per = 0; + fclose(f); + } + if (q <= 0 || per <= 0 || q > (long long)(INT64_MAX / 1000) - per) return 0; + return (int64_t)((q * 1000 + per - 1) / per); +} + +/* Smallest quota over `root`/`cgpath` and every ancestor, as the memory + * limit walk above. */ +static int64_t cgroup_cpu_walk(const char* root, const char* cgpath, bool v2) { + char dir[512]; + if (snprintf(dir, sizeof(dir), "%s%s", root, cgpath) >= (int)sizeof(dir)) return 0; + size_t rootlen = strlen(root); + int64_t m = 0; + for (;;) { + int64_t v = cgroup_cpu_read(dir, v2); + if (v > 0 && (m == 0 || v < m)) m = v; + if (strlen(dir) <= rootlen) break; + char* slash = strrchr(dir, '/'); + if (!slash || (size_t)(slash - dir) < rootlen) break; + *slash = 0; + } + return m; +} + +/* True when the comma-separated controller list names "cpu". */ +static bool cgroup_has_cpu(const char* ctrls) { + while (*ctrls) { + size_t n = strcspn(ctrls, ","); + if (n == 3 && strncmp(ctrls, "cpu", 3) == 0) return true; + ctrls += n; + if (*ctrls == ',') ctrls++; + } + return false; +} + +int64_t ray_os_cgroup_cpu_quota(void) { + FILE* f = fopen("/proc/self/cgroup", "r"); + if (!f) return 0; + int64_t m = 0; + char line[512]; + while (fgets(line, sizeof(line), f)) { + char* c1 = strchr(line, ':'); + char* c2 = c1 ? strchr(c1 + 1, ':') : NULL; + if (!c2) continue; + *c2 = 0; + const char* ctrls = c1 + 1; + char* cgpath = c2 + 1; + cgpath[strcspn(cgpath, "\n")] = 0; + if (strcmp(cgpath, "/") == 0) cgpath[0] = 0; + int64_t v = 0; + if (*ctrls == 0) { + v = cgroup_cpu_walk("/sys/fs/cgroup", cgpath, true); + } else if (cgroup_has_cpu(ctrls)) { + /* v1 mounts the controller as the directory named by its list + * ("cpu,cpuacct"), usually with a "cpu" link to it */ + char root[300]; + snprintf(root, sizeof(root), "/sys/fs/cgroup/%s", ctrls); + v = cgroup_cpu_walk(root, cgpath, false); + if (!v) v = cgroup_cpu_walk("/sys/fs/cgroup/cpu", cgpath, false); + } + if (v > 0 && (m == 0 || v < m)) m = v; + } + fclose(f); + return m; +} + bool ray_os_cgroup_mem_dir(char* buf, size_t n) { if (n) buf[0] = 0; return n && cgroup_mem_limit(buf, n) > 0 && buf[0]; @@ -652,6 +763,7 @@ bool ray_os_cgroup_stat_flush(const char* cg) { } #else int64_t ray_os_cgroup_mem_limit(void) { return 0; } +int64_t ray_os_cgroup_cpu_quota(void) { return 0; } bool ray_os_cgroup_mem_dir(char* buf, size_t n) { if (n) buf[0] = 0; return false; } bool ray_os_cgroup_stat_flush(const char* cg) { (void)cg; return false; } #endif @@ -959,6 +1071,7 @@ uint32_t ray_thread_count(void) { GetSystemInfo(&si); return (uint32_t)si.dwNumberOfProcessors; } +uint32_t ray_cpu_set_count(void) { return ray_thread_count(); } /* Windows: no cheap topology read here — fall back to the logical count * (the SMT-aware default sizing is a POSIX-side optimization). */ @@ -967,6 +1080,7 @@ uint32_t ray_physical_core_count(void) { } int64_t ray_os_cgroup_mem_limit(void) { return 0; } +int64_t ray_os_cgroup_cpu_quota(void) { return 0; } bool ray_os_cgroup_mem_dir(char* buf, size_t n) { if (n) buf[0] = 0; return false; } bool ray_os_cgroup_stat_flush(const char* cg) { (void)cg; return false; } @@ -1156,7 +1270,9 @@ ray_err_t ray_thread_join(ray_thread_t t) { } uint32_t ray_thread_count(void) { return 1; } +uint32_t ray_cpu_set_count(void) { return ray_thread_count(); } int64_t ray_os_cgroup_mem_limit(void) { return 0; } +int64_t ray_os_cgroup_cpu_quota(void) { return 0; } bool ray_os_cgroup_mem_dir(char* buf, size_t n) { if (n) buf[0] = 0; return false; } bool ray_os_cgroup_stat_flush(const char* cg) { (void)cg; return false; } static uint64_t cache_llc_probe(void) { return 0; } diff --git a/src/core/platform.h b/src/core/platform.h index 1379d6cfe..64c749bec 100644 --- a/src/core/platform.h +++ b/src/core/platform.h @@ -205,6 +205,9 @@ bool ray_vm_hugepage(void* ptr, size_t size); ray_err_t ray_thread_create(ray_thread_t* t, ray_thread_fn fn, void* arg); ray_err_t ray_thread_join(ray_thread_t t); uint32_t ray_thread_count(void); +/* The CPUs this process may run on: the affinity mask on Linux (taskset, a + * cpuset or docker --cpuset-cpus narrow it), ray_thread_count elsewhere. */ +uint32_t ray_cpu_set_count(void); /* Physical cores (SMT siblings collapsed); falls back to the logical * count when topology is unavailable. */ uint32_t ray_physical_core_count(void); @@ -217,6 +220,10 @@ uint64_t ray_cache_llc_bytes(void); * the process's cgroup and its ancestors. 0 when none is set, or on * platforms without cgroups. */ int64_t ray_os_cgroup_mem_limit(void); +/* CPU quota of this process's cgroup (docker --cpus, systemd CPUQuota=), in + * millicpus: the smallest v2 cpu.max or v1 cfs quota over its period on the + * cgroup and its ancestors. 0 when none is set, or without cgroups. */ +int64_t ray_os_cgroup_cpu_quota(void); /* The directory of the cgroup that sets that limit, into buf[n]; false when * no limit is set, or on platforms without cgroups. */ bool ray_os_cgroup_mem_dir(char* buf, size_t n); diff --git a/src/core/pool.c b/src/core/pool.c index 8dcd80bcd..24a0e5975 100644 --- a/src/core/pool.c +++ b/src/core/pool.c @@ -32,6 +32,13 @@ #include #include +uint32_t ray_pool_default_total(void) { + uint32_t n = ray_cpu_set_count(); + int64_t q = ray_os_cgroup_cpu_quota(); /* millicpus, 0 = none */ + if (q > 0 && (q + 999) / 1000 < (int64_t)n) n = (uint32_t)((q + 999) / 1000); + return n > 0 ? n : 1; +} + /* Per-worker query-statistics slab (see core/qstats.h). Zero-initialised; * disabled (mode == 0) until a profiling/progress consumer turns it on. */ ray_qstats_t g_qstats; @@ -581,9 +588,12 @@ static ray_err_t ray_pool_create_impl(ray_pool_t* pool, uint32_t n_workers, long v = strtol(env, NULL, 10); n_workers = (v > 0) ? (uint32_t)v : 0; } else { - /* Default to every online logical CPU. Individual operations may - * bound their task count to their workload or memory budget. */ - uint32_t ncpu = ray_thread_count(); + /* Default to every CPU the process may run on (its affinity + * mask: every online CPU unless taskset or a cpuset narrowed + * it), or fewer under a cgroup CPU quota. Individual + * operations may bound their task count to their workload or + * memory budget. */ + uint32_t ncpu = ray_pool_default_total(); n_workers = (ncpu > 1) ? ncpu - 1 : 0; } } diff --git a/src/core/pool.h b/src/core/pool.h index fbf0b956a..bc32283a9 100644 --- a/src/core/pool.h +++ b/src/core/pool.h @@ -284,6 +284,11 @@ ray_err_t ray_pool_init(uint32_t n_workers); /* Initialize by total execution participants, including the main thread. * Pass 0 to auto-detect. Used by the CLI, where -c is a total-core count. */ ray_err_t ray_pool_init_total(uint32_t total_workers); + +/* The default pool size, main included: one thread per CPU the process may + * run on (ray_cpu_set_count), or per CPU of a cgroup CPU quota, rounded up, + * when that is smaller (docker --cpus, systemd CPUQuota=). */ +uint32_t ray_pool_default_total(void); void ray_pool_destroy(void); /* Call fn on the heap of every worker of the live singleton pool; nothing diff --git a/src/ops/system.c b/src/ops/system.c index c6e469c9e..820952e61 100644 --- a/src/ops/system.c +++ b/src/ops/system.c @@ -48,6 +48,7 @@ #include "core/mcast.h" #include "core/poll.h" #include "core/timer.h" +#include "core/pool.h" /* ray_pool_get — the pool size in .sys.info */ /* Forward decls: keep this TU decoupled from core/runtime.h — only * the opaque accessors are needed. */ @@ -1565,9 +1566,10 @@ ray_t* ray_sysinfo_fn(ray_t** args, int64_t n) { if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; } #if !defined(RAY_OS_WINDOWS) + /* cores: the CPUs this process may run on (its affinity mask) */ int64_t s1 = ray_sym_intern("cores", 5); keys = ray_vec_append(keys, &s1); - ray_t* v1 = make_i64(sysconf(_SC_NPROCESSORS_ONLN)); + ray_t* v1 = make_i64((int64_t)ray_cpu_set_count()); vals = ray_list_append(vals, v1); ray_release(v1); int64_t s2 = ray_sym_intern("page-size", 9); @@ -1599,6 +1601,13 @@ ray_t* ray_sysinfo_fn(ray_t** args, int64_t n) { vals = ray_list_append(vals, v3); ray_release(v3); #endif + /* workers: the threads of the execution pool, main included */ + ray_pool_t* pool = ray_pool_get(); + int64_t sw = ray_sym_intern("workers", 7); + keys = ray_vec_append(keys, &sw); + ray_t* vw = make_i64(pool ? (int64_t)ray_pool_total_workers(pool) : 1); + vals = ray_list_append(vals, vw); ray_release(vw); + /* Process and host identity (#573). An embedded process previously had * no way to name itself: pid plus hostname plus a start timestamp is a * diagnosable epoch, where a random token only tells you that two runs diff --git a/test/main.c b/test/main.c index dd5e5a893..9fc8347c9 100644 --- a/test/main.c +++ b/test/main.c @@ -424,6 +424,13 @@ static test_result_t run_rfl_file(const char* path) { SKIP("requires POSIX shell/filesystem"); } #endif + /* ";; @requires: linux": Linux-only facilities (taskset, cgroups). */ +#if !defined(__linux__) + if (strstr(src, ";; @requires: linux")) { + free(src); + SKIP("requires Linux"); + } +#endif int line_no = 0; int assert_count = 0; /* tallies LHS -- RHS and EXPR !- SUBSTR lines */ diff --git a/test/rfl/system/pool_cpuset.rfl b/test/rfl/system/pool_cpuset.rfl new file mode 100644 index 000000000..320c46a17 --- /dev/null +++ b/test/rfl/system/pool_cpuset.rfl @@ -0,0 +1,28 @@ +;; @requires: posix (the instances run as child processes through .sys.exec) +;; @requires: linux (taskset and cgroups; elsewhere the pool takes every online CPU) +;; The default pool has one thread per CPU the process may run on: its +;; affinity mask (taskset, a cpuset), or fewer under a cgroup CPU quota +;; (docker --cpus, systemd CPUQuota=). .sys.info reports the CPU count and +;; the pool size. + +(.sys.exec "rm -rf /tmp/rf_cpuset && mkdir -p /tmp/rf_cpuset") +(.sys.exec "printf '(set i (.sys.info))\\n(println (format \"%% %%\" (at i (quote cores)) (at i (quote workers))))\\n' > /tmp/rf_cpuset/info.rfl") + +;; Under taskset the default pool has one thread per allowed CPU. The CPUs +;; are the first of the runner's own mask; a runner under a CPU quota below +;; two CPUs (q, its own default pool size) caps the pool at that. +;; RAYFORCE_CORES, which the test harness sets, would size the pool itself. +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); [ $# -ge 2 ] || exit 0; q=$(env -u RAYFORCE_CORES ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); q=${q#* }; e=2; [ \"$q\" -lt 2 ] && e=$q; o=$(env -u RAYFORCE_CORES taskset -c $1,$2 ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = \"2 $e\" ]") -- 0 +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); o=$(env -u RAYFORCE_CORES taskset -c $1 ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = '1 1' ]") -- 0 +;; -c past the set is obeyed, with a warning on stderr +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); o=$(env -u RAYFORCE_CORES taskset -c $1 ./rayforce -c 3 -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = '1 3' ]") -- 0 +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); env -u RAYFORCE_CORES taskset -c $1 ./rayforce -c 3 -f /tmp/rf_cpuset/info.rfl 2>&1 >/dev/null | grep -q 'gives 3 threads, more than the 1 CPU this process may use'") -- 0 +;; -c within what the process may use: no warning +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); [ $# -ge 2 ] || exit 0; q=$(env -u RAYFORCE_CORES ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); q=${q#* }; e=2; [ \"$q\" -lt 2 ] && e=$q; ! env -u RAYFORCE_CORES taskset -c $1,$2 ./rayforce -c $e -f /tmp/rf_cpuset/info.rfl 2>&1 >/dev/null | grep -q warning") -- 0 + +;; A CPU quota of 1.5 CPUs gives a pool of 2 threads, while cores still +;; counts every allowed CPU. Runs only where a user systemd can make a +;; scope with a quota (a scope of its own, outside the runner's cgroup). +(.sys.exec "command -v systemd-run >/dev/null || exit 0; [ $(nproc) -ge 3 ] || exit 0; systemd-run --user --scope -q -p CPUQuota=150% true 2>/dev/null || exit 0; o=$(env -u RAYFORCE_CORES systemd-run --user --scope -q -p CPUQuota=150% ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = \"$(nproc) 2\" ]") -- 0 + +(.sys.exec "rm -rf /tmp/rf_cpuset") diff --git a/test/rfl/system/reserved_namespace.rfl b/test/rfl/system/reserved_namespace.rfl index 4a881d429..0ef273098 100644 --- a/test/rfl/system/reserved_namespace.rfl +++ b/test/rfl/system/reserved_namespace.rfl @@ -9,7 +9,7 @@ ;; the implementation can reject unexpected arguments explicitly. (.sys.gc) -- 0 (.sys.gc 0) !- arity -(count (.sys.info)) -- 5 +(count (.sys.info)) -- 6 (count (.sys.mem)) -- 18 (count (.sys.build)) -- 2 ;; The namespace roots themselves are real dicts — typing just diff --git a/test/rfl/system/system_branch_cov2.rfl b/test/rfl/system/system_branch_cov2.rfl index 224dd17e6..5aa674524 100644 --- a/test/rfl/system/system_branch_cov2.rfl +++ b/test/rfl/system/system_branch_cov2.rfl @@ -117,10 +117,12 @@ ;; ══════════════════════════════════════════════════════════════════════ (type (.sys.info)) -- 'DICT (.sys.info 1) !- arity -(count (.sys.info)) -- 5 +(count (.sys.info)) -- 6 (> (at (.sys.info) 'cores) 0) -- true +(type (at (.sys.info) 'workers)) -- 'i64 +(> (at (.sys.info) 'workers) 0) -- true (> (at (.sys.info) 'total-mem) 0) -- true -(count (value (.sys.info))) -- 5 +(count (value (.sys.info))) -- 6 ;; ══════════════════════════════════════════════════════════════════════ ;; ray_hopen_fn (lines 769-825) diff --git a/test/test_pool.c b/test/test_pool.c index d1085f39f..423a6ddc4 100644 --- a/test/test_pool.c +++ b/test/test_pool.c @@ -21,6 +21,9 @@ * SOFTWARE. */ +#if defined(__linux__) && !defined(_GNU_SOURCE) +#define _GNU_SOURCE /* CPU_SET / sched_setaffinity: the CPU-set sizing test */ +#endif #if !defined(_WIN32) && !defined(_POSIX_C_SOURCE) #define _POSIX_C_SOURCE 200809L #endif @@ -1792,7 +1795,48 @@ static test_result_t test_auto_all_logical_cpus(void) { if (rc == RAY_OK) ray_pool_free(&local); if (saved) { setenv("RAYFORCE_CORES", saved, 1); free(saved); } TEST_ASSERT_EQ_I(rc, RAY_OK); - TEST_ASSERT_EQ_I(total, ray_thread_count()); + /* every CPU the process may run on (taskset or a cpuset narrows the + * set on Linux), or fewer under a cgroup CPU quota (millicpus) */ + int expect = (int)ray_cpu_set_count(); + int64_t quota = ray_os_cgroup_cpu_quota(); + if (quota > 0 && (quota + 999) / 1000 < expect) expect = (int)((quota + 999) / 1000); + TEST_ASSERT_EQ_I(total, expect); + PASS(); +} +#endif + +#if defined(__linux__) +#include +/* The default pool follows the CPUs the process may run on: with the + * calling thread narrowed to two of them (as taskset does for a process), + * a default pool has two threads, main included. Under a cgroup CPU quota + * of less than two CPUs it has fewer. */ +static test_result_t test_auto_follows_cpu_set(void) { + cpu_set_t before, two; + CPU_ZERO(&before); + TEST_ASSERT_EQ_I(sched_getaffinity(0, sizeof(before), &before), 0); + if (CPU_COUNT(&before) < 3) SKIP("needs at least 3 CPUs"); + CPU_ZERO(&two); + for (int c = 0, k = 0; c < CPU_SETSIZE && k < 2; c++) + if (CPU_ISSET(c, &before)) { CPU_SET(c, &two); k++; } + const char* current = getenv("RAYFORCE_CORES"); + char* saved = current ? strdup(current) : NULL; + TEST_ASSERT_TRUE(!current || saved); + unsetenv("RAYFORCE_CORES"); + int set_rc = sched_setaffinity(0, sizeof(two), &two); + uint32_t seen = ray_cpu_set_count(); + ray_pool_t local; + ray_err_t rc = set_rc == 0 ? ray_pool_create(&local, 0) : RAY_ERR_IO; + uint32_t total = rc == RAY_OK ? ray_pool_total_workers(&local) : 0; + if (rc == RAY_OK) ray_pool_free(&local); + (void)sched_setaffinity(0, sizeof(before), &before); + if (saved) { setenv("RAYFORCE_CORES", saved, 1); free(saved); } + TEST_ASSERT_EQ_I(set_rc, 0); + TEST_ASSERT_EQ_I(seen, 2); + TEST_ASSERT_EQ_I(rc, RAY_OK); + int64_t quota = ray_os_cgroup_cpu_quota(); + int expect = (quota > 0 && (quota + 999) / 1000 < 2) ? 1 : 2; + TEST_ASSERT_EQ_I(total, expect); PASS(); } #endif @@ -2462,6 +2506,9 @@ static test_result_t test_scan_read_ahead_queries(void) { const test_entry_t pool_entries[] = { #if defined(__linux__) || defined(__APPLE__) { "pool/auto_all_logical_cpus", test_auto_all_logical_cpus, NULL, NULL }, +#endif +#if defined(__linux__) + { "pool/auto_follows_cpu_set", test_auto_follows_cpu_set, NULL, NULL }, #endif { "pool/parallel_sum", test_parallel_sum, NULL, NULL }, { "pool/dispatch_reclaims_worker_blocks", test_dispatch_reclaims_worker_blocks, NULL, NULL }, diff --git a/test/test_repl.c b/test/test_repl.c index 11acf4d01..df624cf6d 100644 --- a/test/test_repl.c +++ b/test/test_repl.c @@ -3017,8 +3017,8 @@ static int run_cli_capture(const char* command, char* output, size_t capacity, /* /proc/self/task observes the exact process-wide total, so this catches both * the ordinary off-by-one and the -c 1 collision with the pool's auto mode. */ -static int run_cli_task_count(const char* cores, long* out_count) { - char command[256], output[512]; +static int run_cli_task_count(const char* cores, long* out_count, bool* warned) { + char command[256], output[1024]; int status; snprintf(command, sizeof(command), "printf '(count (.fs.list \"/proc/self/task\"))\\n'" @@ -3028,30 +3028,42 @@ static int run_cli_task_count(const char* cores, long* out_count) { strstr(output, "ray_pool_init_total") != NULL || strstr(output, "heap-ID pool exhausted") != NULL) return -1; + /* a -c past the CPUs the process may use is warned of on stderr first */ + char* at = output; + if (warned) *warned = strncmp(at, "warning:", 8) == 0; + while (strncmp(at, "warning:", 8) == 0) { + char* nl = strchr(at, '\n'); + if (!nl) return -2; + at = nl + 1; + } char* end = NULL; - long count = strtol(output, &end, 10); - if (end == output) return -2; + long count = strtol(at, &end, 10); + if (end == at) return -2; *out_count = count; return 0; } static test_result_t test_repl_cli_cores_are_total(void) { long count = 0; - int rc = run_cli_task_count("1", &count); + bool warned = false; + int rc = run_cli_task_count("1", &count, &warned); TEST_ASSERT_FMT(rc == 0, "-c 1 launcher probe failed: %d", rc); TEST_ASSERT_FMT(count == 1, "-c 1 created %ld total threads", count); + TEST_ASSERT_FALSE(warned); - rc = run_cli_task_count("3", &count); + rc = run_cli_task_count("3", &count, NULL); TEST_ASSERT_FMT(rc == 0, "-c 3 launcher probe failed: %d", rc); TEST_ASSERT_FMT(count == 3, "-c 3 created %ld total threads", count); - rc = run_cli_task_count("007", &count); + rc = run_cli_task_count("007", &count, NULL); TEST_ASSERT_FMT(rc == 0, "-c 007 launcher probe failed: %d", rc); TEST_ASSERT_FMT(count == 7, "-c 007 created %ld total threads", count); - rc = run_cli_task_count("1023", &count); + /* more threads than any runner has CPUs: obeyed, with a warning */ + rc = run_cli_task_count("1023", &count, &warned); TEST_ASSERT_FMT(rc == 0, "-c 1023 launcher probe failed: %d", rc); TEST_ASSERT_FMT(count == 1023, "-c 1023 created %ld total threads", count); + TEST_ASSERT_TRUE(warned); PASS(); } From faf79f1fc305970fdba1ead4ffcefe9edb04f47d Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Thu, 22 Oct 2026 19:35:37 +0300 Subject: [PATCH 3/5] test: keep the CPU-set child-process checks out of the TSan pool filter The TSan job builds only the test binary and runs the files whose name holds "pool"; pool_cpuset started ./rayforce children, which that job does not build. Renamed to cpu_set_size. --- .../{pool_cpuset.rfl => cpu_set_size.rfl} | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) rename test/rfl/system/{pool_cpuset.rfl => cpu_set_size.rfl} (70%) diff --git a/test/rfl/system/pool_cpuset.rfl b/test/rfl/system/cpu_set_size.rfl similarity index 70% rename from test/rfl/system/pool_cpuset.rfl rename to test/rfl/system/cpu_set_size.rfl index 320c46a17..f8e627673 100644 --- a/test/rfl/system/pool_cpuset.rfl +++ b/test/rfl/system/cpu_set_size.rfl @@ -5,24 +5,24 @@ ;; (docker --cpus, systemd CPUQuota=). .sys.info reports the CPU count and ;; the pool size. -(.sys.exec "rm -rf /tmp/rf_cpuset && mkdir -p /tmp/rf_cpuset") -(.sys.exec "printf '(set i (.sys.info))\\n(println (format \"%% %%\" (at i (quote cores)) (at i (quote workers))))\\n' > /tmp/rf_cpuset/info.rfl") +(.sys.exec "rm -rf /tmp/rf_cpu_set && mkdir -p /tmp/rf_cpu_set") +(.sys.exec "printf '(set i (.sys.info))\\n(println (format \"%% %%\" (at i (quote cores)) (at i (quote workers))))\\n' > /tmp/rf_cpu_set/info.rfl") ;; Under taskset the default pool has one thread per allowed CPU. The CPUs ;; are the first of the runner's own mask; a runner under a CPU quota below ;; two CPUs (q, its own default pool size) caps the pool at that. ;; RAYFORCE_CORES, which the test harness sets, would size the pool itself. -(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); [ $# -ge 2 ] || exit 0; q=$(env -u RAYFORCE_CORES ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); q=${q#* }; e=2; [ \"$q\" -lt 2 ] && e=$q; o=$(env -u RAYFORCE_CORES taskset -c $1,$2 ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = \"2 $e\" ]") -- 0 -(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); o=$(env -u RAYFORCE_CORES taskset -c $1 ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = '1 1' ]") -- 0 +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); [ $# -ge 2 ] || exit 0; q=$(env -u RAYFORCE_CORES ./rayforce -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); q=${q#* }; e=2; [ \"$q\" -lt 2 ] && e=$q; o=$(env -u RAYFORCE_CORES taskset -c $1,$2 ./rayforce -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); [ \"$o\" = \"2 $e\" ]") -- 0 +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); o=$(env -u RAYFORCE_CORES taskset -c $1 ./rayforce -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); [ \"$o\" = '1 1' ]") -- 0 ;; -c past the set is obeyed, with a warning on stderr -(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); o=$(env -u RAYFORCE_CORES taskset -c $1 ./rayforce -c 3 -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = '1 3' ]") -- 0 -(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); env -u RAYFORCE_CORES taskset -c $1 ./rayforce -c 3 -f /tmp/rf_cpuset/info.rfl 2>&1 >/dev/null | grep -q 'gives 3 threads, more than the 1 CPU this process may use'") -- 0 +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); o=$(env -u RAYFORCE_CORES taskset -c $1 ./rayforce -c 3 -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); [ \"$o\" = '1 3' ]") -- 0 +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); env -u RAYFORCE_CORES taskset -c $1 ./rayforce -c 3 -f /tmp/rf_cpu_set/info.rfl 2>&1 >/dev/null | grep -q 'gives 3 threads, more than the 1 CPU this process may use'") -- 0 ;; -c within what the process may use: no warning -(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); [ $# -ge 2 ] || exit 0; q=$(env -u RAYFORCE_CORES ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); q=${q#* }; e=2; [ \"$q\" -lt 2 ] && e=$q; ! env -u RAYFORCE_CORES taskset -c $1,$2 ./rayforce -c $e -f /tmp/rf_cpuset/info.rfl 2>&1 >/dev/null | grep -q warning") -- 0 +(.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); [ $# -ge 2 ] || exit 0; q=$(env -u RAYFORCE_CORES ./rayforce -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); q=${q#* }; e=2; [ \"$q\" -lt 2 ] && e=$q; ! env -u RAYFORCE_CORES taskset -c $1,$2 ./rayforce -c $e -f /tmp/rf_cpu_set/info.rfl 2>&1 >/dev/null | grep -q warning") -- 0 ;; A CPU quota of 1.5 CPUs gives a pool of 2 threads, while cores still ;; counts every allowed CPU. Runs only where a user systemd can make a ;; scope with a quota (a scope of its own, outside the runner's cgroup). -(.sys.exec "command -v systemd-run >/dev/null || exit 0; [ $(nproc) -ge 3 ] || exit 0; systemd-run --user --scope -q -p CPUQuota=150% true 2>/dev/null || exit 0; o=$(env -u RAYFORCE_CORES systemd-run --user --scope -q -p CPUQuota=150% ./rayforce -f /tmp/rf_cpuset/info.rfl 2>/dev/null | tail -1); [ \"$o\" = \"$(nproc) 2\" ]") -- 0 +(.sys.exec "command -v systemd-run >/dev/null || exit 0; [ $(nproc) -ge 3 ] || exit 0; systemd-run --user --scope -q -p CPUQuota=150% true 2>/dev/null || exit 0; o=$(env -u RAYFORCE_CORES systemd-run --user --scope -q -p CPUQuota=150% ./rayforce -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); [ \"$o\" = \"$(nproc) 2\" ]") -- 0 -(.sys.exec "rm -rf /tmp/rf_cpuset") +(.sys.exec "rm -rf /tmp/rf_cpu_set") From cf21e6ea2734f61ca2e2eb2cb12ea69e578d7eb0 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Thu, 22 Oct 2026 22:57:00 +0300 Subject: [PATCH 4/5] fix(pool): find the CPU quota through the cgroup mount; read wide CPU masks The quota was read under /sys/fs/cgroup at the membership path of /proc/self/cgroup. Where a container mounts its own v2 subtree there, or a v1 cpu controller sits elsewhere, those paths miss the quota files and the pool is sized to every allowed CPU again. The hierarchy's mount point and the subtree it shows now come from /proc/self/mountinfo; the membership is translated through that root and its visible ancestors are read up to the mount point. The conventional mounts remain the fallback without mountinfo. On a kernel built for more CPUs than a cpu_set_t holds, sched_getaffinity rejects the fixed buffer with EINVAL however few CPUs are allowed, and the count fell back to every online CPU. It now retries with a larger buffer. .sys.info created the pool to report its size; it now reports the size a default pool would get while none exists, from the rule the pool uses. The quota check of a child moves to pool/cpu_quota_child, which skips visibly without a systemd user manager or ./rayforce and expects at most the size the same child gets in a scope without a quota. --- docs/docs/namespaces/sys.md | 2 +- src/core/platform.c | 164 ++++++++++++++++++++++++++----- src/core/platform.h | 6 ++ src/core/pool.c | 46 +++++---- src/core/pool.h | 3 + src/ops/system.c | 6 +- test/rfl/system/cpu_set_size.rfl | 6 +- test/test_platform.c | 106 ++++++++++++++++++++ test/test_pool.c | 49 +++++++++ 9 files changed, 335 insertions(+), 53 deletions(-) diff --git a/docs/docs/namespaces/sys.md b/docs/docs/namespaces/sys.md index bca88a9b7..77d3c3f79 100644 --- a/docs/docs/namespaces/sys.md +++ b/docs/docs/namespaces/sys.md @@ -78,7 +78,7 @@ Signature: `(.sys.build)`. Returns a dict with `version` (string) and `build-dat Signature: `(.sys.info)`. Returns `{cores: i64, page-size: i64, total-mem: i64, workers: i64, pid: i64, hostname: str}` on every platform (on Windows from `GetSystemInfo` / `GlobalMemoryStatusEx`). - `cores`: the CPUs this process may run on. On Linux that is its affinity mask, which `taskset`, a cpuset or `docker --cpuset-cpus` may narrow below the machine's CPU count. -- `workers`: the threads of the execution pool, main included (see [Running Several Instances](../guides/instances.md)). +- `workers`: the threads of the execution pool, main included; before the pool exists, the size it will get. `.sys.info` does not create the pool (see [Running Several Instances](../guides/instances.md)). ```lisp (.sys.info) diff --git a/src/core/platform.c b/src/core/platform.c index 1e6052a68..44506c186 100644 --- a/src/core/platform.c +++ b/src/core/platform.c @@ -422,13 +422,27 @@ uint32_t ray_thread_count(void) { * online count; the online count elsewhere. The mask read is the calling * thread's: the process's, unless an embedding application narrowed the * thread that first uses the pool (whose workers then inherit it). */ +/* Test hook: a mask buffer below this many bytes fails as a kernel whose + * mask is wider than the buffer fails it (EINVAL); 0 = none. */ +static size_t g_cpu_set_test_min; +void ray_cpu_set_test_min_bytes(size_t n) { g_cpu_set_test_min = n; } + uint32_t ray_cpu_set_count(void) { #if defined(RAY_OS_LINUX) - cpu_set_t set; - CPU_ZERO(&set); - if (sched_getaffinity(0, sizeof(set), &set) == 0) { - int n = CPU_COUNT(&set); - if (n > 0) return (uint32_t)n; + /* A kernel built for more CPUs than a cpu_set_t holds (1024) rejects + * the fixed buffer with EINVAL however few CPUs the process may use: + * retry with a larger one. */ + for (size_t sz = sizeof(cpu_set_t); sz <= ((size_t)1 << 20); sz *= 2) { + cpu_set_t* set = (cpu_set_t*)ray_sys_alloc(sz); + if (!set) break; + memset(set, 0, sz); + int rc, err = EINVAL; + if (sz < g_cpu_set_test_min) rc = -1; + else { rc = sched_getaffinity(0, sz, set); err = errno; } + int n = rc == 0 ? CPU_COUNT_S(sz, set) : 0; + ray_sys_free(set); + if (rc == 0) { if (n > 0) return (uint32_t)n; break; } + if (err != EINVAL) break; } #endif return ray_thread_count(); @@ -685,8 +699,17 @@ static int64_t cgroup_cpu_read(const char* dir, bool v2) { return (int64_t)((q * 1000 + per - 1) / per); } -/* Smallest quota over `root`/`cgpath` and every ancestor, as the memory - * limit walk above. */ +/* Test hook: a directory /proc/self/{cgroup,mountinfo} and the cgroup + * files of the quota reader are taken from (NULL: the real ones). */ +static const char* g_cg_test_root; +void ray_os_cgroup_test_root(const char* dir) { g_cg_test_root = dir; } + +static void cg_path(char* out, size_t n, const char* p) { + snprintf(out, n, "%s%s", g_cg_test_root ? g_cg_test_root : "", p); +} + +/* Smallest quota over `root`/`cgpath` and every ancestor up to `root`, as + * the memory limit walk above. */ static int64_t cgroup_cpu_walk(const char* root, const char* cgpath, bool v2) { char dir[512]; if (snprintf(dir, sizeof(dir), "%s%s", root, cgpath) >= (int)sizeof(dir)) return 0; @@ -703,19 +726,113 @@ static int64_t cgroup_cpu_walk(const char* root, const char* cgpath, bool v2) { return m; } -/* True when the comma-separated controller list names "cpu". */ -static bool cgroup_has_cpu(const char* ctrls) { - while (*ctrls) { - size_t n = strcspn(ctrls, ","); - if (n == 3 && strncmp(ctrls, "cpu", 3) == 0) return true; - ctrls += n; - if (*ctrls == ',') ctrls++; +/* True when the comma-separated list names `want`. */ +static bool cgroup_list_has(const char* list, const char* want) { + size_t wl = strlen(want); + while (*list) { + size_t n = strcspn(list, ","); + if (n == wl && strncmp(list, want, n) == 0) return true; + list += n; + if (*list == ',') list++; } return false; } +/* mountinfo escapes space, tab, newline and backslash as \ooo */ +static void cg_unescape(char* s) { + char* o = s; + for (char* p = s; *p; p++) { + if (p[0] == '\\' && p[1] >= '0' && p[1] <= '3' && p[2] >= '0' && p[2] <= '7' && + p[3] >= '0' && p[3] <= '7') { + *o++ = (char)((p[1] - '0') * 64 + (p[2] - '0') * 8 + (p[3] - '0')); + p += 3; + } else { + *o++ = *p; + } + } + *o = 0; +} + +/* Where a cgroup hierarchy is mounted: the v2 one (`ctrl` NULL), or the v1 + * one whose super options name `ctrl`. Its mount point, and the path in + * the hierarchy that the mount shows (its root: "/" unless a subtree was + * mounted). False when /proc/self/mountinfo lists none. */ +static bool cgroup_mount(const char* ctrl, char* mnt, size_t mn, char* root, size_t rn) { + char path[600]; + cg_path(path, sizeof(path), "/proc/self/mountinfo"); + FILE* f = fopen(path, "r"); + if (!f) return false; + char line[2048]; + bool found = false; + while (!found && fgets(line, sizeof(line), f)) { + /* "ID PARENT MAJ:MIN ROOT MOUNTPOINT OPTS [OPTIONAL...] - FSTYPE SOURCE SUPEROPTS" */ + line[strcspn(line, "\n")] = 0; + char* sep = strstr(line, " - "); + if (!sep) continue; + *sep = 0; + char* f_root = NULL; char* f_mnt = NULL; + char* save = NULL; + int i = 0; + for (char* t = strtok_r(line, " ", &save); t; t = strtok_r(NULL, " ", &save), i++) { + if (i == 3) f_root = t; + else if (i == 4) { f_mnt = t; break; } + } + if (!f_root || !f_mnt) continue; + char* fstype = strtok_r(sep + 3, " ", &save); + char* source = fstype ? strtok_r(NULL, " ", &save) : NULL; + char* super = source ? strtok_r(NULL, " ", &save) : NULL; + if (!fstype) continue; + if (ctrl ? (strcmp(fstype, "cgroup") == 0 && super && cgroup_list_has(super, ctrl)) + : strcmp(fstype, "cgroup2") == 0) { + cg_unescape(f_root); + cg_unescape(f_mnt); + found = snprintf(mnt, mn, "%s", f_mnt) < (int)mn && + snprintf(root, rn, "%s", f_root) < (int)rn; + } + } + fclose(f); + return found; +} + +/* The quota of a membership path `cgpath` ("" for the root) of the v2 + * hierarchy (`ctrl` NULL) or the v1 one carrying `ctrl`: the membership is + * translated through the mount's root to its directory under the mount + * point, and the walk stops at the mount point (ancestors outside the + * mount are not visible). Without mountinfo, the conventional mounts. */ +static int64_t cgroup_cpu_member(const char* ctrl, const char* v1list, const char* cgpath) { + char mnt[512], root[512], base[600]; + if (cgroup_mount(ctrl, mnt, sizeof(mnt), root, sizeof(root))) { + const char* rel = cgpath; + if (strcmp(root, "/") != 0) { + size_t rl = strlen(root); + if (strncmp(cgpath, root, rl) != 0 || (cgpath[rl] != 0 && cgpath[rl] != '/')) + return 0; /* membership outside the visible subtree */ + rel = cgpath + rl; + } + cg_path(base, sizeof(base), mnt); + return cgroup_cpu_walk(base, rel, ctrl == NULL); + } + if (!ctrl) { + cg_path(base, sizeof(base), "/sys/fs/cgroup"); + return cgroup_cpu_walk(base, cgpath, true); + } + /* v1 mounts the controller as the directory named by its list + * ("cpu,cpuacct"), usually with a "cpu" link to it */ + char dir[300]; + snprintf(dir, sizeof(dir), "/sys/fs/cgroup/%s", v1list); + cg_path(base, sizeof(base), dir); + int64_t v = cgroup_cpu_walk(base, cgpath, false); + if (!v) { + cg_path(base, sizeof(base), "/sys/fs/cgroup/cpu"); + v = cgroup_cpu_walk(base, cgpath, false); + } + return v; +} + int64_t ray_os_cgroup_cpu_quota(void) { - FILE* f = fopen("/proc/self/cgroup", "r"); + char path[600]; + cg_path(path, sizeof(path), "/proc/self/cgroup"); + FILE* f = fopen(path, "r"); if (!f) return 0; int64_t m = 0; char line[512]; @@ -729,16 +846,8 @@ int64_t ray_os_cgroup_cpu_quota(void) { cgpath[strcspn(cgpath, "\n")] = 0; if (strcmp(cgpath, "/") == 0) cgpath[0] = 0; int64_t v = 0; - if (*ctrls == 0) { - v = cgroup_cpu_walk("/sys/fs/cgroup", cgpath, true); - } else if (cgroup_has_cpu(ctrls)) { - /* v1 mounts the controller as the directory named by its list - * ("cpu,cpuacct"), usually with a "cpu" link to it */ - char root[300]; - snprintf(root, sizeof(root), "/sys/fs/cgroup/%s", ctrls); - v = cgroup_cpu_walk(root, cgpath, false); - if (!v) v = cgroup_cpu_walk("/sys/fs/cgroup/cpu", cgpath, false); - } + if (*ctrls == 0) v = cgroup_cpu_member(NULL, NULL, cgpath); + else if (cgroup_list_has(ctrls, "cpu")) v = cgroup_cpu_member("cpu", ctrls, cgpath); if (v > 0 && (m == 0 || v < m)) m = v; } fclose(f); @@ -764,6 +873,7 @@ bool ray_os_cgroup_stat_flush(const char* cg) { #else int64_t ray_os_cgroup_mem_limit(void) { return 0; } int64_t ray_os_cgroup_cpu_quota(void) { return 0; } +void ray_os_cgroup_test_root(const char* dir) { (void)dir; } bool ray_os_cgroup_mem_dir(char* buf, size_t n) { if (n) buf[0] = 0; return false; } bool ray_os_cgroup_stat_flush(const char* cg) { (void)cg; return false; } #endif @@ -1072,6 +1182,7 @@ uint32_t ray_thread_count(void) { return (uint32_t)si.dwNumberOfProcessors; } uint32_t ray_cpu_set_count(void) { return ray_thread_count(); } +void ray_cpu_set_test_min_bytes(size_t n) { (void)n; } /* Windows: no cheap topology read here — fall back to the logical count * (the SMT-aware default sizing is a POSIX-side optimization). */ @@ -1081,6 +1192,7 @@ uint32_t ray_physical_core_count(void) { int64_t ray_os_cgroup_mem_limit(void) { return 0; } int64_t ray_os_cgroup_cpu_quota(void) { return 0; } +void ray_os_cgroup_test_root(const char* dir) { (void)dir; } bool ray_os_cgroup_mem_dir(char* buf, size_t n) { if (n) buf[0] = 0; return false; } bool ray_os_cgroup_stat_flush(const char* cg) { (void)cg; return false; } @@ -1271,8 +1383,10 @@ ray_err_t ray_thread_join(ray_thread_t t) { uint32_t ray_thread_count(void) { return 1; } uint32_t ray_cpu_set_count(void) { return ray_thread_count(); } +void ray_cpu_set_test_min_bytes(size_t n) { (void)n; } int64_t ray_os_cgroup_mem_limit(void) { return 0; } int64_t ray_os_cgroup_cpu_quota(void) { return 0; } +void ray_os_cgroup_test_root(const char* dir) { (void)dir; } bool ray_os_cgroup_mem_dir(char* buf, size_t n) { if (n) buf[0] = 0; return false; } bool ray_os_cgroup_stat_flush(const char* cg) { (void)cg; return false; } static uint64_t cache_llc_probe(void) { return 0; } diff --git a/src/core/platform.h b/src/core/platform.h index 64c749bec..2803e0593 100644 --- a/src/core/platform.h +++ b/src/core/platform.h @@ -208,6 +208,9 @@ uint32_t ray_thread_count(void); /* The CPUs this process may run on: the affinity mask on Linux (taskset, a * cpuset or docker --cpuset-cpus narrow it), ray_thread_count elsewhere. */ uint32_t ray_cpu_set_count(void); +/* Test hook: a mask buffer below n bytes fails as on a kernel with a wider + * mask (EINVAL), so the retry with a larger buffer is exercised; 0 = none. */ +void ray_cpu_set_test_min_bytes(size_t n); /* Physical cores (SMT siblings collapsed); falls back to the logical * count when topology is unavailable. */ uint32_t ray_physical_core_count(void); @@ -224,6 +227,9 @@ int64_t ray_os_cgroup_mem_limit(void); * millicpus: the smallest v2 cpu.max or v1 cfs quota over its period on the * cgroup and its ancestors. 0 when none is set, or without cgroups. */ int64_t ray_os_cgroup_cpu_quota(void); +/* Test hook: read /proc/self/{cgroup,mountinfo} and the cgroup files of the + * quota from under `dir` (NULL: the real ones). */ +void ray_os_cgroup_test_root(const char* dir); /* The directory of the cgroup that sets that limit, into buf[n]; false when * no limit is set, or on platforms without cgroups. */ bool ray_os_cgroup_mem_dir(char* buf, size_t n); diff --git a/src/core/pool.c b/src/core/pool.c index 24a0e5975..e34f15a19 100644 --- a/src/core/pool.c +++ b/src/core/pool.c @@ -39,6 +39,25 @@ uint32_t ray_pool_default_total(void) { return n > 0 ? n : 1; } +/* Spawned workers of a default pool. The RAYFORCE_CORES env var overrides + * the size — the test harness sets it (see the Makefile `test` target) so + * neither the in-process runtime nor the server children it spawns via + * .sys.exec each create a thread per CPU on a many-core box. An explicit + * positive -c uses ray_pool_init_total() with exact sizing and bypasses + * this entirely. Otherwise every CPU the process may run on (its affinity + * mask: every online CPU unless taskset or a cpuset narrowed it), or fewer + * under a cgroup CPU quota; individual operations may bound their task + * count to their workload or memory budget. */ +static uint32_t pool_default_workers(void) { + const char* env = getenv("RAYFORCE_CORES"); + if (env && *env) { + long v = strtol(env, NULL, 10); + return (v > 0) ? (uint32_t)v : 0; + } + uint32_t ncpu = ray_pool_default_total(); + return (ncpu > 1) ? ncpu - 1 : 0; +} + /* Per-worker query-statistics slab (see core/qstats.h). Zero-initialised; * disabled (mode == 0) until a profiling/progress consumer turns it on. */ ray_qstats_t g_qstats; @@ -576,26 +595,7 @@ static ray_err_t ray_pool_create_impl(ray_pool_t* pool, uint32_t n_workers, } if (auto_size) { - /* Auto-size to ncpu-1. The RAYFORCE_CORES env var overrides this - * default worker count — the test harness sets it (see the Makefile - * `test` target) so neither the in-process runtime nor the server - * children it spawns via .sys.exec each create ncpu-1 threads on a - * many-core box. An explicit positive -c uses ray_pool_init_total() - * with exact sizing and bypasses this entirely; a non-test run with - * the var unset keeps the historical ncpu-1 default. */ - const char* env = getenv("RAYFORCE_CORES"); - if (env && *env) { - long v = strtol(env, NULL, 10); - n_workers = (v > 0) ? (uint32_t)v : 0; - } else { - /* Default to every CPU the process may run on (its affinity - * mask: every online CPU unless taskset or a cpuset narrowed - * it), or fewer under a cgroup CPU quota. Individual - * operations may bound their task count to their workload or - * memory budget. */ - uint32_t ncpu = ray_pool_default_total(); - n_workers = (ncpu > 1) ? ncpu - 1 : 0; - } + n_workers = pool_default_workers(); } pool->n_workers = n_workers; @@ -1063,6 +1063,12 @@ static void pool_trace_arm(void) { if (!armed && g_pool.trace) { armed = true; atexit(pool_trace_atexit); } } +uint32_t ray_pool_size(void) { + if (atomic_load_explicit(&g_pool_init_state, memory_order_acquire) == 2) + return ray_pool_total_workers(&g_pool); + return pool_default_workers() + 1; +} + ray_pool_t* ray_pool_get(void) { uint32_t state = atomic_load_explicit(&g_pool_init_state, memory_order_acquire); if (state == 2) return &g_pool; diff --git a/src/core/pool.h b/src/core/pool.h index bc32283a9..4420eeba5 100644 --- a/src/core/pool.h +++ b/src/core/pool.h @@ -278,6 +278,9 @@ void ray_pool_dispatch_n(ray_pool_t* pool, ray_pool_fn fn, void* ctx, uint32_t n /* Global pool lifecycle (lazy singleton) */ ray_pool_t* ray_pool_get(void); +/* Threads of the global pool, main included: its size when it exists, else + * the size a default one would get (nothing is created). */ +uint32_t ray_pool_size(void); /* Public pool init/destroy (moved from rayforce.h) */ ray_err_t ray_pool_init(uint32_t n_workers); diff --git a/src/ops/system.c b/src/ops/system.c index 820952e61..56c35c2b6 100644 --- a/src/ops/system.c +++ b/src/ops/system.c @@ -1601,11 +1601,11 @@ ray_t* ray_sysinfo_fn(ray_t** args, int64_t n) { vals = ray_list_append(vals, v3); ray_release(v3); #endif - /* workers: the threads of the execution pool, main included */ - ray_pool_t* pool = ray_pool_get(); + /* workers: the threads of the execution pool, main included (the size + * a default pool would get while none exists; none is created here) */ int64_t sw = ray_sym_intern("workers", 7); keys = ray_vec_append(keys, &sw); - ray_t* vw = make_i64(pool ? (int64_t)ray_pool_total_workers(pool) : 1); + ray_t* vw = make_i64((int64_t)ray_pool_size()); vals = ray_list_append(vals, vw); ray_release(vw); /* Process and host identity (#573). An embedded process previously had diff --git a/test/rfl/system/cpu_set_size.rfl b/test/rfl/system/cpu_set_size.rfl index f8e627673..ef37820cc 100644 --- a/test/rfl/system/cpu_set_size.rfl +++ b/test/rfl/system/cpu_set_size.rfl @@ -20,9 +20,7 @@ ;; -c within what the process may use: no warning (.sys.exec "set -- $(taskset -pc $$ | sed 's/.*: //' | tr ',' '\\n' | awk -F- '{ if (NF == 2) for (i = $1; i <= $2; i++) print i; else print $1 }'); [ $# -ge 2 ] || exit 0; q=$(env -u RAYFORCE_CORES ./rayforce -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); q=${q#* }; e=2; [ \"$q\" -lt 2 ] && e=$q; ! env -u RAYFORCE_CORES taskset -c $1,$2 ./rayforce -c $e -f /tmp/rf_cpu_set/info.rfl 2>&1 >/dev/null | grep -q warning") -- 0 -;; A CPU quota of 1.5 CPUs gives a pool of 2 threads, while cores still -;; counts every allowed CPU. Runs only where a user systemd can make a -;; scope with a quota (a scope of its own, outside the runner's cgroup). -(.sys.exec "command -v systemd-run >/dev/null || exit 0; [ $(nproc) -ge 3 ] || exit 0; systemd-run --user --scope -q -p CPUQuota=150% true 2>/dev/null || exit 0; o=$(env -u RAYFORCE_CORES systemd-run --user --scope -q -p CPUQuota=150% ./rayforce -f /tmp/rf_cpu_set/info.rfl 2>/dev/null | tail -1); [ \"$o\" = \"$(nproc) 2\" ]") -- 0 +;; The CPU quota case runs as pool/cpu_quota_child (test/test_pool.c), where +;; a missing systemd user manager shows as a skip. (.sys.exec "rm -rf /tmp/rf_cpu_set") diff --git a/test/test_platform.c b/test/test_platform.c index 6be7c5525..75641ec6c 100644 --- a/test/test_platform.c +++ b/test/test_platform.c @@ -37,6 +37,9 @@ #include #include #include +#ifdef RAY_OS_LINUX +#include +#endif /* ---- test_vm_alloc_free ------------------------------------------------ */ @@ -167,6 +170,107 @@ static test_result_t test_cgroup_stat_flush(void) { /* ---- Suite definition -------------------------------------------------- */ +#ifdef RAY_OS_LINUX +/* Writes `text` to root/rel, creating its directories. */ +static void cgq_put(const char* root, const char* rel, const char* text) { + char path[512]; + snprintf(path, sizeof(path), "%s/%s", root, rel); + for (char* p = path + strlen(root) + 1; *p; p++) + if (*p == '/') { *p = 0; (void)mkdir(path, 0700); *p = '/'; } + int fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600); + if (fd < 0) return; + ssize_t n = write(fd, text, strlen(text)); + (void)n; + close(fd); +} + +/* The quota of a fake /proc + /sys under a fresh directory. */ +static int64_t cgq_quota(const char* tag, const char* cgroup, const char* mountinfo, + const char* const* files) { + char root[128], cmd[300]; + snprintf(root, sizeof(root), "/tmp/rayforce_cgq_%d_%s", (int)getpid(), tag); + snprintf(cmd, sizeof(cmd), "rm -rf %s", root); + if (system(cmd) != 0) return -2; + (void)mkdir(root, 0700); + cgq_put(root, "proc/self/cgroup", cgroup); + if (mountinfo) cgq_put(root, "proc/self/mountinfo", mountinfo); + for (int i = 0; files[i]; i += 2) cgq_put(root, files[i], files[i + 1]); + ray_os_cgroup_test_root(root); + int64_t q = ray_os_cgroup_cpu_quota(); + ray_os_cgroup_test_root(NULL); + if (system(cmd) != 0) return -3; + return q; +} +#endif + +/* The CPU quota is found through the mount the hierarchy is visible at: + * its mount point and the subtree it shows (/proc/self/mountinfo), for v2 + * and for a v1 cpu controller wherever mounted; the smallest over the + * cgroup and its visible ancestors; the conventional mounts without + * mountinfo. */ +static test_result_t test_cgroup_cpu_quota(void) { + int64_t real = ray_os_cgroup_cpu_quota(); + TEST_ASSERT(real >= 0, "a quota is never negative"); +#ifdef RAY_OS_LINUX + const char* v2mi = "30 25 0:26 / /sys/fs/cgroup rw,nosuid shared:9 - cgroup2 cgroup2 rw\n"; + const char* const canon[] = { "sys/fs/cgroup/a/b/cpu.max", "150000 100000\n", NULL }; + TEST_ASSERT_EQ_I(cgq_quota("v2", "0::/a/b\n", v2mi, canon), 1500); + /* the v2 subtree /tenant mounted at /sys/fs/cgroup: /tenant/job lives at + * /sys/fs/cgroup/job, and the unlimited visible root does not count */ + const char* const sub[] = { "sys/fs/cgroup/job/cpu.max", "150000 100000\n", + "sys/fs/cgroup/cpu.max", "max 100000\n", NULL }; + TEST_ASSERT_EQ_I(cgq_quota("sub", "0::/tenant/job\n", + "30 25 0:26 /tenant /sys/fs/cgroup rw - cgroup2 cgroup2 rw\n", sub), 1500); + /* a membership outside the visible subtree has no readable quota */ + TEST_ASSERT_EQ_I(cgq_quota("out", "0::/other/job\n", + "30 25 0:26 /tenant /sys/fs/cgroup rw - cgroup2 cgroup2 rw\n", sub), 0); + /* a v1 cpu controller mounted off the conventional place */ + const char* const v1[] = { "cgroup/cpu/x/cpu.cfs_quota_us", "50000\n", + "cgroup/cpu/x/cpu.cfs_period_us", "100000\n", NULL }; + TEST_ASSERT_EQ_I(cgq_quota("v1", "4:cpu,cpuacct:/x\n", + "40 25 0:30 / /cgroup/cpu rw - cgroup cgroup rw,cpu,cpuacct\n", v1), 500); + /* a tighter ancestor wins (a quarter of a CPU, in millicpus) */ + const char* const anc[] = { "sys/fs/cgroup/a/b/cpu.max", "300000 100000\n", + "sys/fs/cgroup/a/cpu.max", "25000 100000\n", NULL }; + TEST_ASSERT_EQ_I(cgq_quota("anc", "0::/a/b\n", v2mi, anc), 250); + const char* const none[] = { "sys/fs/cgroup/a/b/cpu.max", "max 100000\n", NULL }; + TEST_ASSERT_EQ_I(cgq_quota("none", "0::/a/b\n", v2mi, none), 0); + /* no mountinfo: the conventional /sys/fs/cgroup */ + TEST_ASSERT_EQ_I(cgq_quota("nomi", "0::/a/b\n", NULL, canon), 1500); + TEST_ASSERT_EQ_I(ray_os_cgroup_cpu_quota(), real); +#else + TEST_ASSERT_EQ_I(real, 0); +#endif + PASS(); +} + +/* A kernel whose CPU mask is wider than a cpu_set_t rejects the fixed + * buffer (EINVAL); the count retries with a larger one instead of falling + * back to every online CPU. */ +static test_result_t test_cpu_set_count_wide_mask(void) { + uint32_t n = ray_cpu_set_count(); + TEST_ASSERT(n >= 1, "at least one CPU"); +#ifdef RAY_OS_LINUX + cpu_set_t set, two; + CPU_ZERO(&set); + TEST_ASSERT_EQ_I(sched_getaffinity(0, sizeof(set), &set), 0); + TEST_ASSERT_EQ_I(n, CPU_COUNT(&set)); + if (CPU_COUNT(&set) < 3) SKIP("needs at least 3 CPUs"); + /* two CPUs of the mask, so the fallback (every online CPU) differs */ + CPU_ZERO(&two); + for (int c = 0, k = 0; c < CPU_SETSIZE && k < 2; c++) + if (CPU_ISSET(c, &set)) { CPU_SET(c, &two); k++; } + int rc = sched_setaffinity(0, sizeof(two), &two); + ray_cpu_set_test_min_bytes(4 * sizeof(cpu_set_t)); + uint32_t wide = ray_cpu_set_count(); + ray_cpu_set_test_min_bytes(0); + (void)sched_setaffinity(0, sizeof(set), &set); + TEST_ASSERT_EQ_I(rc, 0); + TEST_ASSERT_EQ_I(wide, 2); +#endif + PASS(); +} + const test_entry_t platform_entries[] = { { "platform/vm_alloc_free", test_vm_alloc_free, NULL, NULL }, { "platform/vm_alloc_aligned", test_vm_alloc_aligned, NULL, NULL }, @@ -174,6 +278,8 @@ const test_entry_t platform_entries[] = { { "platform/thread_create_join", test_thread_create_join, NULL, NULL }, { "platform/cgroup_mem_limit", test_cgroup_mem_limit, NULL, NULL }, { "platform/cgroup_stat_flush", test_cgroup_stat_flush, NULL, NULL }, + { "platform/cgroup_cpu_quota", test_cgroup_cpu_quota, NULL, NULL }, + { "platform/cpu_set_count_wide_mask", test_cpu_set_count_wide_mask, NULL, NULL }, { NULL, NULL, NULL, NULL }, }; diff --git a/test/test_pool.c b/test/test_pool.c index 423a6ddc4..1e06d0ce9 100644 --- a/test/test_pool.c +++ b/test/test_pool.c @@ -1805,6 +1805,54 @@ static test_result_t test_auto_all_logical_cpus(void) { } #endif +#if defined(__linux__) +/* "cores workers" from a child rayforce run under `prefix`, or false. */ +static bool cq_child(const char* prefix, const char* script, long* cores, long* workers) { + char cmd[512], out[512]; + snprintf(cmd, sizeof(cmd), "env -u RAYFORCE_CORES %s ./rayforce -f %s 2>/dev/null", prefix, script); + FILE* p = popen(cmd, "r"); + if (!p) return false; + size_t n = fread(out, 1, sizeof(out) - 1, p); + out[n] = 0; + if (pclose(p) != 0) return false; + while (n && (out[n - 1] == '\n' || out[n - 1] == ' ')) out[--n] = 0; + char* last = strrchr(out, '\n'); + return sscanf(last ? last + 1 : out, "%ld %ld", cores, workers) == 2; +} +#endif + +/* A child under a CPU quota of 1.5 CPUs (a systemd user scope, as docker + * --cpus would set) sizes its default pool to 2 threads, or fewer when a + * quota above the scope is tighter: the same child in a scope without a + * quota gives that bound. cores still counts every allowed CPU. Skipped + * where no systemd user manager makes scopes, and without ./rayforce (the + * TSan job builds only the test binary). */ +static test_result_t test_cpu_quota_child(void) { +#if !defined(__linux__) + SKIP("cgroup CPU quotas are Linux-only"); +#else + if (access("./rayforce", X_OK) != 0) SKIP("needs the ./rayforce binary"); + if (system("command -v systemd-run >/dev/null 2>&1 && " + "systemd-run --user --scope -q true >/dev/null 2>&1") != 0) + SKIP("no systemd user manager to make a scope with a CPU quota"); + char script[128]; + snprintf(script, sizeof(script), "/tmp/rayforce_cq_%d.rfl", (int)getpid()); + FILE* f = fopen(script, "w"); + TEST_ASSERT_TRUE(f != NULL); + fputs("(set i (.sys.info))\n(println (format \"% %\" (at i 'cores) (at i 'workers)))\n", f); + fclose(f); + long c0 = 0, w0 = 0, c1 = 0, w1 = 0; + bool ok0 = cq_child("systemd-run --user --scope -q", script, &c0, &w0); + bool ok1 = cq_child("systemd-run --user --scope -q -p CPUQuota=150%", script, &c1, &w1); + unlink(script); + TEST_ASSERT_TRUE(ok0); + TEST_ASSERT_TRUE(ok1); + TEST_ASSERT_EQ_I(c1, c0); + TEST_ASSERT_EQ_I(w1, w0 < 2 ? w0 : 2); + PASS(); +#endif +} + #if defined(__linux__) #include /* The default pool follows the CPUs the process may run on: with the @@ -2510,6 +2558,7 @@ const test_entry_t pool_entries[] = { #if defined(__linux__) { "pool/auto_follows_cpu_set", test_auto_follows_cpu_set, NULL, NULL }, #endif + { "pool/cpu_quota_child", test_cpu_quota_child, NULL, NULL }, { "pool/parallel_sum", test_parallel_sum, NULL, NULL }, { "pool/dispatch_reclaims_worker_blocks", test_dispatch_reclaims_worker_blocks, NULL, NULL }, { "pool/parallel_add", test_parallel_add, NULL, NULL }, From e76dab11ef1bd83112251320f3993077d680dd80 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Thu, 22 Oct 2026 23:14:32 +0300 Subject: [PATCH 5/5] fix(pool): take the tightest CPU quota over every view of the cgroup A cgroup hierarchy can be mounted more than once (a subtree and the whole tree, or an unrelated subtree first). The quota reader stopped at the first mount of the hierarchy, readable or not; it now reads every mount whose subtree holds the membership and keeps the tightest quota visible through any of them. The tests snapshot the affinity mask with a fixed buffer, which a kernel with a wider mask rejects: they now skip there. The -c warning test expects the warning exactly when -c passes the default pool size rather than assuming fewer CPUs than 1023. --- src/core/platform.c | 111 +++++++++++++++++++++++-------------------- test/test_platform.c | 17 ++++++- test/test_pool.c | 3 +- test/test_repl.c | 6 ++- 4 files changed, 81 insertions(+), 56 deletions(-) diff --git a/src/core/platform.c b/src/core/platform.c index 44506c186..5ea49c6f2 100644 --- a/src/core/platform.c +++ b/src/core/platform.c @@ -753,65 +753,72 @@ static void cg_unescape(char* s) { *o = 0; } -/* Where a cgroup hierarchy is mounted: the v2 one (`ctrl` NULL), or the v1 - * one whose super options name `ctrl`. Its mount point, and the path in - * the hierarchy that the mount shows (its root: "/" unless a subtree was - * mounted). False when /proc/self/mountinfo lists none. */ -static bool cgroup_mount(const char* ctrl, char* mnt, size_t mn, char* root, size_t rn) { - char path[600]; - cg_path(path, sizeof(path), "/proc/self/mountinfo"); - FILE* f = fopen(path, "r"); - if (!f) return false; - char line[2048]; - bool found = false; - while (!found && fgets(line, sizeof(line), f)) { - /* "ID PARENT MAJ:MIN ROOT MOUNTPOINT OPTS [OPTIONAL...] - FSTYPE SOURCE SUPEROPTS" */ - line[strcspn(line, "\n")] = 0; - char* sep = strstr(line, " - "); - if (!sep) continue; - *sep = 0; - char* f_root = NULL; char* f_mnt = NULL; - char* save = NULL; - int i = 0; - for (char* t = strtok_r(line, " ", &save); t; t = strtok_r(NULL, " ", &save), i++) { - if (i == 3) f_root = t; - else if (i == 4) { f_mnt = t; break; } - } - if (!f_root || !f_mnt) continue; - char* fstype = strtok_r(sep + 3, " ", &save); - char* source = fstype ? strtok_r(NULL, " ", &save) : NULL; - char* super = source ? strtok_r(NULL, " ", &save) : NULL; - if (!fstype) continue; - if (ctrl ? (strcmp(fstype, "cgroup") == 0 && super && cgroup_list_has(super, ctrl)) - : strcmp(fstype, "cgroup2") == 0) { - cg_unescape(f_root); - cg_unescape(f_mnt); - found = snprintf(mnt, mn, "%s", f_mnt) < (int)mn && - snprintf(root, rn, "%s", f_root) < (int)rn; - } +/* One mount of a cgroup hierarchy from a /proc/self/mountinfo line: the + * v2 one (`ctrl` NULL), or the v1 one whose super options name `ctrl`. + * Its mount point, and the path in the hierarchy the mount shows (its + * root: "/" unless a subtree was mounted). The line is consumed. */ +static bool cgroup_mount_line(char* line, const char* ctrl, char** mnt, char** root) { + /* "ID PARENT MAJ:MIN ROOT MOUNTPOINT OPTS [OPTIONAL...] - FSTYPE SOURCE SUPEROPTS" */ + line[strcspn(line, "\n")] = 0; + char* sep = strstr(line, " - "); + if (!sep) return false; + *sep = 0; + char* f_root = NULL; char* f_mnt = NULL; + char* save = NULL; + int i = 0; + for (char* t = strtok_r(line, " ", &save); t; t = strtok_r(NULL, " ", &save), i++) { + if (i == 3) f_root = t; + else if (i == 4) { f_mnt = t; break; } } - fclose(f); - return found; + if (!f_root || !f_mnt) return false; + char* fstype = strtok_r(sep + 3, " ", &save); + char* source = fstype ? strtok_r(NULL, " ", &save) : NULL; + char* super = source ? strtok_r(NULL, " ", &save) : NULL; + if (!fstype) return false; + if (ctrl ? !(strcmp(fstype, "cgroup") == 0 && super && cgroup_list_has(super, ctrl)) + : strcmp(fstype, "cgroup2") != 0) + return false; + cg_unescape(f_root); + cg_unescape(f_mnt); + *mnt = f_mnt; + *root = f_root; + return true; } /* The quota of a membership path `cgpath` ("" for the root) of the v2 - * hierarchy (`ctrl` NULL) or the v1 one carrying `ctrl`: the membership is - * translated through the mount's root to its directory under the mount - * point, and the walk stops at the mount point (ancestors outside the - * mount are not visible). Without mountinfo, the conventional mounts. */ + * hierarchy (`ctrl` NULL) or the v1 one carrying `ctrl`. Every mount of + * the hierarchy whose subtree holds the membership is a view of it: the + * membership is translated through the view's root to its directory under + * the mount point, and the cgroup and its ancestors visible there (up to + * the mount point) are read. The tightest quota of all views counts, as + * each shows a part of the same ancestry. Without mountinfo, or with no + * mount of the hierarchy listed, the conventional mounts. */ static int64_t cgroup_cpu_member(const char* ctrl, const char* v1list, const char* cgpath) { - char mnt[512], root[512], base[600]; - if (cgroup_mount(ctrl, mnt, sizeof(mnt), root, sizeof(root))) { - const char* rel = cgpath; - if (strcmp(root, "/") != 0) { - size_t rl = strlen(root); - if (strncmp(cgpath, root, rl) != 0 || (cgpath[rl] != 0 && cgpath[rl] != '/')) - return 0; /* membership outside the visible subtree */ - rel = cgpath + rl; + char path[600], base[600]; + cg_path(path, sizeof(path), "/proc/self/mountinfo"); + FILE* f = fopen(path, "r"); + bool listed = false; + int64_t m = 0; + if (f) { + char line[2048]; + while (fgets(line, sizeof(line), f)) { + char* mnt; char* root; + if (!cgroup_mount_line(line, ctrl, &mnt, &root)) continue; + listed = true; + const char* rel = cgpath; + if (strcmp(root, "/") != 0) { + size_t rl = strlen(root); + if (strncmp(cgpath, root, rl) != 0 || (cgpath[rl] != 0 && cgpath[rl] != '/')) + continue; /* the membership is outside this view */ + rel = cgpath + rl; + } + cg_path(base, sizeof(base), mnt); + int64_t v = cgroup_cpu_walk(base, rel, ctrl == NULL); + if (v > 0 && (m == 0 || v < m)) m = v; } - cg_path(base, sizeof(base), mnt); - return cgroup_cpu_walk(base, rel, ctrl == NULL); + fclose(f); } + if (listed) return m; if (!ctrl) { cg_path(base, sizeof(base), "/sys/fs/cgroup"); return cgroup_cpu_walk(base, cgpath, true); diff --git a/test/test_platform.c b/test/test_platform.c index 75641ec6c..0175bdda9 100644 --- a/test/test_platform.c +++ b/test/test_platform.c @@ -224,6 +224,20 @@ static test_result_t test_cgroup_cpu_quota(void) { /* a membership outside the visible subtree has no readable quota */ TEST_ASSERT_EQ_I(cgq_quota("out", "0::/other/job\n", "30 25 0:26 /tenant /sys/fs/cgroup rw - cgroup2 cgroup2 rw\n", sub), 0); + /* several views of the hierarchy: the one holding the membership counts, + * whatever comes first in mountinfo */ + const char* const lim[] = { "sys/fs/cgroup/limited/cpu.max", "150000 100000\n", NULL }; + TEST_ASSERT_EQ_I(cgq_quota("multi", "0::/limited\n", + "30 25 0:26 /other /unrelated rw - cgroup2 cgroup2 rw\n" + "31 25 0:26 / /sys/fs/cgroup rw - cgroup2 cgroup2 rw\n", lim), 1500); + /* a narrow view and a full one: the ancestor quota only the full view + * shows is the tighter one */ + const char* const both[] = { "narrow/leaf/cpu.max", "150000 100000\n", + "sys/fs/cgroup/org/tenant/leaf/cpu.max", "150000 100000\n", + "sys/fs/cgroup/org/cpu.max", "50000 100000\n", NULL }; + TEST_ASSERT_EQ_I(cgq_quota("views", "0::/org/tenant/leaf\n", + "30 25 0:26 /org/tenant /narrow rw - cgroup2 cgroup2 rw\n" + "31 25 0:26 / /sys/fs/cgroup rw - cgroup2 cgroup2 rw\n", both), 500); /* a v1 cpu controller mounted off the conventional place */ const char* const v1[] = { "cgroup/cpu/x/cpu.cfs_quota_us", "50000\n", "cgroup/cpu/x/cpu.cfs_period_us", "100000\n", NULL }; @@ -253,7 +267,8 @@ static test_result_t test_cpu_set_count_wide_mask(void) { #ifdef RAY_OS_LINUX cpu_set_t set, two; CPU_ZERO(&set); - TEST_ASSERT_EQ_I(sched_getaffinity(0, sizeof(set), &set), 0); + if (sched_getaffinity(0, sizeof(set), &set) != 0) + SKIP("the kernel's CPU mask is wider than a cpu_set_t: no fixed-size snapshot"); TEST_ASSERT_EQ_I(n, CPU_COUNT(&set)); if (CPU_COUNT(&set) < 3) SKIP("needs at least 3 CPUs"); /* two CPUs of the mask, so the fallback (every online CPU) differs */ diff --git a/test/test_pool.c b/test/test_pool.c index 1e06d0ce9..d315062a7 100644 --- a/test/test_pool.c +++ b/test/test_pool.c @@ -1862,7 +1862,8 @@ static test_result_t test_cpu_quota_child(void) { static test_result_t test_auto_follows_cpu_set(void) { cpu_set_t before, two; CPU_ZERO(&before); - TEST_ASSERT_EQ_I(sched_getaffinity(0, sizeof(before), &before), 0); + if (sched_getaffinity(0, sizeof(before), &before) != 0) + SKIP("the kernel's CPU mask is wider than a cpu_set_t: no fixed-size snapshot"); if (CPU_COUNT(&before) < 3) SKIP("needs at least 3 CPUs"); CPU_ZERO(&two); for (int c = 0, k = 0; c < CPU_SETSIZE && k < 2; c++) diff --git a/test/test_repl.c b/test/test_repl.c index df624cf6d..1dcffc3f0 100644 --- a/test/test_repl.c +++ b/test/test_repl.c @@ -48,6 +48,7 @@ #include "core/ipc.h" #include "core/sock.h" #include "core/platform.h" +#include "core/pool.h" /* ray_pool_default_total: the -c warning test */ #include "core/runtime.h" #include "mem/sys.h" @@ -3059,11 +3060,12 @@ static test_result_t test_repl_cli_cores_are_total(void) { TEST_ASSERT_FMT(rc == 0, "-c 007 launcher probe failed: %d", rc); TEST_ASSERT_FMT(count == 7, "-c 007 created %ld total threads", count); - /* more threads than any runner has CPUs: obeyed, with a warning */ + /* obeyed; warned of exactly when it passes the CPUs the process may use + * (the child runs under the same mask and quota) */ rc = run_cli_task_count("1023", &count, &warned); TEST_ASSERT_FMT(rc == 0, "-c 1023 launcher probe failed: %d", rc); TEST_ASSERT_FMT(count == 1023, "-c 1023 created %ld total threads", count); - TEST_ASSERT_TRUE(warned); + TEST_ASSERT_TRUE(warned == (1023 > ray_pool_default_total())); PASS(); }