Skip to content

Instantly share code, notes, and snippets.

Last active July 6, 2019 22:20
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
What would you like to do?
non-blocking python subprocess
import subprocess
import threading
import queue
import os
import time
class Runner(object):
def __init__(self, cmd: []):
self.cmd = cmd
self.return_code = None
self.process = None # type: subprocess.Popen
self.run_time = 0
def _default_popen_kwargs(self):
return {
"env": os.environ.copy(),
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"shell": True,
"universal_newlines": True,
"bufsize": 1,
def _watch_output(self, process: subprocess.Popen, queue):
for line in iter(process.stderr.readline, ""):
if process.poll() is not None:
def stdout(self):
return self.process.stdout
def stderr(self):
return self.process.stderr
def start(self, wait_limit = 15):
start_time = time.time()
pargs = self._default_popen_kwargs
if self.cwd is not None:
pargs['cwd'] = self.cwd
self.process = subprocess.Popen(self.cmd, **pargs)
self.returned = None
last_output = time.time()
q = queue.Queue()
t = threading.Thread(target=self._watch_output, args=(self.process, q,))
t.daemon = True
while self.returned is None:
self.returned = self.process.poll()
delay = last_output-time.time()
if self.returned is None:
stdout = f"{last_output-time.time()} waited"
stderr = q.get_nowait()
except queue.Empty:
yield stdout, stderr
last_output = time.time()
if delay > wait_limit:
print("Waited 15 seconds, breaking")
self.run_time = time.time() - start_time
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment