如何在没有内存泄漏的情况下同时（线程安全）填充 c++11 std::map<std::string,std::bitset<N>*>？

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 函数不是线程安全的。考虑：

struct A {
  int q;
  void set(int qq) { q = qq; }
  int get() const { return q; }
};

get() 不是线程安全的 - 可能会调用另一个线程集，这将修改 q.如果您想要线程安全，则必须使用原子结构锁定或更新（还有其他多线程问题，如果您不锁定/使用原子结构，则会发生这种情况，但这些超出了您的问题范围 - 您绝对需要那些！）。

现在解决： 因为您需要显式同步对地图结构的访问，所以您的问题不会成为问题：

std::mutex m; // since c++11
...
{
    std::lock_guard _l(m); // since c++11
    if (!my_map.emplace(key, bitset_ptr).second)
        delete bitset_ptr;
}

这会将元素插入到 my_map 中，键为 key，值为 bitset_ptr，但前提是它不存在。它将返回两个元素的元组 - 第一个是创建的元素和先前存在的元素的迭代器，第二个是布尔标志，如果元素已创建，则为 true，如果之前存在，则为 false。因此，如果元素已经插入并且没有内存泄漏，您只需删除 bitset_ptr。请注意，由于同步量的原因，这可能会很慢。

更新： 只要您在多个线程中不断更新，您显然需要使用 mutex m 同步对 my_map 的任何访问。

更新2： op 尝试了最简单的解决方案，但发现速度不够快。让我们更深入。 （注意：最理想的做法是测量应用程序的性能并找到代码花费大部分时间的地方，但我不能这样做；））。减速的“明显”（阅读：可能）原因很少：

• 插入地图 - 插入已排序地图具有 O(ln n) 运行时性能，实际上可能由于缓存不匹配而非常慢。平均而言，在 100 万个元素映射中，您需要与 10 个不同的字符串进行比较，这些字符串（可能）位于完全不同的内存区域，从而始终从处理器缓存中相互清除。
• 从文件中读取 - 从多个文件中读取小块可能（也可能不会！）不利于整体速度。
• 多次分配 - 通常内存分配很慢。此外，大量分配会增加内存碎片并降低局部性。
• 锁定同步 - 这对任何事情都不利...

我假设，您不能轻松（廉价）确定单个文件中的元素数量和总数。先上代码：

using etype = pair<string, bitset<N>*>;
vector<etype> all_elements;
mutex all_elements_mutex;
void parse_single_file_in_thread(...) {
    vector<etype> tmp;
    for(auto element : parse_element_from_file()) 
        tmp.push_back(move(element));
    lock_guard _l(all_elements_mutex);
    for(auto &a : tmp) all_elements.push_back(move(a));
}
map<string, bitset<N>*> parse_all_files() {
    // create threads, parse files in them and wait for them to finish
    std::sort(all_elements.begin(), all_elements.end(), 
        [](const etype &a, const etype &b) { return a.first < b.first; });
    map<string, bitset<N>*> tmp;
    for(auto &a : all_elements) if (!tmp.insert(tmp.end(), etype(move(a.first), a.second)).second) delete a.second;
    all_elements.clear();
    return tmp;
}

它做了几件事： - 首先将键插入向量（忽略检查重复项），稍后将对其进行排序并插入带有位置提示的地图（它们已排序，因此我们总是知道插入下一个元素的正确位置 - 地图的末尾），这是比直接插入地图快得多 - 每个文件的项目首先放入它自己的向量中，并在解析整个文件后移动到全局向量中，这样可以最大限度地减少锁定

这应该足以提高性能。接下来的事情是用其他东西替换字符串，以避免太多字符串到字符串的排序比较。但这很容易超出范围。 ;)

注意：我已经从内存中编写了整个代码，因此它可能无法编译并且可能需要 c++17。

Answer 2

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

任何未同步的非const 访问都会使任何其他访问（const 或非const）非法（程序行为变得未定义）。

有些操作不是const，但就同步而言是const。例如，非 const find 被视为“const”，vector 上的 [] 也是如此。

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

const 并不意味着 线程安全，它意味着 其他const 操作的线程安全。线程安全是两个或多个代码位之间的关系，它不是绝对的。因此，在其他人插入时调用 .count 是不合法的。

一般来说，共享是线程安全的祸根。解决此问题的更简单方法是为每个“任务”提供自己的map 以供使用。然后将这些maps 合并回主地图。

合并发生的复杂程度和频率取决于特定应用程序和复制的数量。

最简单的解决方案是：

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 接受T() 类型的对象并返回future<T>。天真的执行者只是简单地运行函数并返回一个准备好的未来。更高级的 executor 使用std::async 将其线程化。更高级的使用线程池，当等待时使用等待线程来运行任务（如果它尚未运行）。

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