[CS/Python]Multiprocessing(1)

Jay·2023년 1월 30일
python
0
post-thumbnail
post-custom-banner

Multiprocessing 구현

multiprocessing 모듈을 사용하면 각각의 프로세스가 분리된 메모리 공간을 가지게 됩니다. 또한 CPU의 코어 개수만큼 작업들을 병렬적으로 처리할 수 있어 자원을 최대한 활용할 수 있으며, GIL이나 전체 Lock의 영향을 받지 않습니다. 분리된 메모리 공간을 가짐으로써 shared memory를 사용하지 않는 한 자원의 무결성 문제도 사라집니다. 따라서 파이썬에서는 CPU Bound 작업은 multiprocessing 모듈로 구현하는 것이 효율적 입니다.

하지만 multiprocessing 모듈은 스레드보다 많은 메모리를 사용합니다. 또한 분리된 메모리 공간을 가짐으로서 서로 다른 프로세스간 데이터를 공유하기 위해서는 별도의 IPC(Inter Process Communication)을 구현해야 합니다.

함수

import os
import multiprocessing

def worker(count):
    print("\nname : %s, argument : %s" % (multiprocessing.current_process().name, count))
    print("parent pid : %s, pid : %s" % (os.getppid(), os.getpid()))

def main():
    for i in range(5):
        p = multiprocessing.Process(target=worker, name="process %i" % i, args=(i,))	# 프로세스 생성
        p.start()																		# 프로세스 시작

if __name__ == "__main__":
    main()

multiprocessing을 구현하는 방법은 스레드와 유사합니다. multiprocessing.Process클래스를 사용하여 프로세스를 구현하였습니다. worker함수는 현재 실행중인 프로세스의 이름과 인자, 그리고 부모 프로세스와 자신의 프로세스 ID를 출력하는 함수입니다. 작업을 실행시킨 부모 프로세스는 모두 같지만 자식 프로세스는 모두 다른 PID를 가지게 됩니다.

스레드의 구현과 또다른 차이점이 존재합니다. 스레드에서는 예시 코드를 작성하며 임의대로 다른 스레드에게 GIL을 넘기기 위해 I/O 를 실행하거나 time.sleep()을 통해 다른 스레드로 GIL을 넘겨주어야 했습니다. 하지만 multiprocess.Process는 GIL이 아닌 다른 방식을 통해 실행이 제어되고 있습니다.

name : process 0, argument : 0

name : process 1, argument : 1
parent pid : 20444, pid : 30784
parent pid : 20444, pid : 30636

name : process 2, argument : 2
parent pid : 20444, pid : 4788

name : process 3, argument : 3
parent pid : 20444, pid : 2012

name : process 4, argument : 4
parent pid : 20444, pid : 38720

클래스

import os
import multiprocessing

class Worker(multiprocessing.Process):

    def __init__(self, name, args):
        multiprocessing.Process.__init__(self)
        self.name = name
        self.args = args

    def run(self):
        print ("\nname : %s, argument : %s" % (self.name, self.args[0]))
        print ("parent pid : %s, pid : %s" % (os.getppid(), os.getpid()))

def main():
    for i in range(5):
        p = Worker(name="process %i" % i, args=(i,))
        p.start()

if __name__ == "__main__":
    main()

스레드와 마찬가지로 multiprocess.Process 클래스를 상속받아 프로세스를 구현할 수 있습니다. Process 클래스는 threading.Thread의 API를 따릅니다. 따라서 구현, 사용 방법도 Thread의 구현, 사용 방법과 유사합니다.

name : process 0, argument : 0
parent pid : 5528, pid : 30540

name : process 1, argument : 1
parent pid : 5528, pid : 37528

name : process 2, argument : 2

name : process 3, argument : 3
parent pid : 5528, pid : 19916

parent pid : 5528, pid : 31344

name : process 4, argument : 4
parent pid : 5528, pid : 38444

