We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Simple strategy is broken since PR #1519
**Describe the bug**
As of 6d66cab31bda9f0f7a3c0d081ce8cf658c10643a (PR #1519) the following exception has started being logged from the simple strategy, and I think probably simple strategy does not function properly.

```
2020-01-09 13:51:29.561 parsl.dataflow.flow_control:142 [ERROR]  Flow control callback threw an exception
- logging and proceeding anyway
Traceback (most recent call last):
  File "/home/benc/parsl/src/parsl/parsl/dataflow/flow_control.py", line 140, in make_callback
    self.callback(tasks=self._event_buffer, kind=kind)
  File "/home/benc/parsl/src/parsl/parsl/dataflow/strategy.py", line 192, in _strategy_simple
    running = sum([1 for x in status if x.state == JobState.RUNNING])
  File "/home/benc/parsl/src/parsl/parsl/dataflow/strategy.py", line 192, in <listcomp>
    running = sum([1 for x in status if x.state == JobState.RUNNING])
AttributeError: 'int' object has no attribute 'state'
```

It looks like this comes from changing the type of `executor.status()` in #1519 from a List of job statuses to a dict; and so now the keys of the dict are being iterated over rather than the job statuses in the simple strategy, because iteration over a dict happens over keys not values.



**To Reproduce**
```
$ git checkout 6d66cab31bda9f0f7a3c0d081ce8cf658c10643a
$ pytest parsl/ --config parsl/tests/configs/htex_local_alternate.py ; grep "Flow control callback threw" runinfo_*/000/parsl.log
```
**Expected behavior**
This exception should not be happening.

**Environment**
My laptop


--- 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 from parsl.providers.provider_base import JobState
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 = 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, '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 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, tasks, *args, kind=None, **kwargs):
132         """Do nothing.
133 
134         Args:
135             - tasks (task_ids): Not used here.
136 
137         KWargs:
138             - kind (Not used)
139         """
140 
141     def unset_logging(self):
142         """ Mute newly added handlers to the root level, right after calling executor.status
143         """
144         if self.logger_flag is True:
145             return
146 
147         root_logger = logging.getLogger()
148 
149         for hndlr in root_logger.handlers:
150             if hndlr not in self.prior_loghandlers:
151                 hndlr.setLevel(logging.ERROR)
152 
153         self.logger_flag = True
154 
155     def _strategy_simple(self, tasks, *args, kind=None, **kwargs):
156         """Peek at the DFK and the executors specified.
157 
158         We assume here that tasks are not held in a runnable
159         state, and that all tasks from an app would be sent to
160         a single specific executor, i.e tasks cannot be specified
161         to go to one of more executors.
162 
163         Args:
164             - tasks (task_ids): Not used here.
165 
166         KWargs:
167             - kind (Not used)
168         """
169 
170         for label, executor in self.dfk.executors.items():
171             if not executor.scaling_enabled:
172                 continue
173 
174             # Tasks that are either pending completion
175             active_tasks = executor.outstanding
176 
177             status = executor.status()
178             self.unset_logging()
179 
180             # FIXME we need to handle case where provider does not define these
181             # FIXME probably more of this logic should be moved to the provider
182             min_blocks = executor.provider.min_blocks
183             max_blocks = executor.provider.max_blocks
184             if isinstance(executor, IPyParallelExecutor) or isinstance(executor, HighThroughputExecutor):
185                 tasks_per_node = executor.workers_per_node
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.state == JobState.RUNNING])
193             pending = sum([1 for x in status if x.state == JobState.PENDING])
194             active_blocks = running + pending
195             active_slots = active_blocks * tasks_per_node * nodes_per_block
196 
197             if hasattr(executor, 'connected_workers'):
198                 logger.debug('Executor {} has {} active tasks, {}/{} running/pending blocks, and {} connected workers'.format(
199                     label, active_tasks, running, pending, executor.connected_workers))
200             else:
201                 logger.debug('Executor {} has {} active tasks and {}/{} running/pending blocks'.format(
202                     label, active_tasks, running, pending))
203 
204             # reset kill timer if executor has active tasks
205             if active_tasks > 0 and self.executors[executor.label]['idle_since']:
206                 self.executors[executor.label]['idle_since'] = None
207 
208             # Case 1
209             # No tasks.
210             if active_tasks == 0:
211                 # Case 1a
212                 # Fewer blocks that min_blocks
213                 if active_blocks <= min_blocks:
214                     # Ignore
215                     # logger.debug("Strategy: Case.1a")
216                     pass
217 
218                 # Case 1b
219                 # More blocks than min_blocks. Scale down
220                 else:
221                     # We want to make sure that max_idletime is reached
222                     # before killing off resources
223                     if not self.executors[executor.label]['idle_since']:
224                         logger.debug("Executor {} has 0 active tasks; starting kill timer (if idle time exceeds {}s, resources will be removed)".format(
225                             label, self.max_idletime)
226                         )
227                         self.executors[executor.label]['idle_since'] = time.time()
228 
229                     idle_since = self.executors[executor.label]['idle_since']
230                     if (time.time() - idle_since) > self.max_idletime:
231                         # We have resources idle for the max duration,
232                         # we have to scale_in now.
233                         logger.debug("Idle time has reached {}s for executor {}; removing resources".format(
234                             self.max_idletime, label)
235                         )
236                         executor.scale_in(active_blocks - min_blocks)
237 
238                     else:
239                         pass
240                         # logger.debug("Strategy: Case.1b. Waiting for timer : {0}".format(idle_since))
241 
242             # Case 2
243             # More tasks than the available slots.
244             elif (float(active_slots) / active_tasks) < parallelism:
245                 # Case 2a
246                 # We have the max blocks possible
247                 if active_blocks >= max_blocks:
248                     # Ignore since we already have the max nodes
249                     # logger.debug("Strategy: Case.2a")
250                     pass
251 
252                 # Case 2b
253                 else:
254                     # logger.debug("Strategy: Case.2b")
255                     excess = math.ceil((active_tasks * parallelism) - active_slots)
256                     excess_blocks = math.ceil(float(excess) / (tasks_per_node * nodes_per_block))
257                     excess_blocks = min(excess_blocks, max_blocks - active_blocks)
258                     logger.debug("Requesting {} more blocks".format(excess_blocks))
259                     executor.scale_out(excess_blocks)
260 
261             elif active_slots == 0 and active_tasks > 0:
262                 # Case 4
263                 # Check if slots are being lost quickly ?
264                 logger.debug("Requesting single slot")
265                 if active_blocks < max_blocks:
266                     executor.scale_out(1)
267             # Case 3
268             # tasks ~ slots
269             else:
270                 # logger.debug("Strategy: Case 3")
271                 pass
272
```
--- 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
@@ -189,8 +189,8 @@
             nodes_per_block = executor.provider.nodes_per_block
             parallelism = executor.provider.parallelism
 
-            running = sum([1 for x in status if x.state == JobState.RUNNING])
-            pending = sum([1 for x in status if x.state == JobState.PENDING])
+            running = sum([1 for x in status.values() if x.state == JobState.RUNNING])
+            pending = sum([1 for x in status.values() if x.state == JobState.PENDING])
             active_blocks = running + pending
             active_slots = active_blocks * tasks_per_node * nodes_per_block

{"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@@ -189,8 +189,8 @@\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.state == JobState.RUNNING])\n-            pending = sum([1 for x in status if x.state == JobState.PENDING])\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", "issue": "Simple strategy is broken since PR #1519\n**Describe the bug**\r\nAs of 6d66cab31bda9f0f7a3c0d081ce8cf658c10643a (PR #1519) the following exception has started being logged from the simple strategy, and I think probably simple strategy does not function properly.\r\n\r\n```\r\n2020-01-09 13:51:29.561 parsl.dataflow.flow_control:142 [ERROR]  Flow control callback threw an exception\r\n- logging and proceeding anyway\r\nTraceback (most recent call last):\r\n  File \"/home/benc/parsl/src/parsl/parsl/dataflow/flow_control.py\", line 140, in make_callback\r\n    self.callback(tasks=self._event_buffer, kind=kind)\r\n  File \"/home/benc/parsl/src/parsl/parsl/dataflow/strategy.py\", line 192, in _strategy_simple\r\n    running = sum([1 for x in status if x.state == JobState.RUNNING])\r\n  File \"/home/benc/parsl/src/parsl/parsl/dataflow/strategy.py\", line 192, in <listcomp>\r\n    running = sum([1 for x in status if x.state == JobState.RUNNING])\r\nAttributeError: 'int' object has no attribute 'state'\r\n```\r\n\r\nIt looks like this comes from changing the type of `executor.status()` in #1519 from a List of job statuses to a dict; and so now the keys of the dict are being iterated over rather than the job statuses in the simple strategy, because iteration over a dict happens over keys not values.\r\n\r\n\r\n\r\n**To Reproduce**\r\n```\r\n$ git checkout 6d66cab31bda9f0f7a3c0d081ce8cf658c10643a\r\n$ pytest parsl/ --config parsl/tests/configs/htex_local_alternate.py ; grep \"Flow control callback threw\" runinfo_*/000/parsl.log\r\n```\r\n**Expected behavior**\r\nThis exception should not be happening.\r\n\r\n**Environment**\r\nMy laptop\r\n\n", "before_files": [{"content": "import logging\nimport time\nimport math\n\nfrom parsl.executors import IPyParallelExecutor, HighThroughputExecutor, ExtremeScaleExecutor\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          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 = 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, '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 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, 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) or isinstance(executor, HighThroughputExecutor):\n                tasks_per_node = executor.workers_per_node\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.state == JobState.RUNNING])\n            pending = sum([1 for x in status 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                        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                    excess_blocks = min(excess_blocks, max_blocks - active_blocks)\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                if active_blocks < max_blocks:\n                    executor.scale_out(1)\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\n\nfrom parsl.executors import IPyParallelExecutor, HighThroughputExecutor, ExtremeScaleExecutor\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          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 = 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, '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 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, 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) or isinstance(executor, HighThroughputExecutor):\n                tasks_per_node = executor.workers_per_node\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.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                        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                    excess_blocks = min(excess_blocks, max_blocks - active_blocks)\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                if active_blocks < max_blocks:\n                    executor.scale_out(1)\n            # Case 3\n            # tasks ~ slots\n            else:\n                # logger.debug(\"Strategy: Case 3\")\n                pass\n", "path": "parsl/dataflow/strategy.py"}]}