c++ - 如何在没有内存泄漏的情况下同时(线程安全)填充 c++11 std::map*>？

Question

基本上，我需要使用来自同时读取的数千个文件的数百万个关键条目(或多或少 5000 万个订单)填充 std::map。这些键将指向的值将从堆中分配(std::bitset 类型)。

std::map<std::string,std::bitset<BITSET_SIZE>*> my_map;

1. 我的第一个问题是:我不想要两个线程(首先检查一个key 存在，如果不存在，) 从堆中分配空间。因为我只能持有一个指针，而其他分配会导致内存泄漏，因为我无法跟踪它们。

   //count should be thread-safe, since it's defined as const in <map> header file
   if(my_map.count(key) == 0){
       //some other thread may have initialized the key in the mean time
       my_map[key] = new std::bitset<BITSET_SIZE>();
       //Now I will lose the pointer to previous heap allocation from other thread
   }
   

   一个解决方案是有一些互斥机制，比如boost::unique_lock 或 boost::shared_lock 的一些智能组合和 boost::unique_lock 是为了提高性能，我很乐意听取您的想法。

2. 假设我已经完成了第一部分，也就是说；在没有内存泄漏的情况下同时初始化 my_map 的键，任务的第二部分是同时操作值 (std::bitset)。为此，我认为应该没有任何问题，因为根据我的设置，可以保证不会有两个线程同时处理同一个键。 (任何线程都不会为 my_map 的键添加新键或从底层树结构中删除键)

Answer 1

const访问 std::即使没有同步，容器(如 map)也保证在不同线程中是合法的。

任何非 const没有同步的访问会使任何其他访问( const 或非 const )非法(程序行为未定义)。

有些操作不是 const , 但就同步而言是 const .例如，非常量 find被视为“作为 const ”，作为 []在 vector 上.

[]在 map 上不是 const并且不被视为 const .我不确定 []不创建元素的被视为 const ，我将不得不仔细检查标准。而作为find存在并通过明确定义的语义解决了相同的问题，无论如何我都不会在代码中使用它。

const并不意味着线程安全，它意味着线程安全与其他const操作。线程安全是两个或多个代码位之间的关系，它不是绝对的。所以打电话.count而其他人正在插入是不合法的。

一般来说，共享是线程安全的祸根。解决这个问题的更简单方法是为每个“任务”分配自己的 map跟...共事。然后你合并这些 map回到主 map 。

合并的复杂程度和发生频率取决于具体的应用程序以及有多少复制。

最简单的解决方案是:

std::map<std::string, std::unique_ptr<std::bitset<BITSET_SIZE>>>
parse_file( some_file_handle );

然后

std::map<std::string, std::unique_ptr<std::bitset<BITSET_SIZE>>>
parse_files( gsl::span<some_file_handle> handles ) {
  if (handles.size()==0) return {};
  if (handles.size()==1) return parse_file(handles.front());
  auto lhs = parse_files( handles.first(handles.size()/2) );
  auto rhs = parse_files( handles.last(handles.size()-handles.size()/2) );
  return merge_maps(std::move(lhs), std::move(rhs));
}

为我们提供了单线程版本。我们通过以下方式对其进行多线程处理:

std::map<std::string, std::unique_ptr<std::bitset<BITSET_SIZE>>>
parse_files( gsl::span<some_file_handle> handles, executor exec ) {
  if (handles.size()==0) return {};
  if (handles.size()==1) return parse_file(handles.front());
  auto lhs = exec( [handles]{parse_files(handles.first(handles.size()/2) )} );
  auto rhs = exec( [handles]{parse_files(handles.last(handles.size()-handles.size()/2) )} );
  auto retval = exec( [lhs=std::move(lhs], rhs=std::move[rhs]]()mutable{
    return merge_maps(std::move(lhs).get(), std::move(rhs).get() );
  }
  return std::move(retval).get();
}

哪里executor is 采用 T() 类型的对象并返回 future<T> .天真的执行者只是运行函数并返回一个准备好的 future 。一个更高级的执行者使用 std::async把它拧下来。一个更高级的方法是使用线程池，并且在等待时使用等待线程来运行任务(如果任务尚未运行)。

现在，像 ppl 或 Intel 的 TBB 这样的并发库提供了很容易做到这一点的方法。