スズトキヅキ's Vignette

1
#pragma once
2

3
#include <memory>
4
#include <vector>
5
#include <thread>
6
#include <atomic>
7
#include <condition_variable>
8
#include <utility>  // std::move/std::forward
9
#include <cassert>  // assert
10

11
template<typename T>
12
class RingBuffer {
13
private:
14
    // ========== 核心修改：存储unique_ptr ==========
15
    std::vector<std::unique_ptr<T>> buffer_;
16
    size_t capacity_;
17
    size_t head_ = 0;  // 写位置
18
    size_t tail_ = 0;  // 读位置
19
    size_t count_ = 0;
20
    mutable std::mutex mutex_;
21
    std::condition_variable not_empty_;
22
    std::condition_variable not_full_;
23
    std::atomic<bool> stop_ = false; // 停止标志
24

25
    // ========== 移动push：高效转移所有权 ==========
26
    bool push(T&& item, int timeout_ms = 10) {
27
        return emplace(timeout_ms, std::move(item));
28
    }
29

30
    // ========== 非阻塞try_push ==========
31
    bool try_push(const T& item) {
32
        return emplace(0, item);
33
    }
34

35
    // ========== 返回unique_ptr的pop（推荐使用） ==========
36
    std::unique_ptr<T> pop_ptr(int timeout_ms = 10) {
37
        std::unique_lock<std::mutex> lock(mutex_);
38

39
        // ========== 新增：等待条件（含停止判断） ==========
40
        auto waitCondition = [this]() { return count_ > 0 || needStop(); };
41

42
        if (count_ == 0) {
43
            if (timeout_ms == 0) return nullptr;
44
            // ========== 新增：支持timeout_ms=-1 无限等待 ==========
45
            if (timeout_ms == -1) {
46
                not_empty_.wait(lock, waitCondition);
47
            } else {
48
                if (!not_empty_.wait_for(lock, std::chrono::milliseconds(timeout_ms), waitCondition)) {
49
                    return nullptr;
50
                }
51
            }
52
        }
53

54
        // ========== 新增：停止判断 ==========
55
        if (needStop() && count_ == 0) {
56
            return nullptr;
57
        }
58

59
        // ========== 关键优化：转移指针所有权 ==========
60
        std::unique_ptr<T> result = std::move(buffer_[tail_]);
61

62
        tail_ = (tail_ + 1) % capacity_;
63
        count_--;
64

65
        not_full_.notify_one();
66
        return result;
67
    }
68

69
    // ========== 兼容原有接口的pop ==========
70
    bool pop(T& item, int timeout_ms = 10) {
71
        auto ptr = pop_ptr(timeout_ms);
72
        if (!ptr) return false;
73

74
        // 移动对象内容到参数
75
        item = std::move(*ptr);
76
        return true;
77
    }
78

79
    // ========== 查看但不弹出（peek） ==========
80
    std::unique_ptr<T> peek(int timeout_ms = 10) {
81
        std::unique_lock<std::mutex> lock(mutex_);
82

83
        auto waitCondition = [this]() { return count_ > 0 || needStop(); };
84

85
        if (count_ == 0) {
86
            if (timeout_ms == 0) return nullptr;
87
            if (timeout_ms == -1) {
88
                not_empty_.wait(lock, waitCondition);
89
            } else {
90
                if (!not_empty_.wait_for(lock, std::chrono::milliseconds(timeout_ms), waitCondition)) {
91
                    return nullptr;
92
                }
93
            }
94
        }
95

96
        if (needStop() && count_ == 0) {
97
            return nullptr;
98
        }
99

100
        // 创建对象的副本
101
        if (!buffer_[tail_]) return nullptr;
102
        return std::make_unique<T>(*buffer_[tail_]);
103
    }
104

105
    // ========== 辅助方法 ==========
106
    size_t size() const {
107
        std::lock_guard<std::mutex> lock(mutex_);
108
        return count_;
109
    }
110

111
    size_t capacity() const {
112
        return capacity_;
113
    }
114

115
    bool empty() const {
116
        std::lock_guard<std::mutex> lock(mutex_);
117
        return count_ == 0;
118
    }
119

120
    bool full() const {
121
        std::lock_guard<std::mutex> lock(mutex_);
122
        return count_ == capacity_;
123
    }
124

125
    void clear() {
126
        std::lock_guard<std::mutex> lock(mutex_);
127

128
        // 释放所有指针
129
        for (size_t i = 0; i < capacity_; ++i) {
130
            buffer_[i].reset();
131
        }
132

133
        head_ = tail_ = count_ = 0;
134
        not_full_.notify_all();
135
    }
136

137
    bool needStop() const {
138
        return stop_.load(std::memory_order_acquire);
139
    }
140

141
    void stop() {
142
        stop_.store(true, std::memory_order_release);
143
        not_empty_.notify_all();
144
        not_full_.notify_all();
145
    }
146

147
    void reset() {
148
        std::lock_guard<std::mutex> lock(mutex_);
149
        stop_.store(false, std::memory_order_release);
150
        clear();
151
    }
152

153
    // ========== 直接访问方法（谨慎使用） ==========
154
    const T* peek_raw() const {
155
        std::lock_guard<std::mutex> lock(mutex_);
156
        if (count_ == 0) return nullptr;
157
        return buffer_[tail_].get();
158
    }
159

160
    T* peek_raw() {
161
        std::lock_guard<std::mutex> lock(mutex_);
162
        if (count_ == 0) return nullptr;
163
        return buffer_[tail_].get();
164
    }
165

166
    // ========== 预留额外容量（动态扩容） ==========
167
    bool reserve(size_t new_capacity) {
168
        std::lock_guard<std::mutex> lock(mutex_);
169

170
        if (new_capacity <= capacity_) {
171
            return false;  // 只能扩容，不能缩容
172
        }
173

174
        // 创建新缓冲区
175
        std::vector<std::unique_ptr<T>> new_buffer(new_capacity);
176

177
        // 复制现有元素（保持顺序）
178
        for (size_t i = 0; i < count_; ++i) {
179
            size_t src_idx = (tail_ + i) % capacity_;
180
            size_t dst_idx = i;  // 在新缓冲区中从0开始排列
181

182
            if (buffer_[src_idx]) {
183
                new_buffer[dst_idx] = std::move(buffer_[src_idx]);
184
            }
185
        }
186

187
        // 更新状态
188
        buffer_ = std::move(new_buffer);
189
        capacity_ = new_capacity;
190
        head_ = count_;  // 写位置在最后一个元素之后
191
        tail_ = 0;       // 读位置在开头
192

193
        not_full_.notify_all();
194
        return true;
195
    }
196

197

198
};