소개

본 글은 가장 간단한 생산자-소비자 모델을 통해 프로그램 실행 시 CPU 사용률을 관찰하여 뮤텍스와 뮤텍스+조건 변수의 성능을 비교합니다.

이 예제에서는 1개의 생산자와 5개의 소비자로 구성된 생산자-소비자 모델을 사용합니다. 생산자 스레드가 큐에 데이터를 넣고, 5개의 소비자 스레드가 큐에서 데이터를 가져옵니다. 데이터를 가져오기 전에 큐에 데이터가 있는지 확인해야 하며, 이 큐는 전역 공유 자원이므로 뮤텍스로 보호해야 합니다. 즉, 생산자가 큐에 데이터를 넣을 때 다른 소비자는 데이터를 가져올 수 없으며, 그 반대도 마찬가지입니다.

뮤텍스 구현 코드

#include <iostream> // std::cout
#include <deque>    // std::deque
#include <thread>   // std::thread
#include <chrono>   // std::chrono
#include <mutex>    // std::mutex

// 전역 공유 큐
std::deque<int> shared_queue;

// 전역 뮤텍스
std::mutex queue_mutex;

// 생산자 실행 플래그
bool producer_active = true;

// 생산자 스레드 함수
void producer_task()
{
    // 생산할 항목 수
    int item_count = 8;
    
    do
    {
        // 스마트 락: 초기화 시 자동으로 락을 걸고, 범위를 벗어나면 자동으로 해제
        std::unique_lock<std::mutex> lock(queue_mutex);
        // 큐에 데이터 추가
        shared_queue.push_front(item_count);
        // 락을 미리 해제하여 공유 자원에 대한 동기화 범위를 최소화
        lock.unlock(); 

        std::cout << "생산자: 현재 재고 - " << item_count << std::endl;
            
        // 생산 속도 늦추기 (1초 대기)
        std::this_thread::sleep_for(std::chrono::seconds(1));

        // 재고 감소
        item_count--;
    } while(item_count > 0);
    
    // 생산자 종료 플래그 설정
    producer_active = false;

    std::cout << "생산자: 재고가 소진되었습니다. 작업을 종료합니다!" << std::endl;
}

// 소비자 스레드 함수
void consumer_task(int consumer_id)
{
    int acquired_item = 0;

    do
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if(!shared_queue.empty())
        {
            acquired_item = shared_queue.back();
            shared_queue.pop_back();
            lock.unlock();

            std::cout << "소비자[" << consumer_id << "]: 획득한 항목 번호 - " << acquired_item << std::endl;
        }
        else
        {
            lock.unlock();
        }
    } while(producer_active);
    
    std::cout << "소비자[" << consumer_id << "]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!" << std::endl;
}

int main()
{
    std::cout << "1개 생산자 시작..." << std::endl;
    std::thread producer(producer_task);

    std::cout << "5개 소비자 시작..." << std::endl;
    std::thread consumers[5];
    for(int i = 0; i < 5; i++)
    {
        consumers[i] = std::thread(consumer_task, i + 1);
    }

    producer.join();

    for(int i = 0; i < 5; i++)
    {
        consumers[i].join();
    }

    std::cout << "모든 스레드가 종료되었습니다." << std::endl;

    return 0;
}

뮤텍스 구현 실행 결과:

[root@lincoding condition]# g++ -std=c++0x -pthread -D_GLIBCXX_USE_NANOSLEEP main.cpp -o main
[root@lincoding condition]# ./main
1개 생산자 시작...
5개 소비자 시작...
생산자: 현재 재고 - 8
소비자[1]: 획득한 항목 번호 - 8
소비자[1]: 획득한 항목 번호 - 7
생산자: 현재 재고 - 7
생산자: 현재 재고 - 6
소비자[3]: 획득한 항목 번호 - 6
생산자: 현재 재고 - 5
소비자[1]: 획득한 항목 번호 - 5
생산자: 현재 재고 - 4
소비자[2]: 획득한 항목 번호 - 4
생산자: 현재 재고 - 3
소비자[5]: 획득한 항목 번호 - 3
생산자: 현재 재고 - 2
소비자[2]: 획득한 항목 번호 - 2
생산자: 현재 재고 - 1
소비자[1]: 획득한 항목 번호 - 1
생산자: 재고가 소진되었습니다. 작업을 종료합니다!소비자[5]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[2]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[3]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[4]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[1]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
모든 스레드가 종료되었습니다.

뮤텍스는 이 작업을 수행할 수 있지만 성능 문제가 존재합니다.

• producer_task는 생산자 스레드로, 데이터 생산 과정에서 1초씩 대기하므로 생산 속도가 매우 느립니다.
• consumer_task는 소비자 스레드로 while 루프를 가지고 있습니다. 생산자가 실행되지 않을 때까지 루프를 빠져나가지 않으며, 루프 내부에서는 항상 먼저 락을 획득하고, 큐가 비어있지 않은지 확인한 후 큐에서 데이터를 하나 꺼내고 마지막으로 락을 해제합니다. 따라서 생산자가 1초 동안 대기하는 동안 소비자 스레드는 실제로 많은 불필요한 작업을 수행하여 CPU 사용률이 매우 높아집니다!

