Menu
I have a small issue that I'm not quite sure how to solve. Here is a minimal example:
scan_process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) while(some_criterium): line = scan_process.stdout.readline() some_criterium = do_something(line) scan_process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) while(some_criterium): line = scan_process.stdout.readline() if nothing_happens_after_10s: break else: some_criterium = do_something(line) I read a line from a subprocess and do something with it. How can I exit if no line arrived after a fixed time interval?
475
The other day I ran into a use case where I needed to communicate with a subprocess I had started but I needed it to timeout. Unfortunately, Python 2 does not have a way to timeout the communicate method call so it just blocks until it either returns or the process itself closes.
Python 3.5 added the run function which accepts a timeout parameter. According to the documentation, it will be passed to the subprocess’s communicate method and TimeoutExpired exception will be raised should the process time out.
See "Subprocess" in the docs. There is a high-level interface asyncio.create_subprocess_exec () that returns Process objects that allows to read a line asynchroniosly using StreamReader.readline () coroutine (with async / await Python 3.5+ syntax ): Each step could be limited by timeout seconds if necessary. Try the asyncproc module. For example:
Interestingly enough, the timeout parameter was added to the subprocess module in Python 3.3. You can use it in subprocess.call, check_output, and check_call. It's also available in Popen.wait ().
Thanks for all the answers!
I found a way to solve my problem by simply using select.poll to peek into standard output.
import select ... scan_process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) poll_obj = select.poll() poll_obj.register(scan_process.stdout, select.POLLIN) while(some_criterium and not time_limit): poll_result = poll_obj.poll(0) if poll_result: line = scan_process.stdout.readline() some_criterium = do_something(line) update(time_limit) If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With