Merge pull request #6941 from cylc/8.5.x-sync

🤖 Merge 8.5.x-sync into master

Author: hjoliver

changes.d/6848.fix.md

@@ -0,0 +1 @@
+Allow `flow.cylc[runtime][<task>]platform` setting to have a prefix/suffix around a subshell expression.

changes.d/6940.fix.md

@@ -0,0 +1 @@
+Fixed a scheduler crash that could occur after re-running a task in skip mode.

cylc/flow/cfgspec/globalcfg.py

@@ -1046,7 +1046,8 @@ def default_for(
             ''')
             Conf('process check timeout', VDR.V_INTERVAL, DurationFloat(10),
                  desc='''
-                Maximum time for the `cylc play` and `cylc vr` commands to wait
+                Maximum time for the ``cylc play`` and ``cylc vr`` commands
+                to wait
                 for a remote process that checks if an unresponsive scheduler
                 is still alive (for workflows with existing contact files).
 

cylc/flow/cfgspec/workflow.py

@@ -1211,6 +1211,7 @@ def get_script_common_text(this: str, example: Optional[str] = None):
 
                    # run a command to select the platform (or platform group):
                    platform = $(select-platform)
+                   platform = prefix-$(select-platform)-suffix
 
                 .. versionadded:: 8.0.0
             ''')

cylc/flow/platforms.py

@@ -57,7 +57,7 @@
 
 # Regex to check whether a string is a command
 HOST_REC_COMMAND = re.compile(r'(`|\$\()\s*(.*)\s*([`)])$')
-PLATFORM_REC_COMMAND = re.compile(r'(\$\()\s*(.*)\s*([)])$')
+PLATFORM_REC_COMMAND = re.compile(r'(\$\()\s*(.*)\s*(\))')
 
 HOST_SELECTION_METHODS = {
     'definition order': lambda goodhosts: goodhosts[0],

cylc/flow/task_events_mgr.py

@@ -548,7 +548,7 @@ def _get_remote_conf(self, itask, key):
             itask.platform[key]
         )
 
-    def _get_workflow_platforms_conf(self, itask, key):
+    def _get_workflow_platforms_conf(self, itask: 'TaskProxy', key: str):
         """Return top level [runtime] items that default to platforms."""
         overrides = self.broadcast_mgr.get_broadcast(itask.tokens)
         return (
@@ -1844,7 +1844,7 @@ def _get_handler_template_variables(
         }
         # fmt: on
 
-    def _reset_job_timers(self, itask):
+    def _reset_job_timers(self, itask: 'TaskProxy'):
         """Set up poll timer and timeout for task."""
 
         if itask.transient:
@@ -1898,18 +1898,18 @@ def _reset_job_timers(self, itask):
         itask.poll_timer = TaskActionTimer(ctx=ctx, delays=delays)
         # Log timeout and polling schedule
         message = f"health: {timeout_key}={timeout_str}"
-        # Attempt to group identical consecutive delays as N*DELAY,...
         if itask.poll_timer.delays:
-            items = []  # [(number of item - 1, item), ...]
+            # Group identical consecutive delays as N*DELAY,...
+            items: List[List[float]] = []  # [[number of item, item], ...]
             for delay in itask.poll_timer.delays:
                 if items and items[-1][1] == delay:
                     items[-1][0] += 1
                 else:
-                    items.append([0, delay])
+                    items.append([1, delay])
             message += ', polling intervals='
             for num, item in items:
-                if num:
-                    message += '%d*' % (num + 1)
+                if num > 1:
+                    message += f'{num}*'
                 message += '%s,' % intvl_as_str(item)
             message += '...'
         LOG.debug(f"[{itask}] {message}")
@@ -1920,7 +1920,7 @@ def _reset_job_timers(self, itask):
     def process_execution_polling_intervals(
         polling_intervals: List[float],
         time_limit: float,
-        time_limit_polling_intervals: List[float]
+        time_limit_polling_intervals: Optional[List[float]]
     ) -> List[float]:
         """Create a list of polling times.
 
@@ -1952,6 +1952,11 @@ def process_execution_polling_intervals(
         >>> this([], 10, [5])
         [15, 5]
 
+        # There are no execution time limit polling intervals set - just
+        # repeat the execution polling interval until the time limit:
+        >>> this([10], 25, None)
+        [10, 10]
+
         # We have a list of execution time limit polling intervals,
         >>> this([10], 25, [5, 6, 7, 8])
         [10, 10, 10, 6, 7, 8]
@@ -1968,17 +1973,19 @@ def process_execution_polling_intervals(
             size = int((time_limit - sum(delays)) / delays[-1])
             delays.extend([delays[-1]] * size)
 
-        # After the last delay before the execution time limit add the
-        # delay to get to the execution_time_limit
-        if len(time_limit_polling_intervals) == 1:
-            time_limit_polling_intervals.append(
-                time_limit_polling_intervals[0]
-            )
-        time_limit_polling_intervals[0] += time_limit - sum(delays)
+        if time_limit_polling_intervals:
+            # After the last delay before the execution time limit add the
+            # delay to get to the execution_time_limit
+            if len(time_limit_polling_intervals) == 1:
+                time_limit_polling_intervals.append(
+                    time_limit_polling_intervals[0]
+                )
+            time_limit_polling_intervals[0] += time_limit - sum(delays)
+
+            # After the execution time limit, poll at the
+            # execution time limit polling intervals.
+            delays += time_limit_polling_intervals
 
-        # After the execution time limit poll at execution time limit polling
-        # intervals.
-        delays += time_limit_polling_intervals
         return delays
 
     def add_event_timer(self, id_key: EventKey, event_timer) -> None:

cylc/flow/task_remote_mgr.py

@@ -143,17 +143,18 @@ def _subshell_eval(
             return 'localhost'
 
         # Host selection command: $(command) or `command`
-        match = command_pattern.match(eval_str)
+        match = command_pattern.search(eval_str)
         if match:
-            cmd_str = match.groups()[1]
+            cmd_str = match.group(2)
             if cmd_str in self.remote_command_map:
                 # Command recently launched
                 value = self.remote_command_map[cmd_str]
                 if isinstance(value, PlatformError):
                     raise value  # command failed
                 if value is None:
                     return None  # command not yet ready
-                eval_str = value  # command succeeded
+                # command succeeded
+                eval_str = eval_str.replace(match.group(0), value)
             else:
                 # Command not launched (or already reset)
                 self.proc_pool.put_command(

tests/functional/job-submission/19-platform_select.t

@@ -17,7 +17,7 @@
 #-------------------------------------------------------------------------------
 # Test recovery of a failed host select command for a group of tasks.
 . "$(dirname "$0")/test_header"
-set_test_number 6
+set_test_number 7
 
 install_workflow "${TEST_NAME_BASE}"
 
@@ -29,20 +29,24 @@ logfile="${WORKFLOW_RUN_DIR}/log/scheduler/log"
 
 # Check that host = $(cmd) is correctly evaluated
 grep_ok \
-    "1/host_subshell.* evaluated as improbable host name$" \
+    "1/host_subshell/01:.* evaluated as improbable host name$" \
     "${logfile}"
 grep_ok \
-    "1/localhost_subshell.* evaluated as localhost$" \
+    "1/localhost_subshell/01:.* evaluated as localhost$" \
     "${logfile}"
 
 # Check that host = `cmd` is correctly evaluated
 grep_ok \
-    "1/host_subshell_backticks.* evaluated as improbable host name$" \
+    "1/host_subshell_backticks/01:.* evaluated as improbable host name$" \
     "${logfile}"
 
 # Check that platform = $(cmd) correctly evaluated
 grep_ok \
-    "1/platform_subshell.* evaluated as improbable platform name$" \
+    "1/platform_subshell:.* evaluated as improbable platform name$" \
+    "${logfile}"
+
+grep_ok \
+    "1/platform_subshell_suffix:.* evaluated as prefix-middle-suffix$" \
     "${logfile}"
 
 purge

tests/functional/job-submission/19-platform_select/flow.cylc

@@ -15,10 +15,11 @@ purpose = """
         R1 = """
             host_no_subshell
             localhost_subshell
-            platform_subshell:submit-fail? => fin_platform
-            platform_no_subshell:submit-fail? => fin_platform
-            host_subshell:submit-fail? => fin_host
-            host_subshell_backticks:submit-fail? => fin_host
+            platform_subshell:submit-fail?
+            platform_no_subshell:submit-fail?
+            platform_subshell_suffix:submit-fail?
+            host_subshell:submit-fail?
+            host_subshell_backticks:submit-fail?
         """
 
 [runtime]
@@ -35,6 +36,9 @@ purpose = """
     [[platform_subshell]]
         platform = $(echo "improbable platform name")
 
+    [[platform_subshell_suffix]]
+        platform = prefix-$( echo middle )-suffix
+
     [[host_subshell]]
         [[[remote]]]
             host = $(echo "improbable host name")
@@ -46,9 +50,3 @@ purpose = """
     [[localhost_subshell]]
         [[[remote]]]
             host = $(echo "localhost4.localdomain4")
-
-    [[fin_platform]]
-        script = cylc remove "${CYLC_WORKFLOW_ID}//1/platform_*"
-
-    [[fin_host]]
-        script = cylc remove "${CYLC_WORKFLOW_ID}//1/host_subshell*"

tests/functional/job-submission/19-platform_select/reference.log

@@ -4,5 +4,4 @@
 1/host_subshell -triggered off [] in flow 1
 1/host_subshell_backticks -triggered off [] in flow 1
 1/localhost_subshell -triggered off [] in flow 1
-1/fin_platform -triggered off ['1/platform_no_subshell', '1/platform_subshell'] in flow 1
-1/fin_host -triggered off ['1/host_subshell', '1/host_subshell_backticks'] in flow 1
+1/platform_subshell_suffix -triggered off [] in flow 1