C++ thread, mutex, condition_variable

대기시간이 길고 의존관계가 없는 연산들을 병렬적으로 처리하기 위해 쓰레드를 사용한다.

 

C++11 이전에는 표준화된 쓰레드가 없어서 각 플랫폼마다 다른 구현을 사용해야 했지만, C++11부터 표준에 쓰레드가 추가되었다.

 

Thread 생성, join, detach

#include <thread>
#include <iostream>

using namespace std;
using std::thread;

void func1() {
    for (int i = 0; i < 10; i++) {
        cout << "스레드 1 작동중" << endl;
    }
}

void func3() {
 
    // 현재 코드를 실행중인 스레드의 id
    cout << "쓰레드 3 id : " << std::this_thread::get_id() << endl;

    for (int i = 0; i < 10; i++) {
        cout << "스레드 3 작동중" << endl;
    }
}


int main()
{
    // 내 컴퓨터의 논리 프로세서가 몇개인지 출력
    cout << "논리 프로세서의 개수 : " << std::thread::hardware_concurrency() << endl;

    // 현재 코드를 실행중인 스레드의 id
    cout <<  "메인 쓰레드 id : " << std::this_thread::get_id() << endl;

    thread t1(func1);
    thread t2([] {
        for (int i = 0; i < 10; i++) {
            cout << "스레드 2 작동중" << endl;
        }
    });

    thread t3(func3);


    t1.join();
    t2.detach();
    t3.detach();
    cout << "메인 함수 종료" << endl;
}

메인 쓰레드에서 join은 해당 쓰레드가 종료될 때까지 기다리지만, detach는 기다리지 않는다.

 

논리 프로세서의 개수는 작업 관리자에서도 확인할 수 있다.

출력

순간의 OS 환경마다 다른 결과를 출력한다.

 

쓰레드에 인자 전달하기

쓰레드가 실행하는 함수는 리턴 값이 없기 때문에 인자의 포인터를 활용하여 전달해야 한다.

 

void worker(vector<int>::iterator start, vector<int>::iterator end, int* result){
            // 반복자의 시작부터 끝까지의 합을 계산하여 result에 저장하는 코드
}

int main() {
	vector<int> data(10000);

	vector<int> partial_sums(4);
    
    // 각 스레드에서 2500개의 요소를 계산하도록 함
    // 합을 받기 위한 int* 를 인자로 보냄
    for(int i=0; i<4; i++){
    	worker.push_back(thread(worker, data.begin() + i * 2500, data.begin() + (i + 1) * 2500, &partial_sums[i]));
    }
	
}

 

여러 쓰레드에서 같은 메모리 공간(변수)에 접근하는 경우 (mutex lock, unlock, lock_guard)

경쟁 상태(race condition)가 발생한다.

뮤텍스를 사용하여 스레드에서 변수를 사용하기 위해서는 한 번에 한 쓰레드에서만 뮤텍스를 가지고 있도록(lock) 하고, 접근이 끝나면 뮤텍스를 풀어(unlock), 다른 쓰레드가 뮤텍스를 가질 수 있도록 한다.

 

뮤텍스의 lock()과 unlock() 사이에 한 쓰레드만이 유일하게 실행할 수 있는 코드 부분을 임계 영역(critical section)이라고 한다.

 

lock_guard 객체는 뮤텍스를 인자로 받아 생성되는데 생성될 때 뮤텍스의 lock을 실행하고, 범위를 벗어나 소멸되면 뮤텍스의 unlock을 자동으로 호출한다.

#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

void worker(int & result, std::mutex& m){
    // case 1 : lock, unlock을 사용한 경우
    for(int i = 0; i< 10000; i++){
        m.lock();
       	/* 임계 구역 시작 */
        result += 1;
        /* 임계 구역 끝 */
        m.unlock();
    }
    
    /* case 2 : lock_guard를 사용한 경우
    for(int i = 0; i< 10000; i++){
        std::lock_guard<std::mutex> lock(m);
        result += 1;
        
        // scope를 빠져 나가면 lock이 소멸되면서 m을 알아서 unlock 한다.
        // 예외가 발생하는 상황에서도 try-catch 구문으로 잘 감싸주면 자동으로 lock을 해제한다.
    }
    */
}

int main() {
    int counter = 0;
    std::mutex m;

    std::vector<std::thread> workers;
    for (int i = 0; i < 4; i++) {
    	// 쓰레드에 레퍼런스를 넘길 때는 래퍼로 감싸야 한다.
        workers.push_back(std::thread(worker, std::ref(counter), std::ref(m)));
    }

    for (int i = 0; i < 4; i++) {
        workers[i].join();
    }
	
    // 40000을 출력한다.
    std::cout << "Counter 최종 값 : " << counter << std::endl;
}

 

생산자-소비자 패턴을 구현할 때 condition_variable을 사용하기 (+ unique_lock)

 

생산자 쓰레드는 공유하는 큐에 작업을 추가한다. (큐는 공유 자원이므로 임계 영역에서 수정)

소비자 쓰레드는 생산자 쓰레드가 생산한 작업을 처리한다. (큐에 작업이 있는 경우에 pop하여 처리한다.)

 

이때 condition_variable을 사용하지 않고 구현한다고 하면 소비자 쓰레드가 큐가 비었는지를 지속적으로 확인 해야 한다. (while문, std::this_thread::sleep_for(), 시간 객체 chrono, 큐의 empty() 사용) 이는 CPU 낭비다(스핀락). 지속적으로 mutex를 잠그고 큐를 확인해야 하기 때문이다.

 

소비자 쓰레드들을 특정 조건(큐가 empty가 아닌) 이 일어나기 전까지 재우도록(block) 하는 것이 condition_variable의 역할이다. 

 

생산자-소비자 패턴에서의 작업 진행 순서

1. 생산자 쓰레드에서 큐에 작업을 넣고, cv->notify_one()을 호출하여 cv->wait으로 인해 조건을 만족하지 못해 잠자고 있던 소비자 쓰레드중 하나를 깨운다.

 

1-1 cv->wait 함수는 unique_lock과 (조건자 혹은 람다 함수) (인자를 받지 않고 bool값을 리턴하는)을 인자로 받는다.

 

2. 깨어난 소비자 쓰레드에서는 cv->wait의 조건을 통과한다.(lock을 얻는다) 소비자 쓰레드는 큐에서 작업을 pop하고, unique_lock을 unlock() 한다.

 

3. 소비자 쓰레드는 작업을 처리한다.

 

lock_guard 대신 unique_lock을 사용하는 이유

ㄴ lock_guard는 생성자, 소멸자를 통해서만 mutex를 lock, unlock 할 수 있지만 unique_lock은 lock을 얻는 시점을 정할 수 있고 unlock을 직접 호출할 수도 있다.

 

wait 말고 wait_for나 wait_until도 있다.

 

참조:

https://modoocode.com/269