Brpc源码浅析（一）Flatmap实现

源码与官方文档路径

flat_map.h（486行代码）

flat_map_inl.h（657行代码）

flatmap.md

原理简析

Brpc源码浅析（一）Flatmap实现

如图所示，flatmap使用的是开链哈希。使用时先申请一块N*sizeof（Bucket）大小的连续虚拟内存。Bucket本身会存储value值，因此当哈希算法足够优秀时（即不会有拉链或很短的拉链），可以实现接近原生数组的查找性能。

数组大小的选取有两种方式：

数组大小为素数，当除以一个素数时，会产生最分散的余数
数组大小为2的指数倍：这会屏蔽被除数中的位，可能是最糟糕的选取方式。但是好处是可以通过位移运算提高计算余数的效率。

flatmap默认选取的是第二种方式，因为作者认为flatmap选取的哈希算法murmurhash3已经足够好了

主要数据结构介绍

Bucket

如上图所示，我们先看一下Bucket结构的实现，Bucket结构体是在Flatmap类内部实现的：

struct Bucket {
        explicit Bucket(const _K& k) : next(NULL)
        { new (element_spaces) Element(k); }
        Bucket(const Bucket& other) : next(NULL)
        { new (element_spaces) Element(other.element()); }
        bool is_valid() const { return next != (const Bucket*)-1UL; }
        void set_invalid() { next = (Bucket*)-1UL; }
        // NOTE: Only be called when in_valid() is true.
        Element& element() {
            void* spaces = element_spaces; // Suppress strict-aliasing
            return *reinterpret_cast<Element*>(spaces);
        }
        const Element& element() const {
            const void* spaces = element_spaces;
            return *reinterpret_cast<const Element*>(spaces);
        }
        Bucket* next;
        char element_spaces[sizeof(Element)];
    };

在指定分配好的内存位置new对象的用法参考：C++ new 的用法 (总结)

我们可以看到，每个Bucket都有一个指针指向一整个开链，还有一个存储kv对象的element。对于该位置上没有元素的Bucket，我们将其next指针指向-1表示非法，不再需要额外的标记位。

至于这里为什么需要用char数组的方式存储Element对象，不直接存放一个Element对象，个人觉得应该是都可以，用char数组的方式可能看起来更加解耦。

FlatMapElement

也就是Bucket中的Element，这也是个模板类

template <typename K, typename T>
class FlatMapElement {
public:
    typedef std::pair<const K, T> value_type;
    explicit FlatMapElement(const K& k) : _key(k), _value(T()) {}
    const K& first_ref() const { return _key; }
    T& second_ref() { return _value; }
    value_type& value_ref() { return *reinterpret_cast<value_type*>(this); }
    inline static const K& first_ref_from_value(const value_type& v)
    { return v.first; }
    inline static const T& second_ref_from_value(const value_type& v)
    { return v.second; }
private:
    const K _key;
    T _value;
};

模板参数K和V分别表示key和value。

Flatmap

类模板

//_K是key，_T是value，hash是我们选用的hash方法，equal是用来比较key是否相等的方法
//如果hash到数组的元素是很稀疏的，则sparse参数为true，此时在迭代器的实现里面会通过bitmap优化遍历的速度
template <typename _K, typename _T, typename _Hash = DefaultHasher<_K>,typename _Equal = DefaultEqualTo<_K>, bool _Sparse = false>
class FlatMap {}

Flatmap的重要数据成员变量：

size_t _size;  //包括链表上的总的桶数量
    size_t _nbucket; //不包括链表的桶的数量
    Bucket* _buckets;// 桶数组
    uint64_t* _thumbnail;// 用bit存放，每个下标表示桶是否有效
    u_int _load_factor;// 负载因子
    hasher _hashfn;// 哈希方法
    key_equal _eql;// 相等判断方法
    SingleThreadedPool<sizeof(Bucket), 1024, 16> _pool; // 分配桶空间的内存池

初始化flatmap，只用malloc分配内存，并不对每个Bucket做初始化

