Как запустить shell-комманду в новом процессе без блокировки потока?

Question

[Gnifajio None]

Есть асинхронное приложение, в котором нужно вызвать произвольную shell-комманду и получить ее вывод, при этом не заблокировать остальные потоки.
Пока что я додумался только до этого:

from subprocess import Popen, TimeoutExpired, PIPE

async def call_process(command: str, timeout: int = 5) -> None:
    try:
        proc = Popen(command, stdout=PIPE, stderr=PIPE, shell=True)
        stdout, stderr = proc.communicate(timeout=timeout)
        exitcode = proc.returncode
    except TimeoutExpired:
        print(f'Command {command} timed out ({timeout}s)')
    else:
        print(
            f'Exitcode: {exitcode}\n' + (
                "STDERR: " + stderr.decode()
                if stderr else "STDOUT: " + stdout.decode() if stdout else ''
            )
        )

Проблема этого кода в том, что пока оно работает, все приложение не отвечает до того момента, пока команда не выполнится, или не истечет время ожидания.

Я пробовал также multithreading, но так как функция асинхронная, то отрабатывать не хочет.
Также пробовал subprocess.check_output, результат такой же как и от Popen
Предвидя ответы "сделай функцию синхронной" отвечаю: нельзя, вместо print там вызов async функции.

P.S.: И да, я гуглил и читал доки
P.P.S: loop.run_in_executor() не сработал

Answer 1

[Gnifajio None]

В данном случае помогло asyncio run_in_executor
Спасибо Сергей Горностаев и Vindicar, которые посоветовали мне это.
Я преобразовал код вот так:

async def call_process(command: str, timeout: int = 5):
    def caller(cmd: str, t_out):
        try:
            proc = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True)
            stdout, stderr = proc.communicate(timeout=t_out)
            exitcode = proc.returncode
        except TimeoutExpired:
            return f'Command "{cmd}" timed out ({t_out}s)'
        else:
            return f'Exitcode: {exitcode}' + (
                f'STDERR: {stderr.decode()}'
                if stderr else f'STDOUT: {stdout.decode()}' if stdout else ''
                )

    loop = asyncio.get_running_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, caller, command, timeout)
    await some_async_func(result)

Конкретно для моего случая это подходит идеально.

Answer 2

[Vindicar]

communicate() как раз и ждёт. Вместо этого ты можешь попробовать просто читать из proc.stdout.
С асинхронностью сложнее, но ты можешь попробовать запустить функцию в потоке через loop.run_in_executor(). Это позволит завернуть поток в асинхронную задачу. Так как поток будет висеть в ожидании ввода-вывода, он не должен сильно влиять на GIL и мешать другим потокам.

Comments

[Gnifajio None]

На счет communicate я знаю что он ждет. А вот на счет loop.run_in_executor():
Либо

RuntimeWarning: coroutine 'call_process' was never awaited
  handle = None  # Needed to break cycles when an exception occurs.
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

Либо

Traceback (most recent call last):
  File "/usr/lib/python3.10/multiprocessing/queues.py", line 244, in _feed
    obj = _ForkingPickler.dumps(obj)
  File "/usr/lib/python3.10/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: cannot pickle 'weakref.ReferenceType' object

Я думаю что этоо из-за того, что функция, которую я пытаюсь запустить - асинхронная

[Сергей Горностаев]

Gnifajio None, вот поэтому я и посоветовал именно блокирующийся участок кода запихать в блокирующуюся функцию и передать экзекьютору.