c++ - 并发数组检查

Question

我既不是 C++ 专家，也不是并发编程专家。但是，我正在实现一个简单的推理算法，它需要检查许多独立的模型。可能的模型数量很多，所以我想并行检查它们。

为了尽可能简单，我将原来的问题转化为一个非常简单的问题:如何确定数组是否包含非零值？一个简单的顺序解决方案将像这样:

bool containsNonZero (int* arr, int len) {
    for (int i = 0; i < len; ++i)
        if (arr[i]) return true;
    return false;
}

(注意:实际上，len 不能放入 int，但在我原来的问题中，没有数组，只有我生成的许多组合不存储。)

但是，我需要一个并行(高效)的实现。有 t = std::thread::hardware_concurrency() 个线程来搜索数组(注意 t <<len。如果 len % t != 0 那么让最后一个线程处理剩余的值将不是问题)。所以第一个线程将搜索从 0 到 len/t 的索引，第二个线程将搜索从 len/t 到 (2 *len)/t 等。最后一个线程将搜索从 ((t-1)*len)/t 到 len 的索引。如果线程发现非零值，所有线程都将停止并返回 true。否则，它们将等待其他线程完成，如果所有线程都同意，将返回 false。

这看起来很容易，但我在网上找不到任何答案。欢迎使用任何 C++ 版本，但我不想依赖任何第三方库。

Answer 1

我尝试扩展 Davide Spataro 的解决方案，以使用 atomic<bool> 解决 atomic_flag 的同步问题，它“与 std::atomic 的所有特化不同，它保证是无锁的”http://en.cppreference.com/w/cpp/atomic/atomic_flag

编辑:与之前的问题无关，但我已经对哪种方法进行了基准测试，令我惊讶的是 atomic<bool> 比 atomic_flag 快了大约 100。

基准测试结果:

num_threads:2
400000001 iterations flag
401386195 iterations flag
atomic_flag : it took 24.1202 seconds. Result: 1
400000001 iterations bool
375842699 iterations bool
atomic<bool>: it took 0.334785 seconds. Result: 1
num_threads:3
229922451 iterations flag
229712046 iterations flag
233333335 iterations flag
atomic_flag : it took 21.5974 seconds. Result: 1
219564626 iterations bool
233333335 iterations bool
196877803 iterations bool
atomic<bool>: it took 0.200942 seconds. Result: 1
num_threads:4
151745683 iterations flag
150000001 iterations flag
148849108 iterations flag
148933269 iterations flag
atomic_flag : it took 18.6651 seconds. Result: 1
150000001 iterations bool
112825220 iterations bool
151838008 iterations bool
112857688 iterations bool
atomic<bool>: it took 0.167048 seconds. Result: 1

基准代码:

#include <thread>
#include <atomic>
#include <vector>
#include <iostream>
#include <algorithm>



template<typename Iterator>
static void any_of_flag(Iterator & begin, Iterator& end, std::atomic_flag & result)
{
    int counter = 0;
    for (auto it = begin; it != end; ++it)
    {
        counter++;
        if (!result.test_and_set() || (*it) != 0)
        {
            result.clear();
            std::cout << counter << " iterations flag\n";
            return;
        }
    }
}
template<typename Iterator>
static void any_of_atomic(Iterator & begin, Iterator& end, std::atomic<bool> & result)
{
    int counter = 0;
    for (auto it = begin; it != end; ++it)
    {
        counter++;
        if (result || (*it) != 0)
        {
            result = true;
            std::cout << counter << " iterations bool\n";
            return;
        }
    }
}

void test_atomic_flag(std::vector<int>& input, int num_threads)
{

    using namespace std::chrono;

    high_resolution_clock::time_point t1 = high_resolution_clock::now();


    size_t chunk_size = input.size() / num_threads;
    std::atomic_flag result = ATOMIC_FLAG_INIT;
    result.test_and_set();

    std::vector<std::thread> threads;
    for (size_t i = 0; i < num_threads; ++i)
    {
        auto & begin = input.begin() + i *chunk_size;
        auto & end = input.begin() + std::min((i + 1) * chunk_size, input.size());
        // had to use lambda in VS 2017
        threads.emplace_back([&begin, &end, &result] {any_of_flag(begin, end, result); });

    }

    for (auto & thread : threads)
        thread.join();

    bool hasNonZero = !result.test_and_set();


    high_resolution_clock::time_point t2 = high_resolution_clock::now();

    duration<double> time_span = duration_cast<duration<double>>(t2 - t1);

    std::cout << "atomic_flag : it took " << time_span.count() << " seconds. Result: " << hasNonZero << std::endl;
}



void test_atomic_bool(std::vector<int>& input, int num_threads)
{

    using namespace std::chrono;

    high_resolution_clock::time_point t1 = high_resolution_clock::now();


    size_t chunk_size = input.size() / num_threads;
    std::atomic<bool> result(false);

    std::vector<std::thread> threads;
    for (size_t i = 0; i < num_threads; ++i)
    {
        auto & begin = input.begin() + i *chunk_size;
        auto & end = input.begin() + std::min((i + 1) * chunk_size, input.size());
        // had to use lambda in VS 2017
        threads.emplace_back([&begin, &end, &result] {any_of_atomic(begin, end, result); });

    }

    for (auto & thread : threads)
        thread.join();

    bool hasNonZero = result;


    high_resolution_clock::time_point t2 = high_resolution_clock::now();

    duration<double> time_span = duration_cast<duration<double>>(t2 - t1);

    std::cout << "atomic<bool>: it took " << time_span.count() << " seconds. Result: " << hasNonZero << std::endl;
}

int main()
{
    std::vector<int> input(1e9, 0);
    input[1e9 - 1e8] = 1;
    for (int num_threads : {2, 3, 4})
    {
        std::cout << "num_threads:" << num_threads << std::endl;
        test_atomic_flag(input, num_threads);
        test_atomic_bool(input, num_threads);
    }

    int q;
    std::cin >> q;
    return 0;
};

旧帖:我在迭代器的常量性和放置线程方面遇到了一些问题，但核心变化，即 atomic_flag 的使用似乎有效。它不会立即停止所有线程，但在最坏的情况下，每次迭代只会停止一个线程(因为每次迭代只有一个线程会知道它应该已经停止，因为清除了标志)。

#include <thread>
#include <atomic>
#include <vector>
#include <iostream>
#include <algorithm>

template<typename Iterator>
static void any_of(Iterator & begin, Iterator& end, std::atomic_flag & result)
{
    for (auto it = begin; it != end; ++it)
    {
        if (!result.test_and_set() || (*it) != 0)
        {
            result.clear();
            return;
        }
    }
}

int main()
{
    int num_threads = 3;
    std::vector<int> input = { 0,0,0,0,0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0,0,0,0,0, 1,0,0,0,0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0,0,0,0,0,0};
    size_t chunk_size = input.size() / num_threads;
    std::atomic_flag result = ATOMIC_FLAG_INIT;
    result.test_and_set();

    std::vector<std::thread> threads;
    for (size_t i = 0; i < num_threads; ++i)
    {
        auto & begin = input.begin() + i *chunk_size;
        auto & end = input.begin() + std::min((i + 1) * chunk_size, input.size());
        // had to use lambda in VS 2017
        threads.emplace_back([&begin, &end, &result] {any_of(begin, end, result); });

    }

    for (auto & thread : threads)
        thread.join();

    bool hasNonZero = !result.test_and_set();
    return 0;
};