手写线程安全智能指针

Posted on | In

现有 SimpleSharedPtr 的线程安全性分析

在多线程环境下，确保智能指针的线程安全性主要涉及以下几个方面：

  1. 引用计数管理：多个线程可能会同时拷贝、移动或销毁智能指针实例，导致引用计数的修改。若引用计数不是原子操作，则会引发数据竞争和未定义行为。
  2. 指针和控制块的访问：多个线程可能会同时访问或修改同一个智能指针实例的 ptr 和 control 成员，这需要同步机制来保护。

当前 SimpleSharedPtr 的问题：

  • 引用计数非原子：ControlBlock::ref_count 是普通的 int 类型，当多个线程同时修改 ref_count 时，会引发竞态条件。
  • 缺乏同步机制：SimpleSharedPtr 的成员函数（如拷贝构造、赋值操作符等）在修改 ptr 和 control 时没有任何同步机制，导致多个线程同时操作同一个 SimpleSharedPtr 实例时不安全。

实现线程安全的 SimpleSharedPtr

为了解决上述问题，可以从以下几个方面入手：

方法一：使用 std::atomic 管理引用计数

将 ControlBlock::ref_count 从普通的 int 替换为 std::atomic<int>，以确保引用计数的线程安全递增和递减。

优点：

  • 简单高效，避免使用互斥锁带来的性能开销。
  • 类似于标准库中 std::shared_ptr 实现的引用计数管理。

缺点：

  • 只能保证引用计数本身的线程安全，无法保护 ptr 和 control 的同步访问。

方法二：引入互斥锁保护指针操作

在 SimpleSharedPtr 中引入 std::mutex，在所有可能修改 ptr 和 control 的操作中加锁。

优点：

  • 确保 ptr 和 control 在多线程访问时的一致性。
  • 提供更全面的线程安全保障。

缺点：

  • 引入锁机制，可能带来性能开销，特别是在高并发场景下。

方法三：组合使用 std::atomic 和互斥锁

结合使用 std::atomic<int> 进行引用计数的管理，并使用 std::mutex 保护指针和控制块的访问。

优点：

  • 引用计数管理高效且线程安全。
  • 指针和控制块的访问得到完全的同步保护。

缺点：

  • 复杂性较高，需要同时管理原子操作和互斥锁。

完整线程安全的 ThreadSafeSharedPtr 实现

结合上述方法二和方法一，我们可以实现一个名为 ThreadSafeSharedPtr 的类模板，确保在多线程环境下的安全性。以下是具体实现：

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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#include <iostream>
#include <atomic>
#include <mutex>
#include <thread>

// 控制块结构
struct ControlBlock {
    std::atomic<int> ref_count;

    ControlBlock() : ref_count(1) {}
};

// 线程安全的 shared_ptr 实现
template <typename T>
class ThreadSafeSharedPtr {
private:
    T* ptr;                 // 指向管理的对象
    ControlBlock* control;  // 指向控制块

    // 互斥锁，用于保护 ptr 和 control
    mutable std::mutex mtx;

    // 释放当前资源
    void release() {
        if (control) {
            // 原子递减引用计数
            if (--(control->ref_count) == 0) {
                delete ptr;
                delete control;
                std::cout << "Resource and ControlBlock destroyed." << std::endl;
            } else {
                std::cout << "Decremented ref_count to " << control->ref_count.load() << std::endl;
            }
        }
        ptr = nullptr;
        control = nullptr;
    }

public:
    // 默认构造函数
    ThreadSafeSharedPtr() : ptr(nullptr), control(nullptr) {
        std::cout << "Default constructed ThreadSafeSharedPtr (nullptr)." << std::endl;
    }

    // 参数化构造函数
    explicit ThreadSafeSharedPtr(T* p) : ptr(p) {
        if (p) {
            control = new ControlBlock();
            std::cout << "Constructed ThreadSafeSharedPtr, ref_count = " << control->ref_count.load() << std::endl;
        } else {
            control = nullptr;
        }
    }

    // 拷贝构造函数
    ThreadSafeSharedPtr(const ThreadSafeSharedPtr& other) {
        std::lock_guard<std::mutex> lock(other.mtx);
        ptr = other.ptr;
        control = other.control;
        if (control) {
            control->ref_count++;
            std::cout << "Copied ThreadSafeSharedPtr, ref_count = " << control->ref_count.load() << std::endl;
        }
    }

    // 拷贝赋值操作符
    ThreadSafeSharedPtr& operator=(const ThreadSafeSharedPtr& other) {
        if (this != &other) {
            // 为避免死锁，使用 std::scoped_lock 同时锁定两个互斥锁
            std::scoped_lock lock(mtx, other.mtx);
            release();
            ptr = other.ptr;
            control = other.control;
            if (control) {
                control->ref_count++;
                std::cout << "Assigned ThreadSafeSharedPtr, ref_count = " << control->ref_count.load() << std::endl;
            }
        }
        return *this;
    }

