c++ - 使用带有std::multimap的boost::iostreams::mapped_file_source

Question

我要分析的数据很多-每个文件大约5gigs。每个文件具有以下格式:

xxxxx yyyyy

键和值都可以重复，但是键以升序排序。我试图为此目的使用一个内存映射文件，然后找到所需的键并使用它们。这是我写的:

if (data_file != "")
{
    clock_start = clock();
    data_file_mapped.open(data_file);

    data_multimap = (std::multimap<double, unsigned int> *)data_file_mapped.data();
    if (data_multimap != NULL)
    {
        std::multimap<double, unsigned int>::iterator it = data_multimap->find(keys_to_find[4]);
        if (it != data_multimap->end())
        {
            std::cout << "Element found.";
            for (std::multimap<double, unsigned int>::iterator it = data_multimap->lower_bound(keys_to_find[4]); it != data_multimap->upper_bound(keys_to_find[5]); ++it)
            {
                std::cout << it->second;
            }
            std::cout << "\n";
            clock_end = clock();

            std::cout << "Time taken to read in the file: " << (clock_end - clock_start)/CLOCKS_PER_SEC << "\n";
        }
        else
            std::cerr << "Element not found at all" << "\n";
    }
    else
        std::cerr << "Nope - no data received."<< "\n";
}

基本上，我需要找到键的范围并将其拉出以进行处理。我第一次尝试在多图上使用方法时遇到段错误。例如，当调用 find方法时。我也尝试了 upper_bound， lower_bound和其他方法，仍然遇到了段错误。

这是 gdb给我的:

Program received signal SIGSEGV, Segmentation fault.
_M_lower_bound (this=<optimized out>, __k=<optimized out>, __y=<optimized out>, __x=0xa31202030303833) at /usr/include/c++/4.9.2/bits/stl_tree.h:1261
1261            if (!_M_impl._M_key_compare(_S_key(__x), __k))

有人可以指出我做错了吗？我只能在内存映射文件上找到简单的示例-尚无类似的例子。谢谢。

编辑:更多信息:

我上面描述的文件基本上是两列纯文本文件，神经仿真程序将其作为仿真输出提供给我。像这样简单:

$ du -hsc 201501271755.e.ras 
4.9G    201501271755.e.ras
4.9G    total
$ head 201501271755.e.ras 
0.013800  0
0.013800  1
0.013800  10
0.013800  11
0.013800  12
0.013800  13
0.013800  14
0.013800  15
0.013800  16
0.013800  17

第一列是时间，第二列是此时触发的神经元-(这是尖峰时间栅格文件)。实际上，输出是来自每个MPI等级的此类文件，用于运行模拟。使用 sort -g -m将各种文件组合到此主文件中。有关文件格式的更多信息，请参见: http://www.fzenke.net/auryn/doku.php?id=manual:ras

要计算在特定模拟时间的神经元集合的激发速率和其他指标，我需要-在文件中找到时间，在[time -1，time]之间拉出一个块，并运行一些指标，依此类推这个块。该文件非常小，随着仿真变得越来越复杂并且运行时间更长，我希望文件大小会增加很多。这就是为什么我开始研究内存映射文件的原因。我希望这可以澄清问题陈述。我只需要读取输出文件即可处理其中包含的信息。我根本不需要修改此文件。

为了处理数据，我将再次使用多个线程，但是由于我在 map 上的所有操作都是只读的，因此我不希望在那里遇到麻烦。

Answer 1

您正在尝试您不太了解的事情:)没问题。

多个映射不会在内存中顺序排列。 (它们是基于节点的容器，但我离题了)。实际上，即使是这样，布局与文本输入的布局匹配的机会也很小。

基本上有两种方法可以使这项工作:

继续使用multimap，但使用自定义分配器(以便所有分配都在映射的内存区域中完成)。从高级C++的 Angular 来看，这是“最聪明的”，/ but /您将需要更改为文件的二进制格式。

如果可以的话，这就是我的建议。 Boost Container + Boost Interprocess提供了使此过程相对轻松的一切。

您编写了一个直接对映射数据起作用的自定义容器“抽象”。你可以

从任何地方(行尾？)或

识别“xxxx yyyy”对

建立文件中所有行开始的索引。

使用这些可以设计一个插入器(Boost Iterator iterator_facade)，可以用来实现更高级别的操作( lower_bound， upper_bound和 equal_range)。

一旦拥有了这些，就基本上都准备将这个内存映射作为一个只读键值数据库进行查询。

可悲的是，如果您还想支持变异操作( insert， remove)，则这种内存表示形式将对性能造成极大的损害。
如果您有文件的实际样本，我可以对上述两种方法进行演示。

更新资料

快速样本:

使用boost::interprocess，您可以(非常)简单地定义所需的多图:

namespace shared {
    namespace bc = boost::container;

