We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Do not CD to engine_dir
Currently in the IPP submit script we CD to the `engine_dir`. This breaks staging stuff. We should just send engine-related files there rather than changing the PWD.

--- 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/executors/ipp.py`
Content:
```
1 import logging
2 import os
3 import pathlib
4 import uuid
5 
6 from ipyparallel import Client
7 from parsl.providers import LocalProvider
8 from parsl.utils import RepresentationMixin
9 
10 from parsl.dataflow.error import ConfigurationError
11 from parsl.executors.base import ParslExecutor
12 from parsl.executors.errors import *
13 from parsl.executors.ipp_controller import Controller
14 from parsl.utils import wait_for_file
15 
16 logger = logging.getLogger(__name__)
17 
18 
19 class IPyParallelExecutor(ParslExecutor, RepresentationMixin):
20     """The IPython Parallel executor.
21 
22     This executor uses IPythonParallel's pilot execution system to manage multiple processes
23     running locally or remotely.
24 
25     Parameters
26     ----------
27     provider : :class:`~parsl.providers.provider_base.ExecutionProvider`
28         Provider to access computation resources. Can be one of :class:`~parsl.providers.aws.aws.EC2Provider`,
29         :class:`~parsl.providers.azureProvider.azureProvider.AzureProvider`,
30         :class:`~parsl.providers.cobalt.cobalt.Cobalt`,
31         :class:`~parsl.providers.condor.condor.Condor`,
32         :class:`~parsl.providers.googlecloud.googlecloud.GoogleCloud`,
33         :class:`~parsl.providers.gridEngine.gridEngine.GridEngine`,
34         :class:`~parsl.providers.jetstream.jetstream.Jetstream`,
35         :class:`~parsl.providers.local.local.Local`,
36         :class:`~parsl.providers.sge.sge.GridEngine`,
37         :class:`~parsl.providers.slurm.slurm.Slurm`, or
38         :class:`~parsl.providers.torque.torque.Torque`.
39     label : str
40         Label for this executor instance.
41     controller : :class:`~parsl.executors.ipp_controller.Controller`
42         Which Controller instance to use. Default is `Controller()`.
43     container_image : str
44         Launch tasks in a container using this docker image. If set to None, no container is used.
45         Default is None.
46     engine_dir : str
47         Directory where engine logs and configuration files will be stored.
48     working_dir : str
49         Directory where input data should be staged to.
50     storage_access : list of :class:`~parsl.data_provider.scheme.Scheme`
51         Specifications for accessing data this executor remotely. Multiple `Scheme`s are not yet supported.
52     managed : bool
53         If True, parsl will control dynamic scaling of this executor, and be responsible. Otherwise,
54         this is managed by the user.
55     engine_debug_level : int | str
56         Sets engine logging to specified debug level. Choices: (0, 10, 20, 30, 40, 50, 'DEBUG', 'INFO', 'WARN', 'ERROR', 'CRITICAL')
57 
58     .. note::
59            Some deficiencies with this executor are:
60 
61                1. Ipengine's execute one task at a time. This means one engine per core
62                   is necessary to exploit the full parallelism of a node.
63                2. No notion of remaining walltime.
64                3. Lack of throttling means tasks could be queued up on a worker.
65     """
66 
67     def __init__(self,
68                  provider=LocalProvider(),
69                  label='ipp',
70                  working_dir=None,
71                  controller=Controller(),
72                  container_image=None,
73                  engine_dir=None,
74                  storage_access=None,
75                  engine_debug_level=None,
76                  managed=True):
77         self.provider = provider
78         self.label = label
79         self.working_dir = working_dir
80         self.controller = controller
81         self.engine_debug_level = engine_debug_level
82         self.container_image = container_image
83         self.engine_dir = engine_dir
84         self.storage_access = storage_access if storage_access is not None else []
85         if len(self.storage_access) > 1:
86             raise ConfigurationError('Multiple storage access schemes are not yet supported')
87         self.managed = managed
88 
89         self.debug_option = ""
90         if self.engine_debug_level:
91             self.debug_option = "--log-level={}".format(self.engine_debug_level)
92 
93     def start(self):
94         self.controller.profile = self.label
95         self.controller.ipython_dir = self.run_dir
96         if self.engine_dir is None:
97             parent, child = pathlib.Path(self.run_dir).parts[-2:]
98             self.engine_dir = os.path.join(parent, child)
99         self.controller.start()
100 
101         self.engine_file = self.controller.engine_file
102 
103         with wait_for_file(self.controller.client_file, seconds=120):
104             logger.debug("Waiting for {0}".format(self.controller.client_file))
105 
106         if not os.path.exists(self.controller.client_file):
107             raise Exception("Controller client file is missing at {0}".format(self.controller.client_file))
108 
109         command_composer = self.compose_launch_cmd
110 
111         self.executor = Client(url_file=self.controller.client_file)
112         if self.container_image:
113             command_composer = self.compose_containerized_launch_cmd
114             logger.info("Launching IPP with Docker:{0}".format(self.container_image))
115 
116         self.launch_cmd = command_composer(self.engine_file, self.engine_dir, self.container_image)
117         self.engines = []
118 
119         self._scaling_enabled = self.provider.scaling_enabled
120         logger.debug("Starting IPyParallelExecutor with provider:\n%s", self.provider)
121         if hasattr(self.provider, 'init_blocks'):
122             try:
123                 for i in range(self.provider.init_blocks):
124                     engine = self.provider.submit(self.launch_cmd, 1)
125                     logger.debug("Launched block: {0}:{1}".format(i, engine))
126                     if not engine:
127                         raise(ScalingFailed(self.provider.label,
128                                             "Attempts to provision nodes via provider has failed"))
129                     self.engines.extend([engine])
130 
131             except Exception as e:
132                 logger.error("Scaling out failed: %s" % e)
133                 raise e
134 
135         self.lb_view = self.executor.load_balanced_view()
136         logger.debug("Starting executor")
137 
138     def compose_launch_cmd(self, filepath, engine_dir, container_image):
139         """Reads the json contents from filepath and uses that to compose the engine launch command.
140 
141         Args:
142             filepath: Path to the engine file
143             engine_dir: CWD for the engines
144 
145         """
146         self.engine_file = os.path.expanduser(filepath)
147         uid = str(uuid.uuid4())
148         engine_json = None
149         try:
150             with open(self.engine_file, 'r') as f:
151                 engine_json = f.read()
152 
153         except OSError as e:
154             logger.error("Could not open engine_json : ", self.engine_file)
155             raise e
156 
157         return """mkdir -p {0}
158 cd {0}
159 cat <<EOF > ipengine.{uid}.json
160 {1}
161 EOF
162 
163 mkdir -p 'engine_logs'
164 ipengine --file=ipengine.{uid}.json {debug_option} >> engine_logs/$JOBNAME.log 2>&1
165 """.format(engine_dir, engine_json, debug_option=self.debug_option, uid=uid)
166 
167     def compose_containerized_launch_cmd(self, filepath, engine_dir, container_image):
168         """Reads the json contents from filepath and uses that to compose the engine launch command.
169 
170         Notes: Add this to the ipengine launch for debug logs :
171                           --log-to-file --debug
172         Args:
173             filepath (str): Path to the engine file
174             engine_dir (str): CWD for the engines .
175             container_image (str): The container to be used to launch workers
176         """
177         self.engine_file = os.path.expanduser(filepath)
178         uid = str(uuid.uuid4())
179         engine_json = None
180         try:
181             with open(self.engine_file, 'r') as f:
182                 engine_json = f.read()
183 
184         except OSError as e:
185             logger.error("Could not open engine_json : ", self.engine_file)
186             raise e
187 
188         return """mkdir -p {0}
189 cd {0}
190 cat <<EOF > ipengine.{uid}.json
191 {1}
192 EOF
193 
194 DOCKER_ID=$(docker create --network host {2} ipengine --file=/tmp/ipengine.{uid}.json) {debug_option}
195 docker cp ipengine.{uid}.json $DOCKER_ID:/tmp/ipengine.{uid}.json
196 
197 # Copy current dir to the working directory
198 DOCKER_CWD=$(docker image inspect --format='{{{{.Config.WorkingDir}}}}' {2})
199 docker cp -a . $DOCKER_ID:$DOCKER_CWD
200 docker start $DOCKER_ID
201 
202 at_exit() {{
203   echo "Caught SIGTERM/SIGINT signal!"
204   docker stop $DOCKER_ID
205 }}
206 
207 trap at_exit SIGTERM SIGINT
208 sleep infinity
209 """.format(engine_dir, engine_json, container_image, debug_option=self.debug_option, uid=uid)
210 
211     @property
212     def scaling_enabled(self):
213         return self._scaling_enabled
214 
215     def submit(self, *args, **kwargs):
216         """Submits work to the thread pool.
217 
218         This method is simply pass through and behaves like a submit call as described
219         here `Python docs: <https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor>`_
220 
221         Returns:
222               Future
223         """
224         return self.lb_view.apply_async(*args, **kwargs)
225 
226     def scale_out(self, *args, **kwargs):
227         """Scales out the number of active workers by 1.
228 
229         This method is notImplemented for threads and will raise the error if called.
230 
231         """
232         if self.provider:
233             r = self.provider.submit(self.launch_cmd, *args, **kwargs)
234             self.engines.extend([r])
235         else:
236             logger.error("No execution provider available")
237             r = None
238 
239         return r
240 
241     def scale_in(self, blocks):
242         """Scale in the number of active blocks by the specified number.
243 
244         """
245         status = dict(zip(self.engines, self.provider.status(self.engines)))
246 
247         # This works for blocks=0
248         to_kill = [engine for engine in status if status[engine] == "RUNNING"][:blocks]
249 
250         if self.provider:
251             r = self.provider.cancel(to_kill)
252         else:
253             logger.error("No execution provider available")
254             r = None
255 
256         return r
257 
258     def status(self):
259         """Returns the status of the executor via probing the execution providers."""
260         if self.provider:
261             status = self.provider.status(self.engines)
262 
263         else:
264             status = []
265 
266         return status
267 
268     def shutdown(self, hub=True, targets='all', block=False):
269         """Shutdown the executor, including all workers and controllers.
270 
271         The interface documentation for IPP is `here <http://ipyparallel.readthedocs.io/en/latest/api/ipyparallel.html#ipyparallel.Client.shutdown>`_
272 
273         Kwargs:
274             - hub (Bool): Whether the hub should be shutdown, Default:True,
275             - targets (list of ints| 'all'): List of engine id's to kill, Default:'all'
276             - block (Bool): To block for confirmations or not
277 
278         Raises:
279              NotImplementedError
280         """
281         if self.controller:
282             logger.debug("IPP:Shutdown sequence: Attempting controller kill")
283             self.controller.close()
284 
285         # We do not actually do executor.shutdown because
286         # this blocks even when requested to not block, killing the
287         # controller is more effective although impolite.
288         # x = self.executor.shutdown(targets=targets,
289         #                           hub=hub,
290         #                           block=block)
291 
292         logger.debug("Done with executor shutdown")
293         return True
294 
295 
296 if __name__ == "__main__":
297 
298     pool1_config = {"poolname": "pool1",
299                     "queue": "foo"}
300
```
--- 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/executors/ipp.py b/parsl/executors/ipp.py
--- a/parsl/executors/ipp.py
+++ b/parsl/executors/ipp.py
@@ -155,13 +155,12 @@
             raise e
 
         return """mkdir -p {0}
-cd {0}
-cat <<EOF > ipengine.{uid}.json
+cat <<EOF > {0}/ipengine.{uid}.json
 {1}
 EOF
 
-mkdir -p 'engine_logs'
-ipengine --file=ipengine.{uid}.json {debug_option} >> engine_logs/$JOBNAME.log 2>&1
+mkdir -p '{0}/engine_logs'
+ipengine --file={0}/ipengine.{uid}.json {debug_option} >> {0}/engine_logs/$JOBNAME.log 2>&1
 """.format(engine_dir, engine_json, debug_option=self.debug_option, uid=uid)
 
     def compose_containerized_launch_cmd(self, filepath, engine_dir, container_image):

{"golden_diff": "diff --git a/parsl/executors/ipp.py b/parsl/executors/ipp.py\n--- a/parsl/executors/ipp.py\n+++ b/parsl/executors/ipp.py\n@@ -155,13 +155,12 @@\n             raise e\n \n         return \"\"\"mkdir -p {0}\n-cd {0}\n-cat <<EOF > ipengine.{uid}.json\n+cat <<EOF > {0}/ipengine.{uid}.json\n {1}\n EOF\n \n-mkdir -p 'engine_logs'\n-ipengine --file=ipengine.{uid}.json {debug_option} >> engine_logs/$JOBNAME.log 2>&1\n+mkdir -p '{0}/engine_logs'\n+ipengine --file={0}/ipengine.{uid}.json {debug_option} >> {0}/engine_logs/$JOBNAME.log 2>&1\n \"\"\".format(engine_dir, engine_json, debug_option=self.debug_option, uid=uid)\n \n     def compose_containerized_launch_cmd(self, filepath, engine_dir, container_image):\n", "issue": "Do not CD to engine_dir\nCurrently in the IPP submit script we CD to the `engine_dir`. This breaks staging stuff. We should just send engine-related files there rather than changing the PWD.\n", "before_files": [{"content": "import logging\nimport os\nimport pathlib\nimport uuid\n\nfrom ipyparallel import Client\nfrom parsl.providers import LocalProvider\nfrom parsl.utils import RepresentationMixin\n\nfrom parsl.dataflow.error import ConfigurationError\nfrom parsl.executors.base import ParslExecutor\nfrom parsl.executors.errors import *\nfrom parsl.executors.ipp_controller import Controller\nfrom parsl.utils import wait_for_file\n\nlogger = logging.getLogger(__name__)\n\n\nclass IPyParallelExecutor(ParslExecutor, RepresentationMixin):\n    \"\"\"The IPython Parallel executor.\n\n    This executor uses IPythonParallel's pilot execution system to manage multiple processes\n    running locally or remotely.\n\n    Parameters\n    ----------\n    provider : :class:`~parsl.providers.provider_base.ExecutionProvider`\n        Provider to access computation resources. Can be one of :class:`~parsl.providers.aws.aws.EC2Provider`,\n        :class:`~parsl.providers.azureProvider.azureProvider.AzureProvider`,\n        :class:`~parsl.providers.cobalt.cobalt.Cobalt`,\n        :class:`~parsl.providers.condor.condor.Condor`,\n        :class:`~parsl.providers.googlecloud.googlecloud.GoogleCloud`,\n        :class:`~parsl.providers.gridEngine.gridEngine.GridEngine`,\n        :class:`~parsl.providers.jetstream.jetstream.Jetstream`,\n        :class:`~parsl.providers.local.local.Local`,\n        :class:`~parsl.providers.sge.sge.GridEngine`,\n        :class:`~parsl.providers.slurm.slurm.Slurm`, or\n        :class:`~parsl.providers.torque.torque.Torque`.\n    label : str\n        Label for this executor instance.\n    controller : :class:`~parsl.executors.ipp_controller.Controller`\n        Which Controller instance to use. Default is `Controller()`.\n    container_image : str\n        Launch tasks in a container using this docker image. If set to None, no container is used.\n        Default is None.\n    engine_dir : str\n        Directory where engine logs and configuration files will be stored.\n    working_dir : str\n        Directory where input data should be staged to.\n    storage_access : list of :class:`~parsl.data_provider.scheme.Scheme`\n        Specifications for accessing data this executor remotely. Multiple `Scheme`s are not yet supported.\n    managed : bool\n        If True, parsl will control dynamic scaling of this executor, and be responsible. Otherwise,\n        this is managed by the user.\n    engine_debug_level : int | str\n        Sets engine logging to specified debug level. Choices: (0, 10, 20, 30, 40, 50, 'DEBUG', 'INFO', 'WARN', 'ERROR', 'CRITICAL')\n\n    .. note::\n           Some deficiencies with this executor are:\n\n               1. Ipengine's execute one task at a time. This means one engine per core\n                  is necessary to exploit the full parallelism of a node.\n               2. No notion of remaining walltime.\n               3. Lack of throttling means tasks could be queued up on a worker.\n    \"\"\"\n\n    def __init__(self,\n                 provider=LocalProvider(),\n                 label='ipp',\n                 working_dir=None,\n                 controller=Controller(),\n                 container_image=None,\n                 engine_dir=None,\n                 storage_access=None,\n                 engine_debug_level=None,\n                 managed=True):\n        self.provider = provider\n        self.label = label\n        self.working_dir = working_dir\n        self.controller = controller\n        self.engine_debug_level = engine_debug_level\n        self.container_image = container_image\n        self.engine_dir = engine_dir\n        self.storage_access = storage_access if storage_access is not None else []\n        if len(self.storage_access) > 1:\n            raise ConfigurationError('Multiple storage access schemes are not yet supported')\n        self.managed = managed\n\n        self.debug_option = \"\"\n        if self.engine_debug_level:\n            self.debug_option = \"--log-level={}\".format(self.engine_debug_level)\n\n    def start(self):\n        self.controller.profile = self.label\n        self.controller.ipython_dir = self.run_dir\n        if self.engine_dir is None:\n            parent, child = pathlib.Path(self.run_dir).parts[-2:]\n            self.engine_dir = os.path.join(parent, child)\n        self.controller.start()\n\n        self.engine_file = self.controller.engine_file\n\n        with wait_for_file(self.controller.client_file, seconds=120):\n            logger.debug(\"Waiting for {0}\".format(self.controller.client_file))\n\n        if not os.path.exists(self.controller.client_file):\n            raise Exception(\"Controller client file is missing at {0}\".format(self.controller.client_file))\n\n        command_composer = self.compose_launch_cmd\n\n        self.executor = Client(url_file=self.controller.client_file)\n        if self.container_image:\n            command_composer = self.compose_containerized_launch_cmd\n            logger.info(\"Launching IPP with Docker:{0}\".format(self.container_image))\n\n        self.launch_cmd = command_composer(self.engine_file, self.engine_dir, self.container_image)\n        self.engines = []\n\n        self._scaling_enabled = self.provider.scaling_enabled\n        logger.debug(\"Starting IPyParallelExecutor with provider:\\n%s\", self.provider)\n        if hasattr(self.provider, 'init_blocks'):\n            try:\n                for i in range(self.provider.init_blocks):\n                    engine = self.provider.submit(self.launch_cmd, 1)\n                    logger.debug(\"Launched block: {0}:{1}\".format(i, engine))\n                    if not engine:\n                        raise(ScalingFailed(self.provider.label,\n                                            \"Attempts to provision nodes via provider has failed\"))\n                    self.engines.extend([engine])\n\n            except Exception as e:\n                logger.error(\"Scaling out failed: %s\" % e)\n                raise e\n\n        self.lb_view = self.executor.load_balanced_view()\n        logger.debug(\"Starting executor\")\n\n    def compose_launch_cmd(self, filepath, engine_dir, container_image):\n        \"\"\"Reads the json contents from filepath and uses that to compose the engine launch command.\n\n        Args:\n            filepath: Path to the engine file\n            engine_dir: CWD for the engines\n\n        \"\"\"\n        self.engine_file = os.path.expanduser(filepath)\n        uid = str(uuid.uuid4())\n        engine_json = None\n        try:\n            with open(self.engine_file, 'r') as f:\n                engine_json = f.read()\n\n        except OSError as e:\n            logger.error(\"Could not open engine_json : \", self.engine_file)\n            raise e\n\n        return \"\"\"mkdir -p {0}\ncd {0}\ncat <<EOF > ipengine.{uid}.json\n{1}\nEOF\n\nmkdir -p 'engine_logs'\nipengine --file=ipengine.{uid}.json {debug_option} >> engine_logs/$JOBNAME.log 2>&1\n\"\"\".format(engine_dir, engine_json, debug_option=self.debug_option, uid=uid)\n\n    def compose_containerized_launch_cmd(self, filepath, engine_dir, container_image):\n        \"\"\"Reads the json contents from filepath and uses that to compose the engine launch command.\n\n        Notes: Add this to the ipengine launch for debug logs :\n                          --log-to-file --debug\n        Args:\n            filepath (str): Path to the engine file\n            engine_dir (str): CWD for the engines .\n            container_image (str): The container to be used to launch workers\n        \"\"\"\n        self.engine_file = os.path.expanduser(filepath)\n        uid = str(uuid.uuid4())\n        engine_json = None\n        try:\n            with open(self.engine_file, 'r') as f:\n                engine_json = f.read()\n\n        except OSError as e:\n            logger.error(\"Could not open engine_json : \", self.engine_file)\n            raise e\n\n        return \"\"\"mkdir -p {0}\ncd {0}\ncat <<EOF > ipengine.{uid}.json\n{1}\nEOF\n\nDOCKER_ID=$(docker create --network host {2} ipengine --file=/tmp/ipengine.{uid}.json) {debug_option}\ndocker cp ipengine.{uid}.json $DOCKER_ID:/tmp/ipengine.{uid}.json\n\n# Copy current dir to the working directory\nDOCKER_CWD=$(docker image inspect --format='{{{{.Config.WorkingDir}}}}' {2})\ndocker cp -a . $DOCKER_ID:$DOCKER_CWD\ndocker start $DOCKER_ID\n\nat_exit() {{\n  echo \"Caught SIGTERM/SIGINT signal!\"\n  docker stop $DOCKER_ID\n}}\n\ntrap at_exit SIGTERM SIGINT\nsleep infinity\n\"\"\".format(engine_dir, engine_json, container_image, debug_option=self.debug_option, uid=uid)\n\n    @property\n    def scaling_enabled(self):\n        return self._scaling_enabled\n\n    def submit(self, *args, **kwargs):\n        \"\"\"Submits work to the thread pool.\n\n        This method is simply pass through and behaves like a submit call as described\n        here `Python docs: <https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor>`_\n\n        Returns:\n              Future\n        \"\"\"\n        return self.lb_view.apply_async(*args, **kwargs)\n\n    def scale_out(self, *args, **kwargs):\n        \"\"\"Scales out the number of active workers by 1.\n\n        This method is notImplemented for threads and will raise the error if called.\n\n        \"\"\"\n        if self.provider:\n            r = self.provider.submit(self.launch_cmd, *args, **kwargs)\n            self.engines.extend([r])\n        else:\n            logger.error(\"No execution provider available\")\n            r = None\n\n        return r\n\n    def scale_in(self, blocks):\n        \"\"\"Scale in the number of active blocks by the specified number.\n\n        \"\"\"\n        status = dict(zip(self.engines, self.provider.status(self.engines)))\n\n        # This works for blocks=0\n        to_kill = [engine for engine in status if status[engine] == \"RUNNING\"][:blocks]\n\n        if self.provider:\n            r = self.provider.cancel(to_kill)\n        else:\n            logger.error(\"No execution provider available\")\n            r = None\n\n        return r\n\n    def status(self):\n        \"\"\"Returns the status of the executor via probing the execution providers.\"\"\"\n        if self.provider:\n            status = self.provider.status(self.engines)\n\n        else:\n            status = []\n\n        return status\n\n    def shutdown(self, hub=True, targets='all', block=False):\n        \"\"\"Shutdown the executor, including all workers and controllers.\n\n        The interface documentation for IPP is `here <http://ipyparallel.readthedocs.io/en/latest/api/ipyparallel.html#ipyparallel.Client.shutdown>`_\n\n        Kwargs:\n            - hub (Bool): Whether the hub should be shutdown, Default:True,\n            - targets (list of ints| 'all'): List of engine id's to kill, Default:'all'\n            - block (Bool): To block for confirmations or not\n\n        Raises:\n             NotImplementedError\n        \"\"\"\n        if self.controller:\n            logger.debug(\"IPP:Shutdown sequence: Attempting controller kill\")\n            self.controller.close()\n\n        # We do not actually do executor.shutdown because\n        # this blocks even when requested to not block, killing the\n        # controller is more effective although impolite.\n        # x = self.executor.shutdown(targets=targets,\n        #                           hub=hub,\n        #                           block=block)\n\n        logger.debug(\"Done with executor shutdown\")\n        return True\n\n\nif __name__ == \"__main__\":\n\n    pool1_config = {\"poolname\": \"pool1\",\n                    \"queue\": \"foo\"}\n", "path": "parsl/executors/ipp.py"}], "after_files": [{"content": "import logging\nimport os\nimport pathlib\nimport uuid\n\nfrom ipyparallel import Client\nfrom parsl.providers import LocalProvider\nfrom parsl.utils import RepresentationMixin\n\nfrom parsl.dataflow.error import ConfigurationError\nfrom parsl.executors.base import ParslExecutor\nfrom parsl.executors.errors import *\nfrom parsl.executors.ipp_controller import Controller\nfrom parsl.utils import wait_for_file\n\nlogger = logging.getLogger(__name__)\n\n\nclass IPyParallelExecutor(ParslExecutor, RepresentationMixin):\n    \"\"\"The IPython Parallel executor.\n\n    This executor uses IPythonParallel's pilot execution system to manage multiple processes\n    running locally or remotely.\n\n    Parameters\n    ----------\n    provider : :class:`~parsl.providers.provider_base.ExecutionProvider`\n        Provider to access computation resources. Can be one of :class:`~parsl.providers.aws.aws.EC2Provider`,\n        :class:`~parsl.providers.azureProvider.azureProvider.AzureProvider`,\n        :class:`~parsl.providers.cobalt.cobalt.Cobalt`,\n        :class:`~parsl.providers.condor.condor.Condor`,\n        :class:`~parsl.providers.googlecloud.googlecloud.GoogleCloud`,\n        :class:`~parsl.providers.gridEngine.gridEngine.GridEngine`,\n        :class:`~parsl.providers.jetstream.jetstream.Jetstream`,\n        :class:`~parsl.providers.local.local.Local`,\n        :class:`~parsl.providers.sge.sge.GridEngine`,\n        :class:`~parsl.providers.slurm.slurm.Slurm`, or\n        :class:`~parsl.providers.torque.torque.Torque`.\n    label : str\n        Label for this executor instance.\n    controller : :class:`~parsl.executors.ipp_controller.Controller`\n        Which Controller instance to use. Default is `Controller()`.\n    container_image : str\n        Launch tasks in a container using this docker image. If set to None, no container is used.\n        Default is None.\n    engine_dir : str\n        Directory where engine logs and configuration files will be stored.\n    working_dir : str\n        Directory where input data should be staged to.\n    storage_access : list of :class:`~parsl.data_provider.scheme.Scheme`\n        Specifications for accessing data this executor remotely. Multiple `Scheme`s are not yet supported.\n    managed : bool\n        If True, parsl will control dynamic scaling of this executor, and be responsible. Otherwise,\n        this is managed by the user.\n    engine_debug_level : int | str\n        Sets engine logging to specified debug level. Choices: (0, 10, 20, 30, 40, 50, 'DEBUG', 'INFO', 'WARN', 'ERROR', 'CRITICAL')\n\n    .. note::\n           Some deficiencies with this executor are:\n\n               1. Ipengine's execute one task at a time. This means one engine per core\n                  is necessary to exploit the full parallelism of a node.\n               2. No notion of remaining walltime.\n               3. Lack of throttling means tasks could be queued up on a worker.\n    \"\"\"\n\n    def __init__(self,\n                 provider=LocalProvider(),\n                 label='ipp',\n                 working_dir=None,\n                 controller=Controller(),\n                 container_image=None,\n                 engine_dir=None,\n                 storage_access=None,\n                 engine_debug_level=None,\n                 managed=True):\n        self.provider = provider\n        self.label = label\n        self.working_dir = working_dir\n        self.controller = controller\n        self.engine_debug_level = engine_debug_level\n        self.container_image = container_image\n        self.engine_dir = engine_dir\n        self.storage_access = storage_access if storage_access is not None else []\n        if len(self.storage_access) > 1:\n            raise ConfigurationError('Multiple storage access schemes are not yet supported')\n        self.managed = managed\n\n        self.debug_option = \"\"\n        if self.engine_debug_level:\n            self.debug_option = \"--log-level={}\".format(self.engine_debug_level)\n\n    def start(self):\n        self.controller.profile = self.label\n        self.controller.ipython_dir = self.run_dir\n        if self.engine_dir is None:\n            parent, child = pathlib.Path(self.run_dir).parts[-2:]\n            self.engine_dir = os.path.join(parent, child)\n        self.controller.start()\n\n        self.engine_file = self.controller.engine_file\n\n        with wait_for_file(self.controller.client_file, seconds=120):\n            logger.debug(\"Waiting for {0}\".format(self.controller.client_file))\n\n        if not os.path.exists(self.controller.client_file):\n            raise Exception(\"Controller client file is missing at {0}\".format(self.controller.client_file))\n\n        command_composer = self.compose_launch_cmd\n\n        self.executor = Client(url_file=self.controller.client_file)\n        if self.container_image:\n            command_composer = self.compose_containerized_launch_cmd\n            logger.info(\"Launching IPP with Docker:{0}\".format(self.container_image))\n\n        self.launch_cmd = command_composer(self.engine_file, self.engine_dir, self.container_image)\n        self.engines = []\n\n        self._scaling_enabled = self.provider.scaling_enabled\n        logger.debug(\"Starting IPyParallelExecutor with provider:\\n%s\", self.provider)\n        if hasattr(self.provider, 'init_blocks'):\n            try:\n                for i in range(self.provider.init_blocks):\n                    engine = self.provider.submit(self.launch_cmd, 1)\n                    logger.debug(\"Launched block: {0}:{1}\".format(i, engine))\n                    if not engine:\n                        raise(ScalingFailed(self.provider.label,\n                                            \"Attempts to provision nodes via provider has failed\"))\n                    self.engines.extend([engine])\n\n            except Exception as e:\n                logger.error(\"Scaling out failed: %s\" % e)\n                raise e\n\n        self.lb_view = self.executor.load_balanced_view()\n        logger.debug(\"Starting executor\")\n\n    def compose_launch_cmd(self, filepath, engine_dir, container_image):\n        \"\"\"Reads the json contents from filepath and uses that to compose the engine launch command.\n\n        Args:\n            filepath: Path to the engine file\n            engine_dir: CWD for the engines\n\n        \"\"\"\n        self.engine_file = os.path.expanduser(filepath)\n        uid = str(uuid.uuid4())\n        engine_json = None\n        try:\n            with open(self.engine_file, 'r') as f:\n                engine_json = f.read()\n\n        except OSError as e:\n            logger.error(\"Could not open engine_json : \", self.engine_file)\n            raise e\n\n        return \"\"\"mkdir -p {0}\ncat <<EOF > {0}/ipengine.{uid}.json\n{1}\nEOF\n\nmkdir -p '{0}/engine_logs'\nipengine --file={0}/ipengine.{uid}.json {debug_option} >> {0}/engine_logs/$JOBNAME.log 2>&1\n\"\"\".format(engine_dir, engine_json, debug_option=self.debug_option, uid=uid)\n\n    def compose_containerized_launch_cmd(self, filepath, engine_dir, container_image):\n        \"\"\"Reads the json contents from filepath and uses that to compose the engine launch command.\n\n        Notes: Add this to the ipengine launch for debug logs :\n                          --log-to-file --debug\n        Args:\n            filepath (str): Path to the engine file\n            engine_dir (str): CWD for the engines .\n            container_image (str): The container to be used to launch workers\n        \"\"\"\n        self.engine_file = os.path.expanduser(filepath)\n        uid = str(uuid.uuid4())\n        engine_json = None\n        try:\n            with open(self.engine_file, 'r') as f:\n                engine_json = f.read()\n\n        except OSError as e:\n            logger.error(\"Could not open engine_json : \", self.engine_file)\n            raise e\n\n        return \"\"\"mkdir -p {0}\ncd {0}\ncat <<EOF > ipengine.{uid}.json\n{1}\nEOF\n\nDOCKER_ID=$(docker create --network host {2} ipengine --file=/tmp/ipengine.{uid}.json) {debug_option}\ndocker cp ipengine.{uid}.json $DOCKER_ID:/tmp/ipengine.{uid}.json\n\n# Copy current dir to the working directory\nDOCKER_CWD=$(docker image inspect --format='{{{{.Config.WorkingDir}}}}' {2})\ndocker cp -a . $DOCKER_ID:$DOCKER_CWD\ndocker start $DOCKER_ID\n\nat_exit() {{\n  echo \"Caught SIGTERM/SIGINT signal!\"\n  docker stop $DOCKER_ID\n}}\n\ntrap at_exit SIGTERM SIGINT\nsleep infinity\n\"\"\".format(engine_dir, engine_json, container_image, debug_option=self.debug_option, uid=uid)\n\n    @property\n    def scaling_enabled(self):\n        return self._scaling_enabled\n\n    def submit(self, *args, **kwargs):\n        \"\"\"Submits work to the thread pool.\n\n        This method is simply pass through and behaves like a submit call as described\n        here `Python docs: <https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor>`_\n\n        Returns:\n              Future\n        \"\"\"\n        return self.lb_view.apply_async(*args, **kwargs)\n\n    def scale_out(self, *args, **kwargs):\n        \"\"\"Scales out the number of active workers by 1.\n\n        This method is notImplemented for threads and will raise the error if called.\n\n        \"\"\"\n        if self.provider:\n            r = self.provider.submit(self.launch_cmd, *args, **kwargs)\n            self.engines.extend([r])\n        else:\n            logger.error(\"No execution provider available\")\n            r = None\n\n        return r\n\n    def scale_in(self, blocks):\n        \"\"\"Scale in the number of active blocks by the specified number.\n\n        \"\"\"\n        status = dict(zip(self.engines, self.provider.status(self.engines)))\n\n        # This works for blocks=0\n        to_kill = [engine for engine in status if status[engine] == \"RUNNING\"][:blocks]\n\n        if self.provider:\n            r = self.provider.cancel(to_kill)\n        else:\n            logger.error(\"No execution provider available\")\n            r = None\n\n        return r\n\n    def status(self):\n        \"\"\"Returns the status of the executor via probing the execution providers.\"\"\"\n        if self.provider:\n            status = self.provider.status(self.engines)\n\n        else:\n            status = []\n\n        return status\n\n    def shutdown(self, hub=True, targets='all', block=False):\n        \"\"\"Shutdown the executor, including all workers and controllers.\n\n        The interface documentation for IPP is `here <http://ipyparallel.readthedocs.io/en/latest/api/ipyparallel.html#ipyparallel.Client.shutdown>`_\n\n        Kwargs:\n            - hub (Bool): Whether the hub should be shutdown, Default:True,\n            - targets (list of ints| 'all'): List of engine id's to kill, Default:'all'\n            - block (Bool): To block for confirmations or not\n\n        Raises:\n             NotImplementedError\n        \"\"\"\n        if self.controller:\n            logger.debug(\"IPP:Shutdown sequence: Attempting controller kill\")\n            self.controller.close()\n\n        # We do not actually do executor.shutdown because\n        # this blocks even when requested to not block, killing the\n        # controller is more effective although impolite.\n        # x = self.executor.shutdown(targets=targets,\n        #                           hub=hub,\n        #                           block=block)\n\n        logger.debug(\"Done with executor shutdown\")\n        return True\n\n\nif __name__ == \"__main__\":\n\n    pool1_config = {\"poolname\": \"pool1\",\n                    \"queue\": \"foo\"}\n", "path": "parsl/executors/ipp.py"}]}