Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[WIP] Attempt to make job submission async #473

Closed
wants to merge 3 commits into from
Closed

[WIP] Attempt to make job submission async #473

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6c259aa
Make function providing centralizes calls (_call) asynchronous
oshadura Oct 22, 2020
2b182e5
Remove TODO note about making call async friendly
oshadura Nov 16, 2020
3ce0910
Apply black formatting
oshadura Nov 16, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 11 additions & 9 deletions dask_jobqueue/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,10 @@
import os
import re
import shlex
import subprocess
import sys
import weakref
import abc
import asyncio

import dask
from dask.utils import ignoring
Expand Down Expand Up @@ -302,7 +302,6 @@ def job_file(self):
yield fn

async def _submit_job(self, script_filename):
# Should we make this async friendly?
return self._call(shlex.split(self.submit_command) + [script_filename])

@property
Expand Down Expand Up @@ -361,17 +360,17 @@ def _close_job(cls, job_id):
logger.debug("Closed job %s", job_id)

@staticmethod
def _call(cmd, **kwargs):
"""Call a command using subprocess.Popen.
async def _call(cmd, **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.
outputs, logging, and an opportunity to go asynchronous.

Parameters
----------
cmd: List(str))
A command, each of which is a list of strings to hand to
subprocess.Popen
asyncio.subprocess.Popen

Examples
--------
Expand All @@ -390,11 +389,14 @@ def _call(cmd, **kwargs):
"Executing the following command to command line\n{}".format(cmd_str)
)

proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kwargs
proc = await asyncio.create_subprocess_exec(
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
**kwargs
)

out, err = proc.communicate()
out, err = await proc.communicate()
out, err = out.decode(), err.decode()

if proc.returncode != 0:
Expand Down