파이썬 스레딩 프로그래밍 고급 우선순위 큐 PriorityQueue와 기아 역전 대응 방법

⚙️ 파이썬 PriorityQueue의 기본 동작 원리

파이썬의 queue.PriorityQueue는 내부적으로 힙(Heap) 자료구조를 기반으로 동작합니다. 힙은 부모 노드가 자식 노드보다 항상 작거나 큰 성질을 가지는 트리 구조로, 파이썬에서는 heapq 모듈을 통해 구현됩니다. 따라서 PriorityQueue는 삽입된 아이템 중 가장 낮은 값(혹은 가장 높은 우선순위)을 가진 요소를 빠르게 꺼낼 수 있도록 보장합니다.

일반적인 큐는 선입선출(FIFO) 원칙을 따르지만, 우선순위 큐는 작업의 중요도를 기준으로 처리 순서를 바꿉니다. 예를 들어, 긴급 알림 메시지, 시스템 오류 처리 같은 태스크는 높은 우선순위로 설정하여 먼저 실행될 수 있습니다. 반대로 백그라운드 로그 저장 같은 작업은 낮은 우선순위로 설정되어 여유가 있을 때 실행되죠.

📌 PriorityQueue의 기본 사용 예시

import queue

pq = queue.PriorityQueue()

# (우선순위, 작업) 형태로 삽입
pq.put((1, "긴급 작업"))
pq.put((3, "일반 작업"))
pq.put((2, "보통 작업"))

while not pq.empty():
    print(pq.get())

위 코드를 실행하면 우선순위가 낮은 숫자일수록 높은 우선순위를 의미하기 때문에 (1, “긴급 작업”)이 가장 먼저 실행되고, 그다음 (2, “보통 작업”), 마지막으로 (3, “일반 작업”)이 처리됩니다. 이렇게 우선순위를 명시하면 작업의 중요도에 따라 실행 순서를 쉽게 제어할 수 있습니다.

🧵 스레드와 우선순위 큐의 결합 활용

실제 애플리케이션에서는 하나의 스레드가 단일 작업만 처리하지 않습니다. 여러 개의 작업 요청이 동시에 들어올 수 있고, 이를 효율적으로 분배하려면 스레드 풀(Thread Pool)과 PriorityQueue의 조합이 매우 효과적입니다. 프로듀서-컨슈머(Producer-Consumer) 패턴과 결합하면 더욱 안정적으로 동작하게 됩니다.

프로듀서는 작업 요청을 PriorityQueue에 넣고, 여러 개의 컨슈머 스레드는 큐에서 우선순위가 높은 작업을 꺼내 처리합니다. 이 방식은 작업의 중요도를 보장하면서 동시에 여러 스레드가 병렬로 실행될 수 있어 성능을 크게 향상시킵니다.

📌 프로듀서-컨슈머 패턴 예시

import threading, queue, time

task_queue = queue.PriorityQueue()

def producer():
    for priority, task in [(3, "일반 로그"), (1, "긴급 알림"), (2, "중요 데이터 저장")]:
        print(f"생성: {task}")
        task_queue.put((priority, task))
        time.sleep(0.5)

def consumer(name):
    while True:
        priority, task = task_queue.get()
        print(f"{name} 실행: {task}")
        task_queue.task_done()

# 컨슈머 스레드 생성
for i in range(2):
    threading.Thread(target=consumer, args=(f"Worker-{i}",), daemon=True).start()

# 프로듀서 실행
producer()

task_queue.join()

이 코드에서는 두 개의 워커 스레드가 동시에 실행되며, 우선순위가 가장 높은 긴급 알림이 먼저 처리되고 그다음 중요 데이터 저장, 마지막으로 일반 로그가 실행됩니다. 즉, 단순한 멀티스레딩보다 훨씬 정교한 태스크 관리가 가능합니다.

• ⚙️우선순위가 높은 작업은 먼저 실행
• 🧵여러 스레드가 동시에 큐에서 작업을 가져가 실행
• 🚀프로듀서-컨슈머 패턴으로 병렬성과 안정성을 동시에 확보

🚦 기아 문제와 우선순위 역전 현상

멀티스레딩 환경에서 우선순위 큐를 사용할 때 주의해야 할 문제가 있습니다. 바로 기아(starvation)와 우선순위 역전(priority inversion) 현상입니다. 두 현상은 시스템의 안정성과 성능에 직접적인 영향을 미치기 때문에 반드시 이해하고 대비해야 합니다.

📌 기아(starvation) 현상

