高负载下c ++线程工作者失败

如何解决高负载下c ++线程工作者失败

我一直在为一个系统构想，其中我可以有许多由中央计时器类定期触发的工人。我在这里关心的部分是TriggeredWorker，在循环中，它使用mutex和conditionVariable方法等待被告知进行工作。它具有方法trigger，该方法被另一线程调用（由另一个线程调用），该方法触发要完成的工作。这是一个抽象类，必须将其子类化才能实现实际的work方法。

我有一个测试表明该机制有效。但是，随着我通过减少触发间隔来增加负载，测试开始失败。当我在两次触发之间延迟20微秒时，该测试是100％可靠的。当我减少到1微秒时，我开始失败，因为执行的工作数量从1000（预期）减少到986、933、999等值。

我的问题是：（1）出了什么问题？如何捕捉到出了什么问题，以便报告或对此做些什么？并且，（2）有没有更好的方法可以使用呢？我不得不承认，尽管我使用其他语言已经有好几年了，但是我在c ++方面的经验仅限于最近三个月。

非常感谢您阅读...

以下是代码的关键部分：

触发的工作程序头文件：

#ifndef TIMER_TRIGGERED_WORKER_H
#define TIMER_TRIGGERED_WORKER_H

#include <thread>
#include <plog/Log.h>

class TriggeredWorker {
private:
    std::mutex mutex_;
    std::condition_variable condVar_;
    std::atomic<bool> running_{false};
    std::atomic<bool> ready_{false};

    void workLoop();
protected:
    virtual void work() {};
public:
    void start();
    void stop();
    void trigger();
};

#endif //TIMER_TRIGGERED_WORKER_H

触发了工作程序的实现：

#include "TriggeredWorker.h"

void TriggeredWorker::workLoop() {
    PLOGD << "workLoop started...";

    while(true) {
        std::unique_lock<std::mutex> lock(mutex_);
        condVar_.wait(lock,[this]{
            bool ready = this->ready_;
            bool running = this->running_;
            return ready | !running; });
        this->ready_ = false;

        if (!this->running_) {
            break;
        }

        PLOGD << "Calling work()...";
        work();

        lock.unlock();
        condVar_.notify_one();
    }

    PLOGD << "Worker thread completed.";
}

void TriggeredWorker::start() {
    PLOGD << "Worker start...";
    this->running_ = true;
    auto thread = std::thread(&TriggeredWorker::workLoop,this);
    thread.detach();
}

void TriggeredWorker::stop() {
    PLOGD << "Worker stop.";
    this->running_ = false;
}

void TriggeredWorker::trigger() {
    PLOGD << "Trigger.";
    std::unique_lock<std::mutex> lock(mutex_);
    ready_ = true;
    lock.unlock();
    condVar_.notify_one();
}

和测试：

#include "catch.hpp"
#include "TriggeredWorker.h"
#include <thread>

TEST_CASE("Simple worker performs work when triggered") {
    static std::atomic<int> twt_count{0};

    class SimpleTriggeredWorker : public TriggeredWorker {
    protected:
        void work() override {
            PLOGD << "Incrementing counter.";
            twt_count.fetch_add(1);
        }
    };

    SimpleTriggeredWorker worker;

    worker.start();

    for (int i = 0; i < 1000; i++) {
        worker.trigger();
        std::this_thread::sleep_for(std::chrono::microseconds(20));
    }

    std::this_thread::sleep_for(std::chrono::seconds(1));

    CHECK(twt_count == 1000);

    std::this_thread::sleep_for(std::chrono::seconds(1));
    worker.stop();
}

解决方法

在worker.trigger()获得锁之前调用workLoop两次时会发生什么？您松开了其中一个“触发器”。时间间隔越小，意味着测试失败的可能性就越高，这是因为worker.trigger()之前唤醒多个连续的workLoop调用的可能性较高。请注意，没有任何东西可以保证workLoop在worker.trigger()之后但在另一个worker.trigger()发生之前获得锁定，即使这些调用接连发生（即不是并行发生）。这由操作系统调度程序控制，我们无法对其进行控制。

无论如何，核心问题是将ready_ = true设置两次会丢失信息。不同于将整数递增两次。因此，最简单的解决方案是将bool替换为int，并使用== 0检查进行inc / dec。此解决方案也称为信号量。更高级（可能更好，尤其是当您需要将一些数据传递给工作程序时）的方法是使用（有界？）线程安全队列。这取决于您要实现的目标。

顺便说一句1：除了stop()函数（和start()但这并不重要）以外，您所有的读取和更新都在该锁下进行。我建议您也将stop()也置于锁定状态（因为无论如何都很少调用它），并将原子变成非原子。目前，原子有不必要的开销。

顺便说一句2：我建议不要使用thread.detach()。您应该将std::thread对象存储在TriggeredWorker上，并添加对stop执行join的析构函数。这些不是独立的生物，因此如果没有detach()，您的代码将变得更加安全（一个人永远都应该死掉）。

高负载下c ++线程工作者失败

如何解决高负载下c ++线程工作者失败

解决方法

相关推荐