Использование boost для превращения одного потока в многопоточный - PullRequest
Новая
       53

Использование boost для превращения одного потока в многопоточный

0 голосов
/ 30 июня 2019

Я пытаюсь превратить код из одного потока в многопоточный (например, создать 6 потоков вместо 1), следя за тем, чтобы все они начинались и заканчивались без какого-либо вмешательства друг в друга.Каков был бы способ сделать это?Могу ли я просто сделать цикл for, который создает поток, пока я <6?И просто добавить класс мьютекса с помощью lock () и unlock ()? </p> 

#include <iostream>
#include <boost/thread.hpp>
#include <boost/date_time.hpp>

void workerFunc()
{
    boost::posix_time::seconds workTime(3);

    std::cout << "Worker: running" << std::endl;

    // Pretend to do something useful...
    boost::this_thread::sleep(workTime);

    std::cout << "Worker: finished" << std::endl;
}

int main(int argc, char* argv[])
{
    std::cout << "main: startup" << std::endl;

    boost::thread workerThread(workerFunc);

    std::cout << "main: waiting for thread" << std::endl;

    workerThread.join();

    std::cout << "main: done" << std::endl;

    system("pause");
    return 0;
}

Ответы [ 3 ]

2 голосов
/ 30 июня 2019

Да, это возможно.Поскольку вы не хотите никаких помех между ними, предоставьте им уникальные данные для работы, чтобы вам не нужно было синхронизировать доступ к этим данным с std::mutex или сделать его std::atomic.Чтобы дополнительно минимизировать помехи между потоками, выровняйте данные в соответствии с std :: hardware_destructive_interference_size .

Вы можете использовать boost :: thread :: hardware_concurrency () , чтобы получитьколичество аппаратных потоков, доступных в текущей системе, чтобы вам не приходилось жестко задавать количество потоков для запуска.

Передача ссылок на поток может быть сделана с помощью std::ref (иначе поток будетполучить ссылку на копию данных).

Здесь я создаю std::list потоков и std::vector данных для работы. 

#include <cstdint> // std::int64_t
#include <iostream>
#include <list>
#include <new> // std::hardware_destructive_interference_size
#include <vector>
#include <boost/thread.hpp>

unsigned hardware_concurrency() {
    unsigned rv = boost::thread::hardware_concurrency();
    if(rv == 0) rv = 1; // fallback if hardware_concurrency returned 0
    return rv;
}

// if you don't have hardware_destructive_interference_size, use something like this
// instead:
//struct alignas(64) data {
struct alignas(std::hardware_destructive_interference_size) data {
    std::int64_t x;
};

void workerFunc(data& d) {
    // work on the supplied data
    for(int i = 0; i < 1024*1024-1; ++i) d.x -= i;
    for(int i = 0; i < 1024*1024*1024-1; ++i) d.x += i;
}

int main() {
    std::cout << "main: startup" << std::endl;

    size_t number_of_threads = hardware_concurrency();
    std::list<boost::thread> threads;
    std::vector<data> dataset(number_of_threads);

    // create the threads 
    for(size_t idx = 0; idx < number_of_threads; ++idx)
        threads.emplace_back(workerFunc, std::ref(dataset[idx]));

    std::cout << "main: waiting for threads" << std::endl;

    // join all threads
    for(auto& th : threads) th.join();
    // display results
    for(const data& d : dataset) std::cout << d.x << "\n";

    std::cout << "main: done" << std::endl;
}

Если вы используетеC ++ 11 (или более поздняя версия), вместо этого я предлагаю использовать std::thread.

1 голос
/ 30 июня 2019

Запуск и остановка группы потоков Boost 

std::vector<boost::thread> threads;
for (int i = 0; i < numberOfThreads; ++i) {
  boost::thread t(workerFunc);
  threads.push_back(std::move(t));
}

for (auto& t : threads) {
  t.join();
}

Имейте в виду, что join() не завершает потоки, а только ожидает их завершения.

Синхронизация

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

std::queue<int> q;
std::mutex q_mu;

void workerFunc1() {
  // ...
  {
    std::lock_guard<std::mutex> guard(q_mu);
    q.push(foo);
  } // lock guard goes out of scope and automatically unlocks q_mu
  // ...
}

void workerFunc2() {
  // ...
  {
    std::lock_guard<std::mutex> guard(q_mu);
    foo = q.pop();
  } // lock guard goes out of scope and automatically unlocks q_mu
  // ...
}

Это предотвращает неопределенное поведение, такое как чтение элемента из очереди, которая не была записана полностью.Будьте осторожны - гонки данных могут привести к сбою вашей программы или повреждению ваших данных.Я часто использую такие инструменты, как Thread Sanitizer или Helgrind , чтобы убедиться, что я ничего не пропустил.Если вы хотите передать результаты только в основную программу, но вам не нужно обмениваться данными между потоками, вы можете рассмотреть возможность использования std::promise и std::future.

0 голосов
/ 30 июня 2019

Да, создание новых потоков может быть выполнено с помощью простого цикла.Однако вам придется помнить о нескольких вещах:

  1. Если потоки будут работать с общими данными, их нужно будет защитить мьютексами, атомами или каким-либо другим способом, чтобы избежать гонки данных и неопределенныхповедение (имейте в виду, что даже примитивные типы, такие как int должны быть обернуты атомным или мьютексом в соответствии со стандартом).
  2. Вы должны будете убедиться, что в конце концов либо вызовете join()или detach() в каждом порождаемом потоке, прежде чем его объект выходит из области видимости, чтобы предотвратить его внезапное завершение.
  3. Лучше всего выполнить некоторые вычисления в главном потоке, ожидая, пока рабочие потоки не будут использовать это время эффективно вместовпустую.
  4. Как правило, вы хотите порождать на 1 поток меньше, чем общее количество потоков, которое вы хотите, так как программа запускается с одним потоком по умолчанию (основной поток).
...