파이썬 동시성 프로그래밍 - (4) multiprocessing

연재 순서 

1. threading
2. Condition & Semaphore
3. Queue
4. multiprocessing
5. 비동기 (gevent) 
6. 분산 (celery)
7. GPGPU (PyCUDA)
8. 코루틴,asyncio,async/awiat
9. concurrent.future

4. Multiprocessing

멀티 프로세싱 모듈은 Python 버전 2.6에 추가되었습니다. 원래 버전은 Jesse Noller와 Richard Oudkerk에 의해  PEP 371에서 정의되었었습니다. multiprocessing 모듈을 사용하면 스레딩 모듈로 스레드를 생성 할 수있는 것과 동일한 방식으로 프로세스를 생성 할 수 있습니다. 여기서 주요 포인트는 프로세스를 생성하기 때문에 GIL (Global Interpreter Lock)을 피하고 시스템의 여러 프로세서를 최대한 활용할 수 있다는 것입니다.

multiprocessing  패키지에는 또한 스레딩 모듈에 없는 몇 가지 API가 포함되어 있습니다. 예를 들어 여러 입력에서 함수 실행을 병렬화하는 데 사용할 수있는 깔끔한 Pool 클래스가 있습니다. 우리는 이후 섹션에서 Pool을 볼 것입니다. 먼저 multiprocessing   모듈의 Process 클래스부터 시작하겠습니다.

multiprocessing 시작하기

Process 클래스는 스레딩 모듈의 Thread 클래스와 매우 유사합니다. 같은 함수를 호출하는 일련의 프로세스를 만들어 봅시다.

import os
 
from multiprocessing import Process
 
def doubler(number):

    result = number * 2
    proc = os.getpid()
    print('{0} doubled to {1} by process id: {2}'.format(
        number, result, proc))
 
if __name__ == '__main__':
    numbers = [5, 10, 15, 20, 25]
    procs = []
 
    for index, number in enumerate(numbers):
        proc = Process(target=doubler, args=(number,))
        procs.append(proc)
        proc.start()
 
    for proc in procs:
        proc.join()

이 예제에서는 Process를 이용해서 doubler 함수를 실행합니다. 함수 내에서 전달 된 숫자를 두 배로 늘립니다. 또한 파이썬의 os 모듈을 사용하여 현재 프로세스의 ID (또는 pid)를 가져옵니다. 이것은 어떤 프로세스가 함수를 호출하는지 알려줍니다. 그런 다음 아래쪽의 코드 블록에서 일련의 프로세스를 만들고 시작합니다. 마지막 마지막 루프는 각 프로세스에서 join () 메소드를 호출하기 만합니다.이 메소드는 파이썬에게 프로세스가 종료 될 때까지 대기하도록 지시합니다. 프로세스를 중지해야하는 경우 terminate () 메소드를 호출 할 수 있습니다.

결과) 

5 doubled to 10 by process id: 10468
10 doubled to 20 by process id: 10469
15 doubled to 30 by process id: 10470
20 doubled to 40 by process id: 10471
25 doubled to 50 by process id: 10472

때로는 프로세스에 사람이 읽을 수있는 이름을 사용하는 것이 더 좋을 수도 있습니다.

import os
 
from multiprocessing import Process, current_process
 
 
def doubler(number):

    result = number * 2
    proc_name = current_process().name
    print('{0} doubled to {1} by: {2}'.format(
        number, result, proc_name))
 
 
if __name__ == '__main__':
    numbers = [5, 10, 15, 20, 25]
    procs = []
    proc = Process(target=doubler, args=(5,))
 
    for index, number in enumerate(numbers):
        proc = Process(target=doubler, args=(number,))
        procs.append(proc)
        proc.start()
 
    proc = Process(target=doubler, name='Test', args=(2,))
    proc.start()
    procs.append(proc)
 
    for proc in procs:
        proc.join()

이번에는 current_process라는 것을 추가로 가져옵니다. current_process는 기본적으로 스레딩 모듈의 current_thread와 동일합니다. 우리는이 함수를 사용하여 함수를 호출하는 스레드의 이름을 가져옵니다. 처음 5 개 프로세스에 대해서는 이름을 설정하지 않는다는 점에 유의하십시오. 그런 다음 여섯 번째 단계에서 프로세스 이름을 "Test"로 설정합니다. 결과는 다음과 같습니다.

5 doubled to 10 by: Process-2
10 doubled to 20 by: Process-3
15 doubled to 30 by: Process-4
20 doubled to 40 by: Process-5
25 doubled to 50 by: Process-6
2 doubled to 4 by: Test

출력은 multiprocessing 모듈이 기본적으로 이름의 일부로 각 프로세스에 번호를 지정한다는 것을 보여줍니다. 물론 이름을 지정할 때 번호가 추가되지는 않습니다.

multiprocessing.Queue vs multiprocessing.Manager().Queue

가 있는데 개인적으로 Manager().Queue를 모든 경우에서 디폴트로 사용하라고 권하고 싶다. 

Demon

일반적으로 데몬쓰레드라고 하면, 메인이 죽으면 같이 죽는 쓰레드를 말한다.

멀티쓰레드 

데몬 쓰레드란 백그라운드에서 실행되는 쓰레드로 메인 쓰레드가 종료되면 즉시 종료되는 쓰레드이다. 
디폴트는 넌데몬이며, 해당 서브쓰레드는 메인 쓰레드가 종료할 지라도 자신의 작업이 끝날 때까지 계속 실행된다.

멀티프로세스 

 프로세스가 종료되면 그것의 자식 데몬 프로세스들을 강제종료 시킬 것이다. 데몬 프로세스는 자식 프로세스를 생성 할 수 없으며, 유닉스 데몬이나 서비스가 아니다. 디폴트인 넌데몬프로세스 (daemon = False) 일 경우에는 해당 프로세스가 종료 될 때까지 메인프로세스는 종료되지 않는다. 암시적으로 내부에서 join() 하고 있다.  

참고로 메인프로세스가 갑자기 죽은 경우 데몬 자식 프로세스를 종료하지 못한다.

Locks

멀티 프로세싱 모듈은 스레딩 모듈과 거의 같은 방식으로 잠금을 지원합니다. 가져 오기 잠금, 가져 오기, 무언가를 수행하고 해제 만하면됩니다. 보시죠.

from multiprocessing import Process, Lock
 
 
def printer(item, lock):

    lock.acquire()
    try:
        print(item)
    finally:
        lock.release()
 
if __name__ == '__main__':
    lock = Lock()
    items = ['tango', 'foxtrot', 10]
    for item in items:
        p = Process(target=printer, args=(item, lock))
        p.start()

여기에서는 전달한 내용을 print 하는 간단한 print 기능을 만듭니다. 프로세스가 서로 간섭하지 않도록 Lock 객체를 사용합니다. 이 코드는 세 항목의 목록을 반복하고 각 항목에 대한 프로세스를 만듭니다. 각 프로세스는 함수를 호출하고 iterable의 항목 중 하나를 전달합니다. 우리가 잠금을 사용하고 있기 때문에, 다음 라인의 프로세스는 계속하기 전에 잠금이 해제 될 때까지 기다릴 것입니다.

Logging

로깅 프로세스는 로깅 스레드와 약간 다릅니다. 그 이유는 파이썬의 로깅 패키지가 프로세스 공유 잠금을 사용하지 않기 때문에 서로 다른 프로세스의 메시지가 섞여서 끝날 수 있기 때문입니다. 이전 예제에 기본 로깅을 추가해 보겠습니다. 코드는 다음과 같습니다.

import logging
import multiprocessing
 
from multiprocessing import Process, Lock
 
def printer(item, lock):
    """
    Prints out the item that was passed in
    """
    lock.acquire()
    try:
        print(item)
    finally:
        lock.release()
 
if __name__ == '__main__':
    lock = Lock()
    items = ['tango', 'foxtrot', 10]
    multiprocessing.log_to_stderr()
    logger = multiprocessing.get_logger()
    logger.setLevel(logging.INFO)
    for item in items:
        p = Process(target=printer, args=(item, lock))
        p.start()

로그하는 가장 간단한 방법은 모두 stderr로 보내는 것입니다. 우리는 log_to_stderr () 함수를 호출하여 이를 수행 할 수 있습니다. 그런 다음 get_logger 함수를 호출하여 로거에 액세스하고 로깅 수준을 INFO로 설정합니다. 나머지 코드는 동일합니다. 여기서 join () 메서드를 호출하지 않는다는 것에 주목 하십시요. 대신 부모 스레드 (즉, 스크립트)가 종료 될 때 암시적으로 join ()을 호출합니다.

결과는 아래와 같습니다.

[INFO/Process-1] child process calling self.run()
tango
[INFO/Process-1] process shutting down
[INFO/Process-1] process exiting with exitcode 0
[INFO/Process-2] child process calling self.run()
[INFO/MainProcess] process shutting down
foxtrot
[INFO/Process-2] process shutting down
[INFO/Process-3] child process calling self.run()
[INFO/Process-2] process exiting with exitcode 0
10
[INFO/MainProcess] calling join() for process Process-3
[INFO/Process-3] process shutting down
[INFO/Process-3] process exiting with exitcode 0
[INFO/MainProcess] calling join() for process Process-2

이제 로그를 디스크에 저장하려면 좀 더 까다로워집니다. 파이썬의 logging Cookbook에서 그 주제에 관해 읽을 수 있습니다.

The Pool Class

Pool 클래스는 작업자 프로세스 풀을 나타내는 데 사용됩니다. 여기에는 작업을 작업 프로세스로 offload  할 수있는 방법이 있습니다. 정말 간단한 예를 살펴 보겠습니다.

from multiprocessing import Pool
 
