You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by sb...@apache.org on 2026/07/31 19:57:22 UTC
(tooling-trusted-releases) 01/01: Set worker CPU quota as a new soft limit per task
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
commit 2b1566c49191c5b91e87d1fe0fbc7dc82d6c61d5
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Fri Jul 31 20:48:31 2026 +0100
Set worker CPU quota as a new soft limit per task
---
atr/docs/resource-management.md | 2 +-
atr/util.py | 13 +++++++++++++
atr/worker.py | 16 ++--------------
tests/unit/test_resource_limits.py | 16 ++++++++++++++++
4 files changed, 32 insertions(+), 15 deletions(-)
diff --git a/atr/docs/resource-management.md b/atr/docs/resource-management.md
index ed90164c..3e4e1b8f 100644
--- a/atr/docs/resource-management.md
+++ b/atr/docs/resource-management.md
@@ -27,7 +27,7 @@ The five main places where ATR consumes resources are: 1. the task system, 2. we
## Worker pool
-The manager ([`manager`](/ref/atr/manager.py)) tracks four to eight workers, each of which works on 8 to 16 tasks before exiting. Workers claim the oldest known task. The queue is global: there is no user quota, fairness, etc. Workers request, but are not guaranteed, a lifetime limit of 300 processor seconds (which children inherit, with separate accounting), per worker rather than per task. Each worker also runs a watchdog thread, polling every 500ms, which kills the worker's own proces [...]
+The manager ([`manager`](/ref/atr/manager.py)) tracks four to eight workers, each of which works on 8 to 16 tasks before exiting. Workers claim the oldest known task. The queue is global: there is no user quota, fairness, etc. Workers arm a soft CPU limit of 300 processor seconds at the start of every task, so the budget is per task rather than per worker lifetime; children inherit the limit, each with its own fresh accounting. The limit is enforced by the kernel as an uncaught SIGXCPU, [...]
After 300s spent on a task, the manager fails it and SIGTERMs the worker. It does not confirm the effect, and does not send the SIGTERM to the worker's children. The manager drops the worker, replacing only when there are fewer than four, so four to eight is the tracked pool, not a cap on live processes: a worker that hasn't acted on its SIGTERM runs on beside the pool, may overwrite its failed verdict if it finishes, and may run another task twice via the untracked reset.
diff --git a/atr/util.py b/atr/util.py
index ddd78709..00cc645c 100644
--- a/atr/util.py
+++ b/atr/util.py
@@ -32,6 +32,7 @@ import json
import os
import pathlib
import re
+import resource
import socket
import ssl
import tempfile
@@ -619,6 +620,18 @@ async def content_list(
)
+def cpu_limit_arm(budget_seconds: int) -> None:
+ times = os.times()
+ soft = int(times.user + times.system) + budget_seconds + 1
+ hard = resource.getrlimit(resource.RLIMIT_CPU)[1]
+ if hard != resource.RLIM_INFINITY:
+ soft = min(soft, hard)
+ try:
+ resource.setrlimit(resource.RLIMIT_CPU, (soft, hard))
+ except ValueError as e:
+ log.warning(f"Could not set CPU time limit: {e}")
+
+
async def create_hard_link_clone(
source_dir: safe.StatePath,
dest_dir: safe.StatePath,
diff --git a/atr/worker.py b/atr/worker.py
index c6621002..a5ed461f 100644
--- a/atr/worker.py
+++ b/atr/worker.py
@@ -29,7 +29,6 @@ import inspect
import json
import logging
import os
-import resource
import signal
import threading
import time
@@ -97,7 +96,7 @@ def main() -> None:
for s in (signal.SIGTERM, signal.SIGINT):
signal.signal(s, lambda signum, frame: asyncio.create_task(_handle_signal(signum)))
- _worker_resources_limit_set()
+ util.cpu_limit_arm(_CPU_LIMIT_SECONDS)
threading.Thread(target=_memory_watchdog_run, daemon=True).start()
async def _start() -> None:
@@ -459,6 +458,7 @@ async def _worker_loop_run() -> None:
if task:
task_id, task_type, task_args, asf_uid = task
log.add_context(task_id=task_id, task_type=task_type, asf_uid=asf_uid)
+ util.cpu_limit_arm(_CPU_LIMIT_SECONDS)
await _task_process(task_id, task_type, task_args, asf_uid)
processed += 1
# Only process max_to_process tasks and then exit
@@ -475,18 +475,6 @@ async def _worker_loop_run() -> None:
await asyncio.sleep(1)
-def _worker_resources_limit_set() -> None:
- """Set CPU limits for this process."""
- # TODO: https://github.com/apache/tooling-trusted-releases/issues/411
- # # Set CPU time limit
- try:
- resource.setrlimit(resource.RLIMIT_CPU, (_CPU_LIMIT_SECONDS, _CPU_LIMIT_SECONDS))
- log.info(f"Set CPU time limit to {_CPU_LIMIT_SECONDS} seconds")
- except ValueError as e:
- log.warning(f"Could not set CPU time limit: {e}")
- return
-
-
if __name__ == "__main__":
log.info("Starting ATR worker...")
try:
diff --git a/tests/unit/test_resource_limits.py b/tests/unit/test_resource_limits.py
index 31540199..b60669bf 100644
--- a/tests/unit/test_resource_limits.py
+++ b/tests/unit/test_resource_limits.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import resource
import subprocess
import sys
import time
@@ -24,6 +25,21 @@ import psutil
import atr.util as util
+def test_cpu_limit_arm_rearms_soft_limit() -> None:
+ before = resource.getrlimit(resource.RLIMIT_CPU)
+ try:
+ util.cpu_limit_arm(300)
+ soft_one, hard_one = resource.getrlimit(resource.RLIMIT_CPU)
+ util.cpu_limit_arm(600)
+ soft_two, hard_two = resource.getrlimit(resource.RLIMIT_CPU)
+ finally:
+ resource.setrlimit(resource.RLIMIT_CPU, before)
+ assert hard_one == before[1]
+ assert hard_two == before[1]
+ assert soft_one >= 301
+ assert soft_two > soft_one
+
+
def test_process_tree_rss_counts_children() -> None:
child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
try:
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org