    template <typename T> using allocator = bip::allocator<T, bip::managed_mapped_file::segment_manager>;
    template <typename K, typename V>
        using multi_map = bc::flat_multimap<
            K, V, std::less<K>, 
            allocator<typename bc::flat_multimap<K, V>::value_type> >;
}

笔记:

我选择了flatmap(实际上是flat_multimap)，因为它可能更多
存储效率高，与第二种方法相比更具可比性
(如下所示)；

请注意，此选择会影响迭代器/引用的稳定性，并且会
非常喜欢只读操作。如果需要迭代器
稳定性和/或许多变异操作，请使用常规的map(或
非常大的hash_map)，而不是平坦的变体形式。

我为此演示选择了managed_mapped_file段(这样您就可以保持持久性)。该演示演示了如何稀疏地预分配10G，但只有实际分配的空间才在磁盘上使用。您同样可以使用managed_shared_memory。

如果具有二进制持久性，则可能会完全丢弃文本数据文件。

我使用Boost Spirit将文本数据从shared::multi_map<double, unsigned>解析为mapped_file_source。该实现是完全通用的。

无需编写iterator类，start_of_line()，end_of_line()，lower_bound()，upper_bound()，equal_range()或其他任何一种，因为它们已经是multi_map接口(interface)中的标准设置，因此我们只需要编写main:

Live On Coliru

#define NDEBUG
#undef DEBUG
#include <boost/iostreams/device/mapped_file.hpp>
#include <boost/fusion/adapted/std_pair.hpp>
#include <boost/container/flat_map.hpp>
#include <boost/interprocess/managed_mapped_file.hpp>
#include <boost/spirit/include/qi.hpp>
#include <iomanip>

namespace bip = boost::interprocess;
namespace qi = boost::spirit::qi;

namespace shared {
    namespace bc = boost::container;

    template <typename T> using allocator = bip::allocator<T, bip::managed_mapped_file::segment_manager>;
    template <typename K, typename V>
        using multi_map = bc::flat_multimap<
            K, V, std::less<K>, 
            allocator<typename bc::flat_multimap<K, V>::value_type> >;
}

#include <iostream>

bip::managed_mapped_file msm(bip::open_or_create, "lookup.bin", 10ul<<30);

template <typename K, typename V>
shared::multi_map<K,V>& get_or_load(const char* fname) {
    using Map = shared::multi_map<K, V>;
    Map* lookup = msm.find_or_construct<Map>("lookup")(msm.get_segment_manager());

    if (lookup->empty()) { 
        // only read input file if not already loaded
        boost::iostreams::mapped_file_source input(fname);
        auto f(input.data()), l(f + input.size());

        bool ok = qi::phrase_parse(f, l,
                (qi::auto_ >> qi::auto_) % qi::eol >> *qi::eol, 
                qi::blank, *lookup);

        if (!ok || (f!=l))
            throw std::runtime_error("Error during parsing at position #" + std::to_string(f - input.data()));
    }

    return *lookup;
}

int main() {
    // parse text file into shared memory binary representation
    auto const& lookup = get_or_load<double, unsigned int>("input.txt");
    auto const e = lookup.end();

    for(auto&& line : lookup)
    {
        std::cout << line.first << "\t" << line.second << "\n";

        auto er = lookup.equal_range(line.first);

        if (er.first  != e) std::cout << " lower: " << er.first->first  << "\t" << er.first->second  << "\n";
        if (er.second != e) std::cout << " upper: " << er.second->first << "\t" << er.second->second << "\n";
    }
}

我完全按照我的描述实现了它:

在原始const char*区域上映射的简单容器；

使用boost::iterator_facade生成一个迭代器，该迭代器在取消引用时解析文本；

用于打印输入行，我使用boost::string_ref-避免了动态分配来复制字符串。

的解析是通过Spirit Qi完成的:

if (!qi::phrase_parse(
            b, _data.end,
            qi::auto_ >> qi::auto_ >> qi::eoi,
            qi::space,
            _data.key, _data.value)) 

选择Qi是出于速度和通用性的考虑:您可以在实例化时选择Key和Value类型:

text_multi_lookup<double, unsigned int> tml(map.data(), map.data() + map.size());

我已经实现了lower_bound，upper_bound和equal_range成员函数，这些函数利用了底层连续存储的优势。即使“行” iterator不是随机访问，而是双向的，我们仍然可以跳转到此类迭代器范围的mid_point，因为我们可以将start_of_line从任何const char*放入基础映射区域。这使得二进制搜索有效。

请注意，此解决方案在解引用 iterator时解析行。如果相同的行被多次取消引用，则这可能无效。

但是，对于不频繁的查找或在输入数据的同一区域中不常见的查找，这将尽可能地提高效率(仅执行最小限度的解析和 O(log n)二进制搜索)，同时始终完全绕过初始查找而是通过映射文件来加载时间(无访问权限意味着不需要加载任何内容)。

