C++ 11 中的线程处理

Question

我有一个基础 class，它管理线程函数和派生函数，它可以在该基础 class 的帮助下同时设置它的函数运行。但是在我模拟这些功能的快速 starts/stops 的测试中，我的程序崩溃了。看来，问题出在互斥锁上，但我可能是错的。有什么问题？

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

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

C++ 11 中的线程处理

Threads handling in C++ 11

c++

concurrency

multithreading

thread-safety

c++11