C++ - Многопоточность - Связь между потоками

Question

#include <iostream> 
#include <thread> 
#include <condition_variable> 
#include <queue> 
#include <cstdlib> 
#include <chrono> 
#include <ctime> 
#include <random> 

using namespace std; 

//counts every number that is added to the queue 
static long long producer_count = 0; 
//counts every number that is taken out of the queue 
static long long consumer_count = 0; 

void generateNumbers(queue<int> & numbers, condition_variable & cv, mutex & m, bool & workdone){ 
    while(!workdone) { 
     unique_lock<std::mutex> lk(m); 
     int rndNum = rand() % 100; 
     numbers.push(rndNum); 
     producer_count++; 
     cv.notify_one(); 
    } 
} 

void work(queue<int> & numbers, condition_variable & cv, mutex & m, bool & workdone) { 
    while(!workdone) { 
     unique_lock<std::mutex> lk(m); 
     cv.wait(lk); 
     cout << numbers.front() << endl; 
     numbers.pop(); 
     consumer_count++; 

    } 
} 

int main() { 
    condition_variable cv; 
    mutex m; 
    bool workdone = false; 
    queue<int> numbers; 

    //start threads 
    thread producer(generateNumbers, ref(numbers), ref(cv), ref(m),  ref(workdone)); 
    thread consumer(work, ref(numbers), ref(cv), ref(m), ref(workdone)); 

    //wait for 3 seconds, then join the threads 
    this_thread::sleep_for(std::chrono::seconds(3)); 
    workdone = true; 

    producer.join(); 
    consumer.join(); 

    //output the counters 
    cout << producer_count << endl; 
    cout << consumer_count << endl; 

    return 0; 
} 

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

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

Я увеличил счетчик для каждого числа, которое добавлено в очередь, и другого счетчика для каждого числа, которое выведено из очереди.

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

Может кто-нибудь объяснить мне, почему существует такая огромная разница? Должен ли я добавить другую переменную условия, чтобы потоки производителей также ожидали поток потребителей? Спасибо!

Answer 1

Нет необходимости на второй std::condition_variable, просто используйте одно из них. Как уже упоминалось, вы должны использовать std::atomic<bool> вместо простого bool. Но я должен признать, что g ++ с -O3 не оптимизирует его.

#include <iostream> 
#include <thread> 
#include <condition_variable> 
#include <queue> 
#include <cstdlib> 
#include <chrono> 
#include <ctime> 
#include <random> 
#include <atomic> 

//counts every number that is added to the queue 
static long long producer_count = 0; 
//counts every number that is taken out of the queue 
static long long consumer_count = 0; 

void generateNumbers(std::queue<int> & numbers, std::condition_variable & cv, std::mutex & m, std::atomic<bool> & workdone) 
{ 
    while(!workdone.load()) 
    { 
     std::unique_lock<std::mutex> lk(m); 
     int rndNum = rand() % 100; 
     numbers.push(rndNum); 
     producer_count++; 
     cv.notify_one(); // Notify worker 
     cv.wait(lk); // Wait for worker to complete 
    } 
} 

void work(std::queue<int> & numbers, std::condition_variable & cv, std::mutex & m, std::atomic<bool> & workdone) 
{ 
    while(!workdone.load()) 
    { 
     std::unique_lock<std::mutex> lk(m); 
     cv.notify_one(); // Notify generator (placed here to avoid waiting for the lock) 
     cv.wait(lk); // Wait for the generator to complete 
     std::cout << numbers.front() << std::endl; 
     numbers.pop(); 
     consumer_count++; 
    } 
} 

int main() { 
    std::condition_variable cv; 
    std::mutex m; 
    std::atomic<bool> workdone(false); 
    std::queue<int> numbers; 

    //start threads 
    std::thread producer(generateNumbers, std::ref(numbers), std::ref(cv), std::ref(m), std::ref(workdone)); 
    std::thread consumer(work, std::ref(numbers), std::ref(cv), std::ref(m), std::ref(workdone)); 


    //wait for 3 seconds, then join the threads 
    std::this_thread::sleep_for(std::chrono::seconds(3)); 
    workdone = true; 
    cv.notify_all(); // To prevent dead-lock 

    producer.join(); 
    consumer.join(); 

    //output the counters 
    std::cout << producer_count << std::endl; 
    std::cout << consumer_count << std::endl; 

    return 0; 
} 

EDIT:

Чтобы избежать спорадической ошибки на единицу вы могли бы использовать это:

#include <iostream> 
#include <thread> 
#include <condition_variable> 
#include <queue> 
#include <cstdlib> 
#include <chrono> 
#include <ctime> 
#include <random> 
#include <atomic> 

//counts every number that is added to the queue 
static long long producer_count = 0; 
//counts every number that is taken out of the queue 
static long long consumer_count = 0; 

void generateNumbers(std::queue<int> & numbers, std::condition_variable & cv, std::mutex & m, std::atomic<bool> & workdone) 
{ 
    while(!workdone.load()) 
    { 
     std::unique_lock<std::mutex> lk(m); 
     int rndNum = rand() % 100; 
     numbers.push(rndNum); 
     producer_count++; 
     cv.notify_one(); // Notify worker 
     cv.wait(lk); // Wait for worker to complete 
    } 
} 

void work(std::queue<int> & numbers, std::condition_variable & cv, std::mutex & m, std::atomic<bool> & workdone) 
{ 
    while(!workdone.load() or !numbers.empty()) 
    { 
     std::unique_lock<std::mutex> lk(m); 
     cv.notify_one(); // Notify generator (placed here to avoid waiting for the lock) 
     if (numbers.empty()) 
      cv.wait(lk); // Wait for the generator to complete 
     if (numbers.empty()) 
      continue; 
     std::cout << numbers.front() << std::endl; 
     numbers.pop(); 
     consumer_count++; 
    } 
} 

int main() { 
    std::condition_variable cv; 
    std::mutex m; 
    std::atomic<bool> workdone(false); 
    std::queue<int> numbers; 

    //start threads 
    std::thread producer(generateNumbers, std::ref(numbers), std::ref(cv), std::ref(m), std::ref(workdone)); 
    std::thread consumer(work, std::ref(numbers), std::ref(cv), std::ref(m), std::ref(workdone)); 


    //wait for 3 seconds, then join the threads 
    std::this_thread::sleep_for(std::chrono::seconds(1)); 
    workdone = true; 
    cv.notify_all(); // To prevent dead-lock 

    producer.join(); 
    consumer.join(); 

    //output the counters 
    std::cout << producer_count << std::endl; 
    std::cout << consumer_count << std::endl; 

    return 0; 
} 

Answer 2

Обратите внимание, что этот код может работать неправильно. переменная workdone определяется как обычный bool , и совершенно законно для компилятора предположить, что его можно безопасно оптимизировать, поскольку он никогда не изменяется внутри блока кода.

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

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