Synchonizing queue with mutex

Question

I am trying to syncronize the queue between threads using mutex. I have 2 mil random integers in input.txt file. Main thread reads integers then puts them in queue. Other threads are simultaneously taking number from the queue and are doing stuff with it. I succeded to get accurate result each time I run program, but I think there is more efficient way to complete this task Code:

#include <iostream>
#include <thread>
#include <vector>
#include <utility>
#include <string>
#include <future>
#include <queue>
#include <random>
#include <chrono>
#include <sstream>
#include <fstream>
#include <iomanip>

template <typename T>
class SafeQ
{
private:
    std::queue<T> q;
    std::mutex mtx;
    std::condition_variable cv;
public:
    void push(T const& value)
    {
        std::unique_lock<std::mutex> lock(mtx);
        q.push(value);
        lock.unlock();
        cv.notify_one();
    }

    void wait_and_pop(T& value)
    {
        std::unique_lock<std::mutex> lock(mtx);
        while(q.empty())
        {
            cv.wait(lock);
        }
        value = q.front();
        q.pop();
        lock.unlock();
    }

    bool try_pop(T& popped_value)
    {
        
        std::unique_lock<std::mutex> lock(mtx);
        if(q.empty())
        {
            return false;
        }
        
        popped_value = q.front();
        q.pop();
        return true;
    }

    size_t size()
    {
        std::unique_lock<std::mutex> lock(mtx);
        size_t s = q.size();
        return s;
    }

    bool empty()
    {
        std::lock_guard<std::mutex> lock(mtx);
        return q.empty();
    }
};
int producer(std::string filename, SafeQ<int> &q, bool &produce_finished) 
{
    int produced_count = 0;
    std::ifstream ifs(filename);

    while(!ifs.eof())
    {
        int num;
        ifs >> num;
        q.push(num);
        produced_count++;
    }
    ifs.close();
    produce_finished = true;
    return produced_count;
}
void workerThread(SafeQ<int> &q, int &primes, int &nonprimes, double &sum, int &consumed_count, std::vector<int> &number_counts, bool produce_finished, std::mutex &m)
{
    while(!q.empty() && !produce_finished) 
    {
        int num;
        q.wait_and_pop(num);

        std::unique_lock<std::mutex> lock(m);

        consumed_count++;
        if(kernel(num) == 1) {
            primes++;
        }
        else {
            nonprimes++;
        }
        number_counts[num%10]++;
        sum+=num;
        lock.unlock();
        }
}
int main(int argc, char **argv)
{
    int num_threads = std::thread::hardware_concurrency();
    std::mutex m;
    bool no_exec_times = false, only_exec_times = false;;
    std::string filename = "input.txt";
    parse_args(argc, argv, num_threads, filename, no_exec_times, only_exec_times);

    int consumed_count = 0;
    bool produce_finished = false;
    int primes = 0, nonprimes = 0, count = 0;
    double mean = 0.0, sum = 0.0;
    std::vector<int> number_counts(10, 0);

    std::vector<std::thread> workers;
    SafeQ<int> q;

    std::future<int> f = std::async(std::launch::async, producer, filename, std::ref(q), std::ref(produce_finished));

    for(int i = 0; i < num_threads; i++)
    {
        workers.push_back(std::thread(workerThread, std::ref(q), std::ref(primes), std::ref(nonprimes), std::ref(sum), std::ref(consumed_count), std::ref(number_counts), produce_finished, std::ref(m)));
    }

    for (int i = 0; i < num_threads; i++)
    {
        workers[i].join();
    }

    int produced_count = f.get();

    mean = sum/consumed_count;

    if ( produced_count != consumed_count ) {
         std::cout << "[error]: produced_count (" << produced_count << ") != consumed_count (" << consumed_count << ")." <<  std::endl;
    }

    print_output(num_threads, primes, nonprimes, mean, number_counts);

    return 0;
}

Is there way to upgrade this code using mutex and avoid using atomic?

bool kernel(int number)
{
    bool prime = true;
    for (int i = 2; i <= number / 2; ++i)
    {
        if (number % i == 0)
        {
            prime = false;
        }
}
    return prime;
};

Answer 1

Optimize the algorithm before throwing more threads at it

Your kernel is a very inefficient primality checker. It doesn't break out of the loop early, it checks against more divisors than necessary (you only need to go up to the square root of number), and if it is called twice for the same number, it repeats the whole calculation.

The producer might not start before the consumers

There is no guarantee in which order threads are scheduled. The producer thread might not actually run before the consumers have started. If that is the case, the consumers will see an empty queue, and think that the work is already finished. Even if the producer thread starts earlier, it might take some time to open the file and read the first number.

Even if you defer creating the consumer threads until there is at least one item in the queue, consider that the consumer threads might consumer faster than the producer can produce. In that case, it could still happen that the queue appears empty at some point, even if the producer is not yet done.

You wanted to guard against this by using the produce_finished variable. However, in the consumer thread you write:

while(!q.empty() && !produce_finished)

If q.empty() is true then the condition will be false, and thus it will exit. You probably meant:

while(!q.empty() || !produce_finished)

However, these checks are done without a lock being held. That means that it could do q.empty() when the queue was empty, then the producer produces the last item, so q.empty() is no longer true, but it did set produce_finished to true, and then the first thread checks produce_finished. It will then think that it doesn't have to do anything, but that is wrong.

First, by the time q.empty() returns, the return value might no longer be true. Since you cannot rely on what SafeQ::empty() returns, you should not use it at all, and just remove it from SafeQ. (The same goes for SafeQ::size().) Instead, you could use try_pop(), or perhaps even better is a wait_and_pop() that returns a std::optional<T>. The produce_finished flag should then be managed by SafeQ, and have a member function to set it.

Avoid needing a mutex for primes and nonprimes

You can having the workers lock a mutex when counting by using atomic variables.

Alternatively, consider keeping per-thread counters, and after the threads have finished, sum the counters to get the final numbers. This avoids needing any kind of locking. You might want to make sure that the per-thread counters don't share cachelines by aligning them using std::hardware_destructive_interference_size.

Missing error checking

Reading from a file can go wrong at any point. If an error occurs, eof() will not be true, but ifs >> num will not work as you expect, and the producer will go into an infinite loop. What you have to do is loop while ifs.good() is true, and after the loop check that ifs.eof() is true. If the latter is not the case, you didn't read until the end, so something has definitely gone wrong. In that case, make sure you print an error message to std::cerr and exit the program with EXIT_FAILURE.