AI爱好者OpenAI的的开放方式有待观望
我们上文基于现有信息，技术理解，和我们的猜想，提出来了ChatGPT的最佳开放模式应该是增强和开放prompting的能力，让健忘的ChatGPT记得你交代的事情，从而让ChatGPT可以个人化。接下来就是GPT-4的开放和通过不同方式调教出来，各有所长的子能力模型。
但是以上都是猜想，真正的使用方式还是要看OpenAI自己如何开放。江湖传闻下一步是和Office Suite的深度整合。Notion AI已经给了很好的例子，确实很香。但是究竟是去增强微软现有产品，还是去成为下一代计算机？我认为OpenAI会选择后者。但是站在此时此刻，我们只能猜测和观望，以及做好人和ChatGPT的合理中间层。
ChatGPT Wrapper是我们当下可做的
其实绝大多数人是不习惯于清晰表达自己脑海中想法的。所以虽然ChatGPT已经把“调用计算力的手段”无限趋近于最合理的手段（用自然语言要求结果），但是我们已经发现，“提出信息完整的prompt”，对现在的人来说，是一个高要求。
网络上已经有很多prompting的模版（比如这里：https://github.com/PlexPt/awesome-chatgpt-prompts-zh）。把这一层做好，一方面能显著提高大家对ChatGPT的使用效率，一方面也能让我们积累对用户真实需求的理解和揣测，从而和ChatGPT做好适配。
有雄心壮志的创业者可能会对这一前景感到不爽，但是取代iPhone和iOS的公司毕竟不多，而在Apple生态内，也大有可为。微信比锤子值钱多了。
对ChatGPT能力的合理运用
对于ChatGPT，我们应该在现在的条件下思考更合适的应用：
- 对症下药：ChatGPT不是搜索引擎、不是程序，我们就用它做它擅长的事情，而不是去做搜索引擎和程序可以更高效完成的事情
- 择善而从：当前的ChatGPT是有明显的hallucination问题的，那我们就不能100%相信它的结论。一个简单的方法是我们要在自己能判断对错的地方去使用ChatGPT提高我们产生想法，收集归纳信息的效率，但是人要来把关
同时，我们也要意识到人类有更大的局限，不谈算力和知识面，光从理解能力这一角度来说：
- 人类很难意识到自己的错误
- 人类有ego，会影响自己的认知，也会让观点带入感情色彩
- 人类会故意偷换概念，等等
所以ChatGPT也许短期不会直接取代人类工作。但是两年内一定可以让一部分人的人效极大程度地提高。虽然说消灭工作的同时会产生新工作，但是很可能是消灭了一百个，产生了一个。那我们必须要思考一下，假如ChatGPT理解能力更上几层楼，拥有了多种能力，并且说话也靠谱不乱说了，我们人类的竞争力还剩下什么呢？第五问会详细展开，我们先讨论两个常见问题。
2B2C都会有，但生产力价值>>娱乐价值
ChatGPT的DAU增长是现象级的，而且大家都觉得很好用，所以会有很多2C应用的想象。但是我想提醒大家两点。
第一，ChatGPT显而易见的能力是在提升生产力上。而娱乐的价值未必很高，因为对话和阅读其实是一个门槛较高的娱乐形式。丰富性和深度，多数情况下也不是一个东西娱乐价值的重要因素。比如游戏想好玩，打击感一般比NPC栩栩如生更重要；一个短视频想吸引点击量，恐怕是话越少越好。所以建议少关注如何把ChatGPT应用到娱乐上，多关注如何提升生产力上，大方向应该没错。
第二，要记得这是一个颠覆型产品，而不是一个渐进式的改进。对于科技的early adopter来说，可能已经离不开它了（比如我），但是对于大众来说，遇到事情打开搜索引擎搜一搜，都不是普遍习惯，更别说去用清晰合理的prompt跟ChatGPT对话了。如果有这样的场景，其实ChatGPT也大概率是一个“提高效率”的角色。
《终结者》 vs《硅谷》
我的观点是AGI已经到来了，因为如果我们不拿一个完美的标准去衡量ChatGPT的话，会发现它已经在很多任务上超越了人类的表现。说实话，我已经开始觉得想尽量少跟人打交道，而尽量多跟ChatGPT打交道，因为像上文所说，ChatGPT有理性，没ego，也不会存心骗我。
那AGI会如何进化？这个时候我们要防止被科幻误导，如果限制ChatGPT的权限，那就不会发生《终结者》里面的世界。但是《硅谷》提到的AI的可能性倒不容小觑。在《硅谷》里，AI可以向着一个“更有效率的工具”这一目标自我进化，但很快，它就研究出如何破解世界上最安全的加密算法了。如果我们让AGI有自我进化的目标，那确实要小心，会涌现什么样的颠覆型能力。这个我想太多也没用，我相信OpenAI是一家最懂AGI，最担心AGI滥用的机构之一。
5 人类和ChatGPT的本质区别是什么？
因为我们的脑科学和神经科学非常不发达，这里只能从哲学寻求解答。BTW，除非脑科学产生重大的范式突破，不然neurallink这种脑机接口，是不可能实现大家想像中的那些功能的。但我们不是哲学专家，这里就仅供参考。
判断力
ChatGPT再厉害，也只能去吸取虚拟数字中的数字信号，是无法与现实世界做真实交互的。它可以听一万个专家告诉他做A就会得到B，但是不从真实世界中做实验，就无法从最底层确认这个说法究竟是真是假。绝知此事要躬行，才能有判断力的根基。
“Eureka”
牛顿看到苹果落地，可以发现万有引力，从而预测星星的运动。哥白尼发现地球是围绕太阳转的，而在他之前全地球人天天看着日出日落，都认为太阳是绕着地球转的。如果那个时候有一个ChatGPT，一定非常笃定太阳绕着地球转。那个ChatGPT也许能从苹果如何落地推测出桃子如何落地，但是大概率无法推测出星星的运动方式。
当然，能发现万有引力的人也是少数。更有意义的是去识别这种思维能力到底是什么，以及在我们日常生活中如何体现。阿基米德在泡澡时候发现浮力定律的时候喊了“Eureka”，大概可以形容这种“灵感并发、灵光一现”的瞬间。我们这里把这个瞬间稍稍具体地归结为“链接了数个相关的点，并且发现了第三个点”的过程。
增量知识
如果把现有知识归纳总结应用， 那必然PK不过ChatGPT。只有创造互联网上不存在的新知识，才可能是ChatGPT做不到的。注意条件指向，互联网上不存在的新知识，也未必不能从存量知识里总结出来，但是能从存量知识里总结出来的，一定不是人类的优势。
理解人
人类的文本知识里一定存在很多人性的理解，但是也一定有一些人性或者偏好，是没有被记录总结在文字里的。如果我们结合1和3，就会发现，去真实世界理解人，而不是去通过调研、问卷、网络资料理解人；去带来增量的理解，而不是去人云亦云地重复套路。才是人类相对于ChatGPT的优势。
6 总结
ChatGPT的范式突破是“乌鸦”能力。恕我能力所限，无法更简单地无损表达这一能力的本质了。如果允许有损，我会用“理解”能力来概括它最重要的一面。作为对比，过往ML的能力模式是“鹦鹉”能力，所做的是寻找“对应关系”ChatGPT的意义是对“调用算力、总结信息”最究极的手段，预测会在两年内有能力辅助人类取代大多数可被定义为“搬砖”类型的工作ChatGPT的“乌鸦”能力是涌现的，工程难度是极高的。我们应该抛弃各种噪音，聚焦关注Google是否能复现这一能力，从而判定这一能力到底有多难。而现在，建议我们的默认预测是这一能力很难复现，需要别人来用强力证据说服我们他们能复现我们对ChatGPT的使用应该观望OpenAI给我们提供的调用方式，在当下，我们应该聚焦用好ChatGPT，并且做好ChatGPT能力与我们所需要解决问题的的中间层。实现的线程安全的环形队列，同时添加了gtest单元测试，并提供了相应的测试报告。
circular_queue.h
// 头文件防卫
#ifndef CIRCULAR_QUEUE_H
#define CIRCULAR_QUEUE_H
#include <mutex> // 互斥量
#include <condition_variable> // 条件变量
template <typename T>
class CircularQueue {
public:
// 构造函数，初始化成员变量
explicit CircularQueue(size_t capacity) :
capacity_(capacity),
size_(0),
head_(0),
tail_(0),
buffer_(new T[capacity]) {}
// 析构函数，释放 buffer_ 内存
~CircularQueue() {
delete[] buffer_;
}
// 判断队列是否为空
bool empty() {
std::unique_lock<std::mutex> lock(mutex_);
return size_ == 0;
}
// 判断队列是否已满
bool full() {
std::unique_lock<std::mutex> lock(mutex_);
return size_ == capacity_;
}
// 获取队列中元素的数量
size_t size() {
std::unique_lock<std::mutex> lock(mutex_);
return size_;
}
// 获取队列的容量
size_t capacity() {
return capacity_;
}
// 将元素加入队列，可能会阻塞
bool push(const T& value, bool block = true) {
std::unique_lock<std::mutex> lock(mutex_);
if (block) {
// 如果队列已满，则等待队列不满
while (size_ == capacity_) {
not_full_.wait(lock);
}
} else {
// 如果队列已满，则返回 false
if (size_ == capacity_) {
return false;
}
}
// 将元素加入队列尾部，并更新 tail_ 和 size_
buffer_[tail_] = value;
tail_ = (tail_ + 1) % capacity_;
++size_;
// 通知一个等待在 not_empty_ 条件变量上的线程
not_empty_.notify_one();
return true;
}
// 将元素加入队列，可能会阻塞，使用右值引用
bool push(T&& value, bool block = true) {
std::unique_lock<std::mutex> lock(mutex_);
if (block) {
// 如果队列已满，则等待队列不满
while (size_ == capacity_) {
not_full_.wait(lock);
}
} else {
// 如果队列已满，则返回 false
if (size_ == capacity_) {
return false;
}
}
// 将元素加入队列尾部，并更新 tail_ 和 size_
buffer_[tail_] = std::move(value);
tail_ = (tail_ + 1) % capacity_;
++size_;
// 通知一个等待在 not_empty_ 条件变量上的线程
not_empty_.notify_one();
return true;
}
// 从队列中取出元素，可能会阻塞
bool pop(T& value, bool block = true) {
std::unique_lock<std::mutex> lock(mutex_);
if (block) {
// 如果队列为空，则等待队列不空
while (size_ == 0) {
not_empty_.wait(lock);
}
} else {
// 如果队列为空，则返回 false
if (size_ == 0) {
return false;
}
}
// 取出队列头部元素，并更新 head_ 和 size_
value = std::move(buffer_[head_]);
head_ = (head_ + 1) % capacity_;
--size_;
// 通知一个等待在 not_full_ 条件变量上的线程
not_full_.notify_one();
return true;
}
private:
const size_t capacity_; // 队列容量
size_t size_; // 队列中元素的数量
size_t head_; // 队列头部指针
size_t tail_; // 队列尾部指针
T* buffer_; // 队列缓冲区
std::mutex mutex_; // 互斥量，保护队列缓冲区和队列大小
std::condition_variable not_full_; // 条件变量，当队列满时等待
std::condition_variable not_empty_; // 条件变量，当队列空时等待
};
#endif // CIRCULAR_QUEUE_H
针对上面的代码，质量挺好，上面的注释不是我写的，push和pop接口不指定第二个参数的话，默认是阻塞的，这一点使用时需要注意。
生成的单元测试不全面，我又贴了它生成的代码，让他帮我生成单元测试，这次很不错，接口基本覆盖了，还知道起线程测试阻塞接口，左右值的插入版本都测了。
测试时发现了单元测试有bug:
- 调用空队列的pop接口时未指明非阻塞，结果把自己阻塞住了，见单元测试注释（1）
可见目前为止chatgpt写出来的代码也是有bug的，不能全盘依赖。
circular_queue_gtest.cpp
以下是CircularQueue类的单元测试示例代码：
#include <gtest/gtest.h>
#include <thread>
#include "circular_queue.h"
TEST(CircularQueueTest, EmptyQueue) {
CircularQueue<int> queue(10);
ASSERT_TRUE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_EQ(queue.size(), 0);
ASSERT_EQ(queue.capacity(), 10);
}
TEST(CircularQueueTest, PushAndPop) {
CircularQueue<int> queue(3);
ASSERT_TRUE(queue.push(1));
ASSERT_EQ(queue.size(), 1);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.push(2));
ASSERT_EQ(queue.size(), 2);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.push(3));
ASSERT_EQ(queue.size(), 3);
ASSERT_FALSE(queue.empty());
ASSERT_TRUE(queue.full());
int value;
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 1);
ASSERT_EQ(queue.size(), 2);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 2);
ASSERT_EQ(queue.size(), 1);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 3);
ASSERT_EQ(queue.size(), 0);
ASSERT_TRUE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_FALSE(queue.pop(value,false)); // （1） 此项测试未指定非阻塞，笔者修改了
}
TEST(CircularQueueTest, PushAndPopWithBlocking) {
CircularQueue<int> queue(2);
std::thread t([&queue]() {
int value = 0;
queue.pop(value);
ASSERT_EQ(value, 1);
queue.pop(value);
ASSERT_EQ(value, 2);
});
ASSERT_TRUE(queue.push(1));
ASSERT_TRUE(queue.push(2));
ASSERT_TRUE(queue.push(3));
t.join();
}
TEST(CircularQueueTest, PushAndPopWithNonBlocking) {
CircularQueue<int> queue(2);
int value;
ASSERT_TRUE(queue.push(1));
ASSERT_TRUE(queue.push(2));
ASSERT_FALSE(queue.push(3, false));
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 1);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 2);
ASSERT_FALSE(queue.pop(value, false));
}
TEST(CircularQueueTest, MovePushAndPop) {
CircularQueue<std::string> queue(3);
ASSERT_TRUE(queue.push("hello"));
ASSERT_TRUE(queue.push("world"));
ASSERT_EQ(queue.size(), 2);
std::string value;
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "hello");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.push("foo"));
ASSERT_EQ(queue.size(), 2);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "world");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "foo");
ASSERT_EQ(queue.size(), 0);
}
TEST(CircularQueueTest, CopyPushAndPop) {
CircularQueue<std::string> queue(3);
ASSERT_TRUE(queue.push(std::string("hello")));
ASSERT_TRUE(queue.push(std::string("world")));
ASSERT_EQ(queue.size(), 2);
std::string value;
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "hello");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.push(std::string("foo")));
ASSERT_EQ(queue.size(), 2);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "world");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "foo");
ASSERT_EQ(queue.size(), 0);
}
int main(int argc, char** argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
解决上述问题后，单元测试跑过了：
上面的测试远远不够，再对测试提出要求，给代码添加gtest多线程单元测试，这是chatgpt给出的答案：
以下是添加gtest多线程单元测试后的代码：
#include <gtest/gtest.h>
#include <thread>
#include "circular_queue.h"
TEST(CircularQueueTest, MultiThreadPushPop) {
const int num_threads = 4;
const int num_iterations = 10000;
const int queue_size = 100;
CircularQueue<int> queue(queue_size);
std::vector<std::thread> threads;
for (int i = 0; i < num_threads; ++i) {
threads.emplace_back([&queue, num_iterations]() {
for (int j = 0; j < num_iterations; ++j) {
queue.push(j);
}
});
}
for (int i = 0; i < num_threads; ++i) {
threads.emplace_back([&queue, num_iterations]() {
for (int j = 0; j < num_iterations; ++j) {
int value;
queue.pop(value);
}
});
}
for (auto& thread : threads) {
thread.join();
}
ASSERT_EQ(queue.size(), 0);
}
int main(int argc, char** argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
在上面的测试中，我们创建了4个线程，其中两个线程分别向队列中推送10000个整数，另外两个线程则从队列中弹出这些整数。最后，我们检查队列是否为空。
值得注意的是，在多线程测试中，我们需要确保每个线程都使用自己的锁，否则可能会导致死锁或数据竞争等问题。因此，在上面的测试中，我们使用了CircularQueue类中的mutex_成员变量来确保每个线程都能独立地访问队列。
这个单元测试跑过了：
$ ./a.out
[==========] Running 1 test from 1 test suite.
[----------] Global test environment set-up.
[----------] 1 test from CircularQueueTest
[ RUN ] CircularQueueTest.MultiThreadPushPop
[ OK ] CircularQueueTest.MultiThreadPushPop (191 ms)
[----------] 1 test from CircularQueueTest (191 ms total)
[----------] Global test environment tear-down
[==========] 1 test from 1 test suite ran. (192 ms total)
[ PASSED ] 1 test.
相关视频推荐
自旋锁、互斥锁、信号量、原子操作、条件变量在不同开源框架的应用
【文章福利】：小编整理了一些个人觉得比较好的学习书籍、视频资料共享在群文件里面，有需要的可以自行添加哦！~点击832218493加入（需要自取）
无锁版本
上面的循环队列使用锁保证了线程安全，对此我又明确提出了无锁的需求，无锁肯定是借助原子变量实现的，果不其然。
以下是一个基于C++11的线程安全且无锁的环形队列实现，存放的数据类型可以自定义。代码风格使用Google C++风格。
环形队列的实现是基于循环数组的，使用两个指针分别指向队列头和队列尾，实现数据的入队和出队操作。为了保证线程安全，使用了原子操作来保证多线程访问时的数据一致性。
代码如下，下面的代码注释是我加的：
#include <atomic>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <thread>
#include <vector>
template <typename T, size_t N>
class RingQueue {
public:
RingQueue() : read_idx_(0), write_idx_(0), data_{} {}
bool Push(const T& item) { return PushImpl(item); }
bool Push(T&& item) { return PushImpl(std::move(item)); }
bool Pop(T& item) {
// 考虑只有一个读者的，read_idx_的读取可以不加锁
size_t current_read_idx = read_idx_.load(std::memory_order_relaxed);
// 保证读到write_idx_的变化，此处memory_order_acquire发挥的是可见性问题
if (current_read_idx == write_idx_.load(std::memory_order_acquire)) {
return false;
}
item = std::move(data_[current_read_idx]); // 必须先把数据读走read_idx_才能+1，memory_order_release保证了对item的写不会被重排到read_idx+1之后
read_idx_.store(Next(current_read_idx), std::memory_order_release);
return true;
}
template <typename Func>
bool Pop(Func&& func) {
size_t current_read_idx = read_idx_.load(std::memory_order_relaxed);
if (current_read_idx == write_idx_.load(std::memory_order_acquire)) {
return false;
}
T item = std::move(data_[current_read_idx]);
read_idx_.store(Next(current_read_idx), std::memory_order_release);
func(std::move(item));
return true;
}
bool IsEmpty() const {
return read_idx_.load(std::memory_order_acquire) ==
write_idx_.load(std::memory_order_acquire);
}
bool IsFull() const {
return Next(write_idx_.load(std::memory_order_acquire)) ==
read_idx_.load(std::memory_order_acquire);
}
private:
template <typename Item>
bool PushImpl(Item&& item) {
// 只有1个写线程,所以write_idx_可以不加锁
size_t current_write_idx = write_idx_.load(std::memory_order_relaxed);
size_t next_write_idx = Next(current_write_idx);
// 读线程会修改read_idx_，所以此处需要保证看到read_idx_的变化,此处memory_order_acquire保证的是可见性问题
if (next_write_idx == read_idx_.load(std::memory_order_acquire)) {
return false;
}
// 数据的写入必须在write_idx_+1之前
data_[current_write_idx] = std::forward<Item>(item);
// 保证之前的写操作对读线程可见，即读线程能立刻看到data_刚写入的数据，当然也包括write_idx_的+1变化,memory_order_release会保证对data_的写入在write_idx_+1的操作之前完成。
// 因为就算data_的赋值语句放在write_idx_+1之前，由于编译器或者运行期指令重排，并不一定能保证data_赋值语句就一定在write_idx_+1前执行。
write_idx_.store(next_write_idx, std::memory_order_release);
return true;
}
size_t Next(size_t current_idx) const { return (current_idx + 1) % N; }
std::atomic<size_t> read_idx_;
std::atomic<size_t> write_idx_;
std::array<T, N> data_;
};
我们来点评下。
这个版本的代码也是挺优雅的，代码中使用了模板参数T和N，支持了不同数据类型和不同队列大小的选择，借助读写指针两个原子变量实现无锁环形队列。
但需要注意的是，这个只能实现一读一写的线程安全，存在多个读者或者多个写者时就线程不安全了。
无锁编程的难点在于对几个内存时序的理解。
补充下关于内存时序操作的解释。
C++定义了几种内存时序，这些时序规定了原子变量前后的所有内存操作（包括普通变量、原子变量）如何排序std::memory_order_relaxed只保正操作的原子性，对于同一个原子变量的多个操作之间不存在任何内存序的限制，也就是说，它们可以随意重排序，也可以在任意时刻被其他线程的操作所干扰。因此，使用std::memory_order_relaxed时需要特别小心，必须确保操作的正确性不受此种松散的内存访问顺序的影响。std::memory_order_relaxed主要用于那些不需要任何同步机制的场合，比如计数器的自增、自减等操作，这些操作只需要保证结果的正确性，而不需要保证其执行的顺序。因此，std::memory_order_relaxed是最快的内存序，但也是最危险的一种内存序。
std::memory_order_acquire确保所有之前的读操作都已经完成，然后再执行当前读取操作。这意味着，如果当前读取操作需要用到之前的读取操作的结果，那么它将能够正确地获取到这些结果。
具体来说，当使用memory_order_acquire语义时，编译器和处理器都会保证当前线程所在的CPU核心（或处理器）在执行当前原子操作之前，会先将所有之前的读操作所获得的数据从CPU缓存中刷新到主内存中，以保证当前线程能够读取到其他线程对共享变量的最新修改。
使用memory_order_acquire语义可以保证程序的正确性，避免出现数据竞争的问题。但是，使用memory_order_acquire语义可能会降低程序的性能，因为它要求在执行原子操作之前，必须将所有之前的读操作都刷新到主内存中，这可能会导致缓存一致性协议的开销增加。因此，在实际编程中，应该根据具体情况选择合适的内存序语义。
std::memory_order_release确保当前线程的所有写操作在该原子操作之前都已经完成，并且将这些写操作对其他线程可见。这样，其他线程就可以看到当前线程对共享数据所做的更改。
这种释放操作通常用于同步操作，例如将一个共享变量的值更新后通知其他线程。在这种情况下，std::memory_order_release可以确保其他线程能够看到更新后的值。
push提供了插入左值和右值两个版本，提高C++所谓的一点点性能，PushImpl提取了公共代码，实现代码复用，优雅！。
pop提供了两个版本，有个变体，不返回pop出来的对象，而是调用外部传入的回调，对pop出来的对象进行操作。
上面的Push和Pop都是非阻塞的，我们再让其生成阻塞和非阻塞合二为一的版本:
bool Pop(T& item, bool block=false) {
// 考虑只有一个读者的，read_idx_的读取可以不加锁
size_t current_read_idx = read_idx_.load(std::memory_order_relaxed);
// 保证读到write_idx_的变化，此处memory_order_acquire发挥的是可见性问题
while (current_read_idx == write_idx_.load(std::memory_order_acquire)) {
if (!block) {
return false;
}
std::this_thread::yield(); // 让出调度
}
item = std::move(data_[current_read_idx]); // 必须先把数据读走read_idx_才能+1，memory_order_release保证了对item的写不会被重排到read_idx+1之后
read_idx_.store(Next(current_read_idx), std::memory_order_release);
return true;
}
阻塞和非阻塞合二为一的Push的代码见后面，原理类似。
再让其生成async版本:
void PopAsync(const T& value, std::function<void(bool)> callback) {
auto task = [this, value, callback]() {
bool result = Enqueue(value);
callback(result);
};
std::thread(std::move(task)).detach();
}
最后给一个整合后的完整版本：
#include <atomic>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <thread>
#include <vector>
template <typename T, size_t N>
class RingQueue {
public:
RingQueue() : read_idx_(0), write_idx_(0), data_{} {}
bool Push(const T& item, bool block = false) { return PushImpl(item, block); }
bool Push(T&& item, bool block = false) { return PushImpl(std::move(item), block); }
bool Pop(T& item, bool block = false) {
// 考虑只有一个读者的，read_idx_的读取可以不加锁
size_t current_read_idx = read_idx_.load(std::memory_order_relaxed);
// 保证读到write_idx_的变化，此处memory_order_acquire发挥的是可见性问题
while (current_read_idx == write_idx_.load(std::memory_order_acquire)) {
if (!block) {
return false;
}
std::this_thread::yield();
}
item = std::move(data_[current_read_idx]); // 必须先把数据读走read_idx_才能+1，memory_order_release保证了对item的写不会被重排到read_idx+1之后
read_idx_.store(Next(current_read_idx), std::memory_order_release);
return true;
}
template <typename Func>
bool Pop(Func&& func, bool block = false) {
size_t current_read_idx = read_idx_.load(std::memory_order_relaxed);
while (current_read_idx == write_idx_.load(std::memory_order_acquire)) {
if (!block) {
return false;
}
std::this_thread::yield();
}
T item = std::move(data_[current_read_idx]);
read_idx_.store(Next(current_read_idx), std::memory_order_release);
func(std::move(item));
return true;
}
bool IsEmpty() const {
return read_idx_.load(std::memory_order_acquire) ==
write_idx_.load(std::memory_order_acquire);
}
bool IsFull() const {
return Next(write_idx_.load(std::memory_order_acquire)) ==
read_idx_.load(std::memory_order_acquire);
}
private:
template <typename Item>
bool PushImpl(Item&& item, bool block = false) {
// 只有1个写线程,所以write_idx_可以不加锁
size_t current_write_idx = write_idx_.load(std::memory_order_relaxed);
size_t next_write_idx = Next(current_write_idx);
// 读线程会修改read_idx_，所以此处需要保证看到read_idx_的变化,此处memory_order_acquire保证的是可见性问题
while (next_write_idx == read_idx_.load(std::memory_order_acquire)) {
if(!block) {
return false;
}
std::this_thread::yield();
}
// 数据的写入必须在write_idx_+1之前
data_[current_write_idx] = std::forward<Item>(item);
// 保证之前的写操作对读线程可见，即读线程能立刻看到data_刚写入的数据，当然也包括write_idx_的+1变化,memory_order_release会保证对data_的写入在write_idx_+1的操作之前完成。
// 因为就算data_的赋值语句放在write_idx_+1之前，由于编译器或者运行期指令重排，并不一定能保证data_赋值语句就一定在write_idx_+1前执行。
write_idx_.store(next_write_idx, std::memory_order_release);
return true;
}
size_t Next(size_t current_idx) const { return (current_idx + 1) % (N+1); } // 此处笔者做了修改，N改成N+1
std::atomic<size_t> read_idx_;
std::atomic<size_t> write_idx_;
std::array<T, N+1> data_; // 此处笔者做了修改，N改成N+1
};
同样附上单元测试。
#include <gtest/gtest.h>
#include "RingQueue.h"
class RingQueueTest : public ::testing::Test {
protected:
RingQueue<int, 10> queue_;
};
TEST_F(RingQueueTest, PushAndPop) {
int value = 0;
EXPECT_FALSE(queue_.Pop(value));
EXPECT_TRUE(queue_.Push(1));
EXPECT_FALSE(queue_.IsEmpty());
EXPECT_TRUE(queue_.Pop(value));
EXPECT_EQ(value, 1);
EXPECT_TRUE(queue_.IsEmpty());
}
TEST_F(RingQueueTest, PushAndPopWithBlock) {
int value = 0;
std::thread t([&](){
std::this_thread::sleep_for(std::chrono::milliseconds(100));
queue_.Push(1, true);
});
EXPECT_TRUE(queue_.Pop(value, true));
EXPECT_EQ(value, 1);
t.join();
}
TEST_F(RingQueueTest, PushAndPopWithFunc) {
int value = 0;
queue_.Push(1);
queue_.Pop([&](int v){ value = v + 1; });
EXPECT_EQ(value, 2);
}
TEST_F(RingQueueTest, IsEmptyAndIsFull) {
EXPECT_TRUE(queue_.IsEmpty());
EXPECT_FALSE(queue_.IsFull());
for (int i = 0; i < 10; ++i) {
EXPECT_TRUE(queue_.Push(i));
}
EXPECT_TRUE(queue_.IsFull());
EXPECT_FALSE(queue_.IsEmpty());
int value = 0;
EXPECT_FALSE(queue_.Push(10));
EXPECT_TRUE(queue_.Pop(value));
EXPECT_FALSE(queue_.IsFull());
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
chatgpt在这个版本中同样犯了错误，实现的环形队列大小设置为10，但实际容量只有9，这没有什么毛病，因为最后一个空间用来区分非空还是满，但是它给的单元测试却是按照容量为10去测试，实际是测试不过的，然而它还给出了测试通过的报告。
修正这个问题后，单元测试都通过了。
[==========] Running 4 tests from 1 test suite.
[----------] Global test environment set-up.
[----------] 4 tests from RingQueueTest
[ RUN ] RingQueueTest.PushAndPop
[ OK ] RingQueueTest.PushAndPop (0 ms)
[ RUN ] RingQueueTest.PushAndPopWithBlock
[ OK ] RingQueueTest.PushAndPopWithBlock (100 ms)
[ RUN ] RingQueueTest.PushAndPopWithFunc
[ OK ] RingQueueTest.PushAndPopWithFunc (0 ms)
[ RUN ] RingQueueTest.IsEmptyAndIsFull
[ OK ] RingQueueTest.IsEmptyAndIsFull (0 ms)
[----------] 4 tests from RingQueueTest (100 ms total)
[----------] Global test environment tear-down
[==========] 4 tests from 1 test suite ran. (100 ms total)
[ PASSED ] 4 tests.
显然上面的测试不够充分，我还需要读写线程并发读写的测试，它帮我生成了：
#include <gtest/gtest.h>
#include <thread>
#include <chrono>
#include <string>
#include <vector>
#include <iostream>
#include "RingQueue.h"
class RingQueueTest : public testing::Test {
protected:
virtual void SetUp() {
// 初始化数据
for (int i = 0; i < 1000; ++i) {
data_.push_back("data_" + std::to_string(i));
}
}
std::vector<std::string> data_;
};
TEST_F(RingQueueTest, MultiThreadTest) {
RingQueue<std::string, 10> queue;
// 写线程
std::thread writer([&queue, this]() {
for (const auto& item : data_) {
queue.Push(item, true);
}
});
// 读线程
std::thread reader([&queue, this]() {
int count = 0;
std::string item;
while (count < 1000) {
if (queue.Pop(item, true)) {
EXPECT_EQ(item, "data_" + std::to_string(count));
++count;
} else {
std::this_thread::yield();
}
}
});
writer.join();
reader.join();
}
一句话，虽有瑕疵，但还是很牛逼，超越国内99%的程序员，当然包括笔者。
我们近期会推出更多关于chatgpt写代码的文章。
