We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Strategy htex_auto_scale removing blocks far too slow
**Describe the bug**
The "htex_auto_scale" strategy is very slow to remove blocks in case there are more slots than tasks. It removes one block per 5 seconds at most.

**To Reproduce**
Steps to reproduce the behavior, for e.g:
1. Run the test script
2. Check parsl.log and wait for the first handful of tasks are completed
3. The logs shows that it will need around a minute after the idle time out to remove the blocks, even though all tasks but one have been finished. Most of the blocks are idle.

**Expected behavior**
All idle blocks are removed as soon as possible. This becomes really important if you are dealing with, say 500 blocks and most of them are idle.

**Environment**
Parsl 1.2.0, Python 3.8


**Test script**
```python
import parsl
from parsl.providers import LocalProvider
from parsl.channels import LocalChannel
from parsl.config import Config
from parsl.executors import HighThroughputExecutor
from parsl.app.app import python_app


local_htex = Config(
    executors=[
        HighThroughputExecutor(
            label="htex_Local",
            worker_debug=True,
            max_workers=1,
            provider=LocalProvider(
                channel=LocalChannel(),
                init_blocks=2,
                max_blocks=20,
            ),
        )
    ],
    strategy="htex_auto_scale",
    max_idletime=3,
)
parsl.load(local_htex)


@python_app
def long():
    import time
    time.sleep(10000)


@python_app
def short():
    import time
    time.sleep(1)


a = long()
b = [short() for i in range(20)]
a.result()
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `parsl/dataflow/strategy.py`
Content:
```
1 import logging
2 import time
3 import math
4 from typing import List
5 
6 from parsl.dataflow.executor_status import ExecutorStatus
7 from parsl.executors import HighThroughputExecutor
8 from parsl.providers.provider_base import JobState
9 
10 logger = logging.getLogger(__name__)
11 
12 
13 class Strategy(object):
14     """FlowControl strategy.
15 
16     As a workflow dag is processed by Parsl, new tasks are added and completed
17     asynchronously. Parsl interfaces executors with execution providers to construct
18     scalable executors to handle the variable work-load generated by the
19     workflow. This component is responsible for periodically checking outstanding
20     tasks and available compute capacity and trigger scaling events to match
21     workflow needs.
22 
23     Here's a diagram of an executor. An executor consists of blocks, which are usually
24     created by single requests to a Local Resource Manager (LRM) such as slurm,
25     condor, torque, or even AWS API. The blocks could contain several task blocks
26     which are separate instances on workers.
27 
28 
29     .. code:: python
30 
31                 |<--min_blocks     |<-init_blocks              max_blocks-->|
32                 +----------------------------------------------------------+
33                 |  +--------block----------+       +--------block--------+ |
34      executor = |  | task          task    | ...   |    task      task   | |
35                 |  +-----------------------+       +---------------------+ |
36                 +----------------------------------------------------------+
37 
38     The relevant specification options are:
39        1. min_blocks: Minimum number of blocks to maintain
40        2. init_blocks: number of blocks to provision at initialization of workflow
41        3. max_blocks: Maximum number of blocks that can be active due to one workflow
42 
43 
44     .. code:: python
45 
46           active_tasks = pending_tasks + running_tasks
47 
48           Parallelism = slots / tasks
49                       = [0, 1] (i.e,  0 <= p <= 1)
50 
51     For example:
52 
53     When p = 0,
54          => compute with the least resources possible.
55          infinite tasks are stacked per slot.
56 
57          .. code:: python
58 
59                blocks =  min_blocks           { if active_tasks = 0
60                          max(min_blocks, 1)   {  else
61 
62     When p = 1,
63          => compute with the most resources.
64          one task is stacked per slot.
65 
66          .. code:: python
67 
68                blocks = min ( max_blocks,
69                         ceil( active_tasks / slots ) )
70 
71 
72     When p = 1/2,
73          => We stack upto 2 tasks per slot before we overflow
74          and request a new block
75 
76 
77     let's say min:init:max = 0:0:4 and task_blocks=2
78     Consider the following example:
79     min_blocks = 0
80     init_blocks = 0
81     max_blocks = 4
82     tasks_per_node = 2
83     nodes_per_block = 1
84 
85     In the diagram, X <- task
86 
87     at 2 tasks:
88 
89     .. code:: python
90 
91         +---Block---|
92         |           |
93         | X      X  |
94         |slot   slot|
95         +-----------+
96 
97     at 5 tasks, we overflow as the capacity of a single block is fully used.
98 
99     .. code:: python
100 
101         +---Block---|       +---Block---|
102         | X      X  | ----> |           |
103         | X      X  |       | X         |
104         |slot   slot|       |slot   slot|
105         +-----------+       +-----------+
106 
107     """
108 
109     def __init__(self, dfk):
110         """Initialize strategy."""
111         self.dfk = dfk
112         self.config = dfk.config
113         self.executors = {}
114         self.max_idletime = self.dfk.config.max_idletime
115 
116         for e in self.dfk.config.executors:
117             self.executors[e.label] = {'idle_since': None, 'config': e.label}
118 
119         self.strategies = {None: self._strategy_noop,
120                            'simple': self._strategy_simple,
121                            'htex_auto_scale': self._strategy_htex_auto_scale}
122 
123         self.strategize = self.strategies[self.config.strategy]
124 
125         logger.debug("Scaling strategy: {0}".format(self.config.strategy))
126 
127     def add_executors(self, executors):
128         for executor in executors:
129             self.executors[executor.label] = {'idle_since': None, 'config': executor.label}
130 
131     def _strategy_noop(self, status: List[ExecutorStatus], tasks):
132         """Do nothing.
133 
134         Args:
135             - tasks (task_ids): Not used here.
136         """
137 
138     def _strategy_simple(self, status_list, tasks):
139         self._general_strategy(status_list, tasks, strategy_type='simple')
140 
141     def _strategy_htex_auto_scale(self, status_list, tasks):
142         """HTEX specific auto scaling strategy
143 
144         This strategy works only for HTEX. This strategy will scale up by
145         requesting additional compute resources via the provider when the
146         workload requirements exceed the provisioned capacity. The scale out
147         behavior is exactly like the 'simple' strategy.
148 
149         If there are idle blocks during execution, this strategy will terminate
150         those idle blocks specifically. When # of tasks >> # of blocks, HTEX places
151         tasks evenly across blocks, which makes it rather difficult to ensure that
152         some blocks will reach 0% utilization. Consequently, this strategy can be
153         expected to scale down effectively only when # of workers, or tasks executing
154         per block is close to 1.
155 
156         Args:
157             - tasks (task_ids): Not used here.
158         """
159         self._general_strategy(status_list, tasks, strategy_type='htex')
160 
161     def _general_strategy(self, status_list, tasks, *, strategy_type):
162         for exec_status in status_list:
163             executor = exec_status.executor
164             label = executor.label
165             if not executor.scaling_enabled:
166                 continue
167 
168             # Tasks that are either pending completion
169             active_tasks = executor.outstanding
170 
171             status = exec_status.status
172 
173             # FIXME we need to handle case where provider does not define these
174             # FIXME probably more of this logic should be moved to the provider
175             min_blocks = executor.provider.min_blocks
176             max_blocks = executor.provider.max_blocks
177             tasks_per_node = executor.workers_per_node
178 
179             nodes_per_block = executor.provider.nodes_per_block
180             parallelism = executor.provider.parallelism
181 
182             running = sum([1 for x in status.values() if x.state == JobState.RUNNING])
183             pending = sum([1 for x in status.values() if x.state == JobState.PENDING])
184             active_blocks = running + pending
185             active_slots = active_blocks * tasks_per_node * nodes_per_block
186 
187             if hasattr(executor, 'connected_workers'):
188                 logger.debug('Executor {} has {} active tasks, {}/{} running/pending blocks, and {} connected workers'.format(
189                     label, active_tasks, running, pending, executor.connected_workers))
190             else:
191                 logger.debug('Executor {} has {} active tasks and {}/{} running/pending blocks'.format(
192                     label, active_tasks, running, pending))
193 
194             # reset kill timer if executor has active tasks
195             if active_tasks > 0 and self.executors[executor.label]['idle_since']:
196                 self.executors[executor.label]['idle_since'] = None
197 
198             # Case 1
199             # No tasks.
200             if active_tasks == 0:
201                 # Case 1a
202                 # Fewer blocks that min_blocks
203                 if active_blocks <= min_blocks:
204                     # Ignore
205                     # logger.debug("Strategy: Case.1a")
206                     pass
207 
208                 # Case 1b
209                 # More blocks than min_blocks. Scale down
210                 else:
211                     # We want to make sure that max_idletime is reached
212                     # before killing off resources
213                     if not self.executors[executor.label]['idle_since']:
214                         logger.debug("Executor {} has 0 active tasks; starting kill timer (if idle time exceeds {}s, resources will be removed)".format(
215                             label, self.max_idletime)
216                         )
217                         self.executors[executor.label]['idle_since'] = time.time()
218 
219                     idle_since = self.executors[executor.label]['idle_since']
220                     if (time.time() - idle_since) > self.max_idletime:
221                         # We have resources idle for the max duration,
222                         # we have to scale_in now.
223                         logger.debug("Idle time has reached {}s for executor {}; removing resources".format(
224                             self.max_idletime, label)
225                         )
226                         exec_status.scale_in(active_blocks - min_blocks)
227 
228                     else:
229                         pass
230                         # logger.debug("Strategy: Case.1b. Waiting for timer : {0}".format(idle_since))
231 
232             # Case 2
233             # More tasks than the available slots.
234             elif (float(active_slots) / active_tasks) < parallelism:
235                 # Case 2a
236                 # We have the max blocks possible
237                 if active_blocks >= max_blocks:
238                     # Ignore since we already have the max nodes
239                     # logger.debug("Strategy: Case.2a")
240                     pass
241 
242                 # Case 2b
243                 else:
244                     # logger.debug("Strategy: Case.2b")
245                     excess = math.ceil((active_tasks * parallelism) - active_slots)
246                     excess_blocks = math.ceil(float(excess) / (tasks_per_node * nodes_per_block))
247                     excess_blocks = min(excess_blocks, max_blocks - active_blocks)
248                     logger.debug("Requesting {} more blocks".format(excess_blocks))
249                     exec_status.scale_out(excess_blocks)
250 
251             elif active_slots == 0 and active_tasks > 0:
252                 # Case 4
253                 logger.debug("Requesting single slot")
254                 if active_blocks < max_blocks:
255                     exec_status.scale_out(1)
256 
257             # Case 4
258             # More slots than tasks
259             elif active_slots > 0 and active_slots > active_tasks:
260                 if strategy_type == 'htex':
261                     # Scale down for htex
262                     logger.debug("More slots than tasks")
263                     if isinstance(executor, HighThroughputExecutor):
264                         if active_blocks > min_blocks:
265                             exec_status.scale_in(1, force=False, max_idletime=self.max_idletime)
266 
267                 elif strategy_type == 'simple':
268                     # skip for simple strategy
269                     pass
270 
271             # Case 3
272             # tasks ~ slots
273             else:
274                 # logger.debug("Strategy: Case 3")
275                 pass
276
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/parsl/dataflow/strategy.py b/parsl/dataflow/strategy.py
--- a/parsl/dataflow/strategy.py
+++ b/parsl/dataflow/strategy.py
@@ -242,8 +242,8 @@
                 # Case 2b
                 else:
                     # logger.debug("Strategy: Case.2b")
