diff --git a/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst b/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst index 85b11dd2607..fa2f9638762 100644 --- a/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst +++ b/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst @@ -1,3 +1,5 @@ +.. _cs-JobWrapper: + Systems / WorkloadManagement / / JobWrapper - Sub-subsection ============================================================================ @@ -31,3 +33,49 @@ The options used to configure JobWrapper are showed in the table below: +----------------------+-------------------------------------------------+------------------------------+ | *OutputSandboxLimit* | Limit of sandbox output expressed in MB | OutputSandboxLimit = 10 | +----------------------+-------------------------------------------------+------------------------------+ +| *StopMargin* | Wall-clock seconds at the end of the batch | StopMargin = 300 | +| | slot reserved for uploading the outputs and | | +| | the logs. Deducted once, by the agent that | | +| | publishes the slot budget, so consumers of | | +| | /LocalSite/CPUTimeLeft must not deduct it | | +| | again | | ++----------------------+-------------------------------------------------+------------------------------+ + +Graceful stop +------------- + +A payload killed when its slot runs out loses whatever it had produced but not yet +written. An application that knows how to wind down can be signalled instead, early +enough to finish its current unit of work and write its output; the Watchdog stops it +anyway once the budget is spent, so these only buy it the chance. + +The same four options may be set per job in the JDL, under the same names, which takes +precedence. Setting them here is what reaches work already submitted, whose JDL is fixed. + ++------------------------+-------------------------------------------------+------------------------------+ +| **Name** | **Description** | **Example** | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigRegex* | Regular expression matched against the | StopSigRegex = Gauss | +| | command line of the payload processes, | | +| | naming the application to signal. Unset | | +| | means the mechanism is off | | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigNumber* | Signal to send to the matching processes. | StopSigNumber = 10 | +| | Default 2 (SIGINT) | | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigStartWork* | How much work a payload must have done | StopSigStartWork = 8370 | +| | before it is worth interrupting: below it | | +| | there is nothing to save. Defaults to | | +| | StopSigFinishWork, so a payload is never | | +| | stopped having done less work than stopping | | +| | it costs | | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigFinishWork* | How much computation the application needs | StopSigFinishWork = 16740 | +| | to wind down: finish the unit of work in | | +| | progress, close and write its output. In | | +| | CPU work rather than seconds, so that one | | +| | figure means the same on a fast and a slow | | +| | node; divided by this node's | | +| | CPUNormalizationFactor before use. The | | +| | graceful stop does nothing until it is set | | ++------------------------+-------------------------------------------------+------------------------------+ diff --git a/docs/source/AdministratorGuide/Systems/WorkloadManagement/allocatedTime.rst b/docs/source/AdministratorGuide/Systems/WorkloadManagement/allocatedTime.rst new file mode 100644 index 00000000000..72072919e3a --- /dev/null +++ b/docs/source/AdministratorGuide/Systems/WorkloadManagement/allocatedTime.rst @@ -0,0 +1,186 @@ +.. _allocated_time: + +========================================= +Allocated time management +========================================= + +A pilot holds a batch slot for a fixed length of time. Everything that has to happen in +that slot -- matching a job, running the payload, uploading what it produced -- has to fit, +and the pieces are sized by components that must agree on how much is left. This page +describes where that budget comes from, who may spend which part of it, and which options +control it. + +.. contents:: :local: + +Where the budget comes from +=========================== + +The pilot establishes two numbers before any job is matched: how fast this worker node is, +and how long the batch system will let it run. + +.. graphviz:: + + digraph seeding { + rankdir=LR; node [shape=box, fontsize=10]; edge [fontsize=9]; + bench [label="dirac-wms-cpu-normalization\n(benchmarks the node)"]; + norm [label="/LocalSite/\nCPUNormalizationFactor", shape=cylinder]; + batch [label="batch system\n(sacct, qstat, ...)"]; + cs [label="queue maxCPUTime\nin the CS", shape=note]; + queue [label="dirac-wms-get-queue-cpu-time"]; + left [label="/LocalSite/CPUTimeLeft", shape=cylinder]; + + bench -> norm; + batch -> queue [label="seconds left, if reachable"]; + cs -> queue [label="otherwise, minutes x 60"]; + norm -> left [label="multiplied in", style=dashed]; + queue -> left [label="seconds"]; + } + +``dirac-wms-get-queue-cpu-time`` tries three sources in order, and takes the first that +answers: + +#. **The batch system**, through the TimeLeft utility -- ``sacct``, ``qstat``, ``bjobs`` + and so on, selected by ``/LocalSite/BatchSystemInfo``. This is the only source that + knows how much of the slot has already been used. It is unavailable to a payload + running inside a container, which is why the figure is written to the configuration + rather than recomputed later. +#. **The queue's** ``maxCPUTime`` **in the CS**, under + ``/Resources/Sites///CEs//Queues//``. If no queue is identified, + the smallest ``maxCPUTime`` among that CE's queues is used instead. +#. ``/Resources/Computing/CEDefaults/MaxCPUTime``, a global fallback. + +.. warning:: + + These are not in the same unit. ``maxCPUTime`` on a queue is in **minutes** and is + multiplied by 60; ``CEDefaults/MaxCPUTime`` is in **seconds** and is not. A queue + configured as though it were seconds will hand out slots sixty times too long. + +The result, in seconds, is multiplied by ``CPUNormalizationFactor`` and stored as +``/LocalSite/CPUTimeLeft``. That field is therefore **CPU work**, not seconds: wall-clock +seconds times the power of the node. + +.. note:: + + ``CPUNormalizationFactor`` comes from a benchmark run on the node itself, divided by + ``Operations/JobScheduling/CPUNormalizationCorrection``. Everything downstream is + proportional to it, so an inaccurate benchmark scales the whole budget with it. + +Who may spend which part +======================== + +There is one writer of ``/LocalSite/CPUTimeLeft`` after the pilot: the JobAgent. Everything +else reads what it publishes. + +.. graphviz:: + + digraph budget { + rankdir=LR; node [shape=box, fontsize=10]; edge [fontsize=9]; + agent [label="JobAgent", style=filled, fillcolor="#e8e8e8"]; + cfg [label="/LocalSite/CPUTimeLeft", shape=cylinder]; + matcher [label="Matcher"]; + wd [label="Watchdog"]; + payload [label="payload\n(elastic sizing)"]; + + cfg -> agent [label="read once at\ninitialize()"]; + agent -> cfg [label="republished each cycle,\nStopMargin already off"]; + agent -> matcher [label="CPU work advertised\nby the CE"]; + cfg -> wd [label="read at job start"]; + cfg -> payload [label="read by the application"]; + } + +The JobAgent subtracts ``StopMargin`` once, when it first reads the slot. Everything +downstream therefore works from what a payload may actually *consume*, and none of them +needs to know a reserve exists. **Consumers must not subtract it again.** + +Then the end of a slot looks like this: + +.. graphviz:: + + digraph timeline { + rankdir=LR; node [shape=box, fontsize=10, width=1.4]; edge [fontsize=9]; + run [label="payload runs", style=filled, fillcolor="#dff0d8"]; + wind [label="winding down", style=filled, fillcolor="#fcf8e3"]; + up [label="uploads", style=filled, fillcolor="#f2dede"]; + end [label="slot ends", shape=plaintext]; + + run -> wind [label="StopSigNumber sent"]; + wind -> up [label="payload exits,\nor is killed"]; + up -> end; + } + +``StopSigFinishWork`` and ``StopMargin`` both reserve time at the end of the slot, but they +are not interchangeable: + +=================== =============================== ========================== ============== +Reserve Whose work When Unit +=================== =============================== ========================== ============== +StopSigFinishWork the payload's, finishing the before the payload exits CPU work + unit of work in progress +StopMargin the JobWrapper's, uploading after the payload exits wall clock + the outputs and the logs +=================== =============================== ========================== ============== + +They also differ in reach. ``StopMargin`` applies to every job and shrinks the budget that +the matcher and the payload are told about. ``StopSigFinishWork`` changes no budget at all: +it only decides how early to tap an opted-in payload on the shoulder, and comes out of that +payload's own share. + +The units differ because the work differs. Uploading a file takes the same wall clock on any +node, so ``StopMargin`` is in seconds. How much computation a payload has done, and how much +it needs to stop, are both questions about work, so the ``StopSig`` thresholds are in CPU work +and are divided by the node's ``CPUNormalizationFactor`` before use -- which means one figure means the same thing across +a heterogeneous fleet, where a figure in seconds would not: + +========= ========== ============== ========== +CPU work slow node typical node fast node +========= ========== ============== ========== +30 3 s 1 s 1 s +200 20 s 7 s 4 s +1000 100 s 36 s 20 s +8000 800 s 287 s 160 s +========= ========== ============== ========== + +The Watchdog stops the payload when the published budget is spent, whether or not it was +signalled first, so a payload that ignores the signal still cannot eat into the uploads. + +Choosing the values +=================== + +All of these live in ``/Systems/WorkloadManagement//JobWrapper`` and are listed in +:ref:`the configuration reference `. The same names may be set per job in the +JDL, which takes precedence -- but a JDL is fixed at submission, so only the CS reaches work +that is already in the system. + +``StopMargin`` + Measure it rather than guess. The JobWrapper reports ``Completing`` the moment the + payload exits and ``Done`` when it has finished uploading, so the span between those two + records in the JobLoggingDB is exactly what this has to cover. Take a high percentile + rather than the mean, and remember that jobs killed *during* post-processing never reach + ``Done`` and so are absent from the sample: the true requirement is if anything larger + than what is measured. + +``StopSigRegex`` + A regular expression matched against the command line of this job's payload processes. + It names the application to signal and nothing else: signalling the shells, wrappers and + container entrypoints that share the process tree is at best useless and at worst kills + the job outright. Only descendants of this JobWrapper are considered, so other jobs on + the same node are never affected however well their command lines match. Leaving it + unset disables the graceful stop. + +``StopSigNumber`` + Whichever signal the application handles. Verify it rather than assume: an application + that does not handle the signal is killed by it, losing exactly the work the mechanism + exists to save. + +``StopSigStartWork`` + How much work a payload must have done before it is worth interrupting. Below that there + is nothing to save, and a payload matched into a slot shorter than its own wind-down + would otherwise be signalled at once and report success having produced nothing. It + defaults to ``StopSigFinishWork``, so a payload is never stopped having done less work + than stopping it costs. + +``StopSigFinishWork`` + How much computation the application needs to wind down -- typically the cost of one unit + of work, plus what it takes to close and write its output. The graceful stop does nothing + until this is set. Err high: reserving too much costs a few units of work at the end of a + slot, while reserving too little costs everything the payload had not yet written. diff --git a/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst b/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst index 2b0ab2319a7..ec246a72cc0 100644 --- a/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst +++ b/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst @@ -87,5 +87,6 @@ The following sections add some detail for the WMS systems. JobsPriorities JobsMatching tagsAndJobs + allocatedTime multiProcessorJobs InputDataResolution diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py index 3860b6e3ea1..2d4970af692 100755 --- a/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py @@ -108,8 +108,15 @@ def initialize(self): if not result["OK"]: return result - # Read initial CPU work left from config (seeded by pilot via dirac-wms-get-queue-cpu-time) - self.initCPUWork = gConfig.getValue("/LocalSite/CPUTimeLeft", self.initCPUWork) + # This is the factor to convert raw CPU to Normalized units (based on the CPU Model) + self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", self.cpuPower) + + # Read initial CPU work left from config (seeded by pilot via dirac-wms-get-queue-cpu-time), + # less the tail of the slot the JobWrapper needs for the uploads. Taking it off here, once, + # leaves every later use of the slot working from what a payload may actually consume. + # An unbenchmarked node has no cpuPower to express the margin in, and reserves nothing. + stopMargin = gConfig.getValue("/Systems/WorkloadManagement/JobWrapper/StopMargin", 300) + self.initCPUWork = max(0.0, gConfig.getValue("/LocalSite/CPUTimeLeft", 0.0) - stopMargin * self.cpuPower) self.cpuWorkLeft = self.initCPUWork self.initTime = time.time() @@ -118,8 +125,6 @@ def initialize(self): self.pilotReference = gConfig.getValue("/LocalSite/PilotReference", self.pilotReference) self.defaultProxyLength = gConfig.getValue("/Registry/DefaultProxyLifeTime", self.defaultProxyLength) # Agent options - # This is the factor to convert raw CPU to Normalized units (based on the CPU Model) - self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", self.cpuPower) self.jobSubmissionDelay = self.am_getOption("SubmissionDelay", self.jobSubmissionDelay) self.fillingMode = self.am_getOption("FillingModeFlag", self.fillingMode) self.minimumCPUWork = self.am_getOption("MinimumTimeLeft", self.minimumCPUWork) @@ -404,6 +409,9 @@ def _computeCPUWorkLeft(self): via dirac-wms-get-queue-cpu-time). The elapsed wall-clock time is multiplied by the CPU power to get the consumed CPU work. + What is counted down is what a payload may consume: initialize() has already taken + the StopMargin off. + :return: cpu work left (wall-clock time left * cpu power) """ elapsed = time.time() - self.initTime diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py index 95ba8fb64d3..9d731156a39 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py @@ -8,9 +8,10 @@ from pathlib import Path import pytest +from unittest.mock import patch from DIRAC.Core.Security.X509Chain import X509Chain # pylint: disable=import-error -from DIRAC import S_ERROR, S_OK, gLogger +from DIRAC import S_ERROR, S_OK, gConfig, gLogger from DIRAC.Resources.Computing.ComputingElementFactory import ComputingElementFactory from DIRAC.Resources.Computing.test.Test_PoolComputingElement import badJobScript, jobScript from DIRAC.WorkloadManagementSystem.Agent.JobAgent import JobAgent @@ -175,6 +176,27 @@ def test__computeCPUWorkLeft(mocker, initCPUWork, cpuPower, elapsedSeconds, expe assert abs(result - expectedTimeLeft) < 10 +def test_the_upload_margin_is_reserved_once_for_the_whole_slot(mocker): + """Filling mode: two jobs matched into one slot share one margin, not one each. + + initialize() carves it out of initCPUWork and the cycles only count down from there, + so the second match is short by the time the first used and by nothing else. + """ + power, slotSeconds, margin = 27.9, 3600, 300 + mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobAgent.AgentModule.__init__") + jobAgent = JobAgent("Test", "Test1") + jobAgent.log = gLogger + jobAgent.cpuPower = power + # as initialize() computes it, from the pilot's /LocalSite/CPUTimeLeft + jobAgent.initCPUWork = max(0.0, slotSeconds * power - margin * power) + + jobAgent.initTime = time.time() + assert jobAgent._computeCPUWorkLeft() / power == pytest.approx(slotSeconds - margin, abs=2) + + jobAgent.initTime = time.time() - 600 # second match, ten minutes in + assert jobAgent._computeCPUWorkLeft() / power == pytest.approx(slotSeconds - 600 - margin, abs=2) + + @pytest.mark.parametrize( "cpuWorkLeft, fillingMode, expectedResult", [ diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py index 652c6ef938f..0f8800fd59f 100755 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py @@ -13,6 +13,7 @@ import getpass import math import os +import re import socket import time from pathlib import Path @@ -32,8 +33,9 @@ class Watchdog: ############################################################################# def __init__(self, pid, exeThread, spObject, jobCPUTime, memoryLimit=0, processors=1, jobArgs={}): """Constructor, takes system flag as argument.""" - self.stopSigStartSeconds = int(jobArgs.get("StopSigStartSeconds", 1800)) # 30 minutes - self.stopSigFinishSeconds = int(jobArgs.get("StopSigFinishSeconds", 1800)) # 30 minutes + self.jobArgs = jobArgs + self.stopSigStartWork = int(jobArgs.get("StopSigStartWork", 0)) # normalized units + self.stopSigFinishWork = int(jobArgs.get("StopSigFinishWork", 0)) # normalized units self.stopSigNumber = int(jobArgs.get("StopSigNumber", 2)) # SIGINT self.stopSigRegex = jobArgs.get("StopSigRegex", None) self.stopSigSent = False @@ -68,7 +70,6 @@ def __init__(self, pid, exeThread, spObject, jobCPUTime, memoryLimit=0, processo self.pollingTime = 10 # 10 seconds self.checkingTime = 30 * 60 # 30 minute period self.minCheckingTime = 20 * 60 # 20 mins - self.wallClockCheckSeconds = 5 * 60 # 5 minutes self.maxWallClockTime = 3 * 24 * 60 * 60 # e.g. 4 days self.jobPeekFlag = 1 # on / off self.minDiskSpace = 10 # MB @@ -78,12 +79,10 @@ def __init__(self, pid, exeThread, spObject, jobCPUTime, memoryLimit=0, processo self.minCPUWallClockRatio = 5 # ratio age self.nullCPULimit = 5 # After 5 sample times return null CPU consumption kill job self.checkCount = 0 - self.wallClockCheckCount = 0 self.nullCPUCount = 0 self.initialWallClockLeft = 0 self.wallClockLeft = 0 - self.stopMargin = 300 # seconds of wall-clock time to reserve for post-processing self.cpuPower = 1.0 self.processors = processors @@ -127,15 +126,13 @@ def initialize(self): ) self.checkingTime = self.minCheckingTime - self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", 1.0) - self.stopMargin = gConfig.getValue(self.section + "/StopMargin", self.stopMargin) + self.__resolveStopSigOptions() - # Read CPU work left from config (written by JobAgent) and convert to wall-clock seconds + self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", 1.0) + # CPU work left, as published by the JobAgent and so already net of the StopMargin + # it set aside for the uploads. Converted to the wall clock this counts down. cpuWorkLeft = gConfig.getValue("/LocalSite/CPUTimeLeft", 0) - if cpuWorkLeft and self.cpuPower: - self.initialWallClockLeft = cpuWorkLeft / self.cpuPower - else: - self.initialWallClockLeft = 0 + self.initialWallClockLeft = cpuWorkLeft / self.cpuPower if cpuWorkLeft and self.cpuPower else 0 return S_OK() @@ -176,14 +173,6 @@ def execute(self): self.log.info("Process to monitor has completed, Watchdog will exit.") return S_OK("Ended") - # WallClock checks every self.wallClockCheckSeconds, but only if StopSigRegex is defined in JDL - if ( - not self.stopSigSent - and self.stopSigRegex is not None - and (time.time() - self.initialValues["StartTime"]) > self.wallClockCheckSeconds * self.wallClockCheckCount - ): - self.wallClockCheckCount += 1 - # Check time left on every poll cycle (cheap: just reads wall-clock) if self.testTimeLeft: result = self.__checkTimeLeft() @@ -726,21 +715,122 @@ def __checkTimeLeft(self): The initial wall-clock time left is read from the local configuration at initialization (written by the JobAgent). A simple countdown is then used to determine the remaining time. - Returns S_ERROR when the remaining wall-clock time drops below the configurable StopMargin - (default: 300s), leaving enough time for post-processing (output upload, cleanup, etc.). + That budget is already net of the ``StopMargin`` the JobAgent set aside for the + uploads, so the payload is stopped as soon as it is exhausted. + + Returning S_ERROR gets the payload killed, which is too late to ask it to wind + down: that takes time, and what is left by then belongs to the uploads. Hence the + second, earlier threshold on the S_OK path. """ if not self.initialWallClockLeft: return S_OK("TimeLeft not available") - elapsed = time.time() - self.initialValues["StartTime"] - wallClockLeft = self.initialWallClockLeft - elapsed - self.wallClockLeft = wallClockLeft + self.wallClockLeft = self.initialWallClockLeft - (time.time() - self.initialValues["StartTime"]) - if wallClockLeft < self.stopMargin: + if self.wallClockLeft <= 0: return S_ERROR(JobMinorStatus.JOB_EXCEEDED_CPU) + self._askPayloadToStop() return S_OK() + ############################################################################# + def __resolveStopSigOptions(self): + """Take from the CS the graceful-stop settings the JDL did not give. + + A JDL is fixed when the job is submitted, so only the CS can reach work already in + the system. + """ + for option, attribute in ( + ("StopSigStartWork", "stopSigStartWork"), + ("StopSigFinishWork", "stopSigFinishWork"), + ("StopSigNumber", "stopSigNumber"), + ("StopSigRegex", "stopSigRegex"), + ): + if option not in self.jobArgs: + setattr(self, attribute, gConfig.getValue(f"{self.section}/{option}", getattr(self, attribute))) + + ############################################################################# + def _askPayloadToStop(self): + """Ask the payload to stop while there is still time for it to do so. + + A payload killed when its budget runs out loses whatever it had produced but not + written: an event generator stopped mid-event loses the whole file. A job that + knows how to wind down names its application in ``StopSigRegex`` and is sent + ``StopSigNumber`` early enough to finish and write. ``__checkTimeLeft`` remains the + backstop for one that ignores it. + + Always returns S_OK: failing to signal is not a reason to fail the job. + """ + if self.stopSigSent or not self.stopSigRegex or not self.stopSigFinishWork: + return S_OK() + if not self.initialWallClockLeft or not self.cpuPower: + return S_OK() + + # Both thresholds are in CPU work, not seconds: how much a payload has done, and how + # much it needs to stop, are questions about computation, and the same computation + # costs more wall clock on a slower node. + windDown = self.stopSigFinishWork / self.cpuPower + + # Nothing worth saving in a payload that has not done as much work as stopping it + # costs. Without this, one matched into a slot shorter than its wind-down would be + # signalled at once and report success having produced nothing. + startWork = self.stopSigStartWork or self.stopSigFinishWork + if (time.time() - self.initialValues["StartTime"]) < startWork / self.cpuPower: + return S_OK() + if self.wallClockLeft > windDown: + return S_OK() + + self.stopSigSent = True # sent once, whether or not the payload acts on it + signalled = self._signalPayload() + self.log.info( + "Asked the payload to stop while it still can", + f"signal={self.stopSigNumber} processes={signalled} " + f"wallClockLeft={self.wallClockLeft:.0f}s windDown={windDown:.0f}s", + ) + return S_OK() + + ############################################################################# + def _signalPayload(self): + """Send ``StopSigNumber`` to the descendants matching ``StopSigRegex``. + + Only descendants of this JobWrapper, so other jobs sharing the node are untouched + however well their command lines match. Unlike killChild this also spares the parent + -- the JobWrapper still has the outputs to upload -- and does not escalate to + SIGKILL: the point is to ask, and to leave time to answer. + + The regex picks the application out of the shells, wrappers and container entrypoints + sharing its tree, which must not be signalled in its place. + + :return: the number of processes signalled + :rtype: int + """ + try: + children = psutil.Process(self.wrapperPID).children(recursive=True) + except psutil.NoSuchProcess: + return 0 + + try: + matcher = re.compile(self.stopSigRegex) + except re.error as exc: + self.log.error("Not a usable StopSigRegex, signalling nothing", f"{self.stopSigRegex!r}: {exc}") + return 0 + + signalled = 0 + for child in children: + try: + if not matcher.search(" ".join(child.cmdline()) or child.name()): + continue + child.send_signal(self.stopSigNumber) + signalled += 1 + except psutil.NoSuchProcess: + continue + except psutil.Error as exc: + self.log.warn("Could not signal payload process", f"pid={child.pid}: {exc}") + + if not signalled: + self.log.warn("StopSigRegex matched no payload process", f"{self.stopSigRegex!r} of {len(children)}") + return signalled + ############################################################################# def __getUsageSummary(self): """Returns average load, memory etc. over execution of job thread""" diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py index 8501b567466..a329bfb3069 100644 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py @@ -1,11 +1,19 @@ """unit test for Watchdog.py""" import os +import signal +import subprocess +import sys import time from unittest.mock import MagicMock, patch +import psutil +import pytest + # sut from DIRAC.WorkloadManagementSystem.JobWrapper.Watchdog import Watchdog +SECTION = "/Systems/WorkloadManagement/JobWrapper" + mock_exeThread = MagicMock() mock_spObject = MagicMock() @@ -42,11 +50,11 @@ def test__performChecksFull(): class TestCheckTimeLeft: """Tests for the simplified wall-clock countdown time-left logic.""" - def _make_watchdog(self, initialWallClockLeft=0, stopMargin=300, cpuPower=10.0): + def _make_watchdog(self, initialWallClockLeft=0, cpuPower=10.0): + """A Watchdog holding a budget the JobAgent has already taken StopMargin off.""" pid = os.getpid() wd = Watchdog(pid, mock_exeThread, mock_spObject, 5000) wd.initialWallClockLeft = initialWallClockLeft - wd.stopMargin = stopMargin wd.cpuPower = cpuPower wd.initialValues = {"StartTime": time.time()} wd.testTimeLeft = 1 @@ -60,17 +68,27 @@ def test_time_left_not_available(self): def test_plenty_of_time_left(self): """When there's plenty of time, the check should pass.""" - wd = self._make_watchdog(initialWallClockLeft=3600, stopMargin=300) + wd = self._make_watchdog(initialWallClockLeft=3600) result = wd._Watchdog__checkTimeLeft() assert result["OK"] is True assert wd.wallClockLeft > 3000 - def test_below_stop_margin(self): - """When wall-clock left drops below stop margin, the check should fail.""" - wd = self._make_watchdog(initialWallClockLeft=3600, stopMargin=300) - # Pretend the job started 3500 seconds ago (only 100s left, below 300s margin) + def test_budget_not_exhausted(self): + """The payload keeps the whole published budget: it is already net of the margin. + + Under the old behaviour the Watchdog took another StopMargin off here and stopped + the job with 100 s still on its clock. + """ + wd = self._make_watchdog(initialWallClockLeft=3600) wd.initialValues["StartTime"] = time.time() - 3500 result = wd._Watchdog__checkTimeLeft() + assert result["OK"] is True + + def test_budget_exhausted(self): + """Once the published budget runs out, what is left of the slot is the reserve.""" + wd = self._make_watchdog(initialWallClockLeft=3600) + wd.initialValues["StartTime"] = time.time() - 3700 + result = wd._Watchdog__checkTimeLeft() assert not result["OK"] def test_time_left_updates_heartbeat_value(self): @@ -80,11 +98,10 @@ def test_time_left_updates_heartbeat_value(self): # wallClockLeft should be approximately 3600s assert wd.wallClockLeft > 3500 - def test_exact_stop_margin_boundary(self): - """When wall-clock left equals stop margin, the check should fail (< not <=).""" - wd = self._make_watchdog(initialWallClockLeft=1000, stopMargin=300) - # 699s elapsed → 301s left, which is not < 300 → should pass - wd.initialValues["StartTime"] = time.time() - 699 + def test_boundary(self): + """A second still on the clock is a second the payload may use.""" + wd = self._make_watchdog(initialWallClockLeft=1000) + wd.initialValues["StartTime"] = time.time() - 998 result = wd._Watchdog__checkTimeLeft() assert result["OK"] is True @@ -104,3 +121,233 @@ def test_initialize_reads_config(self, mock_gConfig): # 36000 / 10.0 = 3600 wall-clock seconds assert wd.initialWallClockLeft == 3600.0 + + +class TestConcurrentJobsInOneSlot: + """A PoolComputingElement runs several jobs side by side in the same batch slot.""" + + SLOT_SECONDS = 3600 + MARGIN = 300 + + @pytest.mark.parametrize( + "secondsIntoSlot, stillRunning", + [(SLOT_SECONDS - MARGIN - 60, True), (SLOT_SECONDS - MARGIN + 60, False)], + ) + def test_jobs_matched_at_different_times_stop_together(self, secondsIntoSlot, stillRunning): + """Both end at the slot's end, not at their own start plus a whole slot. + + The JobAgent publishes what is left *now*, so the job matched ten minutes later is + handed a budget shorter by exactly those ten minutes. + """ + now = time.time() + for startedAt in (0, 600): # two matches, ten minutes apart + wd = Watchdog(os.getpid(), MagicMock(), MagicMock(), 5000) + wd.initialWallClockLeft = self.SLOT_SECONDS - startedAt - self.MARGIN + wd.initialValues = {"StartTime": now - (secondsIntoSlot - startedAt)} + wd.testTimeLeft = 1 + assert wd._Watchdog__checkTimeLeft()["OK"] is stillRunning, f"job matched at {startedAt}s" + + +#: A job that knows how to wind down: signal it 10 min out, once it has run 5 min. +#: 16740 normalized units is 600 s of wall clock on a node benchmarked at 27.9. +GRACEFUL = {"StopSigRegex": "gauss", "StopSigFinishWork": 16740} + + +@patch.object(Watchdog, "_signalPayload") +class TestGracefulStop: + """Signalling the payload before its budget runs out. Opt-in through StopSigRegex.""" + + def _make_watchdog(self, jobArgs=None, initialWallClockLeft=3600, elapsed=0): + wd = Watchdog(os.getpid(), MagicMock(), MagicMock(), 5000, jobArgs=jobArgs or {}) + wd.initialWallClockLeft = initialWallClockLeft + wd.initialValues = {"StartTime": time.time() - elapsed} + wd.testTimeLeft = 1 + wd.exeThread.is_alive.return_value = True + wd.cpuPower = 27.9 + return wd + + @pytest.mark.parametrize( + "cpuPower, elapsed, expectSignalled", + [ + # 16740 units buys 600 s here, so 700 s of budget left is still comfortable... + (27.9, 2900, False), + (27.9, 3100, True), + # ...but on a node half as fast the same work needs 1200 s, and 700 s is not + # enough. A figure in seconds could not tell those two nodes apart. + (13.95, 2900, True), + ], + ) + def test_the_wind_down_is_budgeted_in_cpu_work(self, mock_signal, cpuPower, elapsed, expectSignalled): + """Finishing the unit of work in progress costs more wall clock on a slower node.""" + wd = self._make_watchdog(jobArgs=GRACEFUL, initialWallClockLeft=3600, elapsed=elapsed) + wd.cpuPower = cpuPower + wd._Watchdog__checkTimeLeft() + assert wd.stopSigSent is expectSignalled + + def test_signals_the_payload_with_budget_still_on_the_clock(self, mock_signal): + """3600 s of budget, 3000 gone: the payload is asked to stop, and told with 600 left. + + Those 600 s are the point: an application asked to stop needs time to stop, so the + ask cannot wait for the budget to run out. + """ + wd = self._make_watchdog(jobArgs=GRACEFUL, elapsed=3000) + wd.execute() + assert wd.stopSigSent is True + assert wd.wallClockLeft == pytest.approx(GRACEFUL["StopSigFinishWork"] / 27.9, abs=2) + mock_signal.assert_called_once_with() + + @pytest.mark.parametrize( + "jobArgs, initialWallClockLeft, elapsed, why", + [ + ({"StopSigFinishWork": 16740}, 3600, 3000, "no StopSigRegex, so opted out"), + ({**GRACEFUL, "StopSigRegex": ""}, 3600, 3000, "an empty regex opts out, it does not match everything"), + (GRACEFUL, 3600, 600, "plenty of budget left"), + (GRACEFUL, 300, 60, "barely started: less work done than stopping it would cost"), + (GRACEFUL, 0, 3000, "no budget published, so no deadline to act on"), + ], + ) + def test_stays_silent(self, mock_signal, jobArgs, initialWallClockLeft, elapsed, why): + wd = self._make_watchdog(jobArgs=jobArgs, initialWallClockLeft=initialWallClockLeft, elapsed=elapsed) + wd._Watchdog__checkTimeLeft() + assert wd.stopSigSent is False, why + mock_signal.assert_not_called() + + @pytest.mark.parametrize( + "elapsed, expectSignalled", + [ + # 16740 units is 600 s here, so the payload must have run 600 s to have done as + # much work as stopping it will cost. 500 s in, it has not, however near the end. + (500, False), + (700, True), + ], + ) + def test_the_start_guard_defaults_to_the_cost_of_stopping(self, mock_signal, elapsed, expectSignalled): + """Unset, StopSigStartWork is what winding down costs: below that there is nothing to save. + + Without it a payload matched into a slot shorter than its own wind-down would be + signalled immediately and report success having produced nothing. + """ + wd = self._make_watchdog(jobArgs=GRACEFUL, initialWallClockLeft=1000, elapsed=elapsed) + assert wd._Watchdog__checkTimeLeft()["OK"] is True + assert wd.stopSigSent is expectSignalled + + def test_signals_once_then_leaves_the_hard_stop_to_finish(self, mock_signal): + """Later polls must not keep signalling, and a payload that ignores it is still killed.""" + wd = self._make_watchdog(jobArgs=GRACEFUL, elapsed=3000) + for _ in range(5): + assert wd._Watchdog__checkTimeLeft()["OK"] is True + assert mock_signal.call_count == 1 + + wd.initialValues["StartTime"] = time.time() - 3700 + assert not wd._Watchdog__checkTimeLeft()["OK"] + + @pytest.mark.parametrize( + "cs, jobArgs, expected", + [ + ({}, {}, (None, 2, 0, 0)), # off, on the built-in defaults + ( # the CS alone turns it on: no JDL change, so productions already submitted are covered + { + f"{SECTION}/StopSigRegex": "Gauss", + f"{SECTION}/StopSigNumber": 10, + f"{SECTION}/StopSigStartWork": 8370, + f"{SECTION}/StopSigFinishWork": 5580, + }, + {}, + ("Gauss", 10, 8370, 5580), + ), + # a job that knows better than the site-wide setting still gets its way + ( + {f"{SECTION}/StopSigRegex": "Gauss"}, + {"StopSigRegex": "myApp", "StopSigNumber": "1"}, + ("myApp", 1, 0, 0), + ), + # and can opt out of it + ({f"{SECTION}/StopSigRegex": "Gauss"}, {"StopSigRegex": ""}, ("", 2, 0, 0)), + ], + ) + def test_settings_come_from_the_jdl_then_the_cs(self, mock_signal, cs, jobArgs, expected): + with patch("DIRAC.WorkloadManagementSystem.JobWrapper.Watchdog.gConfig") as watchdogCfg: + watchdogCfg.getValue.side_effect = lambda key, default=None: cs.get(key, default) + wd = Watchdog(os.getpid(), MagicMock(), MagicMock(), 5000, jobArgs=jobArgs) + wd.calibrate() + wd.initialize() + + assert ( + wd.stopSigRegex, + wd.stopSigNumber, + wd.stopSigStartWork, + wd.stopSigFinishWork, + ) == expected + + +class TestSignalPayload: + """Which processes the signal reaches, and which it must not.""" + + @staticmethod + def _standIn(*payloads): + """A stand-in JobWrapper that spawns the given payloads and outlives them.""" + spawns = "; ".join(f"subprocess.Popen({list(p)!r})" for p in payloads) + wrapper = subprocess.Popen([sys.executable, "-c", f"import subprocess, time; {spawns}; time.sleep(60)"]) + deadline = time.time() + 30 + children = [] + while len(children) < len(payloads) and time.time() < deadline: + children = psutil.Process(wrapper.pid).children(recursive=True) + time.sleep(0.05) + assert len(children) == len(payloads), "the stand-in never spawned its payloads" + return wrapper, {" ".join(c.cmdline()): c for c in children} + + @staticmethod + def _watchdog(pid, nameRegex): + wd = Watchdog(pid, MagicMock(), MagicMock(), 5000, jobArgs={"StopSigRegex": nameRegex}) + wd.stopSigNumber = int(signal.SIGTERM) + return wd + + @staticmethod + def _stopped(process): + deadline = time.time() + 30 + while process.is_running() and process.status() != psutil.STATUS_ZOMBIE and time.time() < deadline: + time.sleep(0.05) + return not process.is_running() or process.status() == psutil.STATUS_ZOMBIE + + def test_signals_the_named_application_only(self): + """Not the parent, which still has the outputs to upload, nor its siblings.""" + wrapper, payloads = self._standIn(["sleep", "61"], ["sleep", "62"]) + try: + assert self._watchdog(wrapper.pid, "sleep 61")._signalPayload() == 1 + assert self._stopped(payloads["sleep 61"]), "the named payload ignored the signal" + assert payloads["sleep 62"].is_running(), "a process the regex did not name was signalled" + assert wrapper.poll() is None, "the wrapper was signalled too" + finally: + wrapper.kill() + wrapper.wait(timeout=30) + + def test_leaves_other_jobs_on_the_node_alone(self): + """A node runs many jobs whose command lines all match. Only our own tree is ours. + + The search starts from this JobWrapper's pid, so another job's payload is out of + reach however well it matches. + """ + ours, ourPayloads = self._standIn(["sleep", "61"]) + theirs, theirPayloads = self._standIn(["sleep", "61"]) + try: + assert self._watchdog(ours.pid, "sleep 61")._signalPayload() == 1 + assert self._stopped(ourPayloads["sleep 61"]) + assert theirPayloads["sleep 61"].is_running(), "signalled another job's payload" + finally: + for wrapper in (ours, theirs): + wrapper.kill() + wrapper.wait(timeout=30) + + @pytest.mark.parametrize("nameRegex", ["no-such-payload", "[unclosed"]) + def test_signals_nothing_it_cannot_name(self, nameRegex): + """A regex matching nothing, or not compiling, must not fall back to signalling all.""" + wrapper, payloads = self._standIn(["sleep", "61"]) + try: + assert self._watchdog(wrapper.pid, nameRegex)._signalPayload() == 0 + assert payloads["sleep 61"].is_running() + finally: + wrapper.kill() + wrapper.wait(timeout=30) + + def test_an_exited_payload_is_not_an_error(self): + assert self._watchdog(999999999, "anything")._signalPayload() == 0