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 모듈이 기본적으로 이름의 일부로 각 프로세스에 번호를 지정한다는 것을 보여줍니다. 물론 이름을 지정할 때 번호가 추가되지는 않습니다.

Demon

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

멀티쓰레드 

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

멀티프로세스 

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 로 하려면 ( 블로그 독자님이 메일로 물어보셔서 찾아봄) 

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

3. Queue

다음 예 에서는 생산자-소비자 패턴의 가장 중요한 큐를 다룰 것이다. 생산자가 아이템을 큐에 넣어주면 소비자들은 그것을 가져와서 사용하는데 , 그것들 사이의 동기화는 Queue 에서 모두 해결해준다. 즉 더이상 생산자가 task 를 추가할 수 없으면 put 에서 대기를 하게되고, 소비자가 더 이상 가져올 task 가 없으면 get 에서 기다리게 된다.

import threading
import time
import logging
import random
import Queue

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)

BUF_SIZE = 10
q = Queue.Queue(BUF_SIZE)

class ProducerThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, verbose=None):
        super(ProducerThread,self).__init__()
        self.target = target
        self.name = name

    def run(self):
        while True:
            if not q.full():
                item = random.randint(1,10)
                q.put(item)
                logging.debug('Putting ' + str(item)  
                              + ' : ' + str(q.qsize()) + ' items in queue')
                time.sleep(random.random())
        return

class ConsumerThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, verbose=None):
        super(ConsumerThread,self).__init__()
        self.target = target
        self.name = name
        return

    def run(self):
        while True:
            if not q.empty():
                item = q.get()
                logging.debug('Getting ' + str(item) 
                              + ' : ' + str(q.qsize()) + ' items in queue')
                time.sleep(random.random())
        return

if __name__ == '__main__':
    
    p = ProducerThread(name='producer')
    c = ConsumerThread(name='consumer')

    p.start()
    time.sleep(2)
    c.start()
    time.sleep(2)

* 주의 사항 : multiprocessing 의  경우에는 이 Queue 말고 , from multiprocessing import Queue 를 사용함.

LifoQueue 

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

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()

레퍼런스:

http://www.bogotobogo.com/python/Multithread/python_multithreading_Synchronization_Condition_Objects_Producer_Consumer.php

2. Condition

파이썬의 Condidtion 은 쉽게 생각하면 Event + Mutex 쯤으로 보면된다.
다음예를보면 소비자 쓰레드들은 Condition 이 set 이 되길 기다리고 있다. 생산자 쓰레드는 이 Condition을 set 해줘서 다른 쓰레드들에게 진행해도 좋다고 고지한다. 기다리고 있던 쓰레드 모두가 통과할 수 는 없고 상호 배제되어 하나씩 통과된다 . 

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)

def consumer(cv):
    logging.debug('Consumer thread started ...')
    with cv:
    	logging.debug('Consumer waiting ...')
        cv.wait()
        time.sleep(3)
        logging.debug('Consumer consumed the resource')

def producer(cv):
    logging.debug('Producer thread started ...')
    with cv:
        logging.debug('Making resource available')
        logging.debug('Notifying to all consumers')
        cv.notifyAll()

if __name__ == '__main__':
    condition = threading.Condition()
    cs1 = threading.Thread(name='consumer1', target=consumer, args=(condition,))
    cs2 = threading.Thread(name='consumer2', target=consumer, args=(condition,))
    pd = threading.Thread(name='producer', target=producer, args=(condition,))

    cs1.start()
    time.sleep(1)
    cs2.start()
    time.sleep(2)
    pd.start()

(consumer1) Consumer thread started ...

(consumer1) Consumer waiting ...

(consumer2) Consumer thread started ...

(consumer2) Consumer waiting ...

(producer ) Producer thread started ...

(producer ) Making resource available

(producer ) Notifying to all consumers

(consumer2) Consumer consumed the resource

(consumer1) Consumer consumed the resource

threading.Condition vs threading.Event

질문:

지금까지 Condition과 Event 에 대한 명확한 차이점에 대한 설명을 찾지 못해서 드리는 질문입니다. 각각의 사용처에 대해 알려 주실 수 있나요? 제가 지금까지 찾은 예제들은 모두 생산자-소비자모델이었는데요. queue.Queue가 더 좋은솔루션인듯요.

답변:

간단하게 말해서 쓰레드들이 다른곳에서 무엇인가 true 가 될 때까지 기다려야한다면 그리고 일단 그것이 true가 된다면 공유된 자원에 엑세스하는것이 배제적이 될때 Condition이 일반적이고, 반면에 Event 는 쓰레드들이 단지 무엇이 true가 되기만을 기다리는것에 흥미를 가지고 있으면 사용된다.

2-2 Semaphore

파이썬의 Semaphore 는 정해진 갯수의 쓰레드만 통과시켜준다. 예를들어 웹크롤링을 하는 쓰레드를 50개정도로 한정지어 놓는데 사용할 수 있다.  아래 예제에서는 10개의 쓰레드 중에서 3개만 일을 하도록 한다. 만약 세마포어를 통과한 3개의 쓰레드가 동일한 리소스를 사용하려고 할 때는 그들끼리 Lock 을 통해서 상호배제되어야 할 것이다.

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)

class ThreadPool(object):
    def __init__(self):
        super(ThreadPool, self).__init__()
        self.active = []
        self.lock = threading.Lock()
    def makeActive(self, name):
        with self.lock:
            self.active.append(name)
            time.sleep(5)
            logging.debug('Running: %s', self.active)
    def makeInactive(self, name):
        with self.lock:
            self.active.remove(name)
            logging.debug('Running: %s', self.active)

def f(s, pool):
    logging.debug('Waiting to join the pool')
    with s:
        name = threading.currentThread().getName()
        pool.makeActive(name)
        time.sleep(1)
        pool.makeInactive(name)

if __name__ == '__main__':
    pool = ThreadPool()
    s = threading.Semaphore(3)
    for i in range(10):
        t = threading.Thread(target=f, name='thread_'+str(i), args=(s, pool))
        t.start()

2-3 Event

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)
                    
def wait_for_event(e):
    logging.debug('wait_for_event starting')
    event_is_set = e.wait()
    logging.debug('event set: %s', event_is_set)

def wait_for_event_timeout(e, t):
    while not e.isSet():
        logging.debug('wait_for_event_timeout starting')
        event_is_set = e.wait(t)
        logging.debug('event set: %s', event_is_set)
        if event_is_set:
            logging.debug('processing event')
        else:
            logging.debug('doing other things')

if __name__ == '__main__':
    e = threading.Event()
    t1 = threading.Thread(name='blocking', 
                      target=wait_for_event,
                      args=(e,))
    t1.start()

    t2 = threading.Thread(name='non-blocking', 
                      target=wait_for_event_timeout, 
                      args=(e, 2))
    t2.start()

    logging.debug('Waiting before calling Event.set()')
    time.sleep(3)
    e.set()
    logging.debug('Event is set')

레퍼런스:

http://www.bogotobogo.com/python/Multithread/python_multithreading_Synchronization_Condition_Objects_Producer_Consumer.php

1. 쓰레드 

파이썬에서의 쓰레드는 보통 Thread 혹은 Threading 모듈을 사용할 수 있다. 
또한 Queue 모듈을 통해서 생산자-소비자 패턴을 구현한다. 여러가지 이유로 Thread 보다는 Threading 모듈을 사용하길 추천한다.따라서 이 글에서는 Threading 과 Queue 에 대해서 알아본다.

daemon 

파이썬에서 thread 는 기본적으로 daemon 속성이 False 인데, 메인 쓰레드가 종료되도 자신의 작업이 끝날 때까지 계속 실행된다. 부모가 종료되면 즉시 끝나게 하려면 True 를 해줘야한다. 데몬속성은 반드시 start 이전에 호출되어야 한다. 

Threading 

Threading 모듈은 아래와 같은 객체들을 가지고 있다.

Thread 클래스

Thread 생성방식

1. 함수를 전달해서

하나는 8초, 하나는 2초동안 실행되는 2개의 쓰레드를  threading.Thread 에 함수를 지정해서 실행한다. 

2. 상속을 통해 

threading.Thread 를 상속받은 클래스에 run 함수를 오버라이드하여 사용한다.

threading.Thread.__init__ 를 반드시 해야한다.

레퍼런스:

코어 파이썬 어플리케이션 프로그래밍

http://stackoverflow.com/questions/7424590/threading-condition-vs-threading-event

http://www.bogotobogo.com/python/Multithread/python_multithreading_Synchronization_Condition_Objects_Producer_Consumer.php

functools.partial 에 대한 설명

def power(base, exponent):
    return base ** exponent

def square(base):
    return power(base, 2)

def cube(base):
    return power(base, 3)

from functools import partial

square = partial(power, exponent=2)
cube = partial(power, exponent=3)

def test_partials():
    assert square(2) == 4
    assert cube(2) == 8

def test_partial_docs():
    assert square.keywords == {"exponent": 2}
    assert square.func == power

    assert cube.keywords == {"exponent": 3}
    assert cube.func == power

conda vs. pip vs. virtualenv

콘다는 패키지매니저이며  가상환경 매니저이기 때문에 편리하고 강력한 콘다로 갑시다~

요기 자세한 정보가 있음 -> 파이선 가상환경 콘다 시작하기