    // 移动构造函数
    ThreadSafeSharedPtr(ThreadSafeSharedPtr&& other) noexcept {
        std::lock_guard<std::mutex> lock(other.mtx);
        ptr = other.ptr;
        control = other.control;
        other.ptr = nullptr;
        other.control = nullptr;
        std::cout << "Moved ThreadSafeSharedPtr." << std::endl;
    }

    // 移动赋值操作符
    ThreadSafeSharedPtr& operator=(ThreadSafeSharedPtr&& other) noexcept {
        if (this != &other) {
            // 为避免死锁，使用 std::scoped_lock 同时锁定两个互斥锁
            std::scoped_lock lock(mtx, other.mtx);
            release();
            ptr = other.ptr;
            control = other.control;
            other.ptr = nullptr;
            other.control = nullptr;
            std::cout << "Move-assigned ThreadSafeSharedPtr." << std::endl;
        }
        return *this;
    }

    // 析构函数
    ~ThreadSafeSharedPtr() {
        release();
    }

    // 解引用操作符
    T& operator*() const {
        std::lock_guard<std::mutex> lock(mtx);
        return *ptr;
    }

    // 箭头操作符
    T* operator->() const {
        std::lock_guard<std::mutex> lock(mtx);
        return ptr;
    }

    // 获取引用计数
    int use_count() const {
        std::lock_guard<std::mutex> lock(mtx);
        return control ? control->ref_count.load() : 0;
    }

    // 获取裸指针
    T* get() const {
        std::lock_guard<std::mutex> lock(mtx);
        return ptr;
    }

    // 重置指针
    void reset(T* p = nullptr) {
        std::lock_guard<std::mutex> lock(mtx);
        release();
        ptr = p;
        if (p) {
            control = new ControlBlock();
            std::cout << "Reset ThreadSafeSharedPtr, ref_count = " << control->ref_count.load() << std::endl;
        } else {
            control = nullptr;
        }
    }
};

// 测试类
class Test {
public:
    Test(int val) : value(val) {
        std::cout << "Test Constructor: " << value << std::endl;
    }
    ~Test() {
        std::cout << "Test Destructor: " << value << std::endl;
    }
    void show() const {
        std::cout << "Value: " << value << std::endl;
    }

private:
    int value;
};

关键改动说明

  1. 引用计数原子化：

    • 将

      1
      ControlBlock::ref_count

      从普通的

      1
      int

      改为

      1
      std::atomic<int>

      ：

      1
      std::atomic<int> ref_count;

    • 使用原子操作管理引用计数，确保多线程下的安全递增和递减：

      1
      2
      control->ref_count++;
      if (--(control->ref_count) == 0) { ... }

    • 使用 ref_count.load() 获取当前引用计数的值。

  2. 引入互斥锁：

    • 在

      1
      ThreadSafeSharedPtr

      中引入

      1
      std::mutex mtx

      ，用于保护

      1
      ptr

      和

      1
      control

      的访问：

      1
      mutable std::mutex mtx;

    • 在所有可能修改或访问

      1
      ptr

      和

      1
      control

      的成员函数中加锁，确保同步：

      1
      std::lock_guard<std::mutex> lock(mtx);

    • 在拷贝构造函数和拷贝赋值操作符中，为避免死锁，使用

      1
      std::scoped_lock

      同时锁定两个互斥锁：

      1
      std::scoped_lock lock(mtx, other.mtx);

  3. 线程安全的成员函数：

    • 对于 operator* 和 operator->，在返回前锁定互斥锁，确保在多线程环境中的安全访问。
    • 其他成员函数如 use_count、get 和 reset 同样在访问共享资源前加锁。

注意事项

  • 避免死锁：在需要同时锁定多个互斥锁时，使用 std::scoped_lock（C++17 引入）可以同时锁定多个互斥锁，避免死锁风险。
  • 性能开销：引入互斥锁会带来一定的性能开销，尤其是在高并发场景下。根据实际需求，权衡线程安全性和性能之间的关系。

测试线程安全的 ThreadSafeSharedPtr

为了验证 ThreadSafeSharedPtr 的线程安全性，我们可以编写一个多线程程序，让多个线程同时拷贝、赋值和销毁智能指针。

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
#include <iostream>
#include <thread>
#include <vector>
#include "ThreadSafeSharedPtr.h" // 假设将上述代码保存为该头文件

// 测试类
class Test {
public:
    Test(int val) : value(val) {
        std::cout << "Test Constructor: " << value << std::endl;
    }
    ~Test() {
        std::cout << "Test Destructor: " << value << std::endl;
    }
    void show() const {
        std::cout << "Value: " << value << std::endl;
    }

private:
    int value;
};

void thread_func_copy(ThreadSafeSharedPtr<Test> sptr, int thread_id) {
    std::cout << "Thread " << thread_id << " is copying shared_ptr." << std::endl;
    ThreadSafeSharedPtr<Test> local_sptr = sptr;
    std::cout << "Thread " << thread_id << " copied shared_ptr, use_count = " << local_sptr.use_count() << std::endl;
    local_sptr->show();
}

