Tried to make a non-blocking command execution function, what's causing this unexpected behavior?

user_9400421 0
Viewed 68

What I wanted to happen:

So my goal was to write a function that leverages subprocess to run a command and read the stdout, whether it be immediate or delayed, line by line as it comes. And to do that in a non-blocking, asynchronous way.

I also wanted to be able to pass a function to be called each time a new stdout line is read.

What happened instead:

Until the process being run is completely finished / killed, the output isn't handled / printed as expected. All the correct output happens, but I expected it to print in real-time as the output is polled. Rather, it waits until the entire process finishes running, then prints all the expected output.

What I tried:

So I wrote a simple test script lab_temp.py to provide some output:

from time import sleep
for i in range(10):
    print('i:', i)
    sleep(1)

And a function set_interval.py which I mostly copied from some SO answer (although I'm sorry I don't recall which answer to give credit):

import threading

def set_interval(func, sec):
    def func_wrapper():
        t = set_interval(func, sec)
        result = func()
        if result == False:
            t.cancel()
    t = threading.Timer(sec, func_wrapper)
    t.start()
    return t

And then a function call_command.py to run the command and asynchronously poll the process at some interval for output, until it's done. I'm only barely experienced with asynchronous code, and that's probably related to my mistake, but I think the async part is being handled behind the scenes by threading.Timer (in set_interval.py).

call_command.py:

import subprocess
from set_interval import set_interval

def call_command(cmd, update_func=None):
    p = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8')
    def polling(): # Replaces "while True:" to convert to non-blocking
        for line in iter(p.stdout.readline, ''):
            if update_func:
                update_func(line.rstrip())
            
        if p.poll() == 0:
            print('False')
            return False # cancel interval
        else:
            print('True')
            return True # continue interval
    set_interval(polling, 1)

And each of these functions have basic tests:

set_interval.test.py (seems to run as expected):

from set_interval import set_interval

i = 0
def func():
    global i
    i += 1
    print(f"Interval: {i}...")
    if i > 5:
        return False
    else:
        return True
    
set_interval(func, 2)
print('non blocking')

call_command.test.py (results in the wrong behavior, as described initially):

from call_command import call_command

def func(out):
    print(out) # <- This will print in one big batch once
               # the entire process is complete.
    
call_command('python3 lab_temp.py', update_func=func)
print('non-blocking') # <- This will print right away, so I 
                      # know it's not blocked / frozen.

What have I gotten wrong here causing the deviation from expectation?

Edit: Continued efforts...

import subprocess
from set_interval import set_interval


def call_command(cmd):
    p = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8')

    def polling():
        line = p.stdout.readline().strip()
        if not line and p.poll() is not None:
            return False
        else:
            print(line)
            return True

    set_interval(polling, 1)

Doesn't work. Nearly identical issues.

python
1 Answers

The problem is located in your command. The lab_temp.py script uses the print function that prints to sys.stdout by default. sys.stdout is block-buffered by default. Since the buffer is large enough to accept the whole script's output, it gets flushed no sooner than at the end.

To fix this, you can either use the sys.stdout's flush method:

from time import sleep
import sys
for i in range(10):
    print('i:', i)
    sys.stdout.flush()
    sleep(1)

or use print's flush parameter:

from time import sleep
for i in range(10):
    print('i:', i, flush=True)
    sleep(1)

or run Python interpreter with the -u option:

call_command(['python3', '-u', 'lab_temp.py'])

or set the PYTHONUNBUFFERED environment variable to a non-empty string

import subprocess
from set_interval import set_interval


def call_command(cmd):
    p = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', env={'PYTHONUNBUFFERED':'1'})

    def polling():
        line = p.stdout.readline().strip()
        if not line and p.poll() is not None:
            return False
        else:
            print(line)
            return True

    set_interval(polling, 1)

BTW, in order to avoid threads, you might want to use asyncio:

import asyncio

async def call_command(cmd):
   p = await asyncio.create_subprocess_exec(cmd[0], *cmd[1:], stderr=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE)
   async for line in p.stdout:
       line = line.strip().decode('utf-8')
       print(line)
user_5421960
Related