Объединение потоков в С++ 11

Ответ 1

Вы можете использовать библиотеку пула потоков С++, https://github.com/vit-vit/ctpl.

Затем код, который вы написали, может быть заменен следующим

#include <ctpl.h>  // or <ctpl_stl.h> if ou do not have Boost library

int main (int argc, char *argv[]) {
    ctpl::thread_pool p(2 /* two threads in the pool */);
    int arr[4] = {0};
    std::vector<std::future<void>> results(4);
    for (int i = 0; i < 8; ++i) { // for 8 iterations,
        for (int j = 0; j < 4; ++j) {
            results[j] = p.push([&arr, j](int){ arr[j] +=2; });
        }
        for (int j = 0; j < 4; ++j) {
            results[j].get();
        }
        arr[4] = std::min_element(arr, arr + 4);
    }
}

Вы получите нужное количество потоков и не будете создавать и удалять их снова и снова на итерациях.

Ответ 2

Пул потоков означает, что все ваши потоки запущены, все время – другими словами, функция потока никогда не возвращается. Чтобы дать потокам что-то значимое, вам нужно создать систему межпоточной связи, как для того, чтобы сообщить потоку, что там что-то делать, так и для передачи фактических рабочих данных.

Как правило, это будет связано с какой-то параллельной структурой данных, и каждый поток, по-видимому, будет спать на какой-то переменной условия, которая будет уведомляться, когда это будет сделано. После получения уведомления один или несколько потоков пробуждаются, восстанавливают задачу из параллельной структуры данных, обрабатывают ее и сохраняют результат аналогичным образом.

Затем поток продолжал проверять, будет ли еще больше работать, а если не вернуться в режим сна.

Результат состоит в том, что вам нужно все это самостоятельно разработать, поскольку нет естественного понятия "работа", которое универсально применимо. Это довольно много работы, и есть некоторые тонкие проблемы, которые вам нужно сделать правильно. (Вы можете программировать в Go, если вам нравится система, которая заботится о управлении потоками для вас за кулисами.)

Ответ 3

Это скопировано из моего ответа в другой очень похожий пост, надеюсь, это поможет:

1) Начните с максимального количества потоков, которое может поддерживать система:

int Num_Threads =  thread::hardware_concurrency();

2) Для эффективной реализации пула потоков, как только потоки создаются в соответствии с Num_Threads, лучше не создавать новые и не уничтожать старые (путем объединения). Это приведет к снижению производительности, может даже заставить ваше приложение работать медленнее, чем серийная версия.

Каждый поток С++ 11 должен работать в своей функции с бесконечным циклом, постоянно ожидая, когда новые задачи будут получены и запущены.

Вот как прикрепить такую функцию к пулу потоков:

int Num_Threads = thread::hardware_concurrency();
vector<thread> Pool;
for(int ii = 0; ii < Num_Threads; ii++)
{  Pool.push_back(thread(Infinite_loop_function));}

3) Функция Infinite_loop_function

Это цикл "while (true)", ожидающий очередь задач

void The_Pool:: Infinite_loop_function()
{
    while(true)
    {
        {
            unique_lock<mutex> lock(Queue_Mutex);

            condition.wait(lock, []{return !Queue.empty() || therminate_pool});
            Job = Queue.front();
            Queue.pop();
        }
        Job(); // function<void()> type
    }
};

4) Сделайте функцию, чтобы добавить работу в вашу очередь

void The_Pool:: Add_Job(function<void()> New_Job)
{
    {
        unique_lock<mutex> lock(Queue_Mutex);
        Queue.push(New_Job);
    }
    condition.notify_one();
}

5) Привязать произвольную функцию к вашей очереди

Pool_Obj.Add_Job(std::bind(&Some_Class::Some_Method, &Some_object));

После того, как вы интегрируете эти ингредиенты, у вас будет собственный динамический пул потоков. Эти потоки всегда работают, ожидая выполнения задания.

Я извиняюсь, если есть какие-то синтаксические ошибки, я набрал этот код, и у меня плохая память. Извините, что я не могу предоставить вам полный код пула потоков, который нарушил бы мою работу.

Редактировать: чтобы завершить пул, вызовите метод shutdown():

XXXX::shutdown(){
{   unique_lock<mutex> lock(threadpool_mutex);
    terminate_pool = true;} // use this flag in condition.wait

condition.notify_all(); // wake up all threads.

// Join all threads.
for(std::thread &every_thread : thread_vector)
{   every_thread.join();}

thread_vector.empty();  
stopped = true; // use this flag in destructor, if not set, call shutdown() 

}

