We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Executor idle timer not reset
TLDR; I think there's a bug in assessing timeout for idle executors.

Parsl log: https://gist.github.com/macintoshpie/1ae87cd93e45e532513ea2833a3f8e60

### Setup
Running sequence of 5 tasks that each last about 3 hours on NSCC Aspire1, on a single executor.

### Problem
The first task completes after about 3 hours but then the executor is killed and the parsl head errors out.
See this line for where things go awry: https://gist.github.com/macintoshpie/1ae87cd93e45e532513ea2833a3f8e60#file-parsl-log-L5197

### Proposed bug
You can see in [this line](https://gist.github.com/macintoshpie/1ae87cd93e45e532513ea2833a3f8e60#file-parsl-log-L88) that the `idle_time` is set early in the process. There's no logging that the kill timer is being reset/updated later on either. Once the first task finishes, it meets the conditions to enter a potential timeout state (`active_tasks == 0 && active_blocks > min_blocks`). It looks at the very old `idle_time` set earlier and kills the executor immediately.

The solution is to reset the timer once it's no longer in the possible timeout state. Here's the problematic area of code: https://github.com/Parsl/parsl/blob/b4c55567b468278e15516e89373bc0a72d445317/parsl/dataflow/strategy.py#L209-L228

--- 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 
5 from parsl.executors import IPyParallelExecutor, HighThroughputExecutor, ExtremeScaleExecutor
6 
7 
8 logger = logging.getLogger(__name__)
9 
10 
11 class Strategy(object):
12     """FlowControl strategy.
13 
14     As a workflow dag is processed by Parsl, new tasks are added and completed
15     asynchronously. Parsl interfaces executors with execution providers to construct
16     scalable executors to handle the variable work-load generated by the
17     workflow. This component is responsible for periodically checking outstanding
18     tasks and available compute capacity and trigger scaling events to match
19     workflow needs.
20 
21     Here's a diagram of an executor. An executor consists of blocks, which are usually
22     created by single requests to a Local Resource Manager (LRM) such as slurm,
23     condor, torque, or even AWS API. The blocks could contain several task blocks
24     which are separate instances on workers.
25 
26 
27     .. code:: python
28 
29                 |<--min_blocks     |<-init_blocks              max_blocks-->|
30                 +----------------------------------------------------------+
31                 |  +--------block----------+       +--------block--------+ |
32      executor = |  | task          task    | ...   |    task      task   | |
33                 |  +-----------------------+       +---------------------+ |
34                 +----------------------------------------------------------+
35 
36     The relevant specification options are:
37        1. min_blocks: Minimum number of blocks to maintain
38        2. init_blocks: number of blocks to provision at initialization of workflow
39        3. max_blocks: Maximum number of blocks that can be active due to one workflow
40 
41 
42     .. code:: python
43 
44           slots = current_capacity * tasks_per_node * nodes_per_block
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 = 60 * 2  # 2 minutes
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, 'simple': self._strategy_simple}
120 
121         self.strategize = self.strategies[self.config.strategy]
122         self.logger_flag = False
123         self.prior_loghandlers = set(logging.getLogger().handlers)
124 
125         logger.debug("Scaling strategy: {0}".format(self.config.strategy))
126 
127     def _strategy_noop(self, tasks, *args, kind=None, **kwargs):
128         """Do nothing.
129 
130         Args:
131             - tasks (task_ids): Not used here.
132 
133         KWargs:
134             - kind (Not used)
135         """
136 
137     def unset_logging(self):
138         """ Mute newly added handlers to the root level, right after calling executor.status
139         """
140         if self.logger_flag is True:
141             return
142 
143         root_logger = logging.getLogger()
144 
145         for hndlr in root_logger.handlers:
146             if hndlr not in self.prior_loghandlers:
147                 hndlr.setLevel(logging.ERROR)
148 
149         self.logger_flag = True
150 
151     def _strategy_simple(self, tasks, *args, kind=None, **kwargs):
152         """Peek at the DFK and the executors specified.
153 
154         We assume here that tasks are not held in a runnable
155         state, and that all tasks from an app would be sent to
156         a single specific executor, i.e tasks cannot be specified
157         to go to one of more executors.
158 
159         Args:
160             - tasks (task_ids): Not used here.
161 
162         KWargs:
163             - kind (Not used)
164         """
165 
166         for label, executor in self.dfk.executors.items():
167             if not executor.scaling_enabled:
168                 continue
169 
170             # Tasks that are either pending completion
171             active_tasks = executor.outstanding
172 
173             status = executor.status()
174             self.unset_logging()
175 
176             # FIXME we need to handle case where provider does not define these
177             # FIXME probably more of this logic should be moved to the provider
178             min_blocks = executor.provider.min_blocks
179             max_blocks = executor.provider.max_blocks
180             if isinstance(executor, IPyParallelExecutor):
181                 tasks_per_node = executor.workers_per_node
182             elif isinstance(executor, HighThroughputExecutor):
183                 # This is probably wrong calculation, we need this to come from the executor
184                 # since we can't know slots ahead of time.
185                 tasks_per_node = 1
186             elif isinstance(executor, ExtremeScaleExecutor):
187                 tasks_per_node = executor.ranks_per_node
188 
189             nodes_per_block = executor.provider.nodes_per_block
190             parallelism = executor.provider.parallelism
191 
192             running = sum([1 for x in status if x == 'RUNNING'])
193             submitting = sum([1 for x in status if x == 'SUBMITTING'])
194             pending = sum([1 for x in status if x == 'PENDING'])
195             active_blocks = running + submitting + pending
196             active_slots = active_blocks * tasks_per_node * nodes_per_block
197 
198             if (isinstance(executor, IPyParallelExecutor) or
199                 isinstance(executor, HighThroughputExecutor) or
200                 isinstance(executor, ExtremeScaleExecutor)):
201                 logger.debug('Executor {} has {} active tasks, {}/{}/{} running/submitted/pending blocks, and {} connected engines'.format(
202                     label, active_tasks, running, submitting, pending, len(executor.connected_workers)))
203             else:
204                 logger.debug('Executor {} has {} active tasks and {}/{}/{} running/submitted/pending blocks'.format(
205                     label, active_tasks, running, submitting, pending))
206 
207             # Case 1
208             # No tasks.
209             if active_tasks == 0:
210                 # Case 1a
211                 # Fewer blocks that min_blocks
212                 if active_blocks <= min_blocks:
213                     # Ignore
214                     # logger.debug("Strategy: Case.1a")
215                     pass
216 
217                 # Case 1b
218                 # More blocks than min_blocks. Scale down
219                 else:
220                     # We want to make sure that max_idletime is reached
221                     # before killing off resources
222                     if not self.executors[executor.label]['idle_since']:
223                         logger.debug("Executor {} has 0 active tasks; starting kill timer (if idle time exceeds {}s, resources will be removed)".format(
224                             label, self.max_idletime)
225                         )
226                         self.executors[executor.label]['idle_since'] = time.time()
227 
228                     idle_since = self.executors[executor.label]['idle_since']
229                     if (time.time() - idle_since) > self.max_idletime:
230                         # We have resources idle for the max duration,
231                         # we have to scale_in now.
232                         logger.debug("Idle time has reached {}s for executor {}; removing resources".format(
233                             self.max_idletime, label)
234                         )
235                         executor.scale_in(active_blocks - min_blocks)
236 
237                     else:
238                         pass
239                         # logger.debug("Strategy: Case.1b. Waiting for timer : {0}".format(idle_since))
240 
241             # Case 2
242             # More tasks than the available slots.
243             elif (float(active_slots) / active_tasks) < parallelism:
244                 # Case 2a
245                 # We have the max blocks possible
246                 if active_blocks >= max_blocks:
247                     # Ignore since we already have the max nodes
248                     # logger.debug("Strategy: Case.2a")
249                     pass
250 
251                 # Case 2b
252                 else:
253                     # logger.debug("Strategy: Case.2b")
254                     excess = math.ceil((active_tasks * parallelism) - active_slots)
255                     excess_blocks = math.ceil(float(excess) / (tasks_per_node * nodes_per_block))
256                     logger.debug("Requesting {} more blocks".format(excess_blocks))
257                     executor.scale_out(excess_blocks)
258 
259             elif active_slots == 0 and active_tasks > 0:
260                 # Case 4
261                 # Check if slots are being lost quickly ?
262                 logger.debug("Requesting single slot")
263                 executor.scale_out(1)
264             # Case 3
265             # tasks ~ slots
266             else:
267                 # logger.debug("Strategy: Case 3")
268                 pass
269 
270 
271 if __name__ == '__main__':
272 
273     pass
274
```
--- 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
@@ -204,6 +204,10 @@
                 logger.debug('Executor {} has {} active tasks and {}/{}/{} running/submitted/pending blocks'.format(
                     label, active_tasks, running, submitting, pending))
 
+            # reset kill timer if executor has active tasks
+            if active_tasks > 0 and self.executors[executor.label]['idle_since']:
+                self.executors[executor.label]['idle_since'] = None
+
             # Case 1
             # No tasks.
             if active_tasks == 0:

{"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@@ -204,6 +204,10 @@\n                 logger.debug('Executor {} has {} active tasks and {}/{}/{} running/submitted/pending blocks'.format(\n                     label, active_tasks, running, submitting, 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", "issue": "Executor idle timer not reset\nTLDR; I think there's a bug in assessing timeout for idle executors.\r\n\r\nParsl log: https://gist.github.com/macintoshpie/1ae87cd93e45e532513ea2833a3f8e60\r\n\r\n### Setup\r\nRunning sequence of 5 tasks that each last about 3 hours on NSCC Aspire1, on a single executor.\r\n\r\n### Problem\r\nThe first task completes after about 3 hours but then the executor is killed and the parsl head errors out.\r\nSee this line for where things go awry: https://gist.github.com/macintoshpie/1ae87cd93e45e532513ea2833a3f8e60#file-parsl-log-L5197\r\n\r\n### Proposed bug\r\nYou can see in [this line](https://gist.github.com/macintoshpie/1ae87cd93e45e532513ea2833a3f8e60#file-parsl-log-L88) that the `idle_time` is set early in the process. There's no logging that the kill timer is being reset/updated later on either. Once the first task finishes, it meets the conditions to enter a potential timeout state (`active_tasks == 0 && active_blocks > min_blocks`). It looks at the very old `idle_time` set earlier and kills the executor immediately.\r\n\r\nThe solution is to reset the timer once it's no longer in the possible timeout state. Here's the problematic area of code: https://github.com/Parsl/parsl/blob/b4c55567b468278e15516e89373bc0a72d445317/parsl/dataflow/strategy.py#L209-L228\n", "before_files": [{"content": "import logging\nimport time\nimport math\n\nfrom parsl.executors import IPyParallelExecutor, HighThroughputExecutor, ExtremeScaleExecutor\n\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          slots = current_capacity * tasks_per_node * nodes_per_block\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 = 60 * 2  # 2 minutes\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, 'simple': self._strategy_simple}\n\n        self.strategize = self.strategies[self.config.strategy]\n        self.logger_flag = False\n        self.prior_loghandlers = set(logging.getLogger().handlers)\n\n        logger.debug(\"Scaling strategy: {0}\".format(self.config.strategy))\n\n    def _strategy_noop(self, tasks, *args, kind=None, **kwargs):\n        \"\"\"Do nothing.\n\n        Args:\n            - tasks (task_ids): Not used here.\n\n        KWargs:\n            - kind (Not used)\n        \"\"\"\n\n    def unset_logging(self):\n        \"\"\" Mute newly added handlers to the root level, right after calling executor.status\n        \"\"\"\n        if self.logger_flag is True:\n            return\n\n        root_logger = logging.getLogger()\n\n        for hndlr in root_logger.handlers:\n            if hndlr not in self.prior_loghandlers:\n                hndlr.setLevel(logging.ERROR)\n\n        self.logger_flag = True\n\n    def _strategy_simple(self, tasks, *args, kind=None, **kwargs):\n        \"\"\"Peek at the DFK and the executors specified.\n\n        We assume here that tasks are not held in a runnable\n        state, and that all tasks from an app would be sent to\n        a single specific executor, i.e tasks cannot be specified\n        to go to one of more executors.\n\n        Args:\n            - tasks (task_ids): Not used here.\n\n        KWargs:\n            - kind (Not used)\n        \"\"\"\n\n        for label, executor in self.dfk.executors.items():\n            if not executor.scaling_enabled:\n                continue\n\n            # Tasks that are either pending completion\n            active_tasks = executor.outstanding\n\n            status = executor.status()\n            self.unset_logging()\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            if isinstance(executor, IPyParallelExecutor):\n                tasks_per_node = executor.workers_per_node\n            elif isinstance(executor, HighThroughputExecutor):\n                # This is probably wrong calculation, we need this to come from the executor\n                # since we can't know slots ahead of time.\n                tasks_per_node = 1\n            elif isinstance(executor, ExtremeScaleExecutor):\n                tasks_per_node = executor.ranks_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 if x == 'RUNNING'])\n            submitting = sum([1 for x in status if x == 'SUBMITTING'])\n            pending = sum([1 for x in status if x == 'PENDING'])\n            active_blocks = running + submitting + pending\n            active_slots = active_blocks * tasks_per_node * nodes_per_block\n\n            if (isinstance(executor, IPyParallelExecutor) or\n                isinstance(executor, HighThroughputExecutor) or\n                isinstance(executor, ExtremeScaleExecutor)):\n                logger.debug('Executor {} has {} active tasks, {}/{}/{} running/submitted/pending blocks, and {} connected engines'.format(\n                    label, active_tasks, running, submitting, pending, len(executor.connected_workers)))\n            else:\n                logger.debug('Executor {} has {} active tasks and {}/{}/{} running/submitted/pending blocks'.format(\n                    label, active_tasks, running, submitting, pending))\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                        executor.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                    logger.debug(\"Requesting {} more blocks\".format(excess_blocks))\n                    executor.scale_out(excess_blocks)\n\n            elif active_slots == 0 and active_tasks > 0:\n                # Case 4\n                # Check if slots are being lost quickly ?\n                logger.debug(\"Requesting single slot\")\n                executor.scale_out(1)\n            # Case 3\n            # tasks ~ slots\n            else:\n                # logger.debug(\"Strategy: Case 3\")\n                pass\n\n\nif __name__ == '__main__':\n\n    pass\n", "path": "parsl/dataflow/strategy.py"}], "after_files": [{"content": "import logging\nimport time\nimport math\n\nfrom parsl.executors import IPyParallelExecutor, HighThroughputExecutor, ExtremeScaleExecutor\n\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          slots = current_capacity * tasks_per_node * nodes_per_block\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 = 60 * 2  # 2 minutes\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, 'simple': self._strategy_simple}\n\n        self.strategize = self.strategies[self.config.strategy]\n        self.logger_flag = False\n        self.prior_loghandlers = set(logging.getLogger().handlers)\n\n        logger.debug(\"Scaling strategy: {0}\".format(self.config.strategy))\n\n    def _strategy_noop(self, tasks, *args, kind=None, **kwargs):\n        \"\"\"Do nothing.\n\n        Args:\n            - tasks (task_ids): Not used here.\n\n        KWargs:\n            - kind (Not used)\n        \"\"\"\n\n    def unset_logging(self):\n        \"\"\" Mute newly added handlers to the root level, right after calling executor.status\n        \"\"\"\n        if self.logger_flag is True:\n            return\n\n        root_logger = logging.getLogger()\n\n        for hndlr in root_logger.handlers:\n            if hndlr not in self.prior_loghandlers:\n                hndlr.setLevel(logging.ERROR)\n\n        self.logger_flag = True\n\n    def _strategy_simple(self, tasks, *args, kind=None, **kwargs):\n        \"\"\"Peek at the DFK and the executors specified.\n\n        We assume here that tasks are not held in a runnable\n        state, and that all tasks from an app would be sent to\n        a single specific executor, i.e tasks cannot be specified\n        to go to one of more executors.\n\n        Args:\n            - tasks (task_ids): Not used here.\n\n        KWargs:\n            - kind (Not used)\n        \"\"\"\n\n        for label, executor in self.dfk.executors.items():\n            if not executor.scaling_enabled:\n                continue\n\n            # Tasks that are either pending completion\n            active_tasks = executor.outstanding\n\n            status = executor.status()\n            self.unset_logging()\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            if isinstance(executor, IPyParallelExecutor):\n                tasks_per_node = executor.workers_per_node\n            elif isinstance(executor, HighThroughputExecutor):\n                # This is probably wrong calculation, we need this to come from the executor\n                # since we can't know slots ahead of time.\n                tasks_per_node = 1\n            elif isinstance(executor, ExtremeScaleExecutor):\n                tasks_per_node = executor.ranks_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 if x == 'RUNNING'])\n            submitting = sum([1 for x in status if x == 'SUBMITTING'])\n            pending = sum([1 for x in status if x == 'PENDING'])\n            active_blocks = running + submitting + pending\n            active_slots = active_blocks * tasks_per_node * nodes_per_block\n\n            if (isinstance(executor, IPyParallelExecutor) or\n                isinstance(executor, HighThroughputExecutor) or\n                isinstance(executor, ExtremeScaleExecutor)):\n                logger.debug('Executor {} has {} active tasks, {}/{}/{} running/submitted/pending blocks, and {} connected engines'.format(\n                    label, active_tasks, running, submitting, pending, len(executor.connected_workers)))\n            else:\n                logger.debug('Executor {} has {} active tasks and {}/{}/{} running/submitted/pending blocks'.format(\n                    label, active_tasks, running, submitting, 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                        executor.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                    logger.debug(\"Requesting {} more blocks\".format(excess_blocks))\n                    executor.scale_out(excess_blocks)\n\n            elif active_slots == 0 and active_tasks > 0:\n                # Case 4\n                # Check if slots are being lost quickly ?\n                logger.debug(\"Requesting single slot\")\n                executor.scale_out(1)\n            # Case 3\n            # tasks ~ slots\n            else:\n                # logger.debug(\"Strategy: Case 3\")\n                pass\n\n\nif __name__ == '__main__':\n\n    pass\n", "path": "parsl/dataflow/strategy.py"}]}