서로 다른 Thread에서 같은 메모리를 공유할 때 같은 자원을 사용하면서 값이 이상하게 되는 것을 race condition이라고 한다.
시작하기에 앞서 모든 STL들은 기본적으로 멀티스레드 환경을 제공하지 않는다고 알아야 한다.
void worker(int& counter) {
for (int i = 0; i < 10000; i++) {
counter += 1;
}
}
int main() {
int counter = 0;
vector<thread> workers;
for (int i = 0; i < 4; i++) {
// 레퍼런스로 전달하려면 ref 함수로 감싸야 한다.
workers.push_back(thread(worker, ref(counter)));
}
for (int i = 0; i < 4; i++) {
workers[i].join();
}
cout << "Counter 최종 값 : " << counter << endl;
}아니 스레드1에서 3을 대입하려고 하고 스레드2에서 5를 대입하려고 하는 상황이 아니고 그냥 서로 더하기만 하는데도 오류가 생긴다고? 이건 뭔가 잘못됐다.
이거를 이해하기 위해서는 counter+=1이 실제로 어떻게 수행되는지를 알아야 한다.
mov rax, qword ptr [rbp - 8]
mov ecx, dword ptr [rax]
add ecx, 1
mov dword ptr [rax], ecxrax, rbp는 둘다 CPU 레지스터를 의미한다.qword는 8바이트를 의미한다. -> 주소값의 크기를 의미.dword는 4바이트를 의미한다. -> int값의 크기 등등[]의 의미는 역참조다. 즉, rbp - 8이라는 주소에 있는 값을 읽어라 라는 뜻이고 c++의 *와 같은 것이다.mov는 이 문장이 어떤 명령을 하는지를 의미하는 것이다.rax = *(int**)(rbp - 8)근데 사실 이런 과정은 한번 다루고 싶어서 한거고 어셈블리를 몰라도 괜찮다.
⚠️ 참고로 push_back()을 하는 과정도 이와 비슷하게 뒤에 추가하는 거지만 멀티 스레드에서는 충돌이 일어나게 된다. 이 이유에 대해서 생각을 해보자면 동적 배열같은 경우는 범위를 넘어서게 되면 그 공간을 복사하고 해제시켜 다른 여유 공간으로 이동하게 될텐데, 다른 스레드도 마찬가지로 영역을 해제하려고 시도하기 때문에 double free문제가 생기게 될것이다.
이러한 race condition을 관리하기 위해서는 특정 데이터에 접근을 할때 한번에 하나의 스레드만 접근을 허락을 해줘야 하는데 이러한 방법으로는 Atomic을 이용한 관리가 있다. 어떻게 보면 가장 순수한 접근 제약이라고 보면 된다.
int SUM = 0;
void Add() {
for (int i = 0; i < 10000; i++) {
SUM++;
}
}
void Sub() {
for (int i = 0; i < 10000; i++) {
SUM--;
}
}
int main() {
Add();
Sub();
cout << "result:" << SUM << endl;
thread t1(Add);
thread t2(Sub);
t1.join();
t2.join();
cout << "thread result:" << SUM << endl;
}#include<atomic>
atomic<int> SUM {0};
void Add() {
for (int i = 0; i < 10000; i++) {
//SUM++;
SUM.fetch_add(1);
}
}
void Sub() {
for (int i = 0; i < 10000; i++) {
//SUM--;
SUM.fetch_sub(1);
}
}
int main() {
Add();
Sub();
cout << "result:" << SUM << endl;
thread t1(Add);
thread t2(Sub);
t1.join();
t2.join();
cout << "thread result:" << SUM << endl;
}result:0
thread result:0⚠️ 참고로 STL같은거는 atomic으로 감쌀 수 없다. 그 이유는 STL에는 여러 기능이 있기 때문에 이 전체를 atomic으로 감싸는건 비효율적이고 STL들은 내부적으로 allocator가 있고 생성자 소멸자 등 원하는 스타일이 다르기 때문이라고 생각하면 된다.
이런 race condition을 제어하는 것 중 가장 대표적인것이 바로 Lock이다.
vector<int> v;
void Push() {
for (int i = 0; i < 100000; ++i) {
v.push_back(i);
}
}
int main() {
thread t1(Push);
thread t2(Push);
t1.join();
t2.join();
cout << "Vector size: " << v.size() << endl;
return 0;
}v.reserve(200000)mutex를 사용하면 된다.한번에 하나의 스레드만 자원에 접근할 수 있게 해주는 잠금 장치 객체.
#include <mutex> // mutex 를 사용하기 위해 필요
void worker(int& result, mutex& m) {
for (int i = 0; i < 10000; i++) {
m.lock();
result += 1;
m.unlock();
}
}
int main() {
int counter = 0;
mutex m; // 우리의 mutex 객체
vector<thread> workers;
for (int i = 0; i < 4; i++) {
workers.push_back(thread(worker, ref(counter), ref(m)));
}
for (int i = 0; i < 4; i++) {
workers[i].join();
}
cout << "Counter 최종 값 : " << counter << endl;
}m.lock();
m.lock();
// TODO 뭔가 처리
function();
m.unlock();
m.unlock();lock_guard이다.void worker(int& result, mutex& m) {
for (int i = 0; i < 10000; i++) {
// lock 생성 시에 m.lock() 을 실행한다고 보면 된다.
lock_guard<std::mutex> lock(m);
result += 1;
// scope 를 빠져 나가면 lock 이 소멸되면서
// m 을 알아서 unlock 한다.
}
}RAII라는 원칙에 따라 동작하는건데 쉽게 말하자면 생성자에서 락을 걸고 소멸자에서 락을 푸는 패턴으로 만들어 졌다고 보면 된다.template<typename T>
class LockGuard {
public:
LockGuard(T& m) {
_mtx = &m;
_mtx->lock();
}
~LockGuard() {
_mtx->unlock();
}
private:
T* _mtx;
};
vector<int> v;
mutex m;
void Push() {
for (int i = 0; i < 100000; ++i) {
LockGuard<mutex> lg(m);
v.push_back(i);
}
}
int main() {
thread t1(Push);
thread t2(Push);
t1.join();
t2.join();
cout << "Vector size: " << v.size() << endl;
return 0;
}void worker1(std::mutex& m1, std::mutex& m2) {
for (int i = 0; i < 10000; i++) {
std::lock_guard<std::mutex> lock1(m1);
std::lock_guard<std::mutex> lock2(m2);
// Do something
}
}
void worker2(std::mutex& m1, std::mutex& m2) {
for (int i = 0; i < 10000; i++) {
std::lock_guard<std::mutex> lock2(m2);
std::lock_guard<std::mutex> lock1(m1);
// Do something
}
}
int main() {
int counter = 0;
std::mutex m1, m2; // 우리의 mutex 객체
std::thread t1(worker1, std::ref(m1), std::ref(m2));
std::thread t2(worker2, std::ref(m1), std::ref(m2));
t1.join();
t2.join();
std::cout << "끝!" << std::endl;
}lock_guard를 사용한다고 해서 데드락이 회피되는 것은 아니다.void worker2(std::mutex& m1, std::mutex& m2) {
for (int i = 0; i < 10000; i++) {
while (true) {
m2.lock();
// m1 이 이미 lock 되어 있다면 m2에 관한 락을 푼다.
if (!m1.try_lock()) {
m2.unlock();
continue;
}
std::cout << "Worker2 Hi! " << i << std::endl;
m1.unlock();
m2.unlock();
break;
}
}
}lock_guard 말고도 unique_lock이라는 wrapper도 존재한다. 큰 차이점이라고 한다면 유연성에 있다.
mutex m;
void Flexible() {
unique_lock<mutex> lock(m, defer_lock); // lock 안 함
if (need_lock) lock.lock(); // 조건 lock
// 작업1
lock.unlock(); // 임시 unlock
DoIO(); // IO 중 mutex 풀기
lock.lock(); // 재lock
// 작업2
} // 자동 unlock이런 락을 구현하는 방법에는 크게 3가지가 존재하게 된다.
class SpinLock {
public:
void lock() {
while (_lock) {
}
_lock = true;
}
void unlock() {
_lock = false;
}
private:
bool _lock = false;
};
int SUM = 0;
SpinLock spinlock;
void Add() {
for (int i = 0; i < 10000; i++) {
spinlock.lock();
SUM++;
spinlock.unlock();
}
}
void Sub() {
for (int i = 0; i < 10000; i++) {
spinlock.lock();
SUM--;
spinlock.unlock();
}
}
int main() {
thread t1(Add);
thread t2(Sub);
t1.join();
t2.join();
cout << "result:" << SUM << endl;
return 0;
}class SpinLock {
public:
void lock() {
bool expected = false;
bool desired = true;
while (_lock.compare_exchange_strong(expected, desired) == false) {
expected = false;
}
}
void unlock() {
_lock.store(false);
}
private:
atomic<bool> _lock{ false };
};compare_exchange_strong이라는 것이 대표적인 CAS 함수이다.if (_lock == expected)
{
expected = _lock;
_lock = desired;
return true;
}
else
{
expected = _lock;
}_lock이 우리가 기대하는 expacted값과 같다면 desired값으로 바꾸는 것이다.expacted가 _lock의 값으로 변환이 된다는 것이다.expacted값을 false로 초기화 해주고 있는 것이다._lock이 false가 되면 이걸 true(desired)로 바꿔주고 반복문을 빠져나가게 되는 것이다.class SpinLock {
public:
void lock() {
bool expected = false;
bool desired = true;
while (_lock.compare_exchange_strong(expected, desired) == false) {
expected = false;
this_thread::sleep_for(100ms);
}
}
void unlock() {
_lock.store(false);
}
private:
atomic<bool> _lock{ false };
};sleep_for()를 이용하면 알아서 원하는 시간만큼 대기하다가 작업을 이어서 하게 된다.this_thread::yield()도 있는데 이거는 그냥 sleep_for(0ms)과 같다고 생각하면 된다.이벤트를 이용한 동기화 방법이 사실 가장 좋다. 그 이유는 웬만한 상황에서 높은 효율을 보여주기 때문이다. 물론 락을 오래 점유하지 않는 상황에서는 context switch비용이 생각보다 크기 때문에 스핀락이 유리하겠지만 그런 경우는 많이 없다.
mutex m;
queue<int32> q;
void Producer()
{
while (true)
{
{
unique_lock<mutex> lock(m);
q.push(100);
}
this_thread::sleep_for(chrono::milliseconds(1));
}
}
void Consumer()
{
while (true)
{
unique_lock<mutex> lock(m);
if (!q.empty())
{
int32 value = q.front();
q.pop();
cout << value << endl;
}
}
}
int main()
{
thread t1(Producer);
thread t2(Consumer);
t1.join();
t2.join();
}mutex m;
queue<int32> q;
HANDLE handle;
void Producer()
{
while (true)
{
{
unique_lock<mutex> lock(m);
q.push(100);
}
::SetEvent(handle); // 신호를 signal 상태로 변경
this_thread::sleep_for(chrono::milliseconds(1));
}
}
void Consumer()
{
while (true)
{
::WaitForSingleObject(handle, INFINITE); // signal 상태가 될 때까지 대기
unique_lock<mutex> lock(m);
if (!q.empty())
{
int32 value = q.front();
q.pop();
cout << value << endl;
}
}
}
int main()
{
// 커널 오브젝트
// Useage Count (이 오브젝트를 몇명이 사용하고 있는지)
// Signal / Non-Signal (신호 / 비신호)
// Auto Reset / Manual Reset (자동 리셋 / 수동 리셋)
handle = ::CreateEvent(NULL/*보안 속성*/, FALSE/*manual Reset*/, FALSE, NULL);
thread t1(Producer);
thread t2(Consumer);
t1.join();
t2.join();
}CreateEvent함수를 이용해서 커널 오브젝트를 만든다.SetEvent를 통해서 신호를 주게 되면, WaitForSigleObjece와 같이 신호를 기다리고 있던 부분에서 통과가 되면서 작업을 수행할 수 있는 것이다.condition variable은 특정 조건이 될 때 까지 스레드를 효율적으로 대기시키는 동기화 도구이다. 참고로 이벤트와 비슷하게 동작을 하는 개념인데, 커널 오브젝트는 아니고 user-level 오브젝트이다. 따라서 뭔가 자주 일어나는 그런 작업이라면 유용하게 사용할 수 있다.
그리고 그냥 커널오브젝트와의 차이점이라고 한다면 보통 락이랑 짝지어서 사용한다. 아래 예시를 살펴보자.
condition_variable cv;
void Producer()
{
while (true)
{
{
unique_lock<mutex> lock(m);
q.push(100);
}
cv.notify_one(); // wait중인 스레드 1개를 깨운다.
}
}
void Consumer()
{
while (true)
{
unique_lock<mutex> lock(m);
cv.wait(lock, []() {return q.empty() == false;});
// 1.락을 건다.
// 2. 조건 확인
// IF 조건이 True -> 빠져나와서 이어서 코드를 진행
// IF 조건이 False -> 락을 풀고 대기 상태로 들어감.
{
int32 value = q.front();
q.pop();
cout << value << endl;
}
}
}대학교 OS 과목에서 배웠던 걸 그대로 다시 공부하니까 참 신기하면서도 진작 할껄 그랬다는 생각이든다.
queue를 이용한다.#include <chrono>
#include <iostream>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <vector>
void producer(std::queue<std::string>* downloaded_pages, std::mutex* m,
int index) {
for (int i = 0; i < 5; i++) {
// 웹사이트를 다운로드 하는데 걸리는 시간이라 생각하면 된다.
// 각 쓰레드 별로 다운로드 하는데 걸리는 시간이 다르다.
std::this_thread::sleep_for(std::chrono::milliseconds(100 * index));
std::string content = "웹사이트 : " + std::to_string(i) + " from thread(" +
std::to_string(index) + ")\n";
// data 는 쓰레드 사이에서 공유되므로 critical section 에 넣어야 한다.
m->lock();
downloaded_pages->push(content);
m->unlock();
}
}
void consumer(std::queue<std::string>* downloaded_pages, std::mutex* m,
int* num_processed) {
// 전체 처리하는 페이지 개수가 5 * 5 = 25 개.
while (*num_processed < 25) {
m->lock();
// 만일 현재 다운로드한 페이지가 없다면 다시 대기.
if (downloaded_pages->empty()) {
m->unlock();
// 10 밀리초 뒤에 다시 확인한다.
std::this_thread::sleep_for(std::chrono::milliseconds(10));
continue;
}
// 맨 앞의 페이지를 읽고 대기 목록에서 제거한다.
std::string content = downloaded_pages->front();
downloaded_pages->pop();
(*num_processed)++;
m->unlock();
// content 를 처리한다.
std::cout << content;
std::this_thread::sleep_for(std::chrono::milliseconds(80));
}
}
int main() {
// 현재 다운로드한 페이지들 리스트로, 아직 처리되지 않은 것들이다.
std::queue<std::string> downloaded_pages;
std::mutex m;
std::vector<std::thread> producers;
for (int i = 0; i < 5; i++) {
producers.push_back(std::thread(producer, &downloaded_pages, &m, i + 1));
}
int num_processed = 0;
std::vector<std::thread> consumers;
for (int i = 0; i < 3; i++) {
consumers.push_back(
std::thread(consumer, &downloaded_pages, &m, &num_processed));
}
for (int i = 0; i < 5; i++) {
producers[i].join();
}
for (int i = 0; i < 3; i++) {
consumers[i].join();
}
}서론 c++을 이용해서 개발을 하다보면 복사 대신에 이동연산을 통해서 여러 이점을 얻는다는 정보를 확인할 수…
서론 C++에서 진짜 너무 많이 등장하는 개념이 바로 Pointer이다. 메모리의 주소를 관리하고~ 뭐시기~ 항상 등장하는데,…
서론 이번시간에는 Casting이라는 개념에 대해서 학습해 보도록 하겠다. 하지만 그 전에 Composition과 Aggregation의 차이에 대해서…
서론 c++ basic을 공부하기 위해서는 다른 언어들과의 기본적인 차이를 이해해야 한다. 그 차이는 바로 메모리…