Ответ 4

В потоковом потоке находится поток потоков, привязанных к функции, работающей как цикл событий. Эти потоки будут бесконечно ждать выполнения задачи или собственного завершения.

Задача threadpool заключается в предоставлении интерфейса для отправки заданий, определения (и, возможно, изменения) политики запуска этих заданий (правил планирования, создания потоков, размера пула) и мониторинга состояния потоков и связанных с ними ресурсов.

Итак, для универсального пула нужно сначала определить, что такое задача, как она запущена, прервана, каков результат (см. понятие обещания и будущего для этого вопроса), какие события будут иметь потоки должны реагировать на то, как они будут обрабатывать их, как эти события должны быть дискриминированы из тех, которые выполняются задачами. Это может стать довольно сложным, как вы можете видеть, и наложить ограничения на то, как потоки будут работать, поскольку решение становится все более и более вовлеченным.

Текущая утилита для обработки событий - это довольно баребоны (*): примитивы, такие как мьютексы, переменные условий и несколько абстракций поверх этого (блокировки, барьеры). Но в некоторых случаях эти абстракции могут оказаться непригодными (см. Ссылку question), и нужно вернуться к использованию примитивов.

Необходимо также справиться с другими проблемами:

• Сигнал
• я/о
• аппаратное обеспечение (сродство процессора, гетерогенная настройка)

Как они будут воспроизводиться в вашей настройке?

Этот ответ к аналогичному вопросу указывает на существующую реализацию, предназначенную для boost и stl.

Я предложил очень грубую реализацию потока для другого вопроса, который не затрагивает многие проблемы, описанные выше. Возможно, вы захотите его создать. Вы также можете захотеть взглянуть на существующие рамки на других языках, чтобы найти вдохновение.

_{(*) Я не вижу в этом проблемы, как бы наоборот. Я думаю, что сам дух С++ унаследован от C.}

Ответ 5

Что-то вроде этого может помочь (взято из рабочего приложения).

#include <memory>
#include <boost/asio.hpp>
#include <boost/thread.hpp>

struct thread_pool {
  typedef std::unique_ptr<boost::asio::io_service::work> asio_worker;

  thread_pool(int threads) :service(), service_worker(new asio_worker::element_type(service)) {
    for (int i = 0; i < threads; ++i) {
      auto worker = [this] { return service.run(); };
      grp.add_thread(new boost::thread(worker));
    }
  }

  template<class F>
  void enqueue(F f) {
    service.post(f);
  }

  ~thread_pool() {
    service_worker.reset();
    grp.join_all();
    service.stop();
  }

private:
  boost::asio::io_service service;
  asio_worker service_worker;
  boost::thread_group grp;
};

Вы можете использовать его следующим образом:

thread_pool pool(2);

pool.enqueue([] {
  std::cout << "Hello from Task 1\n";
});

pool.enqueue([] {
  std::cout << "Hello from Task 2\n";
});

Имейте в виду, что переосмысление эффективного механизма асинхронного очередей не является тривиальным.

Boost:: asio:: io_service - очень эффективная реализация или фактически представляет собой набор оболочек, специфичных для платформы (например, он переносит порты завершения ввода/вывода в Windows).

Ответ 6

Изменить: теперь требуется С++ 17 и понятия. (По состоянию на 9/12/16 достаточно только g++ 6.0+.)

Вывод шаблона намного более точен из-за этого, поэтому он стоит усилий для получения более нового компилятора. Я еще не нашел функцию, которая требует явных аргументов шаблона.

Теперь он также принимает любой подходящий вызываемый объект (и все еще статично видоизменяется!!!).

Теперь он также включает дополнительный пул потоков нитей с зеленым потоком, используя тот же API. Однако этот класс является POSIX. Он использует API ucontext_t для переключения задач в пользовательском пространстве.

Я создал для этого простую библиотеку. Ниже приведен пример использования. (Я отвечаю на это, потому что это была одна из вещей, которые я нашел, прежде чем я решил, что нужно написать ее сам.)

bool is_prime(int n){
  // Determine if n is prime.
}

int main(){
  thread_pool pool(8); // 8 threads

  list<future<bool>> results;
  for(int n = 2;n < 10000;n++){
    // Submit a job to the pool.
    results.emplace_back(pool.async(is_prime, n));
  }

  int n = 2;
  for(auto i = results.begin();i != results.end();i++, n++){
    // i is an iterator pointing to a future representing the result of is_prime(n)
    cout << n << " ";
    bool prime = i->get(); // Wait for the task is_prime(n) to finish and get the result.
    if(prime)
      cout << "is prime";
    else
      cout << "is not prime";
    cout << endl;
  }  
}

