C++ / std::atomic

std::atomic은 해당 객체가 원자적으로 처리될 수 있도록 도와주는 객체다.

#include <atomic>
#include <iostream>
#include <thread>
#include <vector>

void worker(std::atomic<int>& counter) {
    for (int i = 0; i < 10000; i++) {
      counter++;
    }
}

int main() {
    std::atomic<int> counter(0);

    std::vector<std::thread> workers;
    for (int i = 0; i < 4; i++) {
      workers.push_back(std::thread(worker, ref(counter)));
    }

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

    std::cout << "Counter : " << counter << std::endl;
}

일반적으로 counter은 critical section에 해당해서 mutex, semaphore, critical_section 등으로 lock을 걸어야 하지만, 해당 변수가 원자적(atomic)으로 처리되었기 때문에 정상적으로 4만이라는 결과값을 얻을 수 있다.

 

is_lock_free 메서드

is_lock_free 메서드는 해당 객체가 원자적으로 처리될 수 있는지 확인해 볼 수 있다. 만약 해당 객체가 원자적 처리를 할 수 없는 객체라면 false를, 아니면 true를 반환한다.

std::atomic<int> num;
std::cout << "Is lock free?" << boolalpha << num.is_lock_free() << std::endl;

memory_order

atomic 객체들은 memory_order을 활용해서 원자적 연산 수행 시에 어떤 방식으로 접근하는지 지정해 줄 수 있다.

memory_order_relaxed

#include <atomic>
#include <cstdio>
#include <thread>
#include <vector>
using std::memory_order_relaxed;

void t1(std::atomic<int>* a, std::atomic<int>* b) {
    b->store(1, memory_order_relaxed);      // b = 1 (쓰기)
    int x = a->load(memory_order_relaxed);  // x = a (읽기)

    printf("x : %d \n", x);
}

void t2(std::atomic<int>* a, std::atomic<int>* b) {
    a->store(1, memory_order_relaxed);      // a = 1 (쓰기)
    int y = b->load(memory_order_relaxed);  // y = b (읽기)

    printf("y : %d \n", y);
}

int main() {
    std::vector<std::thread> threads;

    std::atomic<int> a(0);
    std::atomic<int> b(0);

    threads.push_back(std::thread(t1, &a, &b));
    threads.push_back(std::thread(t2, &a, &b));

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

memory_order_relaxed는 CPU에게 연산 순서에 대해서 어떠한 제약도 주지 않는 방식이다. 따라서 t1의 a와 t2의 b에 값을 load해서 담는 시점이 store 이후에 이루어진다고 확정지을 수 없게 된다. 즉, load하기 전에 반드시 store된다고 보장할 수 없다. 

memory_order_acquire 과 memory_order_release

memory_order_release는 해당 명령 이전에 오는 해당 변수에 접근하는 모든 명령이 그 이후로 재배치 되는 것을 막아준다.  다른 쓰레드에서 해당 변수를 읽을 때 memory_order_acquire 를 사용한다면 memory_order_acquire에서 접근하는 변수가 memory_order_release 전에 전달된 모든 명령의 결과를 알아낼 수 있어야 한다(명령이 모두 처리되어야 한다는 뜻임).

 

memory_order_acquire은 release와는 반대로 해당 명령 이후에 오는 명령어들이 해당 명령이 수행되기 이전으로 재배치되는 것을 방지한다. 

 

두 키워드를 활용하면 이런 식으로 활용 가능하다.

#include <vector>
#include <iostream>
#include <thread>
#include <atomic>

void producer(std::atomic<bool>* is_ready, std::atomic<int> data[]) {
	data[0].store(1, std::memory_order_relaxed);
	data[1].store(2, std::memory_order_relaxed);
	data[2].store(3, std::memory_order_relaxed);

	is_ready->store(true, std::memory_order_release);
}

void consumer(std::atomic<bool>* is_ready, std::atomic<int> data[]) {
	while (!is_ready->load(std::memory_order_acquire)) {}
	std::cout << "data[0]: " << data[0].load(std::memory_order_relaxed) << std::endl;
	std::cout << "data[1]: " << data[1].load(std::memory_order_relaxed) << std::endl;
	std::cout << "data[2]: " << data[2].load(std::memory_order_relaxed) << std::endl;
}

int main() {
	std::vector<std::thread> threads;

	std::atomic<bool> is_ready(false);
	std::atomic<int> data[3];

	threads.push_back(std::thread(producer, &is_ready, data));
	threads.push_back(std::thread(consumer, &is_ready, data));

	for (int i = 0; i < 2; ++i) {
		threads[i].join();
	}
	return 0;
}

std::memory_order_relaxed는 메모리 접근 순서를 CPU 마음대로 하게 할 수 있지만, memory_order_release는 해당 명령어들이 반드시 해당 명령어 이전에 오도록 강제할 수 있다. 또한, memory_order_acquire로 접근하는 변수가 존재할 경우 memory_order_release가 있는 쓰레드에서 release 이전의 모든 작업을 완료한 다음 해당 변수에 접근한다는 것을 보장받을 수 있고, 그 이후 해당 변수에 접근해서 값을 가져오므로 반드시 1, 2, 3이라는 값을 가져온다고 보장할 수 있다. 

 

memory_order_acq_rel

memory_order_acq_rel 은 이름에서도 알 수 있듯이, acquire 와 release 를 모두 수행하는 것이다. 이는 읽기와 쓰기를 모두 수행하는 명령 fetch_add와 같은 함수에서 사용될 수 있다. 

memory_order_seq_cst

memory_order_seq_cst 는 메모리 명령의 순차적 일관성(sequential consistency) 을 보장해준다. 즉, 모든 쓰레드에서 모든 시점에 동일한 값을 관측할 수 있게 해준다고 보면 된다. 

 

atomic에서 수행하는 메모리 연산의 default가 해당 memory_order이다.