• C++ 实现线程安全的map(OpenHarmony源码实现版)

    概述

    STL容器不是线程安全的。比如对于vector，即使写方（生产者）是单线程写入，但是并发读的时候，由于潜在的内存重新申请和对象复制问题，会导致读方（消费者）的迭代器失效。实际表现也就是招致了core dump。另外一种情况，如果是多个写方，并发的push_back()，也会导致core dump。但可以通过固定vector的大小（调用resize()）避免动态扩容（无push_back）来做到lock-free。

    c++的map的并发操作也是不安全的，c++里边有红黑树实现的std::map和hash表 unordered_map。在《C++并发编程实战》一书中的162页提供了一个细粒度锁的MAP数据结构，使用了 boost的shared_mutex (C++14已经支持，C++11没有)，那上面的实现代码挺长的，这里给出个OpenHarmony源码实现的safe_map，代码精简，值得学习。

    接上篇欣赏了OpenHarmony源码中实现的ThreadPool的实现，链接在这里：

    c++的ThreadPool,OpenHarmony源码实现版赏析和使用

    源码实现

    源码位置：code-v3.0-LTS\OpenHarmony\utils\native\base\include\safe_map.h

    1. /*
    2.  * Copyright (c) 2021 Huawei Device Co., Ltd.
    3.  * Licensed under the Apache License, Version 2.0 (the "License");
    4.  * you may not use this file except in compliance with the License.
    5.  * You may obtain a copy of the License at
    6.  *
    7.  *     http://www.apache.org/licenses/LICENSE-2.0
    8.  *
    9.  * Unless required by applicable law or agreed to in writing, software
    10.  * distributed under the License is distributed on an "AS IS" BASIS,
    11.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12.  * See the License for the specific language governing permissions and
    13.  * limitations under the License.
    14.  */
    15.  
    16. #ifndef UTILS_BASE_SAFE_MAP_H
    17. #define UTILS_BASE_SAFE_MAP_H
    18.  
    19. #include 
    20. #include 
    21.  
    22. namespace OHOS {
    23.  
    24. template <typename K, typename V>
    25. class SafeMap {
    26. public:
    27.     SafeMap() {}
    28.  
    29.     ~SafeMap() {}
    30.  
    31.     SafeMap(const SafeMap& rhs)
    32.     {
    33.         map_ = rhs.map_;
    34.     }
    35.  
    36.     SafeMap& operator=(const SafeMap& rhs)
    37.     {
    38.         if (&rhs != this) {
    39.             map_ = rhs.map_;
    40.         }
    41.  
    42.         return *this;
    43.     }
    44.  
    45.     V& operator[](const K& key)
    46.     {
    47.         return map_[key];
    48.     }
    49.  
    50.     // when multithread calling size() return a tmp status, some threads may insert just after size() call
    51.     int Size()
    52.     {
    53.         std::lock_guard lock(mutex_);
    54.         return map_.size();
    55.     }
    56.  
    57.     // when multithread calling Empty() return a tmp status, some threads may insert just after Empty() call
    58.     bool IsEmpty()
    59.     {
    60.         std::lock_guard lock(mutex_);
    61.         return map_.empty();
    62.     }
    63.  
    64.     bool Insert(const K& key, const V& value)
    65.     {
    66.         std::lock_guard lock(mutex_);
    67.         auto ret = map_.insert(std::pair(key, value));
    68.         return ret.second;
    69.     }
    70.  
    71.     void EnsureInsert(const K& key, const V& value)
    72.     {
    73.         std::lock_guard lock(mutex_);
    74.         auto ret = map_.insert(std::pair(key, value));
    75.         // find key and cannot insert
    76.         if (!ret.second) {
    77.             map_.erase(ret.first);
    78.             map_.insert(std::pair(key, value));
    79.             return;
    80.         }
    81.         return;
    82.     }
    83.  
    84.     bool Find(const K& key, V& value)
    85.     {
    86.         bool ret = false;
    87.         std::lock_guard lock(mutex_);
    88.  
    89.         auto iter = map_.find(key);
    90.         if (iter != map_.end()) {
    91.             value = iter->second;
    92.             ret = true;
    93.         }
    94.  
    95.         return ret;
    96.     }
    97.  
    98.     bool FindOldAndSetNew(const K& key, V& oldValue, const V& newValue)
    99.     {
    100.         bool ret = false;
    101.         std::lock_guard lock(mutex_);
    102.         if (map_.size() > 0) {
    103.             auto iter = map_.find(key);
    104.             if (iter != map_.end()) {
    105.                 oldValue = iter->second;
    106.                 map_.erase(iter);
    107.                 map_.insert(std::pair(key, newValue));
    108.                 ret = true;
    109.             }
    110.         }
    111.  
    112.         return ret;
    113.     }
    114.  
    115.     void Erase(const K& key)
    116.     {
    117.         std::lock_guard lock(mutex_);
    118.         map_.erase(key);
    119.     }
    120.  
    121.     void Clear()
    122.     {
    123.         std::lock_guard lock(mutex_);
    124.         map_.clear();
    125.         return;
    126.     }
    127.  
    128. private:
    129.     std::mutex mutex_;
    130.     std::map map_;
    131. };
    132.  
    133. } // namespace OHOS
    134. #endif

    源码欣赏

    使用模板语法template 让这个map的实现更通用。这是c++模板泛型的强大之处，不用针对每个类型都实现一遍，复用性更强。且模板是在编译期检查的，也降低的出错的可能性。内部实现上，倒是没啥特别的，就是对相应的操作加了锁。锁使用的RAII模型的std::lock_guard写法，这种很常见也很常用。

    自定义实现了几个常用的操作方法如Find,Erase和Clear，每个里面的操作都相应的加了锁。操作符重载实现了[]和赋值=操作。注意这两处的地方没有用锁，你知道为什么吗？如果多个线程只访问容器但不更改其结构，则不需要对容器进行同步。另外一个原因是，对于map,如果只是通过[]的方式修改而不是新插入，则多线程下也不会core dump。

    单元测试

    源码中同样有safe_map的单元测试，单元测试框架使用的是google的gtest。看来gtest还是很强大的，华为也选择使用了它。以下给出源码，可以熟悉下gtest单元测试的用法。

    1. /*
    2.  * Copyright (c) 2021 Huawei Device Co., Ltd.
    3.  * Licensed under the Apache License, Version 2.0 (the "License");
    4.  * you may not use this file except in compliance with the License.
    5.  * You may obtain a copy of the License at
    6.  *
    7.  *     http://www.apache.org/licenses/LICENSE-2.0
    8.  *
    9.  * Unless required by applicable law or agreed to in writing, software
    10.  * distributed under the License is distributed on an "AS IS" BASIS,
    11.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12.  * See the License for the specific language governing permissions and
    13.  * limitations under the License.
    14.  */
    15. #include "safe_map.h"
    16.  
    17. #include 
    18. #include 
    19. #include 
    20. #include 
    21. #include 
    22.  
    23. #include 
    24.  
    25. #include  // std::chrono::seconds
    26. #include  // std::cout
    27. #include  // std::thread, std::this_thread::sleep_for
    28.  
    29. using namespace testing::ext;
    30. using namespace OHOS;
    31. using namespace std;
    32.  
    33. class UtilsSafeMap : public testing::Test {
    34. };
    35.  
    36. /*
    37.  * @tc.name: testUtilsCopyAndAssign001
    38.  * @tc.desc: single thread test the normal feature insert and erase and EnsureInsert
    39.  */
    40. HWTEST_F(UtilsSafeMap, testUtilsCopyAndAssign001, TestSize.Level1)
    41. {
    42.     SafeMapint> demoData;
    43.     // insert new
    44.     demoData.Insert("A", 1);
    45.     ASSERT_FALSE(demoData.IsEmpty());
    46.     ASSERT_EQ(demoData.Size(), 1);
    47.  
    48.     SafeMapint> newdemo = demoData;
    49.     int tar = -1;
    50.     ASSERT_TRUE(newdemo.Find("A", tar));
    51.     ASSERT_EQ(1, tar);
    52.  
    53.     tar = -1;
    54.     SafeMapint> newdemo2;
    55.     newdemo2 = demoData;
    56.     ASSERT_TRUE(newdemo2.Find("A", tar));
    57.     ASSERT_EQ(1, tar);
    58. }
    59.  
    60. /*
    61.  * @tc.name: testUtilsoperator001
    62.  * @tc.desc: SafeMap
    63.  */
    64. HWTEST_F(UtilsSafeMap, testUtilsoperator001, TestSize.Level1)
    65. {
    66.     SafeMapint> demoData;
    67.     // insert new
    68.     demoData.Insert("A", 1);
    69.     ASSERT_FALSE(demoData.IsEmpty());
    70.     ASSERT_EQ(demoData.Size(), 1);
    71.     ASSERT_EQ(demoData["A"], 1);
    72.  
    73.     SafeMapint> newdemo = demoData;
    74.     ASSERT_EQ(newdemo["A"], 1);
    75.  
    76.     int tar = -1;
    77.     newdemo["B"] = 6;
    78.     ASSERT_TRUE(newdemo.Find("B", tar));
    79.     ASSERT_EQ(6, tar);
    80.  
    81.     SafeMapint> newdemo2;
    82.     newdemo2 = newdemo;
    83.     ASSERT_EQ(newdemo2["A"], 1);
    84. }
    85.  
    86. /*
    87.  * @tc.name: testUtilsNormalFeatureInsert001
    88.  * @tc.desc: SafeMap
    89.  */
    90. HWTEST_F(UtilsSafeMap, testUtilsNormalFeatureInsert001, TestSize.Level1)
    91. {
    92.     SafeMapint> demoData;
    93.     ASSERT_TRUE(demoData.IsEmpty());
    94.     // insert new
    95.     demoData.Insert("A", 1);
    96.     ASSERT_FALSE(demoData.IsEmpty());
    97.     ASSERT_EQ(demoData.Size(), 1);
    98.  
    99.     // insert copy one should fail
    100.     ASSERT_FALSE(demoData.Insert("A", 2));
    101.     ASSERT_EQ(demoData.Size(), 1);
    102. }
    103.  
    104. /*
    105.  * @tc.name: testUtilsNormalFeatureEnsureInsert001
    106.  * @tc.desc: SafeMap
    107.  */
    108. HWTEST_F(UtilsSafeMap, testUtilsNormalFeatureEnsureInsert001, TestSize.Level1)
    109. {
    110.     SafeMapint> demoData;
    111.     ASSERT_TRUE(demoData.IsEmpty());
    112.  
    113.     demoData.Insert("A", 1);
    114.     demoData.EnsureInsert("B", 2);
    115.  
    116.     ASSERT_FALSE(demoData.IsEmpty());
    117.     ASSERT_EQ(demoData.Size(), 2);
    118.     // insert copy one and new one
    119.     demoData.EnsureInsert("B", 5);
    120.     demoData.EnsureInsert("C", 6);
    121.  
    122.     ASSERT_EQ(demoData.Size(), 3);
    123. }
    124.  
    125. /*
    126.  * @tc.name: testUtilsNormalFeatureFind001
    127.  * @tc.desc: SafeMap
    128.  */
    129. HWTEST_F(UtilsSafeMap, testUtilsNormalFeatureFind001, TestSize.Level1)
    130. {
    131.     SafeMapint> demoData;
    132.     ASSERT_TRUE(demoData.IsEmpty());
    133.  
    134.     demoData.Insert("A", 1);
    135.     demoData.Insert("B", 10000);
    136.     demoData.EnsureInsert("B", 2);
    137.     demoData.EnsureInsert("C", 6);
    138.  
    139.     ASSERT_FALSE(demoData.IsEmpty());
    140.     ASSERT_EQ(demoData.Size(), 3);
    141.  
    142.     int i = 0;
    143.     ASSERT_TRUE(demoData.Find("A", i));
    144.     ASSERT_EQ(i, 1);
    145.     ASSERT_TRUE(demoData.Find("B", i));
    146.     ASSERT_EQ(i, 2);
    147.  
    148.     ASSERT_TRUE(demoData.Find("C", i));
    149.     ASSERT_EQ(i, 6);
    150. }
    151.  
    152. /*
    153.  * @tc.name: testUtilsNormalFeatureFindAndSet001
    154.  * @tc.desc: SafeMap
    155.  */
    156. HWTEST_F(UtilsSafeMap, testUtilsNormalFeatureFindAndSet001, TestSize.Level1)
    157. {
    158.     SafeMapint> demoData;
    159.     ASSERT_TRUE(demoData.IsEmpty());
    160.  
    161.     demoData.Insert("A", 1);
    162.     demoData.EnsureInsert("B", 2);
    163.  
    164.     int oldvalue = 0;
    165.     int newvalue = 3;
    166.     ASSERT_TRUE(demoData.FindOldAndSetNew("A", oldvalue, newvalue));
    167.     // old value
    168.     ASSERT_EQ(oldvalue, 1);
    169.  
    170.     newvalue = 4;
    171.     ASSERT_TRUE(demoData.FindOldAndSetNew("B", oldvalue, newvalue));
    172.     // old value
    173.     ASSERT_EQ(oldvalue, 2);
    174.     int i = -1;
    175.     ASSERT_TRUE(demoData.Find("A", i));
    176.     // new value
    177.     ASSERT_EQ(i, 3);
    178.     ASSERT_TRUE(demoData.Find("B", i));
    179.     // new value
    180.     ASSERT_EQ(i, 4);
    181. }
    182.  
    183. /*
    184.  * @tc.name: testUtilsNormalFeatureEraseAndClear001
    185.  * @tc.desc: SafeMap
    186.  */
    187. HWTEST_F(UtilsSafeMap, testUtilsNormalFeatureEraseAndClear001, TestSize.Level1)
    188. {
    189.     SafeMapint> demoData;
    190.     ASSERT_TRUE(demoData.IsEmpty());
    191.  
    192.     demoData.Insert("A", 1);
    193.     demoData.EnsureInsert("B", 2);
    194.  
    195.     ASSERT_EQ(demoData.Size(), 2);
    196.     demoData.Erase("A");
    197.     ASSERT_EQ(demoData.Size(), 1);
    198.  
    199.     demoData.Clear();
    200.     ASSERT_EQ(demoData.Size(), 0);
    201. }
    202.  
    203. /*
    204.  * @tc.name: testUtilsConcurrentWriteAndRead001
    205.  * @tc.desc: 100 threads test in writein to the same key of the map, while read at same time  and no throw
    206.  */
    207. const int THREAD_NUM = 100;
    208. HWTEST_F(UtilsSafeMap, testUtilsConcurrentWriteAndRead001, TestSize.Level1)
    209. {
    210.     SafeMapint> demoData;
    211.     std::thread threads[THREAD_NUM];
    212.     std::thread checkThread[THREAD_NUM];
    213.     ASSERT_NO_THROW({
    214.         auto lamfuncInsert = [](SafeMapint>& data, const string& key,
    215.             const int& value, const std::chrono::system_clock::time_point& absTime) {
    216.             std::this_thread::sleep_until(absTime);
    217.             data.EnsureInsert(key, value);
    218.         };
    219.  
    220.         auto lamfuncCheck = [](SafeMapint>& data, const string& key,
    221.             std::chrono::system_clock::time_point absTime) {
    222.             std::this_thread::sleep_until(absTime);
    223.             thread_local int i = -1;
    224.             data.Find(key, i);
    225.         };
    226.  
    227.         using std::chrono::system_clock;
    228.  
    229.         std::time_t timeT = system_clock::to_time_t(system_clock::now());
    230.         timeT += 2;
    231.         string key("A");
    232.  
    233.         for (int i = 0; i < THREAD_NUM; ++i) {
    234.             threads[i] = std::thread(lamfuncInsert, std::ref(demoData), key, i, system_clock::from_time_t(timeT));
    235.             checkThread[i] = std::thread(lamfuncCheck, std::ref(demoData), key, system_clock::from_time_t(timeT));
    236.         }
    237.  
    238.         std::this_thread::sleep_for(std::chrono::seconds(3));
    239.         for (auto& t : threads) {
    240.             t.join();
    241.         }
    242.  
    243.         for (auto& t : checkThread) {
    244.             t.join();
    245.         }
    246.     });
    247. }
    248.  
    249. /*
    250.  * @tc.name: testUtilsConcurrentWriteAndFind001
    251.  * @tc.desc: 100 threads test in writein to the corresponding key of the map,
    252.  * while read at same time  and check the results
    253.  */
    254. HWTEST_F(UtilsSafeMap, testUtilsConcurrentWriteAndFind001, TestSize.Level1)
    255. {
    256.     SafeMapint> demoData;
    257.     std::thread threads[THREAD_NUM];
    258.     std::vectorint>> vcfi;
    259.  
    260.     ASSERT_NO_THROW({
    261.         auto lamfuncInsert = [](SafeMapint>& data, const string& key,
    262.             const int& value, const std::chrono::system_clock::time_point& absTime) {
    263.             std::this_thread::sleep_until(absTime);
    264.             data.EnsureInsert(key, value);
    265.         };
    266.  
    267.         auto lamfuncCheckLoop = [](SafeMapint>& data, const string& key,
    268.             std::chrono::system_clock::time_point absTime) {
    269.             std::this_thread::sleep_until(absTime);
    270.             thread_local int i = -1;
    271.             while (!data.Find(key, i)) {
    272.                 std::this_thread::sleep_for(std::chrono::microseconds(10));
    273.             }
    274.             return i;
    275.         };
    276.  
    277.         using std::chrono::system_clock;
    278.  
    279.         std::time_t timeT = system_clock::to_time_t(system_clock::now());
    280.         timeT += 2;
    281.         string key("A");
    282.  
    283.         for (int i = 0; i < THREAD_NUM; ++i) {
    284.             threads[i] = std::thread(lamfuncInsert, std::ref(demoData),
    285.                 key + std::to_string(i), i, system_clock::from_time_t(timeT));
    286.             vcfi.push_back(std::async(std::launch::async, lamfuncCheckLoop,
    287.                 std::ref(demoData), key + std::to_string(i), system_clock::from_time_t(timeT)));
    288.         }
    289.  
    290.         std::this_thread::sleep_for(std::chrono::seconds(4));
    291.         for (auto& t : threads) {
    292.             t.join();
    293.         }
    294.  
    295.         vector<int> result;
    296.  
    297.         for (auto& t : vcfi) {
    298.             result.push_back(t.get());
    299.         }
    300.  
    301.         std::sort(result.begin(), result.end());
    302.  
    303.         for (int i = 0; i < THREAD_NUM; ++i) {
    304.             ASSERT_EQ(i, result[i]);
    305.         }
    306.     });
    307. }
    308.  
    309. /*
    310.  * @tc.name: testUtilsConcurrentWriteAndFindAndSet001
    311.  * @tc.desc: 100 threads test in writein to the corresponding key of the map,
    312.  * while findandfix at same time  and check the results
    313.  */
    314. HWTEST_F(UtilsSafeMap, testUtilsConcurrentWriteAndFindAndSet001, TestSize.Level1)
    315. {
    316.     SafeMapint> demoData;
    317.     std::thread threads[THREAD_NUM];
    318.     std::vectorint>> vcfi;
    319.  
    320.     ASSERT_NO_THROW({
    321.         auto lamfuncInsert = [](SafeMapint>& data, const string& key,
    322.             const int& value, const std::chrono::system_clock::time_point& absTime) {
    323.             std::this_thread::sleep_until(absTime);
    324.             data.EnsureInsert(key, value);
    325.         };
    326.  
    327.         auto lamfuncCheckLoop = [](SafeMapint>& data, const string& key,
    328.             const int& newvalue, std::chrono::system_clock::time_point absTime) {
    329.             std::this_thread::sleep_until(absTime);
    330.             thread_local int i = -1;
    331.             while (!data.FindOldAndSetNew(key, i, newvalue)) {
    332.                 std::this_thread::sleep_for(std::chrono::microseconds(10));
    333.             }
    334.             return i;
    335.         };
    336.  
    337.         using std::chrono::system_clock;
    338.  
    339.         std::time_t timeT = system_clock::to_time_t(system_clock::now());
    340.         timeT += 2;
    341.         string key("A");
    342.  
    343.         for (int i = 0; i < THREAD_NUM; ++i) {
    344.             threads[i] = std::thread(lamfuncInsert, std::ref(demoData),
    345.                 key + std::to_string(i), i, system_clock::from_time_t(timeT));
    346.             vcfi.push_back(std::async(std::launch::async, lamfuncCheckLoop,
    347.                 std::ref(demoData), key + std::to_string(i), i + 1, system_clock::from_time_t(timeT)));
    348.         }
    349.  
    350.         std::this_thread::sleep_for(std::chrono::seconds(4));
    351.         for (auto& t : threads) {
    352.             t.join();
    353.         }
    354.  
    355.         vector<int> result;
    356.         for (auto& t : vcfi) {
    357.             result.push_back(t.get());
    358.         }
    359.  
    360.         std::sort(result.begin(), result.end());
    361.  
    362.         for (int i = 0; i < THREAD_NUM; ++i) {
    363.             ASSERT_EQ(i, result[i]);
    364.         }
    365.  
    366.         int t = 0;
    367.         result.clear();
    368.         for (int i = 0; i < THREAD_NUM; ++i) {
    369.             t = -1;
    370.             ASSERT_TRUE(demoData.Find("A" + std::to_string(i), t));
    371.             result.push_back(t);
    372.         }
    373.  
    374.         std::sort(result.begin(), result.end());
    375.  
    376.         for (int i = 0; i < THREAD_NUM; ++i) {
    377.             ASSERT_EQ(i + 1, result[i]);
    378.         }
    379.     });
    380. }

    引用

    C++线程安全map （低效率） | 码农家园

    C++11:基于std::queue和std::mutex构建一个线程安全的队列_10km的博客-CSDN博客_std::queue 线程安全

    C++11:基于std::unordered_map和共享锁构建线程安全的map_10km的博客-CSDN博客_c++ map 无锁

    如何设计并实现一个线程安全的 Map ？(上篇) - JavaShuo

    c++线程安全的map_clh01s的博客-CSDN博客_c++ 线程安全的map

    C++ 20 线程安全的Map_学习好烦啊的博客-CSDN博客_c++ map 线程安全

    c++ map 多线程_线程安全原理简析及HashMap多线程并发5种场景异常分析_sumilao的博客-CSDN博客

    C++ STL容器如何解决线程安全的问题？ - 腾讯云开发者社区-腾讯云

    RISC-V MCU中文社区_致力于RISC-V技术的推广，提供一个交流学习的开放平台

  • 相关阅读:
    springboot集成mybatis
    ubuntu18安装caffe(CPU)
    石油数字孪生可视化管理平台，推动石油行业数字化转型与智能化应用
    《10人以下小团队管理手册》推荐书评
    webpack-cli 在 webpack 打包中的作用
    如何实现Spring中服务关闭时对象销毁执行代码
    Rust基本数据类型-字符串
    Vuex源码分析，实现简易版的Vuex
    一句话解释什么是出口IP
    PMP是什么？
  • 原文地址：https://blog.csdn.net/qq8864/article/details/127435180