-                    excess = math.ceil((active_tasks * parallelism) - active_slots)
-                    excess_blocks = math.ceil(float(excess) / (tasks_per_node * nodes_per_block))
+                    excess_slots = math.ceil((active_tasks * parallelism) - active_slots)
+                    excess_blocks = math.ceil(float(excess_slots) / (tasks_per_node * nodes_per_block))
                     excess_blocks = min(excess_blocks, max_blocks - active_blocks)
                     logger.debug("Requesting {} more blocks".format(excess_blocks))
                     exec_status.scale_out(excess_blocks)
@@ -262,7 +262,10 @@
                     logger.debug("More slots than tasks")
                     if isinstance(executor, HighThroughputExecutor):
                         if active_blocks > min_blocks:
-                            exec_status.scale_in(1, force=False, max_idletime=self.max_idletime)
+                            excess_slots = math.ceil(active_slots - (active_tasks * parallelism))
+                            excess_blocks = math.ceil(float(excess_slots) / (tasks_per_node * nodes_per_block))
+                            excess_blocks = min(excess_blocks, active_blocks - min_blocks)
+                            exec_status.scale_in(excess_blocks, force=False, max_idletime=self.max_idletime)
 
                 elif strategy_type == 'simple':
                     # skip for simple strategy

{"golden_diff": "diff --git a/parsl/dataflow/strategy.py b/parsl/dataflow/strategy.py\n--- a/parsl/dataflow/strategy.py\n+++ b/parsl/dataflow/strategy.py\n@@ -242,8 +242,8 @@\n                 # Case 2b\n                 else:\n                     # logger.debug(\"Strategy: Case.2b\")\n-                    excess = math.ceil((active_tasks * parallelism) - active_slots)\n-                    excess_blocks = math.ceil(float(excess) / (tasks_per_node * nodes_per_block))\n+                    excess_slots = math.ceil((active_tasks * parallelism) - active_slots)\n+                    excess_blocks = math.ceil(float(excess_slots) / (tasks_per_node * nodes_per_block))\n                     excess_blocks = min(excess_blocks, max_blocks - active_blocks)\n                     logger.debug(\"Requesting {} more blocks\".format(excess_blocks))\n                     exec_status.scale_out(excess_blocks)\n@@ -262,7 +262,10 @@\n                     logger.debug(\"More slots than tasks\")\n                     if isinstance(executor, HighThroughputExecutor):\n                         if active_blocks > min_blocks:\n-                            exec_status.scale_in(1, force=False, max_idletime=self.max_idletime)\n+                            excess_slots = math.ceil(active_slots - (active_tasks * parallelism))\n+                            excess_blocks = math.ceil(float(excess_slots) / (tasks_per_node * nodes_per_block))\n+                            excess_blocks = min(excess_blocks, active_blocks - min_blocks)\n+                            exec_status.scale_in(excess_blocks, force=False, max_idletime=self.max_idletime)\n \n                 elif strategy_type == 'simple':\n                     # skip for simple strategy\n", "issue": "Strategy htex_auto_scale removing blocks far too slow\n**Describe the bug**\r\nThe \"htex_auto_scale\" strategy is very slow to remove blocks in case there are more slots than tasks. It removes one block per 5 seconds at most.\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior, for e.g:\r\n1. Run the test script\r\n2. Check parsl.log and wait for the first handful of tasks are completed\r\n3. The logs shows that it will need around a minute after the idle time out to remove the blocks, even though all tasks but one have been finished. Most of the blocks are idle.\r\n\r\n**Expected behavior**\r\nAll idle blocks are removed as soon as possible. This becomes really important if you are dealing with, say 500 blocks and most of them are idle.\r\n\r\n**Environment**\r\nParsl 1.2.0, Python 3.8\r\n\r\n\r\n**Test script**\r\n```python\r\nimport parsl\r\nfrom parsl.providers import LocalProvider\r\nfrom parsl.channels import LocalChannel\r\nfrom parsl.config import Config\r\nfrom parsl.executors import HighThroughputExecutor\r\nfrom parsl.app.app import python_app\r\n\r\n\r\nlocal_htex = Config(\r\n    executors=[\r\n        HighThroughputExecutor(\r\n            label=\"htex_Local\",\r\n            worker_debug=True,\r\n            max_workers=1,\r\n            provider=LocalProvider(\r\n                channel=LocalChannel(),\r\n                init_blocks=2,\r\n                max_blocks=20,\r\n            ),\r\n        )\r\n    ],\r\n    strategy=\"htex_auto_scale\",\r\n    max_idletime=3,\r\n)\r\nparsl.load(local_htex)\r\n\r\n\r\n@python_app\r\ndef long():\r\n    import time\r\n    time.sleep(10000)\r\n\r\n\r\n@python_app\r\ndef short():\r\n    import time\r\n    time.sleep(1)\r\n\r\n\r\na = long()\r\nb = [short() for i in range(20)]\r\na.result()\r\n```\n", "before_files": [{"content": "import logging\nimport time\nimport math\nfrom typing import List\n\nfrom parsl.dataflow.executor_status import ExecutorStatus\nfrom parsl.executors import HighThroughputExecutor\nfrom parsl.providers.provider_base import JobState\n\nlogger = logging.getLogger(__name__)\n\n\nclass Strategy(object):\n    \"\"\"FlowControl strategy.\n\n    As a workflow dag is processed by Parsl, new tasks are added and completed\n    asynchronously. Parsl interfaces executors with execution providers to construct\n    scalable executors to handle the variable work-load generated by the\n    workflow. This component is responsible for periodically checking outstanding\n    tasks and available compute capacity and trigger scaling events to match\n    workflow needs.\n\n    Here's a diagram of an executor. An executor consists of blocks, which are usually\n    created by single requests to a Local Resource Manager (LRM) such as slurm,\n    condor, torque, or even AWS API. The blocks could contain several task blocks\n    which are separate instances on workers.\n\n\n    .. code:: python\n\n                |<--min_blocks     |<-init_blocks              max_blocks-->|\n                +----------------------------------------------------------+\n                |  +--------block----------+       +--------block--------+ |\n     executor = |  | task          task    | ...   |    task      task   | |\n                |  +-----------------------+       +---------------------+ |\n                +----------------------------------------------------------+\n\n    The relevant specification options are:\n       1. min_blocks: Minimum number of blocks to maintain\n       2. init_blocks: number of blocks to provision at initialization of workflow\n       3. max_blocks: Maximum number of blocks that can be active due to one workflow\n\n\n    .. code:: python\n\n          active_tasks = pending_tasks + running_tasks\n\n          Parallelism = slots / tasks\n                      = [0, 1] (i.e,  0 <= p <= 1)\n\n    For example:\n\n    When p = 0,\n         => compute with the least resources possible.\n         infinite tasks are stacked per slot.\n\n         .. code:: python\n\n               blocks =  min_blocks           { if active_tasks = 0\n                         max(min_blocks, 1)   {  else\n\n    When p = 1,\n         => compute with the most resources.\n         one task is stacked per slot.\n\n         .. code:: python\n\n               blocks = min ( max_blocks,\n                        ceil( active_tasks / slots ) )\n\n\n    When p = 1/2,\n         => We stack upto 2 tasks per slot before we overflow\n         and request a new block\n\n\n    let's say min:init:max = 0:0:4 and task_blocks=2\n    Consider the following example:\n    min_blocks = 0\n    init_blocks = 0\n    max_blocks = 4\n    tasks_per_node = 2\n    nodes_per_block = 1\n\n    In the diagram, X <- task\n\n    at 2 tasks:\n\n    .. code:: python\n\n        +---Block---|\n        |           |\n        | X      X  |\n        |slot   slot|\n        +-----------+\n\n    at 5 tasks, we overflow as the capacity of a single block is fully used.\n\n    .. code:: python\n\n        +---Block---|       +---Block---|\n        | X      X  | ----> |           |\n        | X      X  |       | X         |\n        |slot   slot|       |slot   slot|\n        +-----------+       +-----------+\n\n    \"\"\"\n\n    def __init__(self, dfk):\n        \"\"\"Initialize strategy.\"\"\"\n        self.dfk = dfk\n        self.config = dfk.config\n        self.executors = {}\n        self.max_idletime = self.dfk.config.max_idletime\n\n        for e in self.dfk.config.executors:\n            self.executors[e.label] = {'idle_since': None, 'config': e.label}\n\n        self.strategies = {None: self._strategy_noop,\n                           'simple': self._strategy_simple,\n                           'htex_auto_scale': self._strategy_htex_auto_scale}\n\n        self.strategize = self.strategies[self.config.strategy]\n\n        logger.debug(\"Scaling strategy: {0}\".format(self.config.strategy))\n\n    def add_executors(self, executors):\n        for executor in executors:\n            self.executors[executor.label] = {'idle_since': None, 'config': executor.label}\n\n    def _strategy_noop(self, status: List[ExecutorStatus], tasks):\n        \"\"\"Do nothing.\n\n        Args:\n            - tasks (task_ids): Not used here.\n        \"\"\"\n\n    def _strategy_simple(self, status_list, tasks):\n        self._general_strategy(status_list, tasks, strategy_type='simple')\n\n    def _strategy_htex_auto_scale(self, status_list, tasks):\n        \"\"\"HTEX specific auto scaling strategy\n\n        This strategy works only for HTEX. This strategy will scale up by\n        requesting additional compute resources via the provider when the\n        workload requirements exceed the provisioned capacity. The scale out\n        behavior is exactly like the 'simple' strategy.\n\n        If there are idle blocks during execution, this strategy will terminate\n        those idle blocks specifically. When # of tasks >> # of blocks, HTEX places\n        tasks evenly across blocks, which makes it rather difficult to ensure that\n        some blocks will reach 0% utilization. Consequently, this strategy can be\n        expected to scale down effectively only when # of workers, or tasks executing\n        per block is close to 1.\n\n        Args:\n            - tasks (task_ids): Not used here.\n        \"\"\"\n        self._general_strategy(status_list, tasks, strategy_type='htex')\n\n    def _general_strategy(self, status_list, tasks, *, strategy_type):\n        for exec_status in status_list:\n            executor = exec_status.executor\n            label = executor.label\n            if not executor.scaling_enabled:\n                continue\n\n            # Tasks that are either pending completion\n            active_tasks = executor.outstanding\n\n            status = exec_status.status\n\n            # FIXME we need to handle case where provider does not define these\n            # FIXME probably more of this logic should be moved to the provider\n            min_blocks = executor.provider.min_blocks\n            max_blocks = executor.provider.max_blocks\n            tasks_per_node = executor.workers_per_node\n\n            nodes_per_block = executor.provider.nodes_per_block\n            parallelism = executor.provider.parallelism\n\n            running = sum([1 for x in status.values() if x.state == JobState.RUNNING])\n            pending = sum([1 for x in status.values() if x.state == JobState.PENDING])\n            active_blocks = running + pending\n            active_slots = active_blocks * tasks_per_node * nodes_per_block\n\n            if hasattr(executor, 'connected_workers'):\n                logger.debug('Executor {} has {} active tasks, {}/{} running/pending blocks, and {} connected workers'.format(\n                    label, active_tasks, running, pending, executor.connected_workers))\n            else:\n                logger.debug('Executor {} has {} active tasks and {}/{} running/pending blocks'.format(\n                    label, active_tasks, running, pending))\n\n            # reset kill timer if executor has active tasks\n            if active_tasks > 0 and self.executors[executor.label]['idle_since']:\n                self.executors[executor.label]['idle_since'] = None\n\n            # Case 1\n            # No tasks.\n            if active_tasks == 0:\n                # Case 1a\n                # Fewer blocks that min_blocks\n                if active_blocks <= min_blocks:\n                    # Ignore\n                    # logger.debug(\"Strategy: Case.1a\")\n                    pass\n\n                # Case 1b\n                # More blocks than min_blocks. Scale down\n                else:\n                    # We want to make sure that max_idletime is reached\n                    # before killing off resources\n                    if not self.executors[executor.label]['idle_since']:\n                        logger.debug(\"Executor {} has 0 active tasks; starting kill timer (if idle time exceeds {}s, resources will be removed)\".format(\n                            label, self.max_idletime)\n                        )\n                        self.executors[executor.label]['idle_since'] = time.time()\n\n                    idle_since = self.executors[executor.label]['idle_since']\n                    if (time.time() - idle_since) > self.max_idletime:\n                        # We have resources idle for the max duration,\n                        # we have to scale_in now.\n                        logger.debug(\"Idle time has reached {}s for executor {}; removing resources\".format(\n                            self.max_idletime, label)\n                        )\n                        exec_status.scale_in(active_blocks - min_blocks)\n\n                    else:\n                        pass\n                        # logger.debug(\"Strategy: Case.1b. Waiting for timer : {0}\".format(idle_since))\n\n            # Case 2\n            # More tasks than the available slots.\n            elif (float(active_slots) / active_tasks) < parallelism:\n                # Case 2a\n                # We have the max blocks possible\n                if active_blocks >= max_blocks:\n                    # Ignore since we already have the max nodes\n                    # logger.debug(\"Strategy: Case.2a\")\n                    pass\n\n                # Case 2b\n                else:\n                    # logger.debug(\"Strategy: Case.2b\")\n                    excess = math.ceil((active_tasks * parallelism) - active_slots)\n                    excess_blocks = math.ceil(float(excess) / (tasks_per_node * nodes_per_block))\n                    excess_blocks = min(excess_blocks, max_blocks - active_blocks)\n                    logger.debug(\"Requesting {} more blocks\".format(excess_blocks))\n                    exec_status.scale_out(excess_blocks)\n\n            elif active_slots == 0 and active_tasks > 0:\n                # Case 4\n                logger.debug(\"Requesting single slot\")\n                if active_blocks < max_blocks:\n                    exec_status.scale_out(1)\n\n            # Case 4\n            # More slots than tasks\n            elif active_slots > 0 and active_slots > active_tasks:\n                if strategy_type == 'htex':\n                    # Scale down for htex\n                    logger.debug(\"More slots than tasks\")\n                    if isinstance(executor, HighThroughputExecutor):\n                        if active_blocks > min_blocks:\n                            exec_status.scale_in(1, force=False, max_idletime=self.max_idletime)\n\n                elif strategy_type == 'simple':\n                    # skip for simple strategy\n                    pass\n\n            # Case 3\n            # tasks ~ slots\n            else:\n                # logger.debug(\"Strategy: Case 3\")\n                pass\n", "path": "parsl/dataflow/strategy.py"}], "after_files": [{"content": "import logging\nimport time\nimport math\nfrom typing import List\n\nfrom parsl.dataflow.executor_status import ExecutorStatus\nfrom parsl.executors import HighThroughputExecutor\nfrom parsl.providers.provider_base import JobState\n\nlogger = logging.getLogger(__name__)\n\n\nclass Strategy(object):\n    \"\"\"FlowControl strategy.\n\n    As a workflow dag is processed by Parsl, new tasks are added and completed\n    asynchronously. Parsl interfaces executors with execution providers to construct\n    scalable executors to handle the variable work-load generated by the\n    workflow. This component is responsible for periodically checking outstanding\n    tasks and available compute capacity and trigger scaling events to match\n    workflow needs.\n\n    Here's a diagram of an executor. An executor consists of blocks, which are usually\n    created by single requests to a Local Resource Manager (LRM) such as slurm,\n    condor, torque, or even AWS API. The blocks could contain several task blocks\n    which are separate instances on workers.\n\n\n    .. code:: python\n\n                |<--min_blocks     |<-init_blocks              max_blocks-->|\n                +----------------------------------------------------------+\n                |  +--------block----------+       +--------block--------+ |\n     executor = |  | task          task    | ...   |    task      task   | |\n                |  +-----------------------+       +---------------------+ |\n                +----------------------------------------------------------+\n\n    The relevant specification options are:\n       1. min_blocks: Minimum number of blocks to maintain\n       2. init_blocks: number of blocks to provision at initialization of workflow\n       3. max_blocks: Maximum number of blocks that can be active due to one workflow\n\n\n    .. code:: python\n\n          active_tasks = pending_tasks + running_tasks\n\n          Parallelism = slots / tasks\n                      = [0, 1] (i.e,  0 <= p <= 1)\n\n    For example:\n\n    When p = 0,\n         => compute with the least resources possible.\n         infinite tasks are stacked per slot.\n\n         .. code:: python\n\n               blocks =  min_blocks           { if active_tasks = 0\n                         max(min_blocks, 1)   {  else\n\n    When p = 1,\n         => compute with the most resources.\n         one task is stacked per slot.\n\n         .. code:: python\n\n               blocks = min ( max_blocks,\n                        ceil( active_tasks / slots ) )\n\n\n    When p = 1/2,\n         => We stack upto 2 tasks per slot before we overflow\n         and request a new block\n\n\n    let's say min:init:max = 0:0:4 and task_blocks=2\n    Consider the following example:\n    min_blocks = 0\n    init_blocks = 0\n    max_blocks = 4\n    tasks_per_node = 2\n    nodes_per_block = 1\n\n    In the diagram, X <- task\n\n    at 2 tasks:\n\n    .. code:: python\n\n        +---Block---|\n        |           |\n        | X      X  |\n        |slot   slot|\n        +-----------+\n\n    at 5 tasks, we overflow as the capacity of a single block is fully used.\n\n    .. code:: python\n\n        +---Block---|       +---Block---|\n        | X      X  | ----> |           |\n        | X      X  |       | X         |\n        |slot   slot|       |slot   slot|\n        +-----------+       +-----------+\n\n    \"\"\"\n\n    def __init__(self, dfk):\n        \"\"\"Initialize strategy.\"\"\"\n        self.dfk = dfk\n        self.config = dfk.config\n        self.executors = {}\n        self.max_idletime = self.dfk.config.max_idletime\n\n        for e in self.dfk.config.executors:\n            self.executors[e.label] = {'idle_since': None, 'config': e.label}\n\n        self.strategies = {None: self._strategy_noop,\n                           'simple': self._strategy_simple,\n                           'htex_auto_scale': self._strategy_htex_auto_scale}\n\n        self.strategize = self.strategies[self.config.strategy]\n\n        logger.debug(\"Scaling strategy: {0}\".format(self.config.strategy))\n\n    def add_executors(self, executors):\n        for executor in executors:\n            self.executors[executor.label] = {'idle_since': None, 'config': executor.label}\n\n    def _strategy_noop(self, status: List[ExecutorStatus], tasks):\n        \"\"\"Do nothing.\n\n        Args:\n            - tasks (task_ids): Not used here.\n        \"\"\"\n\n    def _strategy_simple(self, status_list, tasks):\n        self._general_strategy(status_list, tasks, strategy_type='simple')\n\n    def _strategy_htex_auto_scale(self, status_list, tasks):\n        \"\"\"HTEX specific auto scaling strategy\n\n        This strategy works only for HTEX. This strategy will scale up by\n        requesting additional compute resources via the provider when the\n        workload requirements exceed the provisioned capacity. The scale out\n        behavior is exactly like the 'simple' strategy.\n\n        If there are idle blocks during execution, this strategy will terminate\n        those idle blocks specifically. When # of tasks >> # of blocks, HTEX places\n        tasks evenly across blocks, which makes it rather difficult to ensure that\n        some blocks will reach 0% utilization. Consequently, this strategy can be\n        expected to scale down effectively only when # of workers, or tasks executing\n        per block is close to 1.\n\n        Args:\n            - tasks (task_ids): Not used here.\n        \"\"\"\n        self._general_strategy(status_list, tasks, strategy_type='htex')\n\n    def _general_strategy(self, status_list, tasks, *, strategy_type):\n        for exec_status in status_list:\n            executor = exec_status.executor\n            label = executor.label\n            if not executor.scaling_enabled:\n                continue\n\n            # Tasks that are either pending completion\n            active_tasks = executor.outstanding\n\n            status = exec_status.status\n\n            # FIXME we need to handle case where provider does not define these\n            # FIXME probably more of this logic should be moved to the provider\n            min_blocks = executor.provider.min_blocks\n            max_blocks = executor.provider.max_blocks\n            tasks_per_node = executor.workers_per_node\n\n            nodes_per_block = executor.provider.nodes_per_block\n            parallelism = executor.provider.parallelism\n\n            running = sum([1 for x in status.values() if x.state == JobState.RUNNING])\n            pending = sum([1 for x in status.values() if x.state == JobState.PENDING])\n            active_blocks = running + pending\n            active_slots = active_blocks * tasks_per_node * nodes_per_block\n\n            if hasattr(executor, 'connected_workers'):\n                logger.debug('Executor {} has {} active tasks, {}/{} running/pending blocks, and {} connected workers'.format(\n                    label, active_tasks, running, pending, executor.connected_workers))\n            else:\n                logger.debug('Executor {} has {} active tasks and {}/{} running/pending blocks'.format(\n                    label, active_tasks, running, pending))\n\n            # reset kill timer if executor has active tasks\n            if active_tasks > 0 and self.executors[executor.label]['idle_since']:\n                self.executors[executor.label]['idle_since'] = None\n\n            # Case 1\n            # No tasks.\n            if active_tasks == 0:\n                # Case 1a\n                # Fewer blocks that min_blocks\n                if active_blocks <= min_blocks:\n                    # Ignore\n                    # logger.debug(\"Strategy: Case.1a\")\n                    pass\n\n                # Case 1b\n                # More blocks than min_blocks. Scale down\n                else:\n                    # We want to make sure that max_idletime is reached\n                    # before killing off resources\n                    if not self.executors[executor.label]['idle_since']:\n                        logger.debug(\"Executor {} has 0 active tasks; starting kill timer (if idle time exceeds {}s, resources will be removed)\".format(\n                            label, self.max_idletime)\n                        )\n                        self.executors[executor.label]['idle_since'] = time.time()\n\n                    idle_since = self.executors[executor.label]['idle_since']\n                    if (time.time() - idle_since) > self.max_idletime:\n                        # We have resources idle for the max duration,\n                        # we have to scale_in now.\n                        logger.debug(\"Idle time has reached {}s for executor {}; removing resources\".format(\n                            self.max_idletime, label)\n                        )\n                        exec_status.scale_in(active_blocks - min_blocks)\n\n                    else:\n                        pass\n                        # logger.debug(\"Strategy: Case.1b. Waiting for timer : {0}\".format(idle_since))\n\n            # Case 2\n            # More tasks than the available slots.\n            elif (float(active_slots) / active_tasks) < parallelism:\n                # Case 2a\n                # We have the max blocks possible\n                if active_blocks >= max_blocks:\n                    # Ignore since we already have the max nodes\n                    # logger.debug(\"Strategy: Case.2a\")\n                    pass\n\n                # Case 2b\n                else:\n                    # logger.debug(\"Strategy: Case.2b\")\n                    excess_slots = math.ceil((active_tasks * parallelism) - active_slots)\n                    excess_blocks = math.ceil(float(excess_slots) / (tasks_per_node * nodes_per_block))\n                    excess_blocks = min(excess_blocks, max_blocks - active_blocks)\n                    logger.debug(\"Requesting {} more blocks\".format(excess_blocks))\n                    exec_status.scale_out(excess_blocks)\n\n            elif active_slots == 0 and active_tasks > 0:\n                # Case 4\n                logger.debug(\"Requesting single slot\")\n                if active_blocks < max_blocks:\n                    exec_status.scale_out(1)\n\n            # Case 4\n            # More slots than tasks\n            elif active_slots > 0 and active_slots > active_tasks:\n                if strategy_type == 'htex':\n                    # Scale down for htex\n                    logger.debug(\"More slots than tasks\")\n                    if isinstance(executor, HighThroughputExecutor):\n                        if active_blocks > min_blocks:\n                            excess_slots = math.ceil(active_slots - (active_tasks * parallelism))\n                            excess_blocks = math.ceil(float(excess_slots) / (tasks_per_node * nodes_per_block))\n                            excess_blocks = min(excess_blocks, active_blocks - min_blocks)\n                            exec_status.scale_in(excess_blocks, force=False, max_idletime=self.max_idletime)\n\n                elif strategy_type == 'simple':\n                    # skip for simple strategy\n                    pass\n\n            # Case 3\n            # tasks ~ slots\n            else:\n                # logger.debug(\"Strategy: Case 3\")\n                pass\n", "path": "parsl/dataflow/strategy.py"}]}