Python subprocess 모듈

 

목차 LIST

 

들어가기 전에

Concurrency(병행성) : 컴퓨터가 여러 일을 동시에 하듯이 수행하는 것

예) CPU 코어가 하나인 컴퓨터에서 운영체제는 단일 프로세서에서 실행하는 프로그램을 교대로 빠르게 변경하여 프로그램들이 동시에 실행되는 것처럼 보이게 한다.

 

Parallelism(병렬성) : 실제로 여러 작업을 동시에 실행하는 것

CPU 코어가 여러 개인 컴퓨터는 여러 프로그램을 동시에 실행할 수 있다. 각 CPU 코어가 각기 다른 프로그램의 명령어(instruction)를 실행하여 같은 순간에 여러 프로그램들이 실행될 수 있다.

 

Concurrency 와 Parallelism은 속도 향상의 측면에서 가장 큰 차이점을 가진다.

한 프로그램에서 서로 다른 작업을 병렬로 진행하면 전체 작업에 걸리는 시간이 절반으로 줄어든다(실행 속도가 두 배로 빨라짐)

하지만 병행은 병렬로 수행하는 것처럼 보이게 해주지만 전체 작업 속도는 향상되지 않는다.

 

이번 포스팅에서는 병렬성을 구현하기 위한 방법 중 하나인 subprocess에 대해 살펴보자.

 

핵심

• subprocess 모듈을 통해 자식 프로세스를 실행하고 입출력 스트림을 관리하자
• 자식 프로세스는 파이썬 인터프리터 내에서 병렬로 작동해 CPU 효율을 최적화한다.
• deadlock에 빠지거나 멈추는 상황을 방지하기 위해 communicate에서 timeout 파라미터를 사용하자

 

자식 프로세스를 관리하려면 subprocess를 사용하자

파이썬은 머신의 CPU 코어를 모두 이용해 프로그램의 처리량을 극대화할 수 있다. 파이썬 자체는 CPU 속도에 의존할 수도 있지만, CPU 를 많이 사용하는 작업을 관리하고 조절하기 쉽다.

 

요즘은 파이썬에서 자식 프로세스를 관리하기 위해 내장 모듈 subprocess를 사용할 수 있다. subprocess를 사용하면 병렬로 처리할 수 있다.

import subprocess

proc = subprocess.Popen(
    ['echo', 'Hello from the child!'],
    stdout=subprocess.PIPE
)

# communicate 메서드는 자식 프로세스의 출력을 읽고 자식 프로세스가 종료할 때까지 대기
out, err = proc.communicate()
print(out.decode('utf-8'))

코드 설명

1. subprocess.Popen:
    * subprocess.Popen은 새로운 프로세스를 시작하기 위해 사용된다
    * ['sleep', '0.1']이라는 명령어는 실제로 UNIX나 Linux 시스템에서 sleep 0.1이라는 명령어를 실행하는 것과 동일하다.
여기서 sleep은 실행하려는 명령이고, 0.1은 그 명령에 전달하는 인자입니다.
    * Popen을 호출하면 외부 프로세스가 시작되며, 이 프로세스는 별도의 스레드나 코어에서 독립적으로 실행된다.
2. while proc.poll() is None:
    * proc.poll()은 해당 프로세스가 아직 실행 중인지 여부를 확인한다.
    * 프로세스가 아직 실행 중일 경우 proc.poll()은 None을 반환하며, 프로세스가 종료되면 종료 코드(예: 0이면 정상 종료)를 반환한다.
    * 따라서 이 while 루프는 sleep 0.1 명령이 완료될 때까지 반복적으로 실행된다.
3. print('Working ...'):
    * while 루프 안에서 print('Working ...')는 sleep 0.1 명령이 완료될 때까지 계속 출력된다.
    * 실제로 sleep 0.1은 0.1초 동안 실행되므로, 'Working ...' 메시지는 0.1초 동안 여러 번 출력될 수 있다.

 

부모 프로세스가 자유롭게 여러 자식 프로세스를 병렬로 실행하는 것을 부모에게서 자식프로세스를 떼어낸다고 말한다.

병렬로 실행하기 위해서는 모든 자식 프로세스를 순차적이 아니라 동시에 시작하게 하면 된다.

def run_sleep(period):
    proc = subprocess.Popen(['sleep', str(period)])
    return proc


start = time()
procs = []
for _ in range(10):
    proc = run_sleep(0.1)
    procs.append(proc)

for proc in procs:
    proc.communicate()

end = time()
print('Finished in %.3f seconds' % (end - start))

# 결과
Finished in 0.131 seconds

만약 모든 자식 프로세스를 병렬로 실행하지 않았다면 전체 시간은 약 1초였을 것이다.

 

communicate() : 자식 프로세스들이 I/O를 마치고 종료하기를 기다림

