throttle: fix linear based strategies (#65422)

* throttle tests: fix detection of parallel execution

The test wasn't able to detect if too many workers were running.

On my laptop:
- without this change, the 'throttle' target takes ~20 seconds
- with this change, the 'throttle' target takes ~70 seconds
- 1 second isn't long enough to encounter the issue

* Fix throttle test when strategy is 'free' based

'free' strategy allows multiple tasks to be executed in parallel: use
one 'throttledir' per task.

Use 'linear' strategy with a dedicated play for cleanup/setup tasks

* throttle: reset worker idx before queuing a new task

* TestStrategyBase: define task.throttle

otherwise '1' will be used instead of the default value due to the
following expression being equal to '1':

    int(templar.template(task_mock.throttle))

Co-authored-by: James Cammarata <jimi@sngx.net>

changelogs/fragments/65422-fix-throttle-with-linear-strategy.yml

@@ -0,0 +1,2 @@
+bugfixes:
+- "throttle: the linear strategy didn't always stuck with the throttle limit"

lib/ansible/plugins/strategy/__init__.py

@@ -325,9 +325,26 @@ class StrategyBase:
 
         # and then queue the new task
         try:
+            # Determine the "rewind point" of the worker list. This means we start
+            # iterating over the list of workers until the end of the list is found.
+            # Normally, that is simply the length of the workers list (as determined
+            # by the forks or serial setting), however a task/block/play may "throttle"
+            # that limit down.
+            rewind_point = len(self._workers)
+            if throttle > 0 and self.ALLOW_BASE_THROTTLING:
+                if task.run_once:
+                    display.debug("Ignoring 'throttle' as 'run_once' is also set for '%s'" % task.get_name())
+                else:
+                    if throttle <= rewind_point:
+                        display.debug("task: %s, throttle: %d" % (task.get_name(), throttle))
+                        rewind_point = throttle
+
             queued = False
             starting_worker = self._cur_worker
             while True:
+                if self._cur_worker >= rewind_point:
+                    self._cur_worker = 0
+
                 worker_prc = self._workers[self._cur_worker]
                 if worker_prc is None or not worker_prc.is_alive():
                     self._queued_task_cache[(host.name, task._uuid)] = {
@@ -346,19 +363,6 @@ class StrategyBase:
 
                 self._cur_worker += 1
 
-                # Determine the "rewind point" of the worker list. This means we start
-                # iterating over the list of workers until the end of the list is found.
-                # Normally, that is simply the length of the workers list (as determined
-                # by the forks or serial setting), however a task/block/play may "throttle"
-                # that limit down.
-                rewind_point = len(self._workers)
-                if throttle > 0 and self.ALLOW_BASE_THROTTLING:
-                    if task.run_once:
-                        display.debug("Ignoring 'throttle' as 'run_once' is also set for '%s'" % task.get_name())
-                    else:
-                        if throttle <= rewind_point:
-                            display.debug("task: %s, throttle: %d" % (task.get_name(), throttle))
-                            rewind_point = throttle
                 if self._cur_worker >= rewind_point:
                     self._cur_worker = 0
 

test/integration/targets/throttle/group_vars/all.yml

@@ -0,0 +1,4 @@
+---
+throttledir: '{{ base_throttledir }}/{{ subdir }}'
+base_throttledir: "{{ lookup('env', 'OUTPUT_DIR') }}/throttle.dir"
+subdir: "{{ test_id if lookup('env', 'SELECTED_STRATEGY') in ['free', 'host_pinned'] else '' }}"

test/integration/targets/throttle/runme.sh

@@ -3,5 +3,5 @@
 set -eux
 
 # https://github.com/ansible/ansible/pull/42528
-ANSIBLE_STRATEGY='linear' ansible-playbook test_throttle.yml -vv -i inventory --forks 12 "$@"
+SELECTED_STRATEGY='linear' ansible-playbook test_throttle.yml -vv -i inventory --forks 12 "$@"
-ANSIBLE_STRATEGY='free' ansible-playbook test_throttle.yml -vv -i inventory --forks 12 "$@"
+SELECTED_STRATEGY='free' ansible-playbook test_throttle.yml -vv -i inventory --forks 12 "$@"

test/integration/targets/throttle/test_throttle.py

@@ -24,6 +24,7 @@ try:
     if len(throttlelist) > max_throttle:
         print(throttlelist)
         raise ValueError("Too many concurrent tasks: %d/%d" % (len(throttlelist), max_throttle))
+    time.sleep(1.5)
 finally:
     # remove the file, then wait to make sure it's gone
     os.unlink(throttlefile)

test/integration/targets/throttle/test_throttle.yml

@@ -1,59 +1,84 @@
 ---
 - hosts: localhosts
   gather_facts: false
-  vars:
+  strategy: linear
-    throttledir: "{{ lookup('env', 'OUTPUT_DIR') }}/throttle.dir/"
+  run_once: yes
   tasks:
-    - name: Clean throttledir '{{ throttledir }}'
+    - name: Clean base throttledir '{{ base_throttledir }}'
       file:
         state: absent
-        path: '{{ throttledir }}'
+        path: '{{ base_throttledir }}'
       ignore_errors: yes
-      run_once: yes
+
     - name: Create throttledir '{{ throttledir }}'
       file:
         state: directory
         path: '{{ throttledir }}'
-      run_once: yes
+      loop: "{{ range(1, test_count|int)|list }}"
+      loop_control:
+        loop_var: test_id
+      vars:
+        test_count: "{{ 9 if lookup('env', 'SELECTED_STRATEGY') in ['free', 'host_pinned'] else 2 }}"
+
+- hosts: localhosts
+  gather_facts: false
+  strategy: "{{ lookup('env', 'SELECTED_STRATEGY') }}"
+  tasks:
     - block:
       - name: "Test 1 (max throttle: 3)"
         script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 3"
+        vars:
+          test_id: 1
       throttle: 3
     - block:
       - name: "Test 2 (max throttle: 5)"
         script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 5"
         throttle: 5
+        vars:
+          test_id: 2
       - block:
         - name: "Test 3 (max throttle: 8)"
           script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 8"
           throttle: 8
         throttle: 6
+        vars:
+           test_id: 3
       - block:
         - block:
           - name: "Test 4 (max throttle: 8)"
             script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 8"
             throttle: 8
+            vars:
+              test_id: 4
           throttle: 6
         throttle: 12
       throttle: 15
     - block:
-      - name: "Test 1 (max throttle: 3)"
+      - name: "Teat 5 (max throttle: 3)"
         script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 3"
+        vars:
+          test_id: 5
       throttle: 3
     - block:
-      - name: "Test 2 (max throttle: 5)"
+      - name: "Test 6 (max throttle: 5)"
         script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 5"
         throttle: 5
+        vars:
+          test_id: 6
       - block:
-        - name: "Test 3 (max throttle: 6)"
+        - name: "Test 7 (max throttle: 6)"
           script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 6"
           throttle: 6
+          vars:
+            test_id: 7
         throttle: 3
       - block:
         - block:
-          - name: "Test 4 (max throttle: 8)"
+          - name: "Test 8 (max throttle: 8)"
             script: "test_throttle.py {{throttledir}} {{inventory_hostname}} 8"
             throttle: 8
+            vars:
+              test_id: 8
           throttle: 6
         throttle: 4
       throttle: 2

test/units/plugins/strategy/test_strategy_base.py

@@ -195,6 +195,7 @@ class TestStrategyBase(unittest.TestCase):
 
         mock_task = MagicMock()
         mock_task._uuid = 'abcd'
+        mock_task.throttle = 0
 
         try:
             strategy_base = StrategyBase(tqm=tqm)