기아는 특정 스레드나 작업이 계속해서 낮은 우선순위로 밀려 실행되지 못하는 상황을 의미합니다. 예를 들어, 긴급한 작업이 끊임없이 발생한다면 일반적인 작업은 무기한 대기 상태에 머무를 수 있습니다. 이는 프로그램 전체 성능에 악영향을 미치고, 중요한 태스크가 방치될 수 있는 위험을 만듭니다.

📌 우선순위 역전(priority inversion)

우선순위 역전은 낮은 우선순위의 스레드가 자원을 점유한 상태에서, 높은 우선순위의 스레드가 해당 자원을 기다려야 하는 상황을 의미합니다. 이 경우 중간 우선순위의 스레드가 계속 실행되면서, 정작 높은 우선순위 스레드는 대기 상태에 놓이게 됩니다. 이런 문제는 실시간 시스템이나 운영체제 커널에서 특히 심각하게 다뤄지는 이슈입니다.

💬 실제 사례: NASA의 마스 패스파인더(Mars Pathfinder) 탐사선은 우선순위 역전 문제로 시스템이 멈추는 현상을 겪었으며, 소프트웨어 업데이트를 통해 ‘우선순위 상속(priority inheritance)’ 기법을 적용하여 해결했습니다.

🛠️ 파이썬에서 기아 문제를 해결하는 기법

우선순위 큐를 사용할 때 발생할 수 있는 기아(starvation)와 우선순위 역전(priority inversion) 문제는 운영체제 이론뿐 아니라 파이썬 프로그래밍에서도 고려해야 할 중요한 주제입니다. 이를 해결하기 위해 몇 가지 실질적인 접근법이 존재합니다.

📌 Aging 기법

Aging은 대기 시간이 길어진 작업의 우선순위를 점차 높여주는 방식입니다. 이렇게 하면 낮은 우선순위의 작업도 일정 시간이 지나면 결국 실행될 수 있습니다. 파이썬에서는 큐에 삽입할 때 (priority + waiting_time, task) 형태로 튜플을 넣어 유사하게 구현할 수 있습니다.

📌 우선순위 상속 (Priority Inheritance)

낮은 우선순위의 스레드가 중요한 자원을 보유하고 있는 동안, 더 높은 우선순위 스레드가 대기하게 되면 문제가 발생합니다. 이때 낮은 우선순위 스레드의 우선순위를 일시적으로 상향 조정하여 자원을 빠르게 해제하도록 하는 것이 우선순위 상속입니다. 운영체제 수준에서 많이 활용되며, 파이썬에서는 직접적인 지원은 없지만 락(Lock) 관리 전략으로 유사한 효과를 구현할 수 있습니다.

📌 라운드 로빈(Round Robin) 보완

우선순위 큐와 라운드 로빈 방식을 혼합하는 방법도 있습니다. 우선순위를 기준으로 먼저 실행하되, 동일한 우선순위 그룹 내에서는 라운드 로빈 방식으로 공정하게 분배합니다. 이렇게 하면 특정 스레드가 독점적으로 실행되는 것을 막을 수 있습니다.

💡 실전 코드 예제와 활용 시나리오

이제 실제 상황에서 PriorityQueue를 어떻게 활용할 수 있는지 코드 예제와 함께 살펴보겠습니다. 특히 기아 문제를 예방하기 위해 Aging 기법을 적용한 방식은 실무에서도 효과적입니다.

📌 Aging을 적용한 PriorityQueue 예제

import queue, time

class AgingPriorityQueue(queue.PriorityQueue):
    def put(self, item, block=True, timeout=None):
        priority, task, timestamp = item
        # 대기 시간이 길어질수록 우선순위를 자동 보정
        age = int(time.time() - timestamp)
        adjusted_priority = priority - age * 0.01
        super().put((adjusted_priority, task, timestamp), block, timeout)

pq = AgingPriorityQueue()

# (우선순위, 작업, 삽입시간)
pq.put((3, "낮은 우선순위 작업", time.time()))
pq.put((1, "긴급 작업", time.time()))

time.sleep(2)
pq.put((2, "중간 작업", time.time()))

while not pq.empty():
    print(pq.get())

위 예제에서는 시간이 지남에 따라 낮은 우선순위의 작업도 점차 높은 우선순위를 부여받아 결국 실행될 수 있습니다. 이렇게 하면 특정 태스크가 무한정 대기하는 기아 현상을 예방할 수 있습니다.

📌 활용 시나리오

❓ 자주 묻는 질문 (FAQ)