API

class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None):
  • name : 프로세스의 이름에 해당하는 속성.
  • daemon : 프로세스의 데몬 여부를 저장하는 플래그 속성. start()가 호출되기 전에 설정되어야 한다.
  • pid : 프로세스의 ID 속성. 프로세스가 spawn되기 전까지는 None값을 가진다.
  • exitcode : 자식 프로세스의 종료 코드. 프로세스가 종료되지 않는다면 None값을 가진다. 만약 정상적으로 run()으로 인하여 반환값을 반환한다면 0, sys.exit(N)을 통해 종료된다면 N값을 가지게 된다.
  • run() : 프로세스의 활동을 정의하는 메소드. 서브 클래스에서 run() 메소드를 오버라이딩하여 동작을 재정의할 수 있다.
  • start() : 프로세스를 실행시키는 메소드.
  • join([timeout]) : timeout이 None인 경우 join() 메소드가 호출된 프로세스가 종료될 때까지 블록된다.
  • is_alive() : 프로세스의 생존 여부를 반환하는 메소드. 생존 기간은 대략 프로세스 객체가 start() 메소드가 반환하는 순간부터, 자식 프로세스가 종료될 때까지이다.
  • terminate() : 프로세스를 강제 종료하는 메소드. 단, 종료된 프로세스의 자식 프로세스들은 종료되지 않는다.
  • kill() : terminate()와 같다.
  • close() : 프로세스 객체의 모든 자원들을 해제하는 메소드. 만약 자식 프로세스가 실행 중이라면 ValueError가 발생한다.

Daemon Process

스레드와 마찬가지로 프로세스도 데몬으로 구현하여 동작시킬 수 있습니다. 스레드와 마찬가지로 부모 프로세스는 자식 프로세스가 종료되지 않으면 부모 프로세스도 종료되지 않습니다. 따라서 데몬 프로세스를 사용하여 메인 프로세스에 영향을 주지 않으면서 동작시킬 작업들을 구현할 수 있습니다.

Daemon Process 구현

import time
import multiprocessing

def daemon():
    print ("Start")
    time.sleep(5)
    print ("Exit")

def main():
    d = multiprocessing.Process(name="daemon", target=daemon)
    d.daemon = True										# 데몬 속성 True 로 설정

    d.start()
    time.sleep(3)
    # d.join()											# 데몬 프로세스가 종료되야 메인 프로세스도 종료

if __name__ == "__main__":
    main()

메인 프로세스에서 프로세스를 구현하고, daemon 속성을 True로 변경했습니다. 스레드에서는 setDaemon() 메소드를 통해 데몬으로 설정하였지만, 프로세스에서는 직접적으로 daemon 속성을 설정할 수 있습니다.

Start

프로그램 실행 결과를 보면 데몬 프로세스의 Exit 출력문이 실행되지 않고 프로그램이 종료되었습니다. 이처럼 데몬 프로세스는 메인 프로세스가 실행이 종료되면 자동으로 종료됩니다.

만약 데몬 프로세스가 종료된 후 메인 프로그램이 종료되기를 원한다면, join() 메소드를 사용하여 띄워진 데몬이 종료될 때까지 기다리도록 구현할 수 있습니다.

Process Exit

스레드에서는 생성된 스레드를 종료할 수 있는 방법이 없었습니다. 하지만 프로세스는 생성된 자식 프로세스를 강제로 종료시킬 수 있을 뿐만 아니라, 상태를 확안하거나 프로세스의 수행 결과를 반환 받을 수도 있습니다.

import sys
import time
import multiprocessing

def good_job():				# 정상적으로 실행된 후 0을 반환하며 종료하는 함수
    p = multiprocessing.current_process()
    print ("Start name:%s, pid:%s" % (p.name, p.pid))
    time.sleep(5)
    print ("Exit name:%s, pid:%s" % (p.name, p.pid))
    return 0