Вы можете передать async любую функцию с любым возвращаемым значением (или void) и любыми (или нет) аргументами, и оно вернет соответствующий std::future. Чтобы получить результат (или просто дождаться завершения задачи), вы вызываете get() в будущем.

Здесь github: https://github.com/Tyler-Hardin/thread_pool.

Ответ 7

Это еще одна реализация пула потоков, которая очень проста, легка для понимания и использования, использует только стандартную библиотеку С++ 11 и может быть просмотрена или изменена для ваших нужд, должна быть хорошей отправной точкой, если вы хотите начать использовать поток бассейны:

https://github.com/progschj/ThreadPool

Ответ 8

Follwoing [PhD EcE](https://stackoverflow.com/users/3818417/phd-ece) suggestion, I implemented the thread pool:

function_pool.h

#pragma once
#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <cassert>

class Function_pool
{

private:
    std::queue<std::function<void()>> m_function_queue;
    std::mutex m_lock;
    std::condition_variable m_data_condition;
    std::atomic<bool> m_accept_functions;

public:

    Function_pool();
    ~Function_pool();
    void push(std::function<void()> func);
    void done();
    void infinite_loop_func();
};

function_pool.cpp

#include "function_pool.h"

Function_pool::Function_pool() : m_function_queue(), m_lock(), m_data_condition(), m_accept_functions(true)
{
}

Function_pool::~Function_pool()
{
}

void Function_pool::push(std::function<void()> func)
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_function_queue.push(func);
    // when we send the notification immediately, the consumer will try to get the lock , so unlock asap
    lock.unlock();
    m_data_condition.notify_one();
}

void Function_pool::done()
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_accept_functions = false;
    lock.unlock();
    // when we send the notification immediately, the consumer will try to get the lock , so unlock asap
    m_data_condition.notify_all();
    //notify all waiting threads.
}

void Function_pool::infinite_loop_func()
{
    std::function<void()> func;
    while (true)
    {
        {
            std::unique_lock<std::mutex> lock(m_lock);
            m_data_condition.wait(lock, [this]() {return !m_function_queue.empty() || !m_accept_functions; });
            if (!m_accept_functions && m_function_queue.empty())
            {
                //lock will be release automatically.
                //finish the thread loop and let it join in the main thread.
                return;
            }
            func = m_function_queue.front();
            m_function_queue.pop();
            //release the lock
        }
        func();
    }
}

main.cpp

#include "function_pool.h"
#include <string>
#include <iostream>
#include <mutex>
#include <functional>
#include <thread>
#include <vector>

Function_pool func_pool;

class quit_worker_exception : public std::exception {};

void example_function()
{
    std::cout << "bla" << std::endl;
}

int main()
{
    std::cout << "stating operation" << std::endl;
    int num_threads = std::thread::hardware_concurrency();
    std::cout << "number of threads = " << num_threads << std::endl;
    std::vector<std::thread> thread_pool;
    for (int i = 0; i < num_threads; i++)
    {
        thread_pool.push_back(std::thread(&Function_pool::infinite_loop_func, &func_pool));
    }

    //here we should send our functions
    for (int i = 0; i < 50; i++)
    {
        func_pool.push(example_function);
    }
    func_pool.done();
    for (unsigned int i = 0; i < thread_pool.size(); i++)
    {
        thread_pool.at(i).join();
    }
}

Ответ 9

Пул потоков без зависимостей вне STL вполне возможен. Недавно я написал небольшую библиотеку потоков только для заголовков, чтобы решить ту же проблему. Он поддерживает динамическое изменение размера пула (изменение количества рабочих во время выполнения), ожидание, остановку, приостановку, возобновление и т.д. Я надеюсь, что вы найдете это полезным.

Ответ 10

Вы можете использовать thread_pool из библиотеки Boost:

void my_task(){...}

int main(){
    int threadNumbers = thread::hardware_concurrency();
    boost::asio::thread_pool pool(threadNumbers);

    // Submit a function to the pool.
    boost::asio::post(pool, my_task);

    // Submit a lambda object to the pool.
    boost::asio::post(pool, []() {
      ...
    });
}

Вы также можете использовать пул потоков из сообщества с открытым исходным кодом:

void first_task() {...}    
void second_task() {...}

int main(){
    int threadNumbers = thread::hardware_concurrency();
    pool tp(threadNumbers);

    // Add some tasks to the pool.
    tp.schedule(&first_task);
    tp.schedule(&second_task);
}