在现代编程中，多线程是提高程序性能的一种常见方法。本文将探讨在C++中使用不同同步机制进行多线程计数的性能。将比较使用互斥锁（mutex）、原子操作（atomic）、Windows特有的InterlockedIncrement函数、无锁（lock-free）算法、单线程执行以及每个核心一个锁（one lock per core）的性能。
本文的基准测试代码来源于VC++的“Windows上30个多线程错误”文章，但这些代码分散在不同的部分。本文将这些代码集中在一起，以便更清晰地展示它们。将使用C++17的并行std::for_each来在所有逻辑核心上启动线程，除非另有说明。计数基准测试是在装有Windows 11的Intel i7 870处理器上进行的，它有12个逻辑核心。Ubuntu 24.04的基准测试结果将在最后部分展示。
第一个示例涉及使用C++11的互斥锁来同步多个线程的计数。
void inc_mutex() {
timer watch;
watch.start("inc mutex");
int count = 0;
std::mutex mut;
std::for_each(std::execution::par, Vec.begin(), Vec.end(),
[&mut, &count](size_t num) -> void {
if (is_valid((int)(num))) {
std::lock_guard guard(mut);
++count;
}
});
watch.stop();
std::cout << "total count:" << count << std::endl;
}
Vec被随机初始化到VEC_SIZE的大小，即100000000。rand()函数用于注入不确定性，以防止编译器在编译时计算结果并优化掉代码。
const size_t VEC_SIZE = 100000000U;
void init_vector() {
srand(time(NULL));
for (size_t i = 0; i < VEC_SIZE; ++i)
Vec.push_back(rand() % 20);
}
bool is_valid(int n) {
return (n % 2 == 0);
}
互斥锁的基准测试结果如下。
inc mutex: 3136ms
由于只处理一个整数count变量，可以放弃互斥锁并使用C++11的原子整数来提高性能。
void inc_atomic() {
timer watch;
watch.start("inc atomic");
std::atomic count = 0;
std::for_each(std::execution::par, Vec.begin(), Vec.end(),
[&count](size_t num) -> void {
if (is_valid((int)(num)))
++count;
});
watch.stop();
std::cout << "total count:" << count << std::endl;
}
互斥锁和原子整数的基准测试结果如下。有超过66%的性能提升。
inc mutex: 3136ms
inc atomic: 1007ms
如果在Windows上使用预C++11编译器，可以使用Windows的InterlockedIncrement来实现相同的性能提升。
#ifdef _WIN32
void inc_winInterlocked() {
timer watch;
watch.start("inc win Interlocked");
LONG count = 0;
std::for_each(std::execution::par, Vec.begin(), Vec.end(),
[&count](size_t num) -> void {
if (is_valid((int)(num)))
::InterlockedIncrement(&count);
});
watch.stop();
std::cout << "total count:" << count << std::endl;
}
#endif
互斥锁、原子整数和InterlockedIncrement的基准测试结果如下。
inc mutex: 3136ms
inc atomic: 1007ms
inc win Interlocked: 1005ms
在这个无锁基准测试中，将使用自己编写的loop::parallel_for而不是使用C++17的并行std::for_each。
template
void parallel_for(int numOfThreads, size_t beginIndex, size_t endIndex, Func f) {
// Function implementation...
}
创建一个CountStruct来存储每个线程的计数。buf成员是为了防止伪共享。所有线程结束后，从所有的CountStruct中累加总计数。
struct CountStruct {
CountStruct() : Count(0) {
memset((void*)buf, 0, sizeof(buf));
}
int Count;
char buf[128]; // to prevent false sharing
};
void inc_no_lock() {
int threads = std::thread::hardware_concurrency();
std::vector count(threads);
timer watch;
watch.start("inc no lock");
loop::parallel_for(threads, (size_t)(0), (size_t)(VEC_SIZE),
[&count](int threadIndex, int index) -> void {
if (is_valid(Vec[index]))
++(count[threadIndex].Count);
});
int total = 0;
for (auto& st : count)
total += st.Count;
watch.stop();
std::cout << "total count:" << total << std::endl;
}
无锁的基准测试结果如下。
inc mutex: 3136ms
inc atomic: 1007ms
inc win Interlocked: 1005ms
inc no lock: 73ms
在这一部分，使用普通的for循环来基准测试单线程性能。注意：这个for循环没有前面代码基准测试的线程启动开销。
void inc_single_thread() {
timer watch;
watch.start("inc single_thread");
int count = 0;
for (size_t i = 0; i < Vec.size(); ++i) {
if (is_valid(Vec[i]))
++count;
}
watch.stop();
std::cout << "total count:" << count << std::endl;
}
单线程的基准测试结果如下。
inc mutex: 3136ms
inc atomic: 1007ms
inc win Interlocked: 1005ms
inc no lock: 73ms
inc single_thread: 86ms
在这一部分，使用C++17的并行std::for_each，但放弃了并行std::for_each的便利性，根据逻辑核心数生成线程，然后在每个线程中手动计算工作分配，并循环存储计数到本地temp_count，在每个线程结束时，获取一个互斥锁，然后累积temp_count到count。注意：获取的互斥锁数量正好是逻辑核心数。
void inc_less_mutex_lock() {
timer watch;
watch.start("inc less mutex lock");
const size_t threads = std::thread::hardware_concurrency();
std::vector vecIndex;
for (size_t i = 0; i < threads; ++i)
vecIndex.push_back(i);
int count = 0;
std::mutex mut;
std::for_each(std::execution::par, vecIndex.begin(), vecIndex.end(),
[&mut, &count, threads](size_t index) -> void {
size_t thunk = Vec.size() / threads;
size_t start = thunk * index;
size_t end = start + thunk;
if (index == (threads - 1)) {
size_t remainder = Vec.size() % threads;
end += remainder;
}
int temp_count = 0;
for (int i = start; i < end; ++i) {
if (is_valid((int)(Vec[i]))) {
++temp_count;
}
}
{
std::lock_guard guard(mut);
count += temp_count;
}
});
watch.stop();
std::cout << "total count:" << count << std::endl;
}
令人惊讶的是，在Windows上，每个逻辑核心获取一个互斥锁（在Intel i7 870案例中，12个核心）获得了最佳结果。
inc mutex: 3136ms
inc atomic: 1007ms
inc win Interlocked: 1005ms
inc no lock: 73ms
inc single_thread: 86ms
inc less mutex lock: 33ms
本节列出了在Windows 11 WSL Ubuntu 24.04的G++ 13.2和Clang 18.1.3上的基准测试结果。CPU与之前相同，是Intel i7 870 12个逻辑核心。两个编译器的结果相似。无锁、每个核心一个锁和单线程的性能相似。请注意，对于正常工作负载，比如计数素数，工作时间远远超过写入count的时间。其他两种方法的优势在于可以将工作负载分配到线程中。
inc mutex: 719ms
inc atomic: 473ms
inc no lock: 51ms
inc single_thread: 48ms
inc less mutex lock: 54ms
inc mutex: 745ms
inc atomic: 463ms
inc no lock: 54ms
inc single_thread: 52ms
inc less mutex lock: 58ms
g++ MultithreadedCount.cpp -O3 -std=c++17
clang++ MultithreadedCount.cpp -O3 -std=c++17