Live On Coliru (包括测试数据)

#define NDEBUG
#undef DEBUG
#include <boost/iostreams/device/mapped_file.hpp>
#include <boost/utility/string_ref.hpp>
#include <boost/optional.hpp>
#include <boost/spirit/include/qi.hpp>
#include <thread>
#include <iomanip>

namespace io = boost::iostreams;
namespace qi = boost::spirit::qi;

template <typename Key, typename Value> 
struct text_multi_lookup {
    text_multi_lookup(char const* begin, char const* end)
        : _map_begin(begin), 
          _map_end(end)
    {
    }

  private:
    friend struct iterator;
    enum : char { nl = '\n' };

    using rawit = char const*;
    rawit _map_begin, _map_end;

    rawit start_of_line(rawit it) const {
        while (it > _map_begin) if (*--it == nl) return it+1;
        assert(it == _map_begin);
        return it;
    }

    rawit end_of_line(rawit it) const {
        while (it < _map_end) if (*it++ == nl) return it;
        assert(it == _map_end);
        return it;
    }

  public:
    struct value_type final {
        rawit beg, end;
        Key   key;
        Value value;

        boost::string_ref str() const { return { beg, size_t(end-beg) }; }
    };

    struct iterator : boost::iterator_facade<iterator, boost::string_ref, boost::bidirectional_traversal_tag, value_type> {

        iterator(text_multi_lookup const& d, rawit it) : _region(&d), _data { it, nullptr, Key{}, Value{} } { 
            assert(_data.beg == _region->start_of_line(_data.beg));
        }

      private:
        friend text_multi_lookup;

        text_multi_lookup const* _region;
        value_type mutable _data;

        void ensure_parsed() const {
            if (!_data.end) 
            {
                assert(_data.beg == _region->start_of_line(_data.beg));
                auto b = _data.beg;
                _data.end = _region->end_of_line(_data.beg);

                if (!qi::phrase_parse(
                            b, _data.end,
                            qi::auto_ >> qi::auto_ >> qi::eoi,
                            qi::space,
                            _data.key, _data.value)) 
                {
                    std::cerr << "Problem in: " << std::string(_data.beg, _data.end) 
                              << "at:         " << std::setw(_data.end-_data.beg) << std::right << std::string(_data.beg,_data.end);
                    assert(false);
                }
            }
        }

        static iterator mid_point(iterator const& a, iterator const& b) {
            assert(a._region == b._region);
            return { *a._region, a._region->start_of_line(a._data.beg + (b._data.beg -a._data.beg)/2) };
        }

      public:
        value_type const& dereference() const {
            ensure_parsed();
            return _data;
        }

        bool equal(iterator const& o) const {
            return (_region == o._region) && (_data.beg == o._data.beg);
        }

        void increment() {
            _data = { _region->end_of_line(_data.beg), nullptr, Key{}, Value{} };
            assert(_data.beg == _region->start_of_line(_data.beg));
        }
    };

    using const_iterator = iterator;

    const_iterator begin()  const { return { *this, _map_begin }; }
    const_iterator end()    const { return { *this, _map_end   }; }
    const_iterator cbegin() const { return { *this, _map_begin }; }
    const_iterator cend()   const { return { *this, _map_end   }; }

    template <typename CompatibleKey>
    const_iterator lower_bound(CompatibleKey const& key) const {
        auto f(begin()), l(end());
        while (f!=l) {
            auto m = iterator::mid_point(f,l);

            if (m->key < key) {
                f = m;
                ++f;
            }
            else {
                l = m;
            }
        }
        return f;
    }

    template <typename CompatibleKey>
    const_iterator upper_bound(CompatibleKey const& key) const {
        return upper_bound(key, begin());
    }

  private:
    template <typename CompatibleKey>
    const_iterator upper_bound(CompatibleKey const& key, const_iterator f) const {
        auto l(end());
        while (f!=l) {
            auto m = iterator::mid_point(f,l);

            if (key < m->key) {
                l = m;
            }
            else {
                f = m;
                ++f;
            }
        }
        return f;
    }

  public:
    template <typename CompatibleKey>
    std::pair<const_iterator, const_iterator> equal_range(CompatibleKey const& key) const {
        auto lb = lower_bound(key);
        return { lb, upper_bound(key, lb) };
    }

};

#include <iostream>

int main() {
    io::mapped_file_source map("input.txt");
    text_multi_lookup<double, unsigned int> tml(map.data(), map.data() + map.size());

    auto const e = tml.end();

    for(auto&& line : tml)
    {
        std::cout << line.str();

        auto er = tml.equal_range(line.key);

        if (er.first  != e) std::cout << " lower: " << er.first->str();
        if (er.second != e) std::cout << " upper: " << er.second->str();
    }
}

出于好奇:这里是拆卸过程。请注意如何将所有算法内容直接内联到 main中: http://paste.ubuntu.com/9946135/