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..7e68d4b8c 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 (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 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..eafdb74b5 --- /dev/null +++ b/docs/docs/guides/instances.md @@ -0,0 +1,72 @@ +# 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 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 & +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`. + +**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: 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 take turns on its CPUs (rayforce prints a warning). + +`(.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, 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 & +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") +``` + +**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..ed5809764 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). + +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..77d3c3f79 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; 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) ;; => {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 c4e1390e6..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. This makes concurrent writes safe and enables incremental updates. +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 | |---|---| @@ -243,7 +245,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 +414,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: 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..5ea49c6f2 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,37 @@ 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). */ +/* 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) + /* 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(); +} + /* 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 +667,200 @@ 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); +} + +/* 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; + 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 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; +} + +/* 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; } + } + 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`. 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 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; + } + fclose(f); + } + if (listed) return m; + 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) { + 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]; + 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_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); + 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 +879,8 @@ 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 @@ -959,6 +1188,8 @@ uint32_t ray_thread_count(void) { GetSystemInfo(&si); 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). */ @@ -967,6 +1198,8 @@ 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; } @@ -1156,7 +1389,11 @@ 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 1379d6cfe..2803e0593 100644 --- a/src/core/platform.h +++ b/src/core/platform.h @@ -205,6 +205,12 @@ 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); +/* 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); @@ -217,6 +223,13 @@ 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); +/* 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 8dcd80bcd..e34f15a19 100644 --- a/src/core/pool.c +++ b/src/core/pool.c @@ -32,6 +32,32 @@ #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; +} + +/* 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; @@ -569,23 +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 online logical CPU. Individual operations may - * bound their task count to their workload or memory budget. */ - uint32_t ncpu = ray_thread_count(); - n_workers = (ncpu > 1) ? ncpu - 1 : 0; - } + n_workers = pool_default_workers(); } pool->n_workers = n_workers; @@ -1053,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 fbf0b956a..4420eeba5 100644 --- a/src/core/pool.h +++ b/src/core/pool.h @@ -278,12 +278,20 @@ 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); /* 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..56c35c2b6 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 (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((int64_t)ray_pool_size()); + 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/cpu_set_size.rfl b/test/rfl/system/cpu_set_size.rfl new file mode 100644 index 000000000..ef37820cc --- /dev/null +++ b/test/rfl/system/cpu_set_size.rfl @@ -0,0 +1,26 @@ +;; @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_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_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_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_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 + +;; 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/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_platform.c b/test/test_platform.c index 6be7c5525..0175bdda9 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,122 @@ 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); + /* 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 }; + 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); + 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 */ + 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 +293,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 d1085f39f..d315062a7 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,97 @@ 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__) +/* "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 + * 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); + 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++) + 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 @@ -2463,6 +2556,10 @@ 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/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 }, diff --git a/test/test_repl.c b/test/test_repl.c index 11acf4d01..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" @@ -3017,8 +3018,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 +3029,43 @@ 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); + /* 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 == (1023 > ray_pool_default_total())); PASS(); }