Skip to content

Instantly share code, notes, and snippets.

@jeffbryner
Created January 25, 2014 02:15
Show Gist options
  • Save jeffbryner/8610763 to your computer and use it in GitHub Desktop.
Save jeffbryner/8610763 to your computer and use it in GitHub Desktop.
python code to run an external command with a timeout and return the output
import fcntl
import subprocess
import sys
import shlex
from threading import Timer
import time
import os
'''
python3 code to allow one to run an external command and return the output
'''
def nonBlockRead(output):
try:
fd = output.fileno()
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
out= output.read()
if out !=None:
return out
else:
return b''
except:
return b''
def kill_proc(proc):
try:
proc.stdout.close() # If they are not closed the fds will hang around until
proc.stderr.close() # os.fdlimit is exceeded and cause a nasty exception
proc.kill() # Important to close the fds prior to terminating the process!
except:
pass
def cmdTimeout(cmd, timeout_sec):
#run a command for awhile, timeout if it doesn't complete in the time alloted.
proc = subprocess.Popen(shlex.split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
timer = Timer(timeout_sec, kill_proc, [proc])
timer.start()
stdout = ''
stderr = ''
while proc.poll() is None : # Monitor process
time.sleep(0.1) # Wait a little
# p.std* blocks on read(), which messes up the timeout timer.
# To fix this, we use a nonblocking read()
# Note: Not sure if this is Windows compatible
stdout+= str(nonBlockRead(proc.stdout).decode("utf-8"))
stderr+= str(nonBlockRead(proc.stderr).decode("utf-8"))
timer.cancel()
returncode = proc.returncode
return (returncode, stdout, stderr)
if __name__ == '__main__':
#run a command that finishes
rcode,sout,serr=cmdTimeout("ping -c1 google.com",3)
print(rcode,sout,serr)
#one that will timeout
rcode,sout,serr=cmdTimeout("ping google.com",3)
print(rcode,sout,serr)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment