Fixed LSFCluster stdin job setup not being run in a shell (#661)

Author: aiudirog

dask_jobqueue/core.py

@@ -462,21 +462,25 @@ async def _close_job(cls, job_id, cancel_command):
             logger.debug("Closed job %s", job_id)
 
     @staticmethod
-    async def _call(cmd, **kwargs):
+    async def _call(cmd, *, shell=False, **kwargs):
         """Call a command using asyncio.create_subprocess_exec.
 
         This centralizes calls out to the command line, providing consistent
         outputs, logging, and an opportunity to go asynchronous in the future.
 
         Parameters
         ----------
-        cmd: List(str))
+        cmd: List(str)
             A command, each of which is a list of strings to hand to
             asyncio.create_subprocess_exec
+        shell: bool
+            Use asyncio.create_subprocess_shell instead to run the command
+            in a shell?
 
         Examples
         --------
         >>> self._call(['ls', '/foo'])
+        >>> self._call(['ls /foo'], shell=True)
 
         Returns
         -------
@@ -491,7 +495,12 @@ async def _call(cmd, **kwargs):
             "Executing the following command to command line\n{}".format(cmd_str)
         )
 
-        proc = await asyncio.create_subprocess_exec(
+        if shell:
+            create_subproc = asyncio.create_subprocess_shell
+        else:
+            create_subproc = asyncio.create_subprocess_exec
+
+        proc = await create_subproc(
             *cmd,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,

dask_jobqueue/lsf.py

@@ -108,7 +108,7 @@ def __init__(
     async def _submit_job(self, script_filename):
         if self.use_stdin:
             piped_cmd = [self.submit_command + "< " + script_filename + " 2> /dev/null"]
-            return await self._call(piped_cmd)
+            return await self._call(piped_cmd, shell=True)
         else:
             result = await super()._submit_job(script_filename)
             return result