如何处理多个线程的重复启动？

Question

到目前为止，当使用线程时，我总是在我的程序中立即启动它们，然后让它们等待来自主控制线程的通知。

std::vector<std::thread> threads;
for(int i = 0; i != thread_count; ++i) {
    threads.push_back(std::thread(&MyClass::myfunction, this));
}

/* some time later in the code */
for(auto& t : threads) {
    t.join();
}

现在我想从我的控制线程运行的函数中按需启动线程，但我不确定如何处理线程对象及其连接。

以下内容会在每次调用时将一个新的线程对象推送到向量上，这让我觉得不理想：

std::vector<std::thread> threads;
while(accumulating_data) {
    if(buffer_full) {
        threads.push_back(std::thread(&MyClass::myfunction, this));
    }
}

让向量在连续运行的线程上保持不超过最大数量似乎更可取。我也不知道如何在不阻塞控制线程的情况下加入这里的线程。

如果我改为这样做：

// dummy code, in my real code I have a queue of idle IDs
std::vector<std::thread> threads(thread_count);
while(accumulating_data) {
    if(buffer_full) {
        threads[thread_id] = std::thread(&MyClass::myfunction, this);
        if(++thread_id == thread_count) { thread_id = 0; }
    }
}

...我很快就崩溃了，可能是因为我没有加入或重新分配给已经包含 std::thread 对象的向量元素。

关于如何实现按需启动线程而不是让它们等待的目标的任何提示？

更新：

通过引入std::thread.joinable() 检查，我设法让代码在不崩溃的情况下运行。我仍然对如何更优雅地处理这个问题持开放态度，所以我不会把它作为我自己问题的答案：

std::vector<std::thread> threads(thread_count);
while(accumulating_data) {
    if(buffer_full) {
        if(threads[thread_id].joinable()) {
            threads[thread_id].join(); }
        }
        threads[thread_id] = std::thread(&MyClass::myfunction, this);
        if(++thread_id == thread_count) { thread_id = 0; }
    }
}

Answer 1

不确定这是否是你想要的..

#include <thread>
#include <tuple>
#include <vector>
#include <stdexcept>

class ThreadGroup
{
    private:
        std::uint32_t max_threads;
        std::vector<std::tuple<std::thread::native_handle_type, std::thread, bool*>> data;

    public:
        ThreadGroup() : max_threads(std::thread::hardware_concurrency()), data(max_threads) {}
        ThreadGroup(std::uint32_t max_threads) : max_threads(max_threads), data(max_threads) {}
        ~ThreadGroup();

        template<class Function, class... Args>
        std::thread::native_handle_type insert(bool &terminate, Function&& f, Args&&... args);
        bool remove(std::thread::native_handle_type id);
};

ThreadGroup::~ThreadGroup()
{
    for (auto &it : data)
    {
        if (std::get<0>(it))
        {
            if (!*std::get<2>(it))
            {
                std::get<1>(it).detach();
                continue;
            }

            std::get<1>(it).join();
        }
    }
}

template<class Function, class... Args>
std::thread::native_handle_type ThreadGroup::insert(bool &terminate, Function&& f, Args&&... args)
{
    int i = 0;
    for (auto &it : data)
    {
        if (std::get<0>(it) == 0)
        {
            auto &&t = std::thread(std::forward<Function>(f), std::forward(args)...);
            auto &&tup = std::make_tuple(t.native_handle(), std::forward<std::thread>(t), &terminate);
            data[i] = std::move(tup);
            return std::get<0>(data[i]);
        }
        ++i;
    }
    throw std::length_error("Maximum thread limit reached.");
}

bool ThreadGroup::remove(std::thread::native_handle_type id)
{
    for (auto it = data.begin(); it != data.end(); ++it)
    {
        if (std::get<0>(*it) == id)
        {
            if (std::get<1>(*it).joinable() && *std::get<2>(*it))
            {
                std::get<1>(*it).join();
                std::get<0>(*it) = 0;
                std::get<2>(*it) = nullptr;
                //data.erase(it);
                return true;
            }
            std::get<1>(*it).detach();
            std::get<0>(*it) = 0;
            std::get<2>(*it) = nullptr;
            //data.erase(it);
            return false;
        }
    }

    return false;
}

然后我像这样使用它：

#include <chrono>
#include <iostream>
#include <thread>

bool terminate1 = false, terminate2 = false, terminate3 = false, terminate4 = false, terminate5 = false;

void func1()
{
    while(!terminate1)
    {
        std::cout<<"T1 ";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

void func2()
{
    while(!terminate2)
    {
        std::cout<<"T2 ";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

void func3()
{
    while(!terminate3)
    {
        std::cout<<"T3 ";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

void func4()
{
    while(!terminate4)
    {
        std::cout<<"T4 ";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

void func5()
{
    while(!terminate5)
    {
        std::cout<<"T5 ";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

int main()
{
    ThreadGroup group;
    auto id1 = group.insert(terminate1, func1);
    auto id2 = group.insert(terminate2, func2);
    auto id3 = group.insert(terminate3, func3);
    auto id4 = group.insert(terminate4, func4);

    try
    {
        auto id5 = group.insert(terminate5, func5); //limit in my case is 4. inserting 5 will throw max limit exception..
    }
    catch(std::exception &e)
    {
        std::cout<<"\n\n"<<e.what()<<"\n\n";
    }

    std::this_thread::sleep_for(std::chrono::seconds(3));
    terminate1 = true;  //allow the thread to join..
    group.remove(id1);  //joins if the thread is finished..

    std::this_thread::sleep_for(std::chrono::seconds(3));
    group.remove(id2);  //remove another thread (detaches if the thread isn't finished)..

    auto id5 = group.insert(terminate5, func5); //insert a new thread in any of the old slots..
    std::this_thread::sleep_for(std::chrono::seconds(3));
}

Answer 2

通过引入std::thread.joinable() 检查，我设法让代码在不崩溃的情况下运行。我仍然对如何更优雅地处理这个问题持开放态度:)

std::vector<std::thread> threads(thread_count);
while(accumulating_data) {
    if(buffer_full) {
        /* the following check returns false prior to an assignment */
        if(threads[thread_id].joinable()) {
            threads[thread_id].join(); }
        }
        threads[thread_id] = std::thread(&MyClass::myfunction, this);
        if(++thread_id == thread_count) { thread_id = 0; }
    }
}