Warn at startup when worker or cascade limits are far past the cores

More than four times os.cpu_count() logs one warning; the value is kept.

backend/fluksio/flow/resources.py

@@ -65,12 +65,37 @@ ENGINE_RAM_RESERVE_MB = 1024
 #: Where a container finds its own memory limit (cgroup v2).
 CGROUP_MEMORY_MAX = Path("/sys/fs/cgroup/memory.max")
 
+#: How many times the core count a concurrency limit may reach before startup
+#: warns. Above one on purpose: a node waiting on the network holds no core.
+OVERSUBSCRIBE_FACTOR = 4
+
 
 def machine_cpus() -> int:
     """Cores a node may be given here, when nobody said."""
     return max(1, (os.cpu_count() or 1) - ENGINE_RESERVE)
 
 
+def warn_oversubscribed(limits: dict[str, int]) -> None:
+    """Warn, never clamp, when a concurrency limit is far past the cores.
+
+    A high number can be right for nodes that wait on devices, but one far
+    past the machine is more often a zero too many.
+    """
+    cores = os.cpu_count() or 1
+    high = [
+        f"{name}={value}"
+        for name, value in limits.items()
+        if value > OVERSUBSCRIBE_FACTOR * cores
+    ]
+    if high:
+        logger.warning(
+            "Concurrency above %dx this machine's %d cores: %s",
+            OVERSUBSCRIBE_FACTOR,
+            cores,
+            ", ".join(high),
+        )
+
+
 def machine_gpus() -> int:
     """Cards a node may be given here, when nobody said.
 

backend/fluksio/main.py

@@ -49,6 +49,7 @@ from fluksio.flow.resources import (
     fair_share_env,
     machine_gpus,
     machine_ram_mb,
+    warn_oversubscribed,
 )
 from fluksio.flow.runs import RUN_STATE_TTL, RunService, sweep_artifacts
 from fluksio.flow.secrets import init_secrets
@@ -372,6 +373,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
             accountant.ram_mb,
             settings.FLOW_WORKER_MAX_RSS_MB,
         )
+        warn_oversubscribed(
+            {
+                "FLOW_MAX_WORKERS": settings.FLOW_MAX_WORKERS,
+                "FLOW_MAX_CASCADES": settings.FLOW_MAX_CASCADES,
+            }
+        )
         watchdog = LoopWatchdog(event_bus)
         app.state.watchdog = watchdog
 

backend/tests/flow/test_resources.py

@@ -20,6 +20,7 @@ from fluksio.flow.resources import (
     fair_share_env,
     machine_gpus,
     machine_ram_mb,
+    warn_oversubscribed,
 )
 from fluksio.flow.schemas import NodeDef, Resources
 
@@ -45,6 +46,16 @@ def test_the_cards_are_counted_from_the_devices_the_driver_makes(monkeypatch):
     assert machine_gpus() == 0
 
 
+def test_a_limit_far_past_the_cores_is_warned_about(monkeypatch, caplog):
+    """Said once, never clamped: nodes waiting on devices can want a high one."""
+    monkeypatch.setattr(os, "cpu_count", lambda: 2)
+    warn_oversubscribed({"FLOW_MAX_WORKERS": 8, "FLOW_MAX_CASCADES": 9})
+
+    assert [r.getMessage() for r in caplog.records] == [
+        "Concurrency above 4x this machine's 2 cores: FLOW_MAX_CASCADES=9"
+    ]
+
+
 def test_memory_is_the_containers_limit_where_that_is_lower(monkeypatch, tmp_path):
     """The host's total is memory a container would be killed for using."""
     from fluksio.flow import resources

docs/reference/configuration.md

@@ -137,7 +137,9 @@ Set `TZ` in `.env` for schedules that should follow local time, such as
 | `OBS_RETENTION_DAYS` | `30` | retention for metrics, events and run records |
 
 The three concurrency limits must be at least `1`. They also have CLI flags:
-`--max-workers`, `--max-cascades` and `--max-runs`. Resource flags are `--gpus`,
+`--max-workers`, `--max-cascades` and `--max-runs`. When `FLOW_MAX_WORKERS` or
+`FLOW_MAX_CASCADES` is more than four times the machine's core count, the
+engine logs a warning at startup and uses the value anyway. Resource flags are `--gpus`,
 `--ram` and `--worker-max-rss`.
 
 Nodes with declared resources are scheduled against the available budget.