We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Duplicate log lines when using IPP
The following code fragment emits log lines to STDERR with the default format style even though no logging is requested. I suspect that `ipyparallel` is doing something dirty.

    from parsl import DataFlowKernel
    from parsl.configs.local import localIPP as config

    dfk = DataFlowKernel(config=config)
    dfk.cleanup()

The above code with the minor change of using threads will not emit the log lines.


    from parsl import DataFlowKernel
    from parsl.configs.local import localThreads as config

    dfk = DataFlowKernel(config=config)
    dfk.cleanup()


Please help test by running this with the latest parsl code. 

--- 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 logger = logging.getLogger(__name__)
6 
7 
8 class Strategy(object):
9     """FlowControl strategy.
10 
11     As a workflow dag is processed by Parsl, new tasks are added and completed
12     asynchronously. Parsl interfaces executors with execution providers to construct
13     scalable executors to handle the variable work-load generated by the
14     workflow. This component is responsible for periodically checking outstanding
15     tasks and available compute capacity and trigger scaling events to match
16     workflow needs.
17 
18     Here's a diagram of an executor. An executor consists of blocks, which are usually
19     created by single requests to a Local Resource Manager (LRM) such as slurm,
20     condor, torque, or even AWS API. The blocks could contain several task blocks
21     which are separate instances on workers.
22 
23 
24     .. code:: python
25 
26                 |<--min_blocks     |<-init_blocks              max_blocks-->|
27                 +----------------------------------------------------------+
28                 |  +--------block----------+       +--------block--------+ |
29      executor = |  | task          task    | ...   |    task      task   | |
30                 |  +-----------------------+       +---------------------+ |
31                 +----------------------------------------------------------+
32 
33     The relevant specification options are:
34        1. min_blocks: Minimum number of blocks to maintain
35        2. init_blocks: number of blocks to provision at initialization of workflow
36        3. max_blocks: Maximum number of blocks that can be active due to one workflow
37 
38 
39     .. code:: python
40 
41           slots = current_capacity * tasks_per_node * nodes_per_block
42 
43           active_tasks = pending_tasks + running_tasks
44 
45           Parallelism = slots / tasks
46                       = [0, 1] (i.e,  0 <= p <= 1)
47 
48     For example:
49 
50     When p = 0,
51          => compute with the least resources possible.
52          infinite tasks are stacked per slot.
53 
54          .. code:: python
55 
56                blocks =  min_blocks           { if active_tasks = 0
57                          max(min_blocks, 1)   {  else
58 
59     When p = 1,
60          => compute with the most resources.
61          one task is stacked per slot.
62 
63          .. code:: python
64 
65                blocks = min ( max_blocks,
66                         ceil( active_tasks / slots ) )
67 
68 
69     When p = 1/2,
70          => We stack upto 2 tasks per slot before we overflow
71          and request a new block
72 
73 
74     let's say min:init:max = 0:0:4 and task_blocks=2
75     Consider the following example:
76     min_blocks = 0
77     init_blocks = 0
78     max_blocks = 4
79     tasks_per_node = 2
80     nodes_per_block = 1
81 
82     In the diagram, X <- task
83 
84     at 2 tasks:
85 
86     .. code:: python
87 
88         +---Block---|
89         |           |
90         | X      X  |
91         |slot   slot|
92         +-----------+
93 
94     at 5 tasks, we overflow as the capacity of a single block is fully used.
95 
96     .. code:: python
97 
98         +---Block---|       +---Block---|
99         | X      X  | ----> |           |
100         | X      X  |       | X         |
101         |slot   slot|       |slot   slot|
102         +-----------+       +-----------+
103 
104     """
105 
106     def __init__(self, dfk):
107         """Initialize strategy."""
108         self.dfk = dfk
109         self.config = dfk.config
110         self.executors = {}
111         self.max_idletime = 60 * 2  # 2 minutes
112 
113         for e in self.dfk.config.executors:
114             self.executors[e.label] = {'idle_since': None, 'config': e.label}
115 
116         self.strategies = {None: self._strategy_noop, 'simple': self._strategy_simple}
117 
118         self.strategize = self.strategies[self.config.strategy]
119 
120         logger.debug("Scaling strategy: {0}".format(self.config.strategy))
121 
122     def _strategy_noop(self, tasks, *args, kind=None, **kwargs):
123         """Do nothing.
124 
125         Args:
126             - tasks (task_ids): Not used here.
127 
128         KWargs:
129             - kind (Not used)
130         """
131 
132     def _strategy_simple(self, tasks, *args, kind=None, **kwargs):
133         """Peek at the DFK and the executors specified.
134 
135         We assume here that tasks are not held in a runnable
136         state, and that all tasks from an app would be sent to
137         a single specific executor, i.e tasks cannot be specified
138         to go to one of more executors.
139 
140         Args:
141             - tasks (task_ids): Not used here.
142 
143         KWargs:
144             - kind (Not used)
145         """
146         # Add logic to check executors
147         # for task in tasks :
148         #    if self.dfk.tasks[task]:
149 
150         for label, executor in self.dfk.executors.items():
151             if not executor.scaling_enabled:
152                 continue
153 
154             # Tasks that are either pending completion
155             active_tasks = executor.executor.outstanding
156 
157             status = executor.status()
158 
159             # FIXME we need to handle case where provider does not define these
160             # FIXME probably more of this logic should be moved to the provider
161             min_blocks = executor.provider.min_blocks
162             max_blocks = executor.provider.max_blocks
163             tasks_per_node = executor.provider.tasks_per_node
164             nodes_per_block = executor.provider.nodes_per_block
165             parallelism = executor.provider.parallelism
166 
167             active_blocks = sum([1 for x in status if x in ('RUNNING',
168                                                             'SUBMITTING',
169                                                             'PENDING')])
170             active_slots = active_blocks * tasks_per_node * nodes_per_block
171 
172             # import pdb; pdb.set_trace()
173             logger.debug("Tasks:{} Slots:{} Parallelism:{}".format(len(active_tasks),
174                                                                    active_slots,
175                                                                    parallelism))
176 
177             # Case 1
178             # No tasks.
179             if len(active_tasks) == 0:
180                 # Case 1a
181                 # Fewer blocks that min_blocks
182                 if active_blocks <= min_blocks:
183                     # Ignore
184                     # logger.debug("Strategy: Case.1a")
185                     pass
186 
187                 # Case 1b
188                 # More blocks than min_blocks. Scale down
189                 else:
190                     # We want to make sure that max_idletime is reached
191                     # before killing off resources
192                     if not self.executors[executor.label]['idle_since']:
193                         logger.debug("Strategy: Scale_in, tasks=0 starting kill timer")
194                         self.executors[executor.label]['idle_since'] = time.time()
195 
196                     idle_since = self.executors[executor.label]['idle_since']
197                     if (time.time() - idle_since) > self.max_idletime:
198                         # We have resources idle for the max duration,
199                         # we have to scale_in now.
200                         logger.debug("Strategy: Scale_in, tasks=0")
201                         executor.scale_in(active_blocks - min_blocks)
202 
203                     else:
204                         pass
205                         # logger.debug("Strategy: Case.1b. Waiting for timer : {0}".format(idle_since))
206 
207             # Case 2
208             # More tasks than the available slots.
209             elif (float(active_slots) / len(active_tasks)) < parallelism:
210                 # Case 2a
211                 # We have the max blocks possible
212                 if active_blocks >= max_blocks:
213                     # Ignore since we already have the max nodes
214                     # logger.debug("Strategy: Case.2a")
215                     pass
216 
217                 # Case 2b
218                 else:
219                     # logger.debug("Strategy: Case.2b")
220                     excess = math.ceil((len(active_tasks) * parallelism) - active_slots)
221                     excess_blocks = math.ceil(float(excess) / (tasks_per_node * nodes_per_block))
222                     logger.debug("Requesting {} more blocks".format(excess_blocks))
223                     executor.scale_out(excess_blocks)
224 
225             elif active_slots == 0 and len(active_tasks) > 0:
226                 # Case 4
227                 # Check if slots are being lost quickly ?
228                 logger.debug("Requesting single slot")
229                 executor.scale_out(1)
230             # Case 3
231             # tasks ~ slots
232             else:
233                 # logger.debug("Strategy: Case 3")
234                 pass
235 
236 
237 if __name__ == '__main__':
238 
239     pass
240
```
--- 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
@@ -116,6 +116,8 @@
         self.strategies = {None: self._strategy_noop, 'simple': self._strategy_simple}
 
         self.strategize = self.strategies[self.config.strategy]
+        self.logger_flag = False
+        self.prior_loghandlers = set(logging.getLogger().handlers)
 
         logger.debug("Scaling strategy: {0}".format(self.config.strategy))
 
@@ -129,6 +131,20 @@
             - kind (Not used)
         """
 
+    def unset_logging(self):
+        """ Mute newly added handlers to the root level, right after calling executor.status
+        """
+        if self.logger_flag is True:
+            return
+
+        root_logger = logging.getLogger()
+
+        for hndlr in root_logger.handlers:
+            if hndlr not in self.prior_loghandlers:
+                hndlr.setLevel(logging.ERROR)
+
+        self.logger_flag = True
+
     def _strategy_simple(self, tasks, *args, kind=None, **kwargs):
         """Peek at the DFK and the executors specified.
 
@@ -143,9 +159,6 @@
         KWargs:
             - kind (Not used)
         """
-        # Add logic to check executors
-        # for task in tasks :
-        #    if self.dfk.tasks[task]:
 
         for label, executor in self.dfk.executors.items():
             if not executor.scaling_enabled:
@@ -155,6 +168,7 @@
             active_tasks = executor.executor.outstanding
 
             status = executor.status()
+            self.unset_logging()
 
             # FIXME we need to handle case where provider does not define these
             # FIXME probably more of this logic should be moved to the provider

{"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@@ -116,6 +116,8 @@\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@@ -129,6 +131,20 @@\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@@ -143,9 +159,6 @@\n         KWargs:\n             - kind (Not used)\n         \"\"\"\n-        # Add logic to check executors\n-        # for task in tasks :\n-        #    if self.dfk.tasks[task]:\n \n         for label, executor in self.dfk.executors.items():\n             if not executor.scaling_enabled:\n@@ -155,6 +168,7 @@\n             active_tasks = executor.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", "issue": "Duplicate log lines when using IPP\nThe following code fragment emits log lines to STDERR with the default format style even though no logging is requested. I suspect that `ipyparallel` is doing something dirty.\r\n\r\n    from parsl import DataFlowKernel\r\n    from parsl.configs.local import localIPP as config\r\n\r\n    dfk = DataFlowKernel(config=config)\r\n    dfk.cleanup()\r\n\r\nThe above code with the minor change of using threads will not emit the log lines.\r\n\r\n\r\n    from parsl import DataFlowKernel\r\n    from parsl.configs.local import localThreads as config\r\n\r\n    dfk = DataFlowKernel(config=config)\r\n    dfk.cleanup()\r\n\r\n\r\nPlease help test by running this with the latest parsl code. \n", "before_files": [{"content": "import logging\nimport time\nimport math\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\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 _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        # Add logic to check executors\n        # for task in tasks :\n        #    if self.dfk.tasks[task]:\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.executor.outstanding\n\n            status = executor.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.provider.tasks_per_node\n            nodes_per_block = executor.provider.nodes_per_block\n            parallelism = executor.provider.parallelism\n\n            active_blocks = sum([1 for x in status if x in ('RUNNING',\n                                                            'SUBMITTING',\n                                                            'PENDING')])\n            active_slots = active_blocks * tasks_per_node * nodes_per_block\n\n            # import pdb; pdb.set_trace()\n            logger.debug(\"Tasks:{} Slots:{} Parallelism:{}\".format(len(active_tasks),\n                                                                   active_slots,\n                                                                   parallelism))\n\n            # Case 1\n            # No tasks.\n            if len(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(\"Strategy: Scale_in, tasks=0 starting kill timer\")\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(\"Strategy: Scale_in, tasks=0\")\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) / len(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((len(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 len(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\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.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            tasks_per_node = executor.provider.tasks_per_node\n            nodes_per_block = executor.provider.nodes_per_block\n            parallelism = executor.provider.parallelism\n\n            active_blocks = sum([1 for x in status if x in ('RUNNING',\n                                                            'SUBMITTING',\n                                                            'PENDING')])\n            active_slots = active_blocks * tasks_per_node * nodes_per_block\n\n            # import pdb; pdb.set_trace()\n            logger.debug(\"Tasks:{} Slots:{} Parallelism:{}\".format(len(active_tasks),\n                                                                   active_slots,\n                                                                   parallelism))\n\n            # Case 1\n            # No tasks.\n            if len(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(\"Strategy: Scale_in, tasks=0 starting kill timer\")\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(\"Strategy: Scale_in, tasks=0\")\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) / len(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((len(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 len(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"}]}