comunicate() 는 내부에서 wait() 함수를 호출한다. communicate()나 wait() 함수를 사용하지 않으면 zombie 프로세스가 생길 수 있다. zombie 프로세스는 자식 프로세스가 종료되었지만 부모 프로세스가 그 종료 상태를 아직 확인(회수)하지 않아, 프로세스의 종료 정보가 시스템에 남아있는 상태를 의미한다.

 

참고) communicate()

프로세스와의 입출력을 처리한다. 선택적으로 입력을 제공할 수 있으며, 출력 및 오류를 반환한다.

선택적으로 입력을 제공한다는 말은, 다음과 같다.

# stdin(입력) 제공
proc = subprocess.Popen(['some_command'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout_val, stderr_val = proc.communicate(input=b"My input data.")

# stdin(입력) 제공 X
stdout_val, stderr_val = proc.communicate()

입력 데이터를 제공하면 해당 데이터가 외부 프로세스의 stdin으로 전송된다. 프로세스가 추가 입력을 기다리지 않거나 입력이 필요하지 않은 경우에는 입력을 제공하지 않는다.

 

 

파이프(pipe)를 이용해 데이터를 서브 프로세스르 보낸 다음 서브 프로세스의 결과를 받아올 수 있다.

이 방법을 사용하면 다른 프로그램을 통해 작업을 병렬로 수행할 수 있다.

아래 예시에서는 데이터 암호화를 위해 openssl command를 사용한다. stdin.write()로 데이터를 입력하고 communicate()로 결과를 받아온다.

def run_openssl(data):
    env = os.environ.copy()
    env['password'] = b'\xe24U\n\xd0Ql3S\x11'
    proc = subprocess.Popen(
        ['openssl', 'enc', '-pbkdf2', '-pass', 'env:password'],
        env=env,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE
    )
    proc.stdin.write(data)
    proc.stdin.flush()
    return proc


procs = []
for _ in range(3):
    data = os.urandom(10)
    proc = run_openssl(data)
    procs.append(proc)


for proc in procs:
    out, err = proc.communicate()
    print(out[-10:])

참고) stdin.flush() 함수 : 파이썬에서 파일이나 프로세스와의 통신을 수행할 때, 내부적으로 버퍼가 사용된다. 버퍼링은 I/O 연산의 효율성을 높이기 위해 사용되는데, 이는 여러 데이터 조각들을 모아 한 번에 전송하거나 수신하는 방식으로 동작한다. flush 메서드는 이 버퍼에 남아있는 모든 데이터를 강제로 밀어내는 역할을 해서 버퍼에 있는 모든 데이터가 즉시 외부 프로세스로 전송된다.

 

한 자식 프로세스의 결과를 다른 프로세스의 입력으로 연결하여 병렬 체인을 생성할 수 있다.

다음 예제는 데이터를 암호화하는 openssl 프로세스 집합과 암호화된 결과를 md5로 해시하는 병렬 체인이다.

communicate() 로 자식 프로세스들이 시작되면 이들 사이의 I/O는 자동으로 일어난다. 

def run_openssl(data):
    env = os.environ.copy()
    env['password'] = b'\xe24U\n\xd0Ql3S\x11'
    proc = subprocess.Popen(
        ['openssl', 'enc', '-pbkdf2', '-pass', 'env:password'],
        env=env,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE
    )
    proc.stdin.write(data)
    proc.stdin.flush()
    return proc


def run_md5(input_stdin):
    proc = subprocess.Popen(
        ['md5'],
        stdin=input_stdin,
        stdout=subprocess.PIPE
    )
    return proc


input_procs = []
hash_procs = []
for _ in range(3):
    data = os.urandom(10)
    proc = run_openssl(data)
    input_procs.append(proc)
    hash_proc = run_md5(proc.stdout)
    hash_procs.append(hash_proc)

for proc in input_procs:
    proc.communicate()

for proc in hash_procs:
    out, err = proc.communicate()
    print(out.strip())

 

참고) 파이썬 내장 모듈 hashlib에서 md5 함수를 제공하므로 subprocess를 항상 이렇게 실행할 필요는 없다. 이 예제는 subprocess에서 입력과 출력을 파이프로 연결하는 방법을 보여주기 위함이다.

 

 

자식 프로세스가 종료되지 않거나 입력 또는 출력 파이프에서 block될 우려가 있다면 communicate()에 timeout 파라미터를 넘긴다. 이렇게 하면 자식 프로세스가 일정 시간 내에 응답하지 않으면 예외가 일어나서 오동작한 자식 프로세스를 종료할 기회를 얻는다.

timeout 파라미터는 파이썬 3.3 부터 이후 버전에서 사용 가능하다.

proc = run_sleep(10)
try:
	proc.communicate(timeout=0.1)
expect subprocess.TimeoutExpired:
	proc.terminate()
    proc.wait()
    
print('Exit Status', proc.poll())

 

 

References

파이썬 코딩의 기술 Effective Python, 똑똑하게 코딩하는 법

https://blog.naver.com/sagala_soske/222131573917