高性能并发map的设计与实现

2022-09-15

algorithm, lock, map

map 作为最常用的高性能数据结构, 重要性无需多言, 进一步到并发安全的大 map, 就会涉及到许多可能的优化和差异, 那么是哪些因素导致产生了 10X 以上的性能差距, 在 Java 中又该如何实现高性能的 map 呢?

0x00. 前言

这个系列大概分为几个部分:

介绍/了解现有的并发/无锁 map 实现背景和大体原理
查看对应 paper/源码, 加以理解 (可配简图)
在前人的基础上进行实现/改进和运用
最后会进行实测 + benchmark 看看具体效果对比

首先 map 作为最常用的数据结构, 能提供接近 O(1) 的时间访问数据所以在这个性能为王的时代显得尤为重要, 当它和并发/锁结合起来的时候, 问题就复杂了许多, 不同的设计 & 实现到底哪个最优很难说, 但至少通过了解核心的思想和特点, 总能找到适合自己场景的一款.

开放地址法

传统的语言库内置的 map 实现一般是拉链法(数组+链表/树)为主, 开放地址法听起来简单粗暴, 为何还要单独说呢, 就是因为它的”简单性“使得它也会广泛运用在高性能的 map 实现中, 原因主要是:

裸数组的方式在并发访问时不需要对链表整体上锁
裸数组本身对 cpu-cache hit 更友好

当然它缺点也很明显, 默认的方式拷贝扩容效率很低, 需要尽可能在固化大小的环境下使用 (比如缓存), 或者是考虑使用高效的实现(分批拷贝等)

常见优化

先简单了解一些比较易懂的优化手段, 对后面理解复杂的实现会有不少帮助, 先说一下同步/锁相关的:

降低持有锁时间
降低锁粒度 (分段锁)
锁分离/共享锁 (读写锁)

再看一下具体到 map 中的:

选择高效的 hash 算法, 尽可能降低高频的 hash(key) 操作, 特别是类似字符串这种理论上光读取一遍字符数组就是 O(n)开销
尽可能的选择整形存储 key, 便于做许多压缩/位运算操作, 那么其他类型(比如 str 如何转换 int?)
尽可能使用位运算提升计算性能
使用位图优化存储占用
大 value 可以转成指针存储一个地址 (指向 value), 那 value 存在哪呢?
增量/多次扩容, 而不是一次长时间阻塞操作

0x01. 分段锁

相比最常见的两个方式实现线程安全, 一个是就是用互斥锁(阻塞/挂起其他线程), 另一个就是用自旋的 CAS 方式, 而这两种方式实际都是需要加锁的, 只是开销/粒度不同. 真正接近 Lock-Free (严谨说”无锁”这个词现在肯定是滥用和错用了的, “无阻塞/挂起”才是比较准确的说法), 那么想最大程度接近无锁, 其中最经典的方式就是分段 —- 降低锁粒度 (其实 Map-Reduce 本质也是这样的分段/分片思路)

分段说起来也好理解, 也就是说把整个内存/任务划分为多个相对独立的区域, 比如 A 线程只访问内存的 0x00~0x10, B 线程只访问 0x11 ~ 0x20, 那么 A 和 B 线程同时操作各自内存中的数据, 自然是不存在竞态(race condition)的, 此时自然不需要锁(🔒)的概念. 其实数据库中常说的行锁也就是分段锁粒度 = 1的情况, 此时只要不同时读写一行数据, 那么自然多个线程读写访问都是并行的. 那是不是一拍脑袋 map 这里我们直接基于每个 bucket (数组的每一位) 加个读写锁, 然后就能实现效果了呢? 当然不是:

分段锁/行锁本质也是锁, 区别本质在于是锁整个内存/还是一段内存
如果分段锁按粒度为1, 也就是每个桶划分一下, 试想有几个问题:
- 为何分段锁需要预先固定分段的数目? 也就是最大并发写数, 思考下为何不能动态增长呢? (hash 值)
- 多个线程同时读/写同一个/段 key, 还是会落到同一个临界区, 那么随着 map 容量增大的时候, 锁的粒度也会不断增大
- 因为 Segment 结构体里一般存的是指向的实际存元素 k/v 的数组地址, 那也就意味着访问并不连续,
- 过细粒度的分段, 会导致全局信息获取开销巨大 (比如 map.size()方法需要锁全局的时候, 当然实际会先尝试无锁获取)

看看 HugeGraph 中默认的 IntMapBySegments 的整体设计, 然后再看看内置的固定地址法实现, 略去非核心方法和检查便于阅读, 由于写法里比较多的用到了 Unsafe 包和一些运运算, 需要先提前了解一下:

1. Integer 的特殊方法

numberOfLeadingZeros: 计算一个 int 二进制位高位全 0 的个数, 例如 10 二进制是 1010, 那么它返回 32 - 4 = 28 (前面 28 都是 0)
numberOfTrailingZeros: 和上一个相反, 这个是计算 int 二进制位低位中全 0 的个数, 比如 10 的最低位后只有 1 个 0, 如果是1011 0000 则返回 4

上面两个计算都是通过二分 + 位运算得出, 很巧妙高效 (二分是为了找到最高/最低的全 0 位置)

2. Unsafe 包常用方法/常量

首先关注我们这里使用 Unsafe 的主要两个考量: ① 绕过 JVM 直接修改堆外内存/对象; ② 调用 CAS 方法实现原子修改. 为何要自己管理使用内存呢, 一是因为 JVM 的堆内存是在用户态, 有一些 I/O 操作必须涉及到内核态交互, 而造成了不必要的用户 -> 内核态拷贝, 二是自己管理内存可以避免 JVM 产生 GC, 以及更好的控制申请/释放等 (同 C/C++).

常量:

ARRAY_OBJECT_BASE_OFFSET: 简单说就是得到对象数组的首地址 obj[0] (mac 上输出 16)
ARRAY_OBJECT_INDEX_SCALE: 简单说就是对象数组中每个元素的地址偏移, 这样可以计算出任何一个 obj[i] 的具体地址 (mac 上输出 4)

那么, 我想从内存中直接取出一个数组的某个位置元素地址, 就可以用 base_offset + index_scale * i 来算出

内存方法:

一般来说, Unsafe 里直接访问(读写)内存底层是通过 mmap() 或类似方法申请的, 如果是想在堆外开辟一个缓冲区, 则多类似 DirectByteBuffer (底层也是调用 unsafe 分配) ,需要注意的是, 一般我们说的堆外内存/直接内存并不一定就是 pagecache 这层, 对比堆内来说流程少了一个复制, 只有

堆内内存 –> 堆外内存 –> page_cache –> 磁盘

此时, 对不存在的内存地址进行操作, 比如 getAddress(1) 会导致 JVM 直接崩溃退出(生成日志文件), 类似 C 操作 OS 触发异常

3. 位运算技巧

我们都知道 hash 的最简单的办法就是简单 key % len 的方式, 但是 % 这个操作性能并不算快, 我们如何用&来模拟呢

/**
 * NOTE: IntMapBySegments(backend by IntMapByFixedAddr) is:
 * - slower 5x than IntMapByFixedAddr for single thread;
 * - slower 5x than IntMapByFixedAddr for 4 threads;
 * - faster 10x than ec IntIntHashMap-segment-lock for 4 threads;
 * - faster 20x than ec IntIntHashMap-global-lock for 4 threads;
 */
final class IntMapBySegments implements IntMap {

    final IntMap[] maps; // 核心基础数组, 存储 IntMap 桶
    final long capacity;
    final long unsignedSize;
    final int segmentSize;
    final int segmentShift;
    final int segmentMask;
    final Function<Integer, IntMap> creator;

    static final int DEFAULT_SEGMENTS = 100 * CPUS; // 默认是分 CPU 逻辑核 * 100 段
    static final int BASE_OFFSET = UNSAFE.ARRAY_OBJECT_BASE_OFFSET; // 返回 obj 数组第一个元素的基础地址
    // SHIFT 代表, 后面先算出了数组偏移的高位0的个数, 用31-28
    static final int SHIFT = 31 - Integer.numberOfLeadingZeros( // 返回无符号 int 二进制高位非 0 的个数
                                          UNSAFE.ARRAY_OBJECT_INDEX_SCALE); // 每个数组元素的偏移量

    public IntMapBySegments(int capacity, int segments, Function<Integer, IntMap> creator) {
        this.maps = new IntMap[segments];
        // include signed and unsigned number
        this.unsignedSize = capacity;
        this.capacity = this.unsignedSize * 2L;
        this.segmentSize = IntSet.segmentSize(this.capacity, segments);
        this.segmentShift = Integer.numberOfTrailingZeros(this.segmentSize);
        /*
         * The mask is lower bits of each segment size, like
         * segmentSize=4096 (0x1000), segmentMask=4095 (0xfff),
         * NOTE: `-1 >>> 0` or `-1 >>> 32` is -1.
         */
        this.segmentMask = this.segmentShift == 0 ? 0 : -1 >>> (32 - this.segmentShift);
        this.creator = IntMapByFixedAddr::new; // 默认使用固定/开放地址法的实现
    }

    @Override
    public boolean put(int key, int value) {
        int innerKey = (int) ((key + this.unsignedSize) & this.segmentMask);
        return segment(key).put(innerKey, value);
    }

    @Override
    public boolean remove(int key) {
        int innerKey = (int) ((key + this.unsignedSize) & this.segmentMask);
        return segment(key).remove(innerKey);
    }

    @Override
    public int get(int key) {
        long ukey = key + this.unsignedSize;
        if (ukey >= this.capacity || ukey < 0L) {
            return NULL_VALUE;
        }
        int innerKey = (int) (ukey & this.segmentMask);
        return segment(key).get(innerKey);
    }

    @Override
    public int size() {
        int size = 0;
        for (int i = 0; i < this.maps.length; i++) {
            IntMap map = this.segmentAt(i);
            if (map != null) {
                size += map.size();
            }
        }
        return size;
    }

    private IntMap segment(int key) {
        long ukey = key + this.unsignedSize;
        if (ukey >= this.capacity || ukey < 0L) {
            E.checkArgument(false, "The key %s is out of bound %s",
                            key, this.capacity);
        }

        long index = ukey >>> this.segmentShift;
        IntMap exist = this.maps[(int) index];
        if (exist != null) {
            return exist;
        }

        // volatile get this.maps[index]
        long offset = (index << SHIFT) + BASE_OFFSET;
        Object old = UNSAFE.getObjectVolatile(this.maps, offset);
        if (old != null) {
            return (IntMap) old;
        }

        // set this.maps[index] = new IntMap()
        IntMap map = this.creator.apply(this.segmentSize);
        while (true) {
            if (UNSAFE.compareAndSwapObject(this.maps, offset, null, map)) {
                return map;
            }
            old = UNSAFE.getObjectVolatile(this.maps, offset);
            if (old != null) {
                return (IntMap) old;
            }
        }
    }

    private IntMap segmentAt(int index) {
        // volatile get this.maps[index]
        long offset = ((long) index << SHIFT) + BASE_OFFSET;
        IntMap map = (IntMap) UNSAFE.getObjectVolatile(this.maps, offset);
        return map;
    }
}

0x02. 普通版本

典型的 JDK8(直到19也没变)核心其实是用的行锁的方式, 它

0x03. 其他

参考资料: