New poll rate control mechanism

Author: vkarak

reframe/frontend/cli.py

@@ -807,6 +807,38 @@ def main():
         help='Timeout for advancing the pipeline',
         type=float
     )
+    argparser.add_argument(
+        dest='poll_randomize_ms',
+        envvar='RFM_POLL_RANDOMIZE_MS',
+        configvar='general/poll_randomize_ms',
+        action='store',
+        type=typ.List[int],
+        help='Randomize the sleep interval between polls'
+    )
+    argparser.add_argument(
+        dest='poll_rate_decay',
+        envvar='RFM_POLL_RATE_DECAY',
+        configvar='general/poll_rate_decay',
+        action='store',
+        type=float,
+        help='Poll rate decay'
+    )
+    argparser.add_argument(
+        dest='poll_rate_max',
+        envvar='RFM_POLL_RATE_MAX',
+        configvar='general/poll_rate_max',
+        action='store',
+        type=float,
+        help='Maximum poll rate'
+    )
+    argparser.add_argument(
+        dest='poll_rate_min',
+        envvar='RFM_POLL_RATE_MIN',
+        configvar='general/poll_rate_min',
+        action='store',
+        type=float,
+        help='Minimum poll rate'
+    )
     argparser.add_argument(
         dest='remote_detect',
         envvar='RFM_REMOTE_DETECT',

reframe/frontend/executors/policies.py

@@ -4,14 +4,15 @@
 # SPDX-License-Identifier: BSD-3-Clause
 
 import contextlib
-import math
+import random
 import sys
 import time
 
 import reframe.core.runtime as rt
 import reframe.utility as util
 import reframe.utility.color as color
-from reframe.core.exceptions import (FailureLimitError,
+from reframe.core.exceptions import (ConfigError,
+                                     FailureLimitError,
                                      RunSessionTimeout,
                                      TaskExit)
 from reframe.core.logging import getlogger, level_from_str
@@ -69,33 +70,83 @@ def _print_pipeline_timings(task):
 
 
 class _PollController:
-    SLEEP_MIN = 0.1
-    SLEEP_MAX = 10
-    SLEEP_INC_RATE = 1.1
+    def _validate_poll_params(self):
+        def _check_positive(x, name):
+            if x <= 0:
+                raise ConfigError(f'{name} must be a positive number')
+
+        _check_positive(self._poll_rate_min, 'minimum poll rate')
+        _check_positive(self._poll_rate_max, 'maximum poll rate')
+        if self._poll_rate_max < self._poll_rate_min:
+            raise ConfigError('maximum poll rate must be greater or equal to '
+                              'minimum poll rate')
+
+        if self._poll_rate_decay < 0 or self._poll_rate_decay > 1:
+            raise ConfigError('poll rate decay must be in range [0,1]')
+
+        if self._poll_randomize_range_ms is not None:
+            left, right = self._poll_randomize_range_ms
+            if left > 0:
+                raise ConfigError('left boundary of poll randomization range '
+                                  'must be a negative integer or zero')
+
+            if right < 0:
+                raise ConfigError('right boundary of poll randomization range '
+                                  'must be a positive integer or zero')
 
     def __init__(self):
-        self._num_polls = 0
-        self._sleep_duration = None
-        self._t_init = None
-
-    def reset_snooze_time(self):
-        self._sleep_duration = self.SLEEP_MIN
+        get_option = rt.runtime().get_option
+        self._poll_rate_max = get_option('general/0/poll_rate_max')
+        self._poll_rate_min = get_option('general/0/poll_rate_min')
+        self._poll_rate_decay = get_option('general/0/poll_rate_decay')
+        self._poll_randomize_range_ms = get_option(
+            'general/0/poll_randomize_ms'
+        )
+        self._poll_count_total = 0
+        self._poll_count_interval = 0
+        self._validate_poll_params()
+        self._t_start = None
+        self._t_last_poll = None
+        self._desired_poll_rate = self._poll_rate_max
+
+    def reset_poll_rate(self):
+        getlogger().debug2('[P] reset poll rate')
+        self._poll_count_interval = 0
+        self._desired_poll_rate = self._poll_rate_max
+
+    def _poll_rate(self):
+        now = time.time()
+        return (self._poll_count_total / (now - self._t_start),
+                self._poll_count_interval / (now - self._t_last_poll))
 
     def snooze(self):
-        if self._num_polls == 0:
-            self._t_init = time.time()
-
-        t_elapsed = time.time() - self._t_init
-        self._num_polls += 1
-        poll_rate = self._num_polls / t_elapsed if t_elapsed else math.inf
-        getlogger().debug2(
-            f'Poll rate control: sleeping for {self._sleep_duration}s '
-            f'(current poll rate: {poll_rate} polls/s)'
-        )
-        time.sleep(self._sleep_duration)
-        self._sleep_duration = min(
-            self._sleep_duration*self.SLEEP_INC_RATE, self.SLEEP_MAX
+        if self._poll_count_total == 0:
+            self._t_start = time.time()
+            self._t_last_poll = self._t_start
+            dt_sleep = 1. / self._desired_poll_rate
+        else:
+            dt_next_interval = time.time() - self._t_last_poll
+            dt_sleep = (self._poll_count_interval + 1) / self._desired_poll_rate - dt_next_interval  # noqa: E501
+
+        if self._poll_randomize_range_ms:
+            sleep_eps = random.randrange(*self._poll_randomize_range_ms)
+            dt_sleep += sleep_eps / 1000
+
+        # Make sure sleep time positive
+        dt_sleep = max(0, dt_sleep)
+        time.sleep(dt_sleep)
+
+        self._poll_count_total += 1
+        self._poll_count_interval += 1
+        poll_rate_global, poll_rate_curr = self._poll_rate()
+        getlogger().debug2(f'[P] sleep_time={dt_sleep:.6f}, '
+                           f'pr_desired={self._desired_poll_rate:.6f}, '
+                           f'pr_current={poll_rate_curr:.6f}, '
+                           f'pr_global={poll_rate_global:.6f}')
+        self._desired_poll_rate = max(
+            self._desired_poll_rate*self._poll_rate_decay, self._poll_rate_min
         )
+        self._t_last_poll = time.time()
 
 
 class _PolicyEventListener(TaskEventListener):
@@ -227,7 +278,7 @@ def runcase(self, case):
             else:
                 sched = partition.scheduler
 
-            self._pollctl.reset_snooze_time()
+            self._pollctl.reset_poll_rate()
             while True:
                 if not self.dry_run_mode:
                     sched.poll(task.check.job)
@@ -366,7 +417,7 @@ def exit(self):
         if self._pipeline_statistics:
             self._init_pipeline_progress(len(self._current_tasks))
 
-        self._pollctl.reset_snooze_time()
+        self._pollctl.reset_poll_rate()
         while self._current_tasks:
             try:
                 self._poll_tasks()
@@ -607,7 +658,7 @@ def _abortall(self, cause):
                 task.abort(cause)
 
     def on_task_exit(self, task):
-        self._pollctl.reset_snooze_time()
+        self._pollctl.reset_poll_rate()
 
     def on_task_compile_exit(self, task):
-        self._pollctl.reset_snooze_time()
+        self._pollctl.reset_poll_rate()

reframe/schemas/config.json

@@ -524,6 +524,16 @@
                     "dump_pipeline_progress": {"type": "boolean"},
                     "perf_info_level": {"$ref": "#/defs/loglevel"},
                     "perf_report_spec": {"type": "string"},
+                    "poll_randomize_ms": {"anyOf": [
+                        {"type": "null"},
+                        {"type": "array",
+                         "minItems": 2,
+                         "maxItems": 2,
+                         "items": {"type": "integer"}}
+                    ]},
+                    "poll_rate_decay": {"type": "number"},
+                    "poll_rate_max": {"type": "number"},
+                    "poll_rate_min": {"type": "number"},
                     "pipeline_timeout": {"type": ["number", "null"]},
                     "purge_environment": {"type": "boolean"},
                     "remote_detect": {"type": "boolean"},
@@ -600,6 +610,10 @@
         "general/non_default_craype": false,
         "general/perf_info_level": "info",
         "general/perf_report_spec": "now-1d:now/last:/+job_nodelist+result",
+        "general/poll_randomize_ms": null,
+        "general/poll_rate_decay": 0.9,
+        "general/poll_rate_max": 10,
+        "general/poll_rate_min": 0.1,
         "general/pipeline_timeout": 3,
         "general/purge_environment": false,
         "general/remote_detect": false,

tools/plot_poll_rate.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+
+import io
+import re
+import sys
+
+import polars as pl
+import plotly.express as px
+
+
+def read_logfile(logfile):
+    regex = re.compile(r"\[(\S+)\] debug2:.*sleep_time=(\S+), "
+                       r"pr_desired=(\S+), pr_current=(\S+), pr_global=(\S+)")
+    csv_data = ''
+    with open(logfile) as fp:
+        for line in fp:
+            if m := regex.match(line):
+                csv_data += ",".join(m.groups()) + "\n"
+
+    if not csv_data:
+        return pl.DataFrame()
+
+    df = pl.read_csv(
+        io.StringIO(csv_data),
+        has_header=False,
+        new_columns=['timestamp', 'sleep_time', 'pr_desired',
+                     'pr_current', 'pr_global']
+    ).with_columns(
+        pl.col('timestamp').str.to_datetime()
+    )
+    return df
+
+
+def plot_poll_rates(logfile):
+    px.line(read_logfile(logfile),
+            x='timestamp', y=['pr_desired', 'pr_current', 'pr_global']).show()
+
+
+def plot_poll_histogram(logfiles):
+    dataframes = []
+    for filename in logfiles:
+        if not (df := read_logfile(filename)).is_empty():
+            dataframes.append(
+                df.with_columns(pl.lit(filename).alias('filename'))
+            )
+
+    fig = px.histogram(pl.concat(dataframes).sort('timestamp'), x='timestamp', color='filename', nbins=100)
+    fig.show()
+
+
+def main():
+    if len(sys.argv[1:]) == 1:
+        plot_poll_rates(sys.argv[1])
+    else:
+        plot_poll_histogram(sys.argv[1:])
+
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

unittests/test_logging.py

@@ -186,7 +186,7 @@ def test_rfc3339_timezone_extension(logfile, logger_with_check,
 def test_rfc3339_timezone_wrong_directive(logfile, logger_without_check):
     formatter = rlog.RFC3339Formatter(
         fmt='[%(asctime)s] %(levelname)s: %(check_name)s: %(message)s',
-        datefmt='%FT%T:z')
+        datefmt=r'%FT%T:z')
     logger_without_check.logger.handlers[0].setFormatter(formatter)
     logger_without_check.info('foo')
     assert _pattern_in_logfile(':z', logfile)