We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
IPP `engine_dir` has no effect if indicated dir does not exist
Reported by @daheise. 

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

{"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@@ -160,7 +160,8 @@\n             logger.error(\"Could not open engine_json : \", self.engine_file)\n             raise e\n \n-        return \"\"\"cd {0}\n+        return \"\"\"mkdir -p {0}\n+cd {0}\n cat <<EOF > ipengine.{uid}.json\n {1}\n EOF\n@@ -190,7 +191,8 @@\n             logger.error(\"Could not open engine_json : \", self.engine_file)\n             raise e\n \n-        return \"\"\"cd {0}\n+        return \"\"\"mkdir -p {0}\n+cd {0}\n cat <<EOF > ipengine.{uid}.json\n {1}\n EOF\n", "issue": "IPP `engine_dir` has no effect if indicated dir does not exist\nReported by @daheise. \n", "before_files": [{"content": "import logging\nimport os\nimport uuid\n\nfrom ipyparallel import Client\nfrom libsubmit.providers import LocalProvider\nfrom libsubmit.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:`~libsubmit.providers.provider_base.ExecutionProvider`\n        Provider to access computation resources. Can be one of :class:`~libsubmit.providers.aws.aws.EC2Provider`,\n        :class:`~libsubmit.providers.azureProvider.azureProvider.AzureProvider`,\n        :class:`~libsubmit.providers.cobalt.cobalt.Cobalt`,\n        :class:`~libsubmit.providers.condor.condor.Condor`,\n        :class:`~libsubmit.providers.googlecloud.googlecloud.GoogleCloud`,\n        :class:`~libsubmit.providers.gridEngine.gridEngine.GridEngine`,\n        :class:`~libsubmit.providers.jetstream.jetstream.Jetstream`,\n        :class:`~libsubmit.providers.local.local.Local`,\n        :class:`~libsubmit.providers.sge.sge.GridEngine`,\n        :class:`~libsubmit.providers.slurm.slurm.Slurm`, or\n        :class:`~libsubmit.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_file : str\n        Path to json engine file that will be used to compose ipp launch commands at\n        scaling events. Default is '~/.ipython/profile_default/security/ipcontroller-engine.json'.\n    engine_dir : str\n        Alternative to above, specify the engine_dir\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                 engine_file='~/.ipython/profile_default/security/ipcontroller-engine.json',\n                 engine_dir='.',\n                 working_dir=None,\n                 controller=Controller(),\n                 container_image=None,\n                 storage_access=None,\n                 engine_debug_level=None,\n                 managed=True):\n        self.provider = provider\n        self.label = label\n        self.engine_file = engine_file\n        self.engine_dir = engine_dir\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.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        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        if self.provider:\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        else:\n            self._scaling_enabled = False\n            logger.debug(\"Starting IpyParallelExecutor with no provider\")\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 \"\"\"cd {0}\ncat <<EOF > ipengine.{uid}.json\n{1}\nEOF\n\nmkdir -p '.ipengine_logs'\nipengine --file=ipengine.{uid}.json {debug_option} >> .ipengine_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 \"\"\"cd {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 uuid\n\nfrom ipyparallel import Client\nfrom libsubmit.providers import LocalProvider\nfrom libsubmit.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:`~libsubmit.providers.provider_base.ExecutionProvider`\n        Provider to access computation resources. Can be one of :class:`~libsubmit.providers.aws.aws.EC2Provider`,\n        :class:`~libsubmit.providers.azureProvider.azureProvider.AzureProvider`,\n        :class:`~libsubmit.providers.cobalt.cobalt.Cobalt`,\n        :class:`~libsubmit.providers.condor.condor.Condor`,\n        :class:`~libsubmit.providers.googlecloud.googlecloud.GoogleCloud`,\n        :class:`~libsubmit.providers.gridEngine.gridEngine.GridEngine`,\n        :class:`~libsubmit.providers.jetstream.jetstream.Jetstream`,\n        :class:`~libsubmit.providers.local.local.Local`,\n        :class:`~libsubmit.providers.sge.sge.GridEngine`,\n        :class:`~libsubmit.providers.slurm.slurm.Slurm`, or\n        :class:`~libsubmit.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_file : str\n        Path to json engine file that will be used to compose ipp launch commands at\n        scaling events. Default is '~/.ipython/profile_default/security/ipcontroller-engine.json'.\n    engine_dir : str\n        Alternative to above, specify the engine_dir\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                 engine_file='~/.ipython/profile_default/security/ipcontroller-engine.json',\n                 engine_dir='.',\n                 working_dir=None,\n                 controller=Controller(),\n                 container_image=None,\n                 storage_access=None,\n                 engine_debug_level=None,\n                 managed=True):\n        self.provider = provider\n        self.label = label\n        self.engine_file = engine_file\n        self.engine_dir = engine_dir\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.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        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        if self.provider:\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        else:\n            self._scaling_enabled = False\n            logger.debug(\"Starting IpyParallelExecutor with no provider\")\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 '.ipengine_logs'\nipengine --file=ipengine.{uid}.json {debug_option} >> .ipengine_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"}]}