queue between processes - goddes4/python-study-wiki GitHub Wiki

Process 간 Queue 통신 문제

• 증상 1 : uwsgi process가 2개 이상일 경우 Queue가 밀려서 처리됨 # uwsgi --http-socket :8081 -w wsgi:app --process 2 --master --catch-exceptions 해결책 : --enable-threads 을 반드시 추가 해줘야 함

• 증상 2 : uwsgi process가 1개 이상이고 max-request에 의해 queue 를 사용하는 worker 들이 respawn 될 경우 일정 확률로 Queue 가 망가짐 (Multiple Producer 가 아니더라도 문제 발생) # uwsgi --http-socket :8081 -w wsgi:app --process 1 --master --catch-exceptions --max-request=3 --enable-threads 의문점 : 프로세스가 queue에 데이터 전송 완료 전에 respawn 되는 경우 ?

wsgi.py

from test_web import main, app

main()

test_web.py

import os
import time

from multiprocessing import Process, Queue
from flask import Flask

from sub.process import other_process


app = Flask(__name__)
tx_seq = 0


@app.route("/")
def index():
    global tx_seq
    tx_seq += 1
    other_process.queue_put(("MSG - pid : {}, seq : {}".format(os.getpid(), tx_seq), "WEB_API"))
    print("PUT - pid : {}, seq : {}".format(os.getpid(), tx_seq))
    return "hello"


def main():
    other_process.init_process()
	

sub/process/other_process.py

import os
import time

from threading import Thread
from multiprocessing import Process, Queue

msg_queue = Queue()


def queue_put(msg):
    msg_queue.put(msg)
    msg_queue.put('STOP')


def background(queue, source):
    seq = 0
    while True:
        try:
            queue_put(("MSG - pid : {}, seq : {}".format(os.getpid(), seq), source))
            seq += 1
            time.sleep(5)
        except Exception as e:
            print(e)


def consumer(queue):
    print("CONSUMER - PID : ", os.getpid())
    t = Thread(target=background, args=(msg_queue, 'by Thread', )).start()

    while True:
        try:
            for msg, source in iter(queue.get, 'STOP'):
                print('===> recv pid : {}, '.format(os.getpid()), msg, ", ", source)
        except Exception as e:
            print(e)


def init_process():
    p1 = Process(target=consumer, args=(msg_queue,))
    p1.daemon = True
    p1.start()

    p2 = Process(target=background, args=(msg_queue, 'by Process', ))
    p2.daemon = True
    p2.start()