并发阵列检查

Concurrent Array Checking

本文关键字：检查 阵列 并发      更新时间：2023-10-16

我既不是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++版本，但我不想依赖任何第三方库。

我试图扩展Davide Spataro的解决方案，以解决使用atomic_flagatomic<bool>同步问题，即"与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 flagn";
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 booln";
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;
};

下面这样的东西呢？

每个工作线程检查其范围内的元素是否为非零，或者是否设置了原子标志(意味着其他一些线程已经找到了它(。

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

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以检索结果。

请注意，通过将一元谓词函数传递给辅助角色以使其更通用，此方法很容易扩展。

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;
}
}
相关文章：