4코어 CPU 환경에서 실행:

[root@lincoding ~]# grep 'model name' /proc/cpuinfo | wc -l
4

top 명령어로 CPU 사용량 확인 시 순수 뮤텍스 사용 시 CPU 오버헤드가 매우 큼을 알 수 있습니다. main 프로세스의 CPU 사용률은 357.5%CPU이며, 시스템 오버헤드 CPU는 54.5%sy, 사용자 오버헤드 CPU는 18.2%us입니다.

[root@lincoding ~]# top
top - 19:13:41 up 36 min,  3 users,  load average: 0.06, 0.05, 0.01
Tasks: 179 total,   1 running, 178 sleeping,   0 stopped,   0 zombie
Cpu(s): 18.2%us, 54.5%sy,  0.0%ni, 27.3%id,  0.0%wa,  0.0%hi,  0.0%si,  0.0%st
Mem:   1004412k total,   313492k used,   690920k free,    41424k buffers
Swap:  2031608k total,        0k used,  2031608k free,    79968k cached

   PID USER      PR  NI  VIRT  RES  SHR S %CPU %MEM    TIME+  COMMAND                                                                                                                       
 35346 root      20   0  137m 3288 1024 S 357.5  0.3   0:05.92 main                                                                                                                          
     1 root      20   0 19232 1492 1224 S  0.0  0.1   0:02.16 init                                                                                                                           
     2 root      20   0     0    0    0 S  0.0  0.0   0:00.01 kthreadd                                                                                                                       
     3 root      RT   0     0    0    0 S  0.0  0.0   0:00.68 migration/0  

해결책 중 하나는 소비자에게도 작은 지연을 추가하는 것입니다. 소비자가 데이터를 가져오지 못할 때 500ms 정도 대기하면 뮤텍스로 인한 CPU 오버헤드를 줄일 수 있습니다.

// 소비자 스레드 함수
void consumer_task(int consumer_id)
{
    int acquired_item = 0;

    do
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if(!shared_queue.empty())
        {
            acquired_item = shared_queue.back();
            shared_queue.pop_back();
            lock.unlock();

            std::cout << "소비자[" << consumer_id << "]: 획득한 항목 번호 - " << acquired_item << std::endl;
        }
        else
        {
            lock.unlock();
            // 소비자가 데이터를 가져오지 못할 때 500ms 대기
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
    } while(producer_active);
    
    std::cout << "소비자[" << consumer_id << "]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!" << std::endl;
}

실행 결과에서 CPU 사용률이 크게 감소한 것을 확인할 수 있습니다.

[root@lincoding ~]# ps aux | grep -v grep | grep main
USER        PID %CPU %MEM    VSZ   RSS TTY      STAT START   TIME COMMAND
root      61296  0.0  0.1 141068  1244 pts/1    Sl+  19:40   0:00 ./main

조건 변수 + 뮤텍스 구현 코드

그렇다면 소비자 지연(대기) 시간을 얼마로 설정해야 할까요?

• 생산자가 매우 빠르게 생산하는데 소비자가 500ms를 대기하면 좋지 않습니다.
• 생산자가 더 느리게 생산한다면 소비자를 500ms 대기시켜도 불필요한 작업이 발생하여 CPU를 사용하게 됩니다.

이러한 문제를 해결하기 위해 std::condition_variable 조건 변수를 도입해야 합니다. 생산자-소비자 모델에서 조건 변수를 적용하면, 생산자가 데이터를 하나 생산한 후 notify_one()을 통해 wait() 중인 소비자 스레드를 깨우고, 소비자가 큐에서 데이터를 꺼내도록 할 수 있습니다.

#include <iostream> // std::cout
#include <deque>    // std::deque
#include <thread>   // std::thread
#include <chrono>   // std::chrono
#include <mutex>    // std::mutex
#include <condition_variable> // std::condition_variable

// 전역 공유 큐
std::deque<int> shared_queue;

// 전역 뮤텍스
std::mutex queue_mutex;

// 전역 조건 변수
std::condition_variable queue_cond;

// 생산자 실행 플래그
bool producer_active = true;

