DIRACGrid · aldbr · Aug 25, 2026
@@ -1,3 +1,5 @@
+.. _cs-JobWrapper:
+
 Systems / WorkloadManagement / <INSTANCE> / 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      |                              |
++------------------------+-------------------------------------------------+------------------------------+
@@ -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/<grid>/<site>/CEs/<ce>/Queues/<queue>/``. 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/<INSTANCE>/JobWrapper`` and are listed in
+:ref:`the configuration reference <cs-JobWrapper>`. 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.
@@ -87,5 +87,6 @@ The following sections add some detail for the WMS systems.
    JobsPriorities
    JobsMatching
    tagsAndJobs
+   allocatedTime
    multiProcessorJobs
    InputDataResolution
@@ -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

@@ -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",
     [