[メモ] subprocess.Popenで実行したコマンドにSIGINTが効かない場合がある

subprocess.Popenでコマンドを実行した際、SIGINTが無視されることがある。

サンプルコードで再現

10秒スリープするコマンドを5秒後にSIGINTで停止させたい。

以下3パターンのコマンドでサンプルコードを作成して実験した。

• fn1: ['sleep', '10'], shell=False (default)
• fn2: sleep 10, shell=True
• fn3: sleep 10; echo "hello", shell=True

main.py

import os
import time
import signal
import subprocess


def fn1():
    proc = subprocess.Popen(['sleep', '10'])
    time.sleep(5)
    proc.send_signal(signal.SIGINT)
    proc.communicate()


def fn2():
    proc = subprocess.Popen('sleep 10', shell=True)
    time.sleep(5)
    proc.send_signal(signal.SIGINT)
    proc.communicate()


def fn3():
    proc = subprocess.Popen('sleep 10; echo "hello"', shell=True)
    time.sleep(5)
    proc.send_signal(signal.SIGINT)
    proc.communicate()


def test_time_counter(fn) -> int:
    start = time.perf_counter()
    fn()
    end = time.perf_counter()
    return end - start


def main():
    t = test_time_counter(fn1)
    print(f'fn1: {t}')
    t = test_time_counter(fn2)
    print(f'fn2: {t}')
    t = test_time_counter(fn3)
    print(f'fn3: {t}')


if __name__ == '__main__':
    main()

実行結果

$ python main.py
fn1: 5.0063532770000005
fn2: 5.006125719999999
hello
fn3: 10.018092612999999

fn3だけなぜかSIGINTが効いてくれない。

原因と対策

私には原因はよくわからないが、以下によると、shellがSIGINTを無視するということらしい。
https://github.com/python/cpython/issues/119059#issuecomment-2111355552

しかしshell=Trueが原因だとするとサンプルコードfn2が想定通りに終わっているのが気になるがここでは無視。

回避策を試す

• fn3_py311: process_group=0を使用
  • Python3.11以降で使用できる
• fn3_py32: start_new_session=Trueを使用
  • Python3.2以降で使用できる

def fn3_py311():
    proc = subprocess.Popen('sleep 10; echo "hello"',
                            shell=True, process_group=0)
    time.sleep(5)
    os.killpg(os.getpgid(proc.pid), signal.SIGINT)
    proc.communicate()


def fn3_py32():
    proc = subprocess.Popen('sleep 10; echo "hello"',
                            shell=True, start_new_session=True)
    time.sleep(5)
    os.killpg(os.getpgid(proc.pid), signal.SIGINT)
    proc.communicate()


def main():
    t = test_time_counter(fn3_py311)
    print(f'fn3_py311: {t}')
    t = test_time_counter(fn3_py32)
    print(f'fn3_py32: {t}')

実行結果

$ python3 main.py
fn3_py311: 5.00557909719646
fn3_py32: 5.007664474658668

どちらも想定通りに動作してくれている🙆🙆

AppiumでWebサイトをクロールする様子をffmpegでスクリーンキャプチャするサンプル

ffmpegでスクリーンキャプチャする際に、時間を指定せずに任意のタイミングでキャプチャを停止する必要があって今回の問題にぶつかった。
せっかくなのでサンプルコードを貼っておく。

• macOS Monterey 12.7.6
• Python 3.9
• ffmpeg version 7.0.2
• appium 2.11.4
  • chromium@1.3.37 [installed (npm)]

インストールなどは省略。

import os
import subprocess
import signal
import time
from appium import webdriver
from appium.options.common.base import AppiumOptions
from appium.webdriver.common.appiumby import AppiumBy


def capture():
    url = 'https://techblog.lycorp.co.jp/ja'
    caps = {
        "platformName": "mac",
        "browserName": "chrome",
        "appium:automationName": "Chromium",
    }
    options = AppiumOptions()
    options.load_capabilities(caps)
    driver = webdriver.Remote('http://localhost:4723', options=options)
    cmd = [
        'ffmpeg',
        '-f',
        'avfoundation',
        '-i', '1',  # 1: screen capture
        'output.mkv',
    ]
    
    # start capturing
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        start_new_session=True
    )
    
    # do something
    driver.get(url)
    time.sleep(1)
    # click first entry
    elem = driver.find_element(AppiumBy.XPATH, './/a[@class="link list_item"]')
    elem.click()
    time.sleep(3)
    driver.quit()
    
    # send SIGINT to stop capturing
    os.killpg(os.getpgid(proc.pid), signal.SIGINT)
    stdout, stderr = proc.communicate()
    print(f'stdout:{stdout}')
    print(f'stderr:{stderr}')


def main():
    capture()


if __name__ == '__main__':
    main()