[Concurrency Patterns] 11. 공유 회피

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class ImmutableString {
private:
    const std::string data;

public:
    ImmutableString(const std::string& s) : data(s) {}

    // 읽기만 가능
    std::string_view get() const { return data; }
    size_t length() const { return data.length(); }

    // "수정"은 새 객체 반환
    ImmutableString toUpper() const {
        std::string upper = data;
        std::transform(upper.begin(), upper.end(), upper.begin(),
                      [](unsigned char c) { return std::toupper(c); });
        return ImmutableString(upper);
    }
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

ImmutableString original("hello");

std::vector<std::thread> readers;
for (int i = 0; i < 10; ++i) {
    readers.emplace_back([original] {
        // 안전: 동시에 읽어도 문제 없음
        std::cout << original.get() << '\n';
    });
}

// "수정"은 새 객체 생성
ImmutableString modified = original.toUpper();  // "HELLO"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

// cow_config.cpp
// 빌드: g++ -std=c++20 -pthread -Wall -Wextra -O2 cow_config.cpp -o cow_config
#include <atomic>
#include <iostream>
#include <map>
#include <memory>
#include <string>
#include <thread>
#include <vector>

// 설정(Config) 객체: 한 번 만들어지면 절대 수정하지 않는다 (Immutable).
struct Config {
    std::map<std::string, std::string> values;
};

// CoW 컨테이너: "현재 설정을 가리키는 shared_ptr"을 atomic하게 교체한다.
class ConfigStore {
public:
    explicit ConfigStore(std::shared_ptr<const Config> initial)
        : current_(std::move(initial)) {}

    // 읽기: 락 없이 현재 버전의 shared_ptr을 얻는다.
    // 이후 reload()가 일어나도 이 스냅샷은 안전하게 유지된다.
    std::shared_ptr<const Config> get() const {
        return std::atomic_load_explicit(&current_, std::memory_order_acquire);
    }

    // 쓰기: 기존 설정을 복사 + 수정한 "새 객체"를 만들고, 포인터만 교체한다.
    void update(const std::string& key, const std::string& value) {
        std::shared_ptr<const Config> oldCfg, newCfg;
        do {
            oldCfg = get();
            auto copy = std::make_shared<Config>(*oldCfg);  // 복사 (Copy)
            copy->values[key] = value;                       // 수정은 복사본에
            newCfg = std::move(copy);
        } while (!std::atomic_compare_exchange_weak_explicit(
            &current_, &oldCfg, newCfg,
            std::memory_order_release, std::memory_order_relaxed));
    }

private:
    mutable std::shared_ptr<const Config> current_;
};

int main() {
    auto initial = std::make_shared<Config>();
    initial->values["timeout"] = "30s";
    ConfigStore store(initial);

    // 읽기 스레드: 락 없이 현재 설정의 스냅샷을 읽는다.
    std::vector<std::thread> readers;
    for (int i = 0; i < 4; ++i) {
        readers.emplace_back([&store, i] {
            for (int j = 0; j < 5; ++j) {
                auto cfg = store.get();  // 스냅샷 — 이후 update에 영향받지 않음
                std::cout << "reader " << i << ": timeout="
                          << cfg->values.at("timeout") << '\n';
            }
        });
    }

    // 쓰기 스레드: 새 버전을 만들어 포인터만 교체한다.
    std::thread writer([&store] {
        store.update("timeout", "60s");
    });

    for (auto& t : readers) t.join();
    writer.join();
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// thread_local_rng.cpp
// 빌드: g++ -std=c++20 -pthread -Wall -Wextra -O2 thread_local_rng.cpp -o tlrng
#include <iostream>
#include <random>
#include <thread>
#include <vector>

// 각 스레드가 자신만의 시드로 초기화된 엔진을 가진다.
// std::random_device로 시드를 섞어, 스레드마다 다른 시퀀스를 보장한다.
int threadLocalRandomInt(int low, int high) {
    static thread_local std::mt19937 engine{
        std::random_device{}() ^
        static_cast<unsigned>(
            std::hash<std::thread::id>{}(std::this_thread::get_id()))
    };
    std::uniform_int_distribution<int> dist(low, high);
    return dist(engine);
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back([i] {
            for (int j = 0; j < 3; ++j) {
                std::cout << "thread " << i << ": "
                          << threadLocalRandomInt(1, 100) << '\n';
            }
        });
    }
    for (auto& t : threads) t.join();
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class ThreadLocalCounter {
private:
    static thread_local int count;

public:
    static void increment() { ++count; }  // 각 스레드만의 count
    static int get() { return count; }
};

thread_local int ThreadLocalCounter::count = 0;

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back([] {
            for (int j = 0; j < 1000; ++j) ThreadLocalCounter::increment();
            std::cout << "Thread " << std::this_thread::get_id()
                      << " count: " << ThreadLocalCounter::get() << '\n';
        });
    }
    for (auto& t : threads) t.join();
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

// spsc_queue.cpp
// 빌드: g++ -std=c++20 -pthread -Wall -Wextra -O2 spsc_queue.cpp -o spsc
#include <array>
#include <atomic>
#include <iostream>
#include <optional>
#include <thread>

// 교육용 SPSC(Single-Producer Single-Consumer) Lock-Free 큐.
// 생산자 1개, 소비자 1개라는 전제 덕분에 head/tail을 각각
// 한 스레드만 "쓰고", 다른 스레드는 "읽기만" 한다 — 그래서
// compare_exchange 루프 없이 단순 load/store로 충분하다.
template <typename T, size_t Capacity>
class SpscQueue {
public:
    // 생산자 스레드에서만 호출
    bool push(const T& value) {
        size_t tail = tail_.load(std::memory_order_relaxed);
        size_t nextTail = (tail + 1) % Capacity;
        if (nextTail == head_.load(std::memory_order_acquire)) {
            return false;  // 큐 가득 참
        }
        buffer_[tail] = value;
        // release: 이 쓰기 이전의 모든 쓰기(buffer_[tail] 대입)가
        // consumer의 acquire 읽기에 보이도록 강제한다.
        tail_.store(nextTail, std::memory_order_release);
        return true;
    }

    // 소비자 스레드에서만 호출
    std::optional<T> pop() {
        size_t head = head_.load(std::memory_order_relaxed);
        if (head == tail_.load(std::memory_order_acquire)) {
            return std::nullopt;  // 큐 비어 있음
        }
        T value = buffer_[head];
        head_.store((head + 1) % Capacity, std::memory_order_release);
        return value;
    }

private:
    std::array<T, Capacity> buffer_{};
    std::atomic<size_t> head_{0};  // 소비자가 쓰고, 생산자가 읽음
    std::atomic<size_t> tail_{0};  // 생산자가 쓰고, 소비자가 읽음
};

int main() {
    SpscQueue<int, 1024> queue;
    constexpr int kCount = 100000;

    std::thread producer([&queue] {
        for (int i = 0; i < kCount; ++i) {
            while (!queue.push(i)) {
                std::this_thread::yield();  // 큐가 가득 차면 잠시 양보
            }
        }
    });

    std::thread consumer([&queue] {
        for (int i = 0; i < kCount; ++i) {
            std::optional<int> value;
            while (!(value = queue.pop())) {
                std::this_thread::yield();  // 큐가 비어 있으면 잠시 양보
            }
            if (*value != i) {
                std::cerr << "순서 오류: expected " << i << ", got " << *value << '\n';
                std::abort();
            }
        }
    });

    producer.join();
    consumer.join();
    std::cout << "OK: " << kCount << "개 모두 순서대로 전달됨\n";
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

1. 공유 상태가 필요한가?
   → 아니오: thread_local 사용
   
2. 쓰기가 드문가?
   → 예: Copy-on-Write
   
3. 쓰기가 많은가?
   → 예: mutex (이전 장들)
   
4. 극도의 성능이 필요한가?
   → 예: lock-free (라이브러리 사용)