template <typename _K, typename _T, typename _H, typename _E, bool _S>
int FlatMap<_K, _T, _H, _E, _S>::init(size_t nbucket, u_int load_factor) {
    if (initialized()) {
        LOG(ERROR) << "Already initialized";
        return -1;
    }
    if (load_factor < 10 || load_factor > 100) {
        LOG(ERROR) << "Invalid load_factor=" << load_factor;
        return -1;
    }
    _size = 0;
    _nbucket = flatmap_round(nbucket);
    _load_factor = load_factor;
     //多分配了一块内存，迭代器能够知道bucket在哪里结束                           
    _buckets = (Bucket*)malloc(sizeof(Bucket) * (_nbucket + 1));
    if (NULL == _buckets) {
        LOG(ERROR) << "Fail to new _buckets";
        return -1;
    }
    for (size_t i = 0; i < _nbucket; ++i) {
        _buckets[i].set_invalid();
    }
    _buckets[_nbucket].next = NULL;

    if (_S) {
        _thumbnail = bit_array_malloc(_nbucket);
        if (NULL == _thumbnail) {
            LOG(ERROR) << "Fail to new _thumbnail";
            return -1;
        }
        bit_array_clear(_thumbnail, _nbucket);
    }
    return 0;
}

拷贝构造函数和赋值运算符重载

template <typename _K, typename _T, typename _H, typename _E, bool _S>
FlatMap<_K, _T, _H, _E, _S>::FlatMap(const FlatMap& rhs)
    : _size(0)
    , _nbucket(0)
    , _buckets(NULL)
    , _thumbnail(NULL)
    , _load_factor(rhs._load_factor)
    , _hashfn(rhs._hashfn)
    , _eql(rhs._eql) {
    operator=(rhs);
}

//将rhs中的元素赋值过来，如果本来已经初始化过了，不会拷贝其他flatmap的私有变量
template <typename _K, typename _T, typename _H, typename _E, bool _S>
void FlatMap<_K, _T, _H, _E, _S>::operator=(const FlatMap<_K, _T, _H, _E, _S>& rhs) {
    if (this == &rhs) {
        return;
    }
    // NOTE: assignment does not change _load_factor/_hashfn/_eql if |this| is
    // initialized
    clear();
    if (rhs.empty()) {
        return;
    }
    if (!initialized()) {
        _load_factor = rhs._load_factor;
    }
    if (_buckets == NULL || is_too_crowded(rhs._size)) {
        free(_buckets);
        _nbucket = rhs._nbucket;
        // note: need an extra bucket to let iterator know where buckets end
        _buckets = (Bucket*)malloc(sizeof(Bucket) * (_nbucket + 1/*note*/));
        if (NULL == _buckets) {
            LOG(ERROR) << "Fail to new _buckets";
            return;
        }
        if (_S) {
            free(_thumbnail);
            _thumbnail = bit_array_malloc(_nbucket);
            if (NULL == _thumbnail) {
                LOG(ERROR) << "Fail to new _thumbnail";
                return;
            }
            bit_array_clear(_thumbnail, _nbucket);
        }
    }
    if (_nbucket == rhs._nbucket) {
        // For equivalent _nbucket, walking through _buckets instead of using
        // iterators is more efficient.
        for (size_t i = 0; i < rhs._nbucket; ++i) {
            if (!rhs._buckets[i].is_valid()) {
                _buckets[i].set_invalid();
            } else {
                if (_S) {
                    bit_array_set(_thumbnail, i);
                }
                new (&_buckets[i]) Bucket(rhs._buckets[i]);
                Bucket* p1 = &_buckets[i];
                Bucket* p2 = rhs._buckets[i].next;
                while (p2) {
                    p1->next = new (_pool.get()) Bucket(*p2);
                    p1 = p1->next;
                    p2 = p2->next;
                }
            }
        }
        _buckets[rhs._nbucket].next = NULL;
        _size = rhs._size;
    } else {
        for (const_iterator it = rhs.begin(); it != rhs.end(); ++it) {
            operator[](Element::first_ref_from_value(*it)) = 
                Element::second_ref_from_value(*it);
        }
    }
}
//将所有的Bucket中的元素清空
template <typename _K, typename _T, typename _H, typename _E, bool _S>
void FlatMap<_K, _T, _H, _E, _S>::clear() {
    if (0 == _size) {
        return;
    }
    _size = 0;
    if (NULL != _buckets) {
        for (size_t i = 0; i < _nbucket; ++i) {
            Bucket& first_node = _buckets[i];
            if (first_node.is_valid()) {
                first_node.element().~Element();
                Bucket* p = first_node.next;
                while (p) {
                    Bucket* next_p = p->next;
                    p->element().~Element();
                    _pool.back(p);
                    p = next_p;
                }
                first_node.set_invalid();
            }
        }
    }
    if (NULL != _thumbnail) {
        bit_array_clear(_thumbnail, _nbucket);
    }
}

插入新元素

template <typename _K, typename _T, typename _H, typename _E, bool _S>
_T* FlatMap<_K, _T, _H, _E, _S>::insert(const key_type& key,
                                        const mapped_type& value) {
    mapped_type *p = &operator[](key);
    *p = value;
    return p;
}

template <typename _K, typename _T, typename _H, typename _E, bool _S>
_T& FlatMap<_K, _T, _H, _E, _S>::operator[](const key_type& key) {
    const size_t index = flatmap_mod(_hashfn(key), _nbucket);
    Bucket& first_node = _buckets[index];
    if (!first_node.is_valid()) {
        ++_size;
        if (_S) {
            bit_array_set(_thumbnail, index);
        }
        //在指定的指针指向的已分配好空间初始化对象
        new (&first_node) Bucket(key); //new了之后才是真正可以用的空间
        return first_node.element().second_ref();
    }
    if (_eql(first_node.element().first_ref(), key)) {
        return first_node.element().second_ref();
    }
    Bucket *p = first_node.next;
    if (NULL == p) {
        if (is_too_crowded(_size)) {
            if (resize(_nbucket + 1)) {
                return operator[](key);
            }
            // fail to resize is OK
        }
        ++_size;
        Bucket* newp = new (_pool.get()) Bucket(key);
        first_node.next = newp;
        return newp->element().second_ref();
    }
    while (1) {
        if (_eql(p->element().first_ref(), key)) {
            return p->element().second_ref();
        }
        if (NULL == p->next) {
            if (is_too_crowded(_size)) {
                if (resize(_nbucket + 1)) {
                    return operator[](key);
                }
                // fail to resize is OK
            }
            ++_size;
            Bucket* newp = new (_pool.get()) Bucket(key);
            p->next = newp;
            return newp->element().second_ref();
        }
        p = p->next;
    }
}

template <typename _K, typename _T, typename _H, typename _E, bool _S>
bool FlatMap<_K, _T, _H, _E, _S>::resize(size_t nbucket2) {
    nbucket2 = flatmap_round(nbucket2);
    if (_nbucket == nbucket2) {
        return false;
    }

    FlatMap new_map;
    if (new_map.init(nbucket2, _load_factor) != 0) {
        LOG(ERROR) << "Fail to init new_map, nbucket=" << nbucket2;
        return false;
    }
    for (iterator it = begin(); it != end(); ++it) {
        new_map[Element::first_ref_from_value(*it)] = 
            Element::second_ref_from_value(*it);
    }
    new_map.swap(*this);
    return true;
}

删除元素

template <typename _K, typename _T, typename _H, typename _E, bool _S>
template <typename K2>
size_t FlatMap<_K, _T, _H, _E, _S>::erase(const K2& key, _T* old_value) {
    if (!initialized()) {
        return 0;
    }
    // TODO: Do we need auto collapsing here?
    const size_t index = flatmap_mod(_hashfn(key), _nbucket);
    Bucket& first_node = _buckets[index];
    if (!first_node.is_valid()) {
        return 0;
    }
    if (_eql(first_node.element().first_ref(), key)) {
        if (old_value) {
            *old_value = first_node.element().second_ref();
        }
        if (first_node.next == NULL) {
            first_node.element().~Element();
            first_node.set_invalid();
            if (_S) {
                bit_array_unset(_thumbnail, index);
            }
        } else {
            Bucket* p = first_node.next;
            first_node.next = p->next;
            const_cast<_K&>(first_node.element().first_ref()) =
                p->element().first_ref();
            first_node.element().second_ref() = p->element().second_ref();
            p->element().~Element();//一定要析构，可能里面有些成员变量需要析构
            _pool.back(p);
        }
        --_size;
        return 1UL;
    }
    Bucket *p = first_node.next;
    Bucket *last_p = &first_node;
    while (p) {
        if (_eql(p->element().first_ref(), key)) {
            if (old_value) {
                *old_value = p->element().second_ref();
            }
            last_p->next = p->next;
            p->element().~Element();
            _pool.back(p);
            --_size;
            return 1UL;
        }
        last_p = p;
        p = p->next;
    }
    return 0;
}

删除的流程图如下：

Brpc源码浅析（一）Flatmap实现

元素查找：

template <typename _K, typename _T, typename _H, typename _E, bool _S>
template <typename K2>
_T* FlatMap<_K, _T, _H, _E, _S>::seek(const K2& key) const {
    if (!initialized()) {
        return NULL;
    }
    Bucket& first_node = _buckets[flatmap_mod(_hashfn(key), _nbucket)];
    if (!first_node.is_valid()) {
        return NULL;
    }
    if (_eql(first_node.element().first_ref(), key)) {
        return &first_node.element().second_ref();
    }
    Bucket *p = first_node.next;
    while (p) {
        if (_eql(p->element().first_ref(), key)) {
            return &p->element().second_ref();
        }
        p = p->next;
    }
    return NULL;
}

对应的还有FlatMapIterator和SparseFlatMapIterator的实现，这里就不展开了。

关注点

flatmap存储的对象必须是可拷贝的，关于不可拷贝对象详见：如何禁止C++ 类支持拷贝
flatmap的实现中还有以下几点值得我们学习：
1. 定义一个通用数据结构，要有一个接口能反映数据结构内部目前的使用情况如BucketInfo结构体对应的接口
2. 数据结构的遍历接口考虑将大循环遍历拆分成多个小循环，降低锁的粒度。如PositionHint对应的接口save_iterator和restore_iterator。没有办法不加锁，就要降低锁的粒度
3. flatmap和vector等容器遍历速度问题，用下标遍历速度更快：遍历vector容器的效率问题

与std::map和std::unorder_map的对比

std::unorder_map的数组大小为素数，且数组元素只包含指针不包含元素值。因此每次查找至少要有两次随机访存操作，好处是占用的连续内存会更少一些。

std::unorder_map和flatmap实质都是hash_map，这与红黑树实现的map相比：

map可以有序遍历，hash_map不可以
访问速度上一般hash_map更快一些
内存消耗上一般hash_map也更多
unordered_map的效率在很大的程度上由它的hash函数算法决定，而红黑树的效率是一个稳定值。
如果需要对数据排序则选择map，否则一般首选hash_map

Brpc源码浅析（一）Flatmap实现

继续阅读

C语言第四章自述2第四章选择结构程序设计

面试题:vector和map的区别，异同。空间分布，100万数据存哪个比较合适。一、迭代器区别二、vector三、Map、Set四、vector_map 为什么比map效率高五、如何选择六、容器选择原则七、效率对比

C++ 多线程用条件变量确定线程的执行顺序而不是使用 sleep(1)

POJ 1284 Primitive Roots (欧拉函数&原根定理)

CQ V1.0分词bates(基于双数组tire树)—应该是目前最快的中文分词算法

成员函数初始化列表

2021-08-13c++——类之操作符重载

swmm与lisflood-fp源码如何一起编译 CMake命令

Windows下VS开发环境环境安装工程项目设置关于Debug和Release的提示

一文看懂字符串的加减乘除

C++ 第十五周报告1--《冒泡法排序》

C++实现简单顺序表

C经典书籍笔记——C陷阱与缺陷②(语法陷阱之优先级)一、错误案列二、优先级规律

线性表之顺序表的实现

C++判断素数、求最大公约数代码判断一个数是否为素数求两个数的最大公约数

SequoiaDB巨杉数据库C++驱动概述