线程安全的 std::map，速度媲美无锁 Map。
线程安全指针和无争用共享互斥体的用法和测试示例。
- 所有这些文件都在 GitHub 上
- 下载 safe_ptr - 5 KB
- 下载 benchmark - 10.1 KB
- 下载 CDS_test - 2.6 MB
- 下载 Real_app_bench - 2.6 MB
三篇相关文章
- 我们让任何对象都线程安全
- 我们让 std::shared_mutex 速度提升 10 倍
- 线程安全的
std::map，兼具无锁 map 的速度
引言
线程安全指针和无竞争共享互斥锁的使用和测试示例
在本文中，我们将展示我们开发的线程安全指针的额外优化、使用示例和测试，该指针使用优化的共享互斥锁 contfree_safe_ptr<T> - 这等同于 safe_ptr<T, contention_free_shared_mutex<>>。
最后，我们将展示我们的线程安全指针测试与 libCDS 中一些最佳无锁算法在 Intel Core i5/i7、Xeon、2 x Xeon 处理器上的对比图。
您可以通过以下链接找到 libCDS：https://github.com/khizmax/libcds。
本文中的所有基准测试都使用了 libCDS 的此提交版本 https://github.com/khizmax/libcds/tree/5e2a9854b0a8625183818eb8ad91e5511fed5898。
背景
不同粒度的锁
首先，我们将以处理表（结构数组）为例，展示如何最佳地使用共享互斥锁。让我们借鉴工业 RDBMS 的经验。例如，MSSQL RDBMS 使用不同粒度的锁：锁定一行或多行、一个页面、一个范围、表的一个分区、一个索引、整个表、整个数据库。 锁粒度和层次结构
确实，如果我们要长时间处理一行，并且在此期间不希望该行被其他线程更改，那么没有必要一直锁定整个表——只需锁定一行就足够了。
- 使用共享锁锁定整个表
- 查找所需的一行或多行
- 然后锁定找到的行
- 解锁表
- 并处理锁定的行
然后其他线程可以并行处理其余的行。
到目前为止，我们只使用了表级别的锁，即我们锁定了一个或多个表。
或者表达式中使用的所有表都被自动锁定，直到它完全完成。
(*safe_map_1)["apple"].first = "fruit"; // locked Table-1 (safe_map_1)
// unlocked Table-1
safe_map_1->at("apple").second =
safe_map_1->at("apple").second * 2; // locked Table-1 (safe_map_1)
// unlocked Table-1
safe_map_1->at("apple").second =
safe_map_2->at("apple").second*2; // locked Table-1(safe_map_1) and Table-2(safe_map_2)
// unlocked Table-1 and Table-2
在其他情况下，我们通过使用 RAII 锁对象手动锁定了一个或多个表，直到这些锁的块作用域过期（直到它们被销毁）。
{
std::lock_guard< decltype(safe_map_1) > lock(safe_map_1); // locked Table-1 (safe_map_1)
(*safe_map_1)["apple"].first = "fruit";
safe_map_1->at("apple").second =
safe_map_1->at("apple").second * 2;
// unlocked Table-1
}
{
lock_timed_any_infinity lock_all(safe_map_1, safe_map_2);
// locked Table-1(safe_map_1) and Table-2(safe_map_2)
safe_map_1->at("apple").second =
safe_map_2->at("apple").second*2; // locked Table-1(safe_map_1) and Table-2(safe_map_2)
safe_map_1->at("potato").second =
safe_map_2->at("potato").second*2; // locked Table-1(safe_map_1) and Table-2(safe_map_2)
// unlocked Table-1 and Table-2
}
我们来考虑一个示例，在该示例中，我们随机选择一个要插入的索引，然后随机选择四种操作（insert、delete、update、read）中的一种，并对类型为 contfree_safe_ptr<std::map> 的线程安全对象执行该操作。
示例：[37] http://coliru.stacked-crooked.com/a/5b68b65bf2c5abeb
在这种情况下，我们将对表进行以下锁定
Insert- 排他锁Delete- 排他锁Update- 排他锁Read- 共享锁
对于 Update 或 Read 操作，我们执行以下操作
- 锁定整个表（
Update为xlock，Read为slock） - 搜索必要的行，读取或更改它
- 解锁表
我们示例 1 中一次迭代的代码
int const rnd_index = index_distribution(generator); // 0 - container_size
int const num_op = operation_distribution(generator);// insert_op, update_op,
// delete_op, read_op
switch (num_op) {
case insert_op: {
safe_map->emplace(rnd_index,(field_t(rnd_index, rnd_index))); // insert with
// X-lock on Table
}
break;
case delete_op: {
size_t erased_elements = safe_map->erase(rnd_index); // erase with X-lock
// on Table
}
break;
case update_op: {
auto x_safe_map = xlock_safe_ptr(safe_map); // X-lock on Table
auto it = x_safe_map->find(rnd_index);
if (it != x_safe_map->cend()) {
it->second.money += rnd_index; // still X-lock on Table
// (must necessarily be)
}
}
break;
case read_op: {
auto s_safe_map = slock_safe_ptr(safe_map); // S-lock on Table
auto it = s_safe_map->find(rnd_index);
if (it != s_safe_map->cend()) {
volatile int money = it->second.money; // still S-lock on Table
// (must necessarily be)
// volatile here only to avoid optimization for unused money-variable
}
}
break;
default: std::cout << "\n wrong way! \n"; break;
}
现在，我们将执行以下操作：在 Update 操作期间，通过读锁（共享）而不是通过更改锁（排他）来锁定表。这将大大加快使用我们之前开发的“无写竞争共享互斥锁”时的 Update 操作。
在这种情况下，多个线程可以同时对一个表执行 Update 和 Read 操作。例如，一个线程读取一行，另一个线程更改另一行。但是，如果一个线程尝试更改另一个线程正在读取的同一行，那么为了避免数据竞争，我们必须在读取和更改该行时锁定该行本身。
示例：[38] http://coliru.stacked-crooked.com/a/89f5ebd2d96296d3
现在对于 Update 或 Read 操作，我们执行以下操作
- 使用共享锁锁定整个表
- 查找所需的一行或多行
- 然后锁定找到的行（
Update为xlock，Read为slock） - 并处理锁定的行（X/S 锁）和锁定的表（S 锁）
- 解锁行
- 解锁表
差异（我们已更改的）
我们示例 2 中一次迭代的代码
int const rnd_index = index_distribution(generator); // 0 - container_size
int const num_op = operation_distribution(generator);// insert_op, update_op,
// delete_op, read_op
switch (num_op) {
case insert_op: {
safe_map->emplace(rnd_index, (field_t(rnd_index, rnd_index)));// insert with
// X-lock on Table
}
break;
case delete_op: {
size_t erased_elements = safe_map->erase(rnd_index); // erase with X-lock
// on Table
}
break;
case update_op: {
auto s_safe_map = slock_safe_ptr(safe_map); // S-lock on Table
auto it = s_safe_map->find(rnd_index);
if (it != s_safe_map->cend()) {
auto x_field = xlock_safe_ptr(it->second);
x_field->money += rnd_index; // X-lock on field, still S-lock on Table
// (must necessarily be)
}
}
break;
case read_op: {
auto s_safe_map = slock_safe_ptr(safe_map); // S-lock on Table
auto it = s_safe_map->find(rnd_index);
if (it != s_safe_map->cend()) {
auto s_field = slock_safe_ptr(it->second);
volatile int money = s_field->money; // S-lock on field,
// still S-lock on Table
// (must necessarily be)
// volatile here only to avoid optimization for unused money-variable
}
}
break;
default: std::cout << "\n wrong way! \n"; break;
}
在这里，为了线程安全地处理行，我们使用了 safe_obj。safe_obj<T> 内部有一个 T 类型的对象，而不是像 safe_ptr<T> 那样是指向它的指针。因此，使用 safe_obj 时，您不需要像 safe_ptr 中那样单独为对象本身分配内存并更改原子引用计数。因此，使用 safe_obj 执行 Insert 和 Delete 操作比使用 safe_ptr 快得多。
值得注意的是，在复制 safe_obj 时，复制的不是指向对象的指针，而是对象本身。在这里，源 safe_obj 和目标 safe_obj 都会被预先锁定。
注意：严格来说，我们没有锁定整行，而是锁定了该行的所有字段，除了我们用于搜索的行索引。因此，我们将我们的对象称为“field”，而不是“row”。此外，为了强调这种方式我们不仅可以锁定一行，甚至可以锁定一行中的单个字段，如果我们将其放置在单独的 safe_obj 对象中。这将允许不同的线程锁定不同的字段并并行处理它们。
现在我们对不同的操作使用以下锁
这个例子对于大量的短期操作非常快。但是，在更改或读取行（字段）时，我们仍然持有表的读锁（共享）。如果我们在表的行上有罕见但非常长的操作，那么整个表的锁将一直保持很长时间。
但是，如果根据您的任务逻辑，一行可能被一个线程删除，而另一个线程正在读取或更改同一行，这无关紧要，那么只需在查找行的时间内锁定表即可。为了避免访问已释放的内存（当另一个线程删除该行时），我们需要使用 std::shared_ptr<T> - 带有原子线程安全引用计数的指针。在这种情况下，内存只有在没有线程指向该行时才会被释放。
我们将使用 safe_ptr 而不是 safe_obj 来保护行。这将允许我们复制指向行的指针，并使用 safe_ptr 中包含的 std::shared_ptr<T> 中的线程安全引用计数。
示例：[39] http://coliru.stacked-crooked.com/a/f2a051abfbfd2716
现在，对于 Update 或 Read 操作，我们执行以下操作
- 使用共享锁锁定整个表
- 查找所需的一行或多行
- 然后锁定找到的行（
Update为xlock，Read为slock） - 解锁表
- 并根据需要处理锁定的行（X/S 锁）
- 解锁行
差异（我们已更改的）
示例 3
int const rnd_index = index_distribution(generator); // 0 - container_size
int const num_op = operation_distribution(generator); // insert_op, update_op,
// delete_op, read_op
safe_ptr_field_t safe_nullptr;
switch (num_op) {
case insert_op: {
safe_map->emplace(rnd_index, (field_t(rnd_index, rnd_index))); //insert with
//X-lock on Table
}
break;
case delete_op: {
size_t erased_elements = safe_map->erase(rnd_index); // erase with X-lock
// on Table
}
break;
case update_op: {
auto pair_result = [&]() {
auto s_safe_map = slock_safe_ptr(safe_map); // S-lock on Table
auto it = s_safe_map->find(rnd_index);
if (it != s_safe_map->cend()) return std::make_pair(it->second, true); // found
else return std::make_pair(safe_nullptr, false); // null-value
}(); // unlock Table
if(pair_result.second) {
auto x_field = xlock_safe_ptr(pair_result.first); // X-lock on field
x_field->money += rnd_index; // if a long time is read
} // unlock field
}
break;
case read_op: {
auto pair_result = [&]() {
auto s_safe_map = slock_safe_ptr(safe_map); // S-lock on Table
auto it = s_safe_map->find(rnd_index);
if (it != s_safe_map->cend()) return std::make_pair(it->second, true); // found
else return std::make_pair(safe_nullptr, false); // null-value
}(); // unlock Table
if(pair_result.second) {
auto s_field = slock_safe_ptr(pair_result.first); // S-lock on field
volatile int money = s_field->money; // if a long time is read
// volatile here only to avoid optimization for unused money-variable
} // unlock field
}
break;
default: std::cout << "\n wrong way! \n"; break;
}
精心设计的多线程程序对共享对象使用短调用。因此，未来我们将使用倒数第二个示例进行短读操作，而不是最后一个。
“环绕执行”习语的缺点
让我们考虑可能的问题并批评我们的代码。
在上一章中，我们考虑了一个相当方便且高性能的例子，通过函数显式指定 Update 和 Read 操作的锁定类型：
slock_safe_ptr()- 只读xlock_safe_ptr()- 用于读写
在这里，锁一直保持到这些函数返回的 lock_obj 对象的生命周期结束：auto lock_obj = slock_safe_ptr (sf_p);
然而，对于 Insert 和 Delete 操作使用了隐式锁。我们的 safe_ptr<std::map> 对象通过“环绕执行指针”习语自动锁定，并在 Insert 或 Delete 操作终止后立即解锁。
示例：[40] http://coliru.stacked-crooked.com/a/89f5ebd2d96296d3
但是您可能会忘记在 Update 或 Read 操作中使用显式锁。在这种情况下，safe_ptr<std::map> 将在搜索操作完成后立即解锁，然后您继续使用
- 找到的迭代器，它可能被另一个线程无效
- 或找到的元素，它可能被另一个线程删除
为了部分解决这个问题，您可以使用 safe_hide_ptr<> 和 safe_hide_obj<> 来代替 safe_ptr<> 和 safe_obj<> — 它们不使用“环绕执行指针”习语，并且您只能在显式锁定后才能访问类成员。
safe_hide_obj<field_t, spinlock_t=""> field_hide_tmp;
//safe_obj<field_t, spinlock_t=""> &field_tmp = field_hide_tmp; // conversion denied -
// compile-time error
//field_hide_tmp->money = 10; // access denied - compile-time error
auto x_field = xlock_safe_ptr(field_hide_tmp); // locked until x_field is alive
x_field->money = 10; // access granted
</field_t,></field_t,>
示例：[41] http://coliru.stacked-crooked.com/a/d65deacecfe2636b
以前，您可能会犯错误并写出以下——错误代码
auto it = safe_map->find(rnd_index); // X-lock, find(), X-unlock
if (it != s_safe_map->cend())
volatile int money = it->second ->money; // X-lock, operator=(), X-unlock
但现在，这样的请求无法编译，并且需要显式锁定对象——正确的代码
auto s_safe_map = slock_safe_ptr(safe_map); // S-lock on Table
auto it = s_safe_map->find(rnd_index);
if (it != s_safe_map->cend()) {
auto s_field = slock_safe_ptr(it->second);
volatile int money = s_field->money; // S-lock on field, still S-lock on Table
// volatile here only to avoid optimization for unused money-variable
} // S-unlock Table, S-unlock field
然而，您仍然面临将锁用作临时对象的危险——不正确
auto it = slock_safe_ptr(safe_map)->find(rnd_index); // S-lock, find(), S-unlock
if (it != s_safe_map->cend()) {
volatile int money = slock_safe_ptr(it->second)->money; // S-lock, =, S-unlock
}
您有选择
- 使用
safe_ptr和safe_obj以便能够显式或自动（环绕执行习语）锁定您的对象 - 或者使用
safe_hide_ptr和safe_hide_obj，只保留显式锁定对象的能力
由您决定如何选择
- 使用方便的自动锁定（环绕执行习语）
- 或通过显式锁定对象略微减少出错的可能性
此外，以下函数计划被纳入 C++17 标准，用于 std::map<>
insert_or_assign()- 如果存在元素，则赋值；如果不存在，则插入try_emplace()- 如果元素不存在，则创建它merge()- 将两个 map 合并为一个extract()- 获取元素并将其从 map 中移除
引入此类函数允许在不使用迭代器的情况下执行常用复合操作——在这种情况下，使用“环绕执行”习语将始终保证这些操作的线程安全性。总的来说，避免对所有容器（除了数组 std::array 和 std::vector）使用迭代器，对于开发多线程程序大有裨益。您使用迭代器的频率越低，访问被任一线程无效的迭代器的机会就越少。
但无论您选择什么：使用“环绕执行”习语，为 safe_hide_ptr 使用显式锁，放弃它们并使用标准的 std::mutex 锁...或者编写自己的无锁算法——您总是有很多机会犯错误。
通过分区表提高性能
让我们回顾一下工业关系数据库的经验。例如，在 RDBMS 中，可以使用表分区来提高性能。在这种情况下，您可以只锁定使用的分区，而不是整个表：https://technet.microsoft.com/en-us/library/ms187504(v=sql.105).aspx
尽管在 RDBMS 中，Delete 和 Insert 操作通常不会锁定整个表（对于 Delete 操作总是如此），但对于 Insert 操作，可以非常快速地将数据加载到表中，其强制条件是独占锁定表。
- MS SQL
(dbcc traceon (610, -1)): INSERT INTO sales_hist WITH (TABLOCKX) - Oracle:
INSERT /*+ APPEND */ INTO sales_hist
https://docs.oracle.com/cd/E18283_01/server.112/e17120/tables004.htm#i1009887
Direct-Path INSERT 的锁定注意事项
在直接路径 INSERT 期间，数据库对表（或分区表的所有分区）获取独占锁。因此，用户无法对表执行任何并发的 insert、update 或 delete 操作，也不允许并发的索引创建和构建操作。但是，支持并发查询，但查询只会返回 insert 操作之前的信息。
由于我们的任务是创建最快的多线程容器，我们也在 Insert / Delete 操作中完全锁定了容器。但现在让我们尝试只锁定容器的一部分。
让我们尝试实现我们自己的分区有序关联数组 partitioned_map，看看性能会提高多少。我们只锁定当前需要的那个部分。
根据上下文，它是 std::map< safe_ptr<std::map<>> >。
这里，第一个 std::map 将是常量，并将包含分区（子表）。
这将是一个非常简化的示例，其中分区的数量在构造函数中指定，并且不再更改。
每个分区都是一个线程安全的关联数组 safe_ptr<std::map<>>。
为了提供最大的性能，分区的数量和范围应使得访问每个分区的概率相同。如果您的键范围从 0 到 1000000，并且对范围开头进行 read/update/insert/delete 调用的概率大于对范围末尾的概率，则键值较小的分区数量应更多，其范围应更小。例如，3 个分区：[0 - 100000]，[100001 - 400000]，[400001 - 1000000]。
但在我们的示例中，我们将假设查询键具有均匀分布。
分区范围可以通过两种方式分配
safe_map_partitioned_t<std::string, int> safe_map { "a", "f", "k", "p", "u" };
// 分配分区限制
safe_map_partitioned_t<int, int> (0, 100000, 10000);
// 分配值限制 0 – 100000 和每个分区的步长 10000
如果访问容器时，键超出了分区的限制，查询将指向最近的分区，即程序将正常工作。
示例：[42] http://coliru.stacked-crooked.com/a/fc148b08eb4b0580
此外，为了获得最大性能，需要使用我们之前在 safe_ptr<> 内部实现的“无竞争共享互斥锁”。从字面意义上讲，这是：std::map<contfree_safe_ptr<std::map<>> >
让我们从上一个示例中获取代码，并从上一章中添加 contfree_safe_ptr 代码。
替换：safe_map_partitioned_t<std::string, std::pair<std::string, int>>
替换为：safe_map_partitioned_t<std::string, std::pair<std::string, int>,contfree_safe_ptr>
示例：[43] http://coliru.stacked-crooked.com/a/23a1f7a3982063a1
我们创建这个 safe_map_partitioned_t<> 类“只是为了好玩”，也就是说，不建议在实际程序中使用它。我们只是展示了一个基于 contfree_safe_ptr<> 指针和 contention_free_shared_mutex<> 锁实现您自己的算法的示例。
Using the Code
如何使用
首先，您应该下载文件 - safe_ptr.h
并将其包含到您的 cpp 文件中：#include "safe_ptr.h"
作为最佳用法，我们将停留在上面显示的示例 2 — 它既简单又高性能。
struct field_t {
int money, time;
field_t(int m, int t) : money(m), time(t) {}
field_t() : money(0), time(0) {}
};
typedef safe_obj<field_t, spinlock_t> safe_obj_field_t;
// thread-safe ordered std::map by using execute around pointer idiom
// with contention-free shared-lock
contfree_safe_ptr< std::map<int, safe_obj_field_t > > safe_map_contfree;
bool success_op;
switch (num_op) {
case insert_op:
slock_safe_ptr(test_map)->find(rnd_index); // find for pre-cache to L1
// with temporary S-lock
test_map->emplace(rnd_index, field_t(rnd_index, rnd_index));
break;
case delete_op:
slock_safe_ptr(test_map)->find(rnd_index); // find for pre-cache to L1
// with temporary S-lock
success_op = test_map->erase(rnd_index);
break;
case read_op: {
auto s_safe_map = slock_safe_ptr(test_map); // S-lock on Table (must necessarily be)
auto it = s_safe_map->find(rnd_index);
if (it != s_safe_map->cend()) {
auto x_field = xlock_safe_ptr(it->second);
volatile int money = x_field->money; // get value
x_field->money += 10; // update value
}
}
break;
default: std::cout << "\n wrong way! \n"; break;
}
插入和删除 - map 的修改： 由于我们的共享锁 slock_safe_ptr() 是最快的，因此在修改（insert_op 或 delete_op）之前，我们使用快速共享锁查找我们需要删除的项目或插入的最近项目，该锁在 find() 之后立即解锁。我们自己不使用找到的项目。但是，这会告诉 CPU 哪些数据在后续的 map 修改中最好缓存到 L1。然后我们执行 test_map->emplace() 进行插入或 test_map->erase() 进行删除，它会自动调用排他锁。Insert/delete 发生得很快，因为所有数据都已缓存到 CPU-L1 中，我们不需要将共享锁升级为排他锁（这会大大增加死锁的可能性 - 冻结程序）。
读取和更新 - 读取或修改项：我们所说的读取（read_op）是指修改 map 中的一个项（map 的一行）。因此，在读取之前，我们对 map 进行共享锁 slock_safe_ptr()，以便其他线程无法删除或替换此项。并在存在对象 s_safe_map 时持有共享锁，然后我们仅对找到的一个项施加排他锁 xlock_safe_ptr()，然后读取此项并进行更改。之后，当离开块作用域 {} 时，x_field 对象首先被销毁，并且项的排他锁被移除，然后销毁下一个对象 s_safe_map，并且 map 的共享锁被移除。
编译器允许您对 test_map 执行任何操作——读写和调用 test_map 的任何方法——在这种情况下，会自动应用排他锁。
但编译器只允许您读取对象并调用 slock_safe_ptr(test_map) 的 const 方法，例如 slock_safe_ptr(test_map)->find();，但不允许 slock_safe_ptr(test_map)->emplace();some_ptr（其中 auto const& some_ptr = test_map;），也允许同样的操作，例如 some_ptr->find() ——在这种情况下，会自动应用共享锁。
所有这些，甚至更多——在第一篇文章中进行了详细而严谨的探讨。
已实现的 safe_ptr 的性能比较
让我们总结一下中期结果。让我们看看由于我们的优化，性能提升的水平。
这里是性能图表——每秒百万次操作 (MOps) 的数量，修改操作的百分比范围从 0% 到 90%。在修改过程中，将以相等的概率执行三项操作之一：insert、delete、update（尽管 Update 操作本身不会改变 std::map 树，只会改变其中的一行）。例如，15% 的修改涉及以下操作：5% insert、5% delete、5% update、85% read。编译器：g++ 4.9.2 (Linux Ubuntu) x86_64
- 下载此基准测试适用于 Linux (GCC 6.3.0) 和 Windows (MSVS2015)：benchmark.zip - 10.1 KB
要在 Linux 上使用 Clang++ 3.8.0 编译器进行编译，您需要修改 Makefile。
这是一个在单个服务器 CPU Intel Xeon E5-2660 v3 2.6 GHz 上对 16 个线程进行测试的示例。首先，我们对橙色线感兴趣：safe<map, contfree>&rowlock
如果您的同一主板上有多个 Xeon CPU，您可以通过以下方式运行测试来重现此结果：
numactl --localalloc --cpunodebind=0 ./benchmark 16
结论
- 有趣的是，我们
contfree_safe_ptr<map>中的共享无竞争锁比safe_ptr <map, std::shared_mutex>中的标准std::shared_mutex快得多。 - 同样有趣的是，从 15% 的更改开始，
std::mutex比std::shared_mutex更快（即safe_ptr<map, std::mutex>比safe_ptr<map, std::shared_mutex>更快）。 - 同样奇怪的是，从 30% 的更改开始，
std:map-1thread的性能高于contfree_safe_ptr<map>&rowlock的性能。但实际任务不仅仅包括关联数组的操作和线程同步。通常，线程间数据交换只占任务的一小部分，而主要部分可以很好地并行化到多个核心。 - 分区的关联数组
safe_map_partitioned<,,contfree_safe_ptr>表现出良好的性能，但它应该只作为“Just-for-fun”使用——如果所需的分区数量及其范围是预先知道的。 - 对于 15% 的修改，我们的共享互斥锁（作为
«contfree_safe_ptr<map> & rowlock»的一部分）表现出 8.37 MOps 的性能，比标准std::shared_mutex（作为«safe_ptr<map, std::shared_mutex>»的一部分）的 0.74 MOps 快 10 倍。
我们的锁“无竞争共享互斥锁”适用于任意数量的线程：对于前 70 个线程，它作为无竞争锁工作，对于后续线程，它作为排他锁工作。从图表中可以看出，即使是排他锁 «std::mutex» 对于 15% 的修改也比 «std::shared_mutex» 快。无竞争锁的线程数 70 由模板参数指定，并且可以更改。
此外，我们还将展示中位延迟图（以微秒为单位），即一半请求的延迟低于此值。
在测试代码 main.cpp 中，您应该设置：const bool measure_latency = true;
- 有趣的是，
std::map和std::mutex的范围与safe_ptr<map, std::mutex>相似。也就是说，通常使用safe_ptr<>不会产生任何额外的开销成本，也不会降低性能。 - 此外，有趣的是，从 60% 的变化开始，
contfree_safe_ptr<map>&rowlock表现出比contfree_safe_ptr<map>更高的延迟。但从前一个图中，我们了解到，对于任何百分比的修改操作，contfree_safe_ptr<map>&rowlock的总体性能仍然更高。
contfree_safe_ptr<std::map> 与 CDS-lib 中容器在不同桌面 CPU 上的性能比较。
使用所有核心在不同桌面 CPU 上的性能比较。
- 有趣的是，对于笔记本电脑和台式机，
contfree_safe_ptr<map>在 CDS-lib 的无锁 map 容器中显示出最高的性能。 - 我们“仅供娱乐”版本的关联容器，分区数量固定为
safe_map_partitioned<,,contfree_safe_ptr>，平均性能提升了 1.7 倍。
contfree_safe_ptr<std::map> 与 CDS-lib 中容器在服务器 CPU 上的性能比较。
在单个服务器 CPU 上使用不同数量线程的不同多线程关联数组的性能比较。
- 下载此基准测试适用于 Linux (GCC 6.3.0) 和 Windows (MSVS2015)：CDS_test.zip - 2.6 MB
要在 Linux 上使用 Clang++ 3.8.0 编译器进行编译，您需要修改 Makefile。
如果您的同一主板上有多个 Xeon CPU，您可以通过以下方式运行测试来重现此结果：
numactl --localalloc --cpunodebind=0 ./benchmark 16
- 如您所见，如果您在单个 CPU 上使用 16 个或更多并行线程，或者如果您使用两个以上的 CPU，那么 CDS-lib 中的无锁容器会比
contfree_safe_ptr<map>表现出更好的性能。也就是说，在强竞争的情况下，无锁容器更好。但是，即使您有超过 16 个线程，但一次访问容器的线程不超过 8 个，那么contfree_safe_ptr<map>仍然会比无锁容器更快。 - 此外，具有固定数量
safe_map_partitioned<,,contfree_safe_ptr>分区的“Just-for-fun”容器在任何线程数下都显示出更好的性能，但它只能在预先知道所需分区数量和范围的情况下使用。
中位延迟是指 50% 的查询完成时间低于该值的时间。
在测试代码 main.cpp 中，您应该设置：const bool measure_latency = true;
对于 contfree_safe_ptr<map>，中位延迟也几乎等于无锁容器在最多 8 个同时竞争线程下的延迟，但在 16 个竞争线程的情况下会更差。
实际应用程序性能
实际应用程序不仅仅包含线程之间的数据交换。实际应用程序的主要工作在每个线程中单独执行，并且只有偶尔才发生线程之间的数据交换。
为了模拟应用程序的实际工作，我们在每次访问线程安全容器之间添加一个不可优化的循环 for(volatile int i = 0; i <9000; ++ i);。在测试开始时，我们将执行此循环 100,000 次，并测量此循环的平均执行时间。在 Intel Xeon E5-2686 v4 2.3 GHz CPU 上，这种实际工作模拟大约需要 20.5 微秒。
- 下载此基准测试适用于 Linux (GCC 6.3.0) 和 Windows (MSVS2015)：Real_app_bench.zip - 2.6 MB
要在 Linux 上使用 Clang++ 3.8.0 编译器进行编译，您需要修改 Makefile。
我们将在双处理器服务器上执行测试：2 x Intel Xeon E5-2686 v4 2.3 GHz (Broadwell)，总核心数：36 个物理核心和 72 个逻辑核心（超线程）。
要在 Linux 上构建和测试，请执行
cd libcds
make
cd ..
make
./benchmark
- 使用标准互斥锁——
std::mutex和std::shared_mutex——来保护std::map，其性能在最多 16 个线程时接近无锁 map 容器。但随后，std::mutex & std::map和std::shared_mutex & std::map的性能开始落后，并在 32 个线程之后开始下降。 - 我们优化的线程安全指针
contfree_safe_ptr<std::map<>>的性能几乎与 libCDS 库中无锁 map 容器的性能相等（适用于 1 到 64 个线程的任何数量）。这在实际应用程序中成立，前提是线程间的交换平均每 20 微秒或更少发生一次。
要测量中位延迟——在测试代码 main.cpp 中，您应该设置
const bool measure_latency = true;
只需在 Linux 上启动基准测试：./benchmark
- 有趣的是，在 64 个线程下，
std::mutex的性能和中位延迟都高于std::shared_mutex。 - 我们优化的线程安全指针
contfree_safe_ptr<std::map<>>在 1 到 64 个线程的任何数量下，具有与 libCDS 中的无锁 map 容器相同的延迟。这也适用于实际应用程序中线程之间的交换每 20 微秒或更少发生一次的情况。 - 对于 64 个线程，延迟为 30 微秒，其中 20 微秒是应用程序延迟，10 微秒是多线程延迟。因此，即使多线程延迟占总延迟的 30%，我们的线程安全智能指针
contfree_safe_ptr<T>也具有与 libCDS 的无锁 map 容器相同的性能（MOps 和中位延迟）。
libCDS 中更简单的无锁和无等待容器（队列、栈……）的延迟明显低于其基于任何锁的实现。
关注点
这些文章涵盖的问题
在这些文章中，我们详细探讨了 C++ 语言结构在单个线程中通过“环绕执行指针”习语执行的顺序。我们还以无竞争共享互斥锁为例，探讨了多线程程序中读写操作的执行顺序。我们提供了 libCDS 中无锁算法和我们开发的基于锁的算法的高性能使用示例。
我们工作的成果 - safe_ptr.h
结论
- 如果线程对同一容器存在强竞争（即，一次有超过 8 个线程访问容器），那么最好的解决方案是使用 libCDS 库中的容器：https://github.com/khizmax/libcds
- 如果 libCDS 具有您需要的线程安全容器，那么就使用它。
- 如果您的程序执行其他操作，而不仅仅是线程之间的数据交换，并且线程之间的交换不超过总时间的 30%，那么即使您使用几十个线程，我们的
contfree_safe_ptr<>指针也比 libCDS 中的 map 容器更快。 - 如果您使用自己的容器或数据结构（不在 libCDS 中）进行线程间数据交换，那么
contfree_safe_ptr<T>将是一个快速简便的解决方案。这比为您的自定义数据结构编写无锁方法的实现要容易得多。
如果您想了解如何构建包含数百个 CPU、GPU 和 FPGA 的 HPC 系统，您可以查阅这篇文章：http://www.alexeyab.com/2017/04/the-fastest-interconnect-for-hundreds.html。
您希望在 Boost 库中看到本文中分析的算法的修改形式吗？
历史
- 2017年4月24日 - 初始版本
- 2017年5月1日 - 添加了GitHub链接