def doubler(number):
    return number * 2
 
if __name__ == '__main__':
    numbers = [5, 10, 20]
    pool = Pool(processes=3)
    print(pool.map(doubler, numbers))

기본적으로 여기서 발생하는 것은 Pool의 인스턴스를 만들고 세 개의 작업자 프로세스를 생성하도록 지시한다는 것입니다. 그런 다음 map 메소드를 사용하여 함수와 반복 가능한 것을 각 프로세스에 매핑합니다. 마지막으로 결과를 인쇄합니다.이 경우 실제로 목록입니다 : [10, 20, 40].

또한 apply_async 메소드를 사용하여 풀에서 프로세스의 결과를 얻을 수 있습니다.

from multiprocessing import Pool
 
def doubler(number):
    return number * 2
 
if __name__ == '__main__':
    pool = Pool(processes=3)
    result = pool.apply_async(doubler, (25,))
    print(result.get(timeout=1))

우리가 할 수있는 것은 process 의 결과를 요구하는 것입니다. get 함수가 그것을 하죠. 우리가 호출 한 기능에 문제가 생길 경우를 대비해서 타임 아웃이 설정되었음을 알 수 있습니다. 우리는 무기한 차단되면 문제가 생길 수도 있으니까요.

Process Communication

communicating 모듈은 프로세스 간 통신시 Queues 와 Pipes라는 두 가지 기본 방법을 사용합니다. 큐는 실제로 스레드와 프로세스에서 빈번히 사용되며 잘 구현되어 있습니다. 큐에 대해 매우 간단한 예제를 살펴 보겠습니다.

* 주의 사항 : multiprocessing 의 Queue 입니다. 그냥 import Queue 는 쓰레드간에 사용됨. 

from multiprocessing import Process, Queue
 
sentinel = -1
 
def creator(data, q):
    """
    Creates data to be consumed and waits for the consumer
    to finish processing
    """
    print('Creating data and putting it on the queue')
    for item in data:
 
        q.put(item)
 
 
def my_consumer(q):

    while True:
        data = q.get()
        print('data found to be processed: {}'.format(data))
        processed = data * 2
        print(processed)
 
        if data is sentinel:
            break
 
 
if __name__ == '__main__':
    q = Queue()
    data = [5, 10, 13, -1]
    process_one = Process(target=creator, args=(data, q))
    process_two = Process(target=my_consumer, args=(q,))
    process_one.start()
    process_two.start()
 
    q.close()
    q.join_thread()
 
    process_one.join()
    process_two.join()

먼저 Queue 및 Process를 import 합니다. 그런 다음 데이터를 생성하여 큐에 추가하고 데이터를 소비하고 처리하는 두 가지 기능을 수행합니다. Queue에 데이터를 추가하는 것은 Queue의 put () 메소드를 사용하는 반면 Queue에서 데이터를 가져 오는 것은 get 메소드를 통해 수행됩니다. 코드의 마지막 덩어리는 Queue 객체와 두 개의 프로세스를 생성 한 다음 실행합니다. Queue 자체보다는 프로세스 객체에 대해 join ()을 호출한다는 것을 알 수 있습니다.

*  쓰레드를 위한 Queue 는 따로 있음을 명심하십시요. 그 큐는 Queue.Queue(10) 이렇게 사용합니다.

LifoQueue 

참고로 멀티프로세서용 Queue 는 FIFO 인데, LIFO 로 하려면 ( 블로그 독자님이 메일로 물어보셔서 찾아봄) 

multiprocessing-managers 라는 것을 통해서 해결 할 수 있을거 같다.

아래는 LifoQueue 예제입니다. LifQueue 자체는 멀티프로세서에서 사용못하지만 
managers 를 통해서 사용 할 수 있게 하는거 같습니다. 테스트를 통해 확인 하고 사용하세요.

from multiprocessing import Process
from multiprocessing.managers import BaseManager
from time import sleep
from queue import LifoQueue


def run(lifo):
    """Wait for three messages and print them out"""
    num_msgs = 0
    while num_msgs < 3:
        # get next message or wait until one is available
        s = lifo.get()
        print(s)
        num_msgs += 1


# create manager that knows how to create and manage LifoQueues
class MyManager(BaseManager):
    pass
MyManager.register('LifoQueue', LifoQueue)


if __name__ == "__main__":

    manager = MyManager()
    manager.start()
    lifo = manager.LifoQueue()
    lifo.put("first")
    lifo.put("second")

    # expected order is "second", "first", "third"
    p = Process(target=run, args=[lifo])
    p.start()

    # wait for lifoqueue to be emptied
    sleep(0.25)
    lifo.put("third")

    p.join()

---

레퍼런스:

https://www.blog.pythonlibrary.org/2016/08/02/python-201-a-multiprocessing-tutorial/

https://pymotw.com/2/multiprocessing/basics.html

https://docs.python.org/2/library/multiprocessing.html#multiprocessing.Process.daemon