def fail_job():				# sys.exit()을 통해 1을 반환하며 종료되는 함수
    p = multiprocessing.current_process()
    print ("Start name:%s, pid:%s" % (p.name, p.pid))
    time.sleep(5)
    print ("Exit name:%s, pid:%s" % (p.name, p.pid))
    sys.exit(1)

def kill_job():
    p = multiprocessing.current_process()
    print ("Start name:%s, pid:%s" % (p.name, p.pid))
    time.sleep(10)
    print ("Exit name:%s, pid:%s" % (p.name, p.pid))
    return 0

def main():
    process_list = []
    for func in [good_job, fail_job, kill_job]:
        p = multiprocessing.Process(name=func.__name__, target=func)
        process_list.append(p)

        print ("Process check : %s, %s" % (p, p.is_alive()))
        p.start()
        time.sleep(0.3)

    for p in process_list:		# 프로세스를 실행시킨 후 상태 출력
        print ("Process check : %s, %s" % (p, p.is_alive()))

    time.sleep(7)				# 기다리는 동안 프로세스 1과 2는 return과 sys.exit()을 통해 종료된다.
    							# 마지막 프로세스는 종료되지 못한 live 상태

    for p in process_list:
        print ("Process check : %s, %s" % (p, p.is_alive()))

        if p.is_alive():
            print ("Terminate process : %s" % p)
            p.terminate()		# 살아있는 프로세스를 다른 프로세스에서 종료시킬 수 있다.

    for p in process_list:
        print ("Process name : %s, exit code : %s" % (p.name, p.exitcode))

if __name__ == "__main__":
    main()

위 프로그램의 실행흐름은 다음과 같습니다.

우선 프로세스 3개를 생성한 후, 각 프로세스의 상태를 출력하고 프로세스를 실행시킵니다. 각 프로세스는 실행되며 메인 프로세스가 기다리는 동안 good_job 프로세스와 fail_job 프로세스는 return과 sys.exit()을 통해 프로세스가 종료되게 됩니다. 하지만 kill_job 프로세스는 메인 프로세스에서 기다리는 동안 종료되지가 않았고, 그 결과 메인 프로세스에서 terminate() 메소드를 통해 종료되게 됩니다.

이처럼 프로세스 클래스에서는 프로세스를 조작하기 위한 is_alive(), exitcode, terminate() 를 제공합니다. 이를 통해 프로세스의 상태, 동작 여부, 반환값을 확인하고, 프로세스 상태에 따른 동작과 로직을 구현할 수 있습니다.

Process check : <Process name='good_job' parent=16132 initial>, False
Process check : <Process name='fail_job' parent=16132 initial>, False
Process check : <Process name='kill_job' parent=16132 initial>, False

Process check : <Process name='good_job' pid=25032 parent=16132 started>, True
Process check : <Process name='fail_job' pid=3520 parent=16132 started>, True
Process check : <Process name='kill_job' pid=15512 parent=16132 started>, True

Start name:good_job, pid:25032
Start name:fail_job, pid:3520
Start name:kill_job, pid:15512

Exit name:good_job, pid:25032
Exit name:fail_job, pid:3520

Process check : <Process name='good_job' pid=25032 parent=16132 stopped exitcode=0>, False
Process check : <Process name='fail_job' pid=3520 parent=16132 stopped exitcode=1>, False
Process check : <Process name='kill_job' pid=15512 parent=16132 started>, True

Terminate process : <Process name='kill_job' pid=15512 parent=16132 started>

Process name : good_job, exit code : 0
Process name : fail_job, exit code : 1
Process name : kill_job, exit code : None




reference

https://docs.python.org/ko/3.9/library/multiprocessing.html

profile
Jay
이전 포스트

[CS/Python]Thread(3)-Condition, Lock, Mutex, Semaphore, Local Data

다음 포스트

[CS/Python]Multiprocessing(2)

post-custom-banner

0개의 댓글