// 생산자 스레드 함수
void producer_task()
{
    // 생산할 항목 수
    int item_count = 8;
    
    do
    {
        // 스마트 락: 초기화 시 자동으로 락을 걸고, 범위를 벗어나면 자동으로 해제
        std::unique_lock<std::mutex> lock(queue_mutex);
        // 큐에 데이터 추가
        shared_queue.push_front(item_count);
        // 락을 미리 해제하여 공유 자원에 대한 동기화 범위를 최소화
        lock.unlock(); 

        std::cout << "생산자: 현재 재고 - " << item_count << std::endl;
        
        // 하나의 스레드를 깨움
        queue_cond.notify_one();
        
        // 1초 대기
        std::this_thread::sleep_for(std::chrono::seconds(1));

        // 재고 감소
        item_count--;
    } while(item_count > 0);
    
    // 생산자 종료 플래그 설정
    producer_active = false;
    
    // 모든 소비 스레드를 깨움
    queue_cond.notify_all();
    
    std::cout << "생산자: 재고가 소진되었습니다. 작업을 종료합니다!" << std::endl;
}

// 소비자 스레드 함수
void consumer_task(int consumer_id)
{
    // 획득한 항목 번호
    int acquired_item = 0;

    do
    {
        // 스마트 락: 초기화 시 자동으로 락을 걸고, 범위를 벗어나면 자동으로 해제
        std::unique_lock<std::mutex> lock(queue_mutex);
        
        // wait() 함수는 먼저 뮤텍스의 unlock()을 호출한 후 스스로를 잠들고, 
        // 깨어나면 다시 락을 획득하여 이후 큐 작업을 보호합니다
        // lock_guard가 아닌 unique_lock을 사용해야 합니다. lock_guard는 lock과 unlock 인터페이스를 제공하지 않지만,
        // unique_lock은 두 인터페이스 모두 제공합니다
        queue_cond.wait(lock);
        
        // 큐가 비어있지 않은 경우
        if(!shared_queue.empty())
        {
            // 큐의 마지막 데이터를 가져옴
            acquired_item = shared_queue.back();
            
            // 큐의 마지막 데이터를 삭제
            shared_queue.pop_back();
            
            // 락을 미리 해제하여 공유 자원에 대한 동기화 범위를 최소화
            lock.unlock(); 

            std::cout << "소비자[" << consumer_id << "]: 획득한 항목 번호 - " << acquired_item << std::endl;
        }
        // 큐가 비어있는 경우
        else
        {
            lock.unlock();
        }
    
    } while(producer_active);
    
    std::cout << "소비자[" << consumer_id << "]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!" << std::endl;
}

int main()
{
    std::cout << "1개 생산자 시작..." << std::endl;
    std::thread producer(producer_task);

    std::cout << "5개 소비자 시작..." << std::endl;
    std::thread consumers[5];
    for(int i = 0; i < 5; i++)
    {
        consumers[i] = std::thread(consumer_task, i + 1);
    }

    producer.join();

    for(int i = 0; i < 5; i++)
    {
        consumers[i].join();
    }

    std::cout << "모든 스레드가 종료되었습니다." << std::endl;

    return 0;
}

조건 변수 + 뮤텍스 실행 결과

[root@lincoding condition]# g++ -std=c++0x -pthread -D_GLIBCXX_USE_NANOSLEEP main.cpp -o main
[root@lincoding condition]# ./main
1개 생산자 시작...
5개 소비자 시작...
생산자: 현재 재고 - 8
소비자[4]: 획득한 항목 번호 - 8
생산자: 현재 재고 - 7
소비자[2]: 획득한 항목 번호 - 7
생산자: 현재 재고 - 6
소비자[3]: 획득한 항목 번호 - 6
생산자: 현재 재고 - 5
소비자[5]: 획득한 항목 번호 - 5
생산자: 현재 재고 - 4
소비자[1]: 획득한 항목 번호 - 4
생산자: 현재 재고 - 3
소비자[4]: 획득한 항목 번호 - 3
생산자: 현재 재고 - 2
소비자[2]: 획득한 항목 번호 - 2
생산자: 현재 재고 - 1
소비자[3]: 획득한 항목 번호 - 1
생산자: 재고가 소진되었습니다. 작업을 종료합니다!
소비자[5]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[1]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[4]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[2]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
소비자[3]: 판매자가 재고를 소진했습니다. 다음에 다시 오겠습니다!
모든 스레드가 종료되었습니다.

CPU 오버헤드가 매우 작습니다.

[root@lincoding ~]# ps aux | grep -v grep | grep main
USER        PID %CPU %MEM    VSZ   RSS TTY      STAT START   TIME COMMAND
root      73838  0.0  0.1 141068  1256 pts/1    Sl+  19:54   0:00 ./main

결론

생산자의 생산 속도가 빠를지 느릴지 불확실한 상황에서는 공유 데이터를 보호하기 위해 순수 뮤텍스만 사용하는 것은 CPU 성능 오버헤드를 크게 발생시킬 수 있습니다. 이럴 때는 뮤텍스와 조건 변수를 함께 사용하는 방식이 좋습니다. 생산자 스레드가 데이터를 하나 생산할 때마다 소비자 스레드를 깨워서 데이터를 소비하게 하여 불필요한 작업으로 인한 성능 오버헤드를 방지할 수 있습니다.