c++ - C++ 并发队列内存泄漏

Question

考虑下面的程序，它实现了一个消费者，多个生产者并发队列。

在 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;
}

建议通过@1201ProgramAlarm 编辑推送方法

    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 的任何想法？

标签： c++multithreadingconcurrency

Answer 1

您的示例存在许多问题。

主要的是它不是线程安全的：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::request_stop()condition_variable::wait()自动加入和终止线程。