C++ 11 中的线程处理答案

【问题标题】：Threads handling in C++ 11C++ 11 中的线程处理
【发布时间】：2015-06-26 12:16:29
【问题描述】：

我有一个基类，它管理线程函数和派生函数，它可以设置它的函数在该基类的帮助下并发运行。但是在我模拟这些功能的快速启动/停止的测试期间，我的程序崩溃了。似乎，问题出在互斥体上，但我可能是错的。有什么问题？

class ThreadedBase
{
public:
    ThreadedBase(){}
    virtual ~ThreadedBase() {
        for (int i = 0; i < m_threadedTasks.size(); ++i) {
            *m_threadedTasks[i]->run = false;
            m_threadedTasks[i]->thread->join();
        }

        while (m_threadedTasks.size()) {
            m_threadedTasks.erase(m_threadedTasks.begin());
        }
    }

    void addTask(std::function<void()> f) {
        static int i = 0;
        m_threadedTasks.insert({ i++, new ThreadedTask(f)});
    }

    void startN(int n) {
        std::lock_guard<std::mutex> lock(m_threadedTasks[n]->mtx);
        std::cout << "StartN\n";
        if (*m_threadedTasks[n]->run == true) return;
        *m_threadedTasks[n]->run = true;
         m_threadedTasks[n]->thread = new std::thread((std::mem_fn(&ThreadedBase::run)), this, n);
    }

    void stopN(int n) {
        std::lock_guard<std::mutex> lock(m_threadedTasks[n]->mtx);
        std::cout << "StopN\n";
        *m_threadedTasks[n]->run = false;
    }

private:
    struct ThreadedTask
    {
        ThreadedTask(std::function<void()> f) : function(f), thread(nullptr), run(new bool(false)) {}
        ~ThreadedTask() {
            thread->join();
            delete thread;
            delete run;
        }
        std::function<void()> function;
        std::thread *thread;
        mutable std::mutex mtx;
        bool *run;
    };
    std::unordered_map<int, ThreadedTask*> m_threadedTasks;

    void run(int n) {
        while (*m_threadedTasks[n]->run) {
            std::lock_guard<std::mutex> lock(m_threadedTasks[n]->mtx);
            m_threadedTasks[n]->function();
            std::cout << "yet another loop\n";
        } 
        // todo: erase if needed
        std::lock_guard<std::mutex> lock(m_threadedTasks[n]->mtx);
        *m_threadedTasks[n]->run = false;
        delete m_threadedTasks[n]->thread;
        m_threadedTasks[n]->thread = nullptr;
    }
};

class Test : public ThreadedBase
{
public:
    Test() : ThreadedBase() {
        addTask(std::bind(std::mem_fn(&Test::threadedBlock1), this));
        addTask(std::bind(std::mem_fn(&Test::threadedBlock2), this));
    };
    void threadedBlock1() {
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
        std::cout << "Task1 finished ";
    }
    void threadedBlock2() {
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
        std::cout << "Task2 finished ";
    }
    ~Test() { std::cout << "Deleting\n"; }

    void startN(int n) {
        ThreadedBase::startN(n);
    }
    void stopN(int n) {
        ThreadedBase::stopN(n);
    }
};

int main()
{
    Test *t = new Test();

    srand(time(NULL));
    while (true) {
        (0 + rand() % 2 == 0) ? t->startN(0) : t->stopN(0);
    }

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

    return 0;
}

【问题讨论】：

加入构造函数可能是个坏主意。
在ThreadedTask类中，为什么会有一个指向bool的指针？为什么你有一个指向std::thread 的指针？现在你很少需要指针，尤其是对于单一的原始类型（比如你的run 变量）。
我也没有看到您在任何地方在ThreadedTask 中构造互斥锁？在调用std::thread 构造函数的addTask 时，您不应该需要std::mem_fn。
由于stopN没有加入线程，你可以在run的循环和它的“todo”块之间调用startN，所以在那个块m_threadedTasks[n]可以引用新创建的对象而不是应该停止的对象。（你有很多内存泄漏。）
std::thread 是 moveable，这意味着你可以做例如std::thread t; t = std::thread(...); t.join();

标签： c++ multithreading c++11 concurrency thread-safety

【解决方案1】：

感谢 Joachim Pileborg，我使这段代码正常工作。现在如下：

class ThreadedBase
{
public:
    ThreadedBase(){}
    virtual ~ThreadedBase() {
        for (int i = 0; i < m_threadedTasks.size(); ++i) {
            m_threadedTasks[i]->run = false;
            m_threadedTasks[i]->thread.join();
        }

        while (m_threadedTasks.size()) {
            m_threadedTasks.erase(m_threadedTasks.begin());
        }
    }

    void addTask(std::function<void()> f) {
        static int i = 0;
        m_threadedTasks.insert({ i++, new ThreadedTask(f) });
    }

    void startN(int n) {
        std::lock_guard<std::mutex> lock(m_threadedTasks[n]->mtx);
        if (m_threadedTasks[n]->run == true) return;
        m_threadedTasks[n]->run = true;
        m_threadedTasks[n]->thread = std::thread(&ThreadedBase::run, this, n);
    }

    void stopN(int n) {
        m_threadedTasks[n]->run = false;
        if (m_threadedTasks[n]->thread.joinable())
            m_threadedTasks[n]->thread.join();
    }

private:
    struct ThreadedTask
    {
        ThreadedTask(std::function<void()> f) : function(f), thread(), mtx(), run(false) {}
        ~ThreadedTask() {
            thread.join();
        }
        std::function<void()> function;
        std::thread thread;
        mutable std::mutex mtx;
        bool run;
    };
    std::unordered_map<int, ThreadedTask*> m_threadedTasks;

    void run(int n) {
        while (m_threadedTasks[n]->run) {
            std::lock_guard<std::mutex> lock(m_threadedTasks[n]->mtx);
            m_threadedTasks[n]->function();
            std::cout << "yet another loop\n";
        }
        // todo: erase if needed
        m_threadedTasks[n]->run = false;
    }
};

class Test : public ThreadedBase
{
public:
    Test() : ThreadedBase() {
        addTask(std::bind(&Test::threadedBlock1, this));
        addTask(std::bind(&Test::threadedBlock2, this));
    };
    void threadedBlock1() {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        std::cout << "Task1 finished ";
    }
    void threadedBlock2() {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        std::cout << "Task2 finished ";
    }
    ~Test() { std::cout << "Deleting\n"; }

    void startN(int n) {
        ThreadedBase::startN(n);
    }
    void stopN(int n) {
        ThreadedBase::stopN(n);
    }
};

int main()
{
    Test *t = new Test();

    srand(time(NULL));
    while (true) {
        (0 + rand() % 2 == 0) ? t->startN(0) : t->stopN(0);
    }

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

    return 0;
}

【讨论】：