void thread_func_reset(ThreadSafeSharedPtr<Test>& sptr, int new_val, int thread_id) {
    std::cout << "Thread " << thread_id << " is resetting shared_ptr." << std::endl;
    sptr.reset(new Test(new_val));
    std::cout << "Thread " << thread_id << " reset shared_ptr, use_count = " << sptr.use_count() << std::endl;
    sptr->show();
}

int main() {
    std::cout << "Creating ThreadSafeSharedPtr with Test(100)." << std::endl;
    ThreadSafeSharedPtr<Test> sptr(new Test(100));
    std::cout << "Initial use_count: " << sptr.use_count() << std::endl;

    // 创建多个线程进行拷贝操作
    const int num_threads = 5;
    std::vector<std::thread> threads_copy;

    for(int i = 0; i < num_threads; ++i) {
        threads_copy.emplace_back(thread_func_copy, sptr, i);
    }

    for(auto& t : threads_copy) {
        t.join();
    }

    std::cout << "After copy threads, use_count: " << sptr.use_count() << std::endl;

    // 创建多个线程进行 reset 操作
    std::vector<std::thread> threads_reset;

    for(int i = 0; i < num_threads; ++i) {
        threads_reset.emplace_back(thread_func_reset, std::ref(sptr), 200 + i, i);
    }

    for(auto& t : threads_reset) {
        t.join();
    }

    std::cout << "After reset threads, final use_count: " << sptr.use_count() << std::endl;

    std::cout << "Exiting main." << std::endl;
    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
Creating ThreadSafeSharedPtr with Test(100).
Test Constructor: 100
Constructed ThreadSafeSharedPtr, ref_count = 1
Initial use_count: 1
Thread 0 is copying shared_ptr.
Copied ThreadSafeSharedPtr, ref_count = 2
Thread 0 copied shared_ptr, use_count = 2
Value: 100
Thread 1 is copying shared_ptr.
Copied ThreadSafeSharedPtr, ref_count = 3
Thread 1 copied shared_ptr, use_count = 3
Value: 100
...
After copy threads, use_count: 6
Thread 0 is resetting shared_ptr.
Decremented ref_count to 5
Resource and ControlBlock destroyed.
Test Constructor: 200
Reset ThreadSafeSharedPtr, ref_count = 1
Value: 200
...
After reset threads, final use_count: 1
Exiting main.
Test Destructor: 200

说明：

  • 多个线程同时拷贝 sptr，引用计数正确递增。
  • 多个线程同时重置 sptr，确保引用计数和资源管理的正确性。
  • 最终，只有最新分配的对象存在，引用计数为 1。

注意事项和最佳实践

  1. 引用计数的原子性：
    • 使用 std::atomic<int> 来保证引用计数的线程安全递增和递减。
    • 避免使用普通的 int，因为在多线程环境下会导致数据竞争。
  2. 互斥锁的使用：
    • 使用 std::mutex 来保护 ptr 和 control 的访问，防止多个线程同时修改智能指针实例。
    • 尽量缩小锁的范围，避免在互斥锁保护的临界区内执行耗时操作，以减少性能开销。
  3. 避免死锁：
    • 在需要同时锁定多个互斥锁时，使用 std::scoped_lock 来一次性锁定，确保锁的顺序一致，避免死锁风险。
  4. 尽量遵循 RAII 原则：
    • 使用 std::lock_guard 或 std::scoped_lock 等 RAII 机制来管理互斥锁，确保在异常抛出时自动释放锁，防止死锁。
  5. 避免多重管理：
    • 确保不通过裸指针绕过智能指针的引用计数管理，避免资源泄漏或重复释放。
  6. 性能考虑：
    • 在高并发场景下，频繁的锁操作可能成为性能瓶颈。根据实际需求，可以考虑使用更轻量级的同步机制，如 std::shared_mutex（C++17）用于读多写少的场景。

总结

通过将 ControlBlock::ref_count 改为 std::atomic<int>，并在 ThreadSafeSharedPtr 中引入互斥锁来保护 ptr 和 control 的访问，可以实现一个线程安全的智能指针。这种实现确保了在多线程环境下，多个线程可以安全地拷贝、赋值和销毁智能指针，同时正确管理引用计数和资源。

关键点总结：

  • 引用计数的原子性：使用 std::atomic<int> 保证引用计数操作的线程安全。
  • 互斥锁保护：使用 std::mutex 保护智能指针实例的内部状态，防止多个线程同时修改。
  • RAII 机制：利用 std::lock_guard 和 std::scoped_lock 等 RAII 机制，确保锁的正确管理和释放。
  • 避免死锁：在需要同时锁定多个互斥锁时，使用 std::scoped_lock 以避免死锁风险。
  • 性能优化：平衡线程安全性和性能，避免不必要的锁竞争。