并发数组检查

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;
};

Answer 2

类似下面的内容呢？

每个工作人员检查其范围内的元素是否非零或是否设置了 atomic 标志（意味着其他一些线程已经找到它）。

以下是每个线程执行的函数（每个线程都分配了不同的范围）

 template<typename Iterator>
static void any_of(Iterator & begin, Iterator& end, std::atomic<bool> & result) 
    {
        for (const auto & it=begin; it!=end; ++it)
        {
            if (result || (*it)!=0)
            {
                result= true;
                return;
            }
       }

你可以这样称呼它

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)
{
    const auto & begin = input.begin() + i *chunk_size;
    const auto & end = input.begin() + std::min((i+1) * chunk_size, input.size());
    threads.emplace_back(any_element_of,begin,end,result);
}

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

在此之后，您可以安全地检查return 以检索您的结果。

请注意，通过将一元谓词函数传递给 worker 使其更通用，这种方法很容易扩展。

 template<typename Iterator, typename Predicate>
static void any_of(Iterator & begin, Iterator& end, Predicate pred, std::atomic<bool> & result) 
    {
        for (const auto & it=begin; it!=end; ++it)
        {
            if (result || pred(*it))
            {
                result= true;
                return;
            }
       }