Merge branch 'main' of github.com:dask/dask-jobqueue into slurm-runner

ci/environment.yml

@@ -3,7 +3,7 @@ channels:
   - conda-forge
   - defaults
 dependencies:
-  - python=3.9
+  - python=3.10
   - dask
   - distributed
   - flake8

ci/sge/docker-compose.yml

@@ -41,8 +41,6 @@ services:
     build:
       context: .
       target: slave
-      args:
-        PYTHON_VERSION: 3.9
     container_name: slave_two
     hostname: slave_two
     #network_mode: host

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

setup.py

@@ -28,7 +28,7 @@
     cmdclass=versioneer.get_cmdclass(),
     description="Deploy Dask on job queuing systems like PBS, Slurm, SGE or LSF",
     url="https://jobqueue.dask.org",
-    python_requires=">=3.9",
+    python_requires=">=3.10",
     license="BSD 3-Clause",
     packages=["dask_jobqueue"],
     include_package_data=True,