考虑下面的程序，它实现了一个单消费者、多生产者并发队列。
在1个消费者和1个生产者的情况下运行良好。
然而，设置第二个消费者(不对下面的行进行注释(会导致内存泄漏，我不明白为什么。。。
使用Queue的T=std:：shared_ptr，并修改pop以返回shared_ptr修复了内存泄漏，那么我在下面的代码中缺少什么呢？
#include <functional>
#include <iostream>
#include <mutex>
#include <thread>
template<typename T>
class Queue {
private:
static constexpr unsigned mSize = 256; //power of two only
static constexpr unsigned mRoundRobinMask = mSize - 1;
static const T mEmpty;
T mData[mSize];
std::mutex mtx;
unsigned mReadP = 0;
unsigned mWriteP = 0;
public:
const T pop() {
if (!peek()) {
return mEmpty; // copy
}
std::lock_guard<std::mutex> lock(mtx);
T& ret = mData[mReadP & mRoundRobinMask]; // get a ref
mReadP++;
return ret; // copy of ref
}
void push(const T& aItemRef) {
start:
if (!wait()) {
throw std::runtime_error("!Queue FULL!");
}
std::lock_guard<std::mutex> lock(mtx);
if(size() == mSize) {
goto start;
}
mData[mWriteP & mRoundRobinMask] = aItemRef;
mWriteP++;
}
bool peek() const {
return mWriteP != mReadP;
}
unsigned size() const {
return mWriteP > mReadP ? mWriteP - mReadP : mReadP - mWriteP; // mod (Read-Write)
}
bool wait() {
unsigned it = 0;
while (size() == mSize) {
if (it++ > 1000000) { return false; }
}
return true;
}
};
template<typename T>
const T Queue<T>::mEmpty = T{ };
int main(int, char**) {
using Method = std::function<void()>;
Queue<Method*> queue;
std::thread consumer([ & ] {
while (true) {
if (queue.peek()) {
auto task = queue.pop();
(*task)();
delete task;
}
}
});
std::thread producer1([ & ] {
unsigned index = 0;
while (true) {
auto id = index++;
auto task = new Method([ = ] {
std::cout << "Running task " << id << std::endl;
});
queue.push(task);
}
});
// std::thread producer2 ([ & ] {
// unsigned index = 0;
// while (true) {
// auto id = index++;
// auto task = new Method([ = ] {
// std::cout << "Running task " << id<< std::endl;
// });
// queue.push(task);
// }
// });
consumer.join();
producer1.join();
// producer2.join();
return 0;
}
建议通过@11201ProgramAlarm 编辑推送方法
void push(const T& aItemRef) {
start:
if (!wait()) {
throw std::runtime_error("!Queue FULL!");
}
std::lock_guard<std::mutex> lock(mtx);
if(getCount() == mSize) {
goto start;
}
mData[mWriteP & mRoundRobinMask] = aItemRef;
mWriteP++;
}
这项工作，没有更多的泄漏，但GOTO：(：(……有关于如何避免使用GOTO的想法吗？
您的示例存在许多问题。
主要是它不是线程安全的：push()和pop()都修改非原子成员变量mReadP和mWriteP，而不受mutex的保护。
第二个不太重要的问题是，等待项目弹出或释放空间进行推送通常是通过使用condition_variables来完成的，它会挂起线程，直到达到条件。
请尝试下面的版本，因为我用这些更改更新了它。
我还添加了一个终止条件，以显示如何安全地退出所有线程，并放慢整个过程的速度以显示正在发生的事情。
#include <array>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <thread>
#include <optional>
template<typename T>
class Queue {
private:
static constexpr unsigned mSize = 256; //power of two only
static constexpr unsigned mRoundRobinMask = mSize - 1;
std::array<T, mSize> mData;
std::mutex mtx;
unsigned mReadP = 0;
unsigned mWriteP = 0;
std::condition_variable notFull;
std::condition_variable notEmpty;
bool stopped = false;
public:
const std::optional<T> pop() {
// Always grab the mutex before accessing any shared members
std::unique_lock<std::mutex> lock(mtx);
// Wait until there is an item in the queue.
notEmpty.wait(lock, [&] {return stopped || mWriteP != mReadP; });
if(stopped)
return std::nullopt;
T& ret = mData[mReadP & mRoundRobinMask]; // get a ref
mReadP++;
// Wake any threads waiting on full buffer
notFull.notify_one();
return ret; // copy of ref
}
void push(const T& pItem) {
std::unique_lock<std::mutex> lock(mtx);
// Wait until there is space to put at least one item
notFull.wait(lock, [&] { return stopped || getCount() < mSize; });
if(stopped)
return;
mData[mWriteP & mRoundRobinMask] = pItem;
mWriteP++;
// Wake any threads waiting on empty buffer
notEmpty.notify_one();
}
unsigned getCount() const {
return mWriteP > mReadP ?
mWriteP - mReadP : mReadP - mWriteP; // mod (Read-Write)
}
void stop() {
// Signal the stop condition
stopped = true;
// Grabbing the lock before notifying is essential to make sure
// any worker threads waiting on the condition_variables.
std::unique_lock<std::mutex> lock(mtx);
// Wake all waiting threads
notFull.notify_all();
notEmpty.notify_all();
}
};
int main(int, char**) {
using Method = std::function<void()>;
Queue<Method> queue;
bool running = true;
std::thread consumer([ & ] {
while (running) {
auto task = queue.pop();
if(task) {
// If there was a task, execute it.
(*task)();
} else {
// No task means we are done.
return;
}
}
});
std::thread producer1([ & ] {
unsigned index = 0;
while (running) {
auto id = index++;
queue.push([ = ] {
std::cout << "Running task " << id << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(100));
});
}
});
std::this_thread::sleep_for(std::chrono::seconds(2));
// Use pre-c++-20 mechanisms to signal the worker threads to stop their loops
running = false;
// If they're in the queue stop that too.
queue.stop();
consumer.join();
producer1.join();
return EXIT_SUCCESS;
}
请注意，如果您可以使用C++20，那么您完全应该使用它，因为它具有std：：jthread，它具有更优雅的机制，如自动线程连接和通过std：jthread:：requestrongtop((终止condition_variable::wait()。
您的push不是线程安全的。
当队列中只有一个插槽可用时，由两个线程调用时，两个线程都可以通过wait，导致其中一个线程可能会覆盖队列中的现有元素。被覆盖的元素将不会被释放，从而导致内存泄漏。
解决方案是在获得锁后检查队列是否再次满。如果是，您需要释放锁，然后再次等待，直到有可用的插槽。
顺便说一句，通过在while循环中包含sleep(0)调用，wait函数可以变得更友好。这将减少等待时的功耗和CPU资源的使用。