MPSC Queue 源码分析

开发/后端 · 阅读 1362 · 点赞 0

在阅读Netty源码中发现在创建EventLoop的时候,内部的taskQueue使用了MpscQueue,MPSC的全称是Multi producer single consumer,单消费者多生产者模型,多生产者会发生竞争要做到无锁使用了CAS,而单一消费者则不需要保证线程安全

而内部大量是用二进制的偶次方特性提高运算效率,并且解决False Sharing问题

创建一个PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks)

 private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {
        return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
                : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
    }

传入的大小和 MAX_ALLOWED_MPSC_CAPACITY (1 << 30 2的30次方) 取小然后和 2048 取大值,也就是初始的chunk大小为2048最大则为1 << 30,初始的chunk为1024

  static <T> Queue<T> newMpscQueue(final int maxCapacity) {
            final int capacity = max(min(maxCapacity, MAX_ALLOWED_MPSC_CAPACITY), MIN_MAX_MPSC_CAPACITY);
            //MPSC_CHUNK_SIZE 1024 
            return newChunkedMpscQueue(MPSC_CHUNK_SIZE, capacity);
        }

继续调用父类构造器,再次判断初始chuck和capacity不能超过最大2的偶次方(2^31)

 public static int roundToPowerOfTwo(final int value) {
        if (value > MAX_POW2) {
            throw new IllegalArgumentException("There is no larger power of 2 int for value:"+value+" since it exceeds 2^31.");
        }
        if (value < 0) {
            throw new IllegalArgumentException("Given value:"+value+". Expecting value >= 0.");
        }
        final int nextPow2 = 1 << (32 - Integer.numberOfLeadingZeros(value - 1));
        return nextPow2;
    }

获取下一个偶次方的值,去掉高位0还剩下多少位 取它的偶次方,value – 1 代表如果value恰好是偶次方则返回他本身,如果不是则取下一个距离最近的偶次方

 final int nextPow2 = 1 << (32 - Integer.numberOfLeadingZeros(value - 1));

numberOfLeadingZeros是integer提供的获取高位0的位运算二分算法,因为int是4个字节32位,第一个if无符号右移16位(刚好是一半,高位补0)如果等于0则高16位没有1,所以n + 16 , 然后左移16位将剩下的16位低位放到高位,下个判断处理1/4部分 以此类推

public static int numberOfLeadingZeros(int i) {
        // 如果是0就是32个0
        if (i == 0)
            return 32;
        // 不是0则至少有一个1
        int n = 1;
        //用2举个例子 00000000 00000000 00000000 00000010 2
        if (i >>> 16 == 0) { n += 16; i <<= 16; }
        //00000000 00000000 00000000 00000000 i >>> 16
        //00000000 00000010 00000000 00000000 i <<= 16
        // n = 1 + 16
        if (i >>> 24 == 0) { n +=  8; i <<=  8; }
        //00000000 00000000 00000000 00000000 i >>> 24
        //00000010 00000000 00000000 00000000 i <<= 8
        // n = 1 + 16 + 8
        if (i >>> 28 == 0) { n +=  4; i <<=  4; }
        //不满足
        if (i >>> 30 == 0) { n +=  2; i <<=  2; }
        //00000000 00000000 00000000 00000000 i >>> 30
        //00001000 00000000 00000000 00000000 i <<=  2
        // n = 1 + 16 + 8 + 2
        n -= i >>> 31;
        // 00001000 00000000 00000000 00000000
        // 00000000 00000000 00000000 00000000 >>> 31
        // n = n - 0
        return n;
    }

最后获取到最近的偶次方两倍的大小给到maxQueueCapacity

maxQueueCapacity = ((long) Pow2.roundToPowerOfTwo(maxCapacity)) << 1;

回过头来看一下MpscChunkedArrayQueue的结构,里面使用了大量的128byte padding,看到的第一眼就知道是用来填充cache line解决False Sharing问题,可问题是为什么使用128 byte而不是64byte?

我们知道市面上绝大部分cpu使用的是64 byte的cache line,下面简单证明一下,macos和linux均使用Intel cpu,可以看到均为64 byte 并且与L1、L2、L3等level无关

getconf LEVEL1_DCACHE_LINESIZE
64
#当然你也可以直接查看device
cat /sys/devices/system/cpu/cpu0/cache/index0/coherency_line_size 
64
# MAC OS
sysctl machdep.cpu.cache.linesize
machdep.cpu.cache.linesize: 64

MpscChunkedArrayQueue padding结构如下

128 byte padding ----
8 byte long maxQueueCapacity //最大长度偶次幂
16 byte JUMP = new Object() //填充生产者数组,消费时候使用
16 byte BUFFER_CONSUMED = new Object() //填充消费者数组,消费时使用
4 byte int CONTINUE_TO_P_INDEX_CAS 
4 byte int RETRY //重试
4 byte int QUEUE_FULL //队列已满
4 byte int QUEUE_RESIZE //扩容
8 byte long producerLimit //生产上线
8 byte long producerMask // 生产者掩码
8 byte long P_LIMIT_OFFSET  //生产者索引位
128 byte padding ----
8 byte long C_INDEX_OFFSET //消费者索引位
8 byte long consumerIndex //消费者下标
8 byte long consumerMask //消费者掩码
128 byte padding ----
8 byte P_INDEX_OFFSET
8 byte producerIndex
128 byte padding ----

突然想起来jdk8+提供了@Contended注解用来自动填充padding,于是我使用JOL查看了如下对象的布局

  public class JavaContended{
        private long var0;
        @Contended
        private long var1;
    }

可以看到@Contended的实现也是基于128 byte对齐的,为啥呢? 其实这只是默认的填充大小可能是为了适配128cache line的cpu 如果恰好是cache line是128 byte而padding是64的话依然会有False Sharing问题,所以牺牲了更大的空间来换取时间

io.netty.example.echo.test.JavaContended object internals:
OFF  SZ   TYPE DESCRIPTION               VALUE
  0   8        (object header: mark)     N/A
  8   4        (object header: class)    N/A
 12   4        (alignment/padding gap)   
 16   8   long JavaContended.var0        N/A
 24 128        (alignment/padding gap)   
152   8   long JavaContended.var1        N/A
160 128        (object alignment gap)    
Instance size: 288 bytes
Space losses: 132 bytes internal + 128 bytes external = 260 bytes total

Jvm提供的关于Contended一些参数

-XX:-RestrictContended Jdk1.8启用Contended
-XX:ContendedPaddingWidth 调整padding大小
-XX:-EnableContended 开启关闭

继续看mpsc queue的源码,初始化最大容量后前还会super调用父类构造器

    // 距离初始容量最近的偶次幂
    int p2capacity = Pow2.roundToPowerOfTwo(initialCapacity);
    // 掩码做位运算用 保证永远是偶数低位为0
    long mask = (p2capacity - 1) << 1;
    // 实际存储数据的数组 + 1 本质是为了添加jump对象做链表跳转否则越界了
    E[] buffer = allocateRefArray(p2capacity + 1);
    //生产者buf
    producerBuffer = buffer;
    producerMask = mask;
    //消费者buf
    consumerBuffer = buffer;
    consumerMask = mask;
    //producerLimit = mask本质是控制数组能添加几次元素
    //因为pindex每次 + 2 所以当小于等于mask时都会直接添加
    //本质就是控制数组添加数量为数组长度-2时进入扩容方法
    soProducerLimit(mask);

添加元素的方法

 @Override
    public boolean offer(final E e)
    {
        if (null == e)
        {
            throw new NullPointerException();
        }

        long mask;
        E[] buffer;
        long pIndex;

        while (true)
        {
            //生产上线值
            long producerLimit = lvProducerLimit();
            //生产者index 初始是0
            pIndex = lvProducerIndex();
            //此处仅在处于扩容期间QUEUE_RESIZE cas pindex期间 其他线程进入会等于1
            if ((pIndex & 1) == 1)
            {
                continue;
            }
           
            mask = this.producerMask;
            buffer = this.producerBuffer;
            //消费者index <= 消费上线 触发扩容,否则pIndex+2
            if (producerLimit <= pIndex)
            {
                int result = offerSlowPath(mask, pIndex, producerLimit);
                switch (result)
                {
                    case CONTINUE_TO_P_INDEX_CAS://跳出Switch执行casProducerIndex
                        break;
                    case RETRY: //重试
                        continue;
                    case QUEUE_FULL: //满了入队失败
                        return false;
                    case QUEUE_RESIZE: //尝试扩容或入队
                        resize(mask, buffer, pIndex, e, null);
                        return true;
                }
            }

            if (casProducerIndex(pIndex, pIndex + 2))
            {
                break;
            }
        }
        //获取下标对应的数组偏移量
        final long offset = modifiedCalcCircularRefElementOffset(pIndex, mask);
        //invalid缓存行使其它线程从内存重新读取
        soRefElement(buffer, offset, e); // release element e
        return true;
    }

这个方法就是用来判断switch的分支

 private int offerSlowPath(long mask, long pIndex, long producerLimit)
    {
        //生产index 没poll过所以是0
        final long cIndex = lvConsumerIndex();
        long bufferCapacity = getCurrentBufferCapacity(mask);

        if (cIndex + bufferCapacity > pIndex)
        {
            if (!casProducerLimit(producerLimit, cIndex + bufferCapacity))
            {
                // retry from top
                return RETRY;
            }
            else
            {
                // continue to pIndex CAS
                return CONTINUE_TO_P_INDEX_CAS;
            }
        }
        // maxQueueCapacity - (pIndex - cIndex) 满了
        else if (availableInQueue(pIndex, cIndex) <= 0)
        {
            // offer should return false;
            return QUEUE_FULL;
        }
        // 否则就把pIndex + 1然后返回resize
        else if (casProducerIndex(pIndex, pIndex + 1))
        {
            // trigger a resize
            return QUEUE_RESIZE;
        }
        else
        {
            // failed resize attempt, retry from top
            return RETRY;
        }
    }

扩容方法,本质是在构建一个链表,总结一下如果初始长度为2那么则是数组套链表每一次都会新建一个链表放到旧数组里,因为pindex + 2 永远和plimit相等直到容量满了返回false,如果长度大于2那么则会填满数组(例如初始长度4(其实+1是5)当添加的元素数量等于长度-2时则会进行扩容,将当前元素放入新数组中并放入旧数组,然后在前面添加一个jump此时数组长度刚好为5

private void resize(long oldMask, E[] oldBuffer, long pIndex, E e, Supplier<E> s)
    {
        assert (e != null && s == null) || (e == null || s != null);
        //oldBuffer的长度
        int newBufferLength = getNextBufferSize(oldBuffer);
        final E[] newBuffer;
        try
        {
            //创建一个和原来长度一样的数组
            newBuffer = allocateRefArray(newBufferLength);
        }
        catch (OutOfMemoryError oom)
        {
            assert lvProducerIndex() == pIndex + 1;
            soProducerIndex(pIndex);
            throw oom;
        }
        // 将新数组指向 producerBuffer 也就是每次resize传递进来的都是newBuffer
        producerBuffer = newBuffer;
        final int newMask = (newBufferLength - 2) << 1;
        producerMask = newMask;
        //原数组的内存地址偏移量
        final long offsetInOld = modifiedCalcCircularRefElementOffset(pIndex, oldMask);
        //新数组的内存地址偏移量
        final long offsetInNew = modifiedCalcCircularRefElementOffset(pIndex, newMask);
        //将元素e添加到新数组中
        soRefElement(newBuffer, offsetInNew, e == null ? s.get() : e);// element in new array
        //将新数组添加到旧数组的下一个索引位
        //比如 offer 1,2
        //长这个样子[1,[2](new)](old)
        soRefElement(oldBuffer, nextArrayOffset(oldMask), newBuffer);// buffer linked

        // ASSERT code
        final long cIndex = lvConsumerIndex();
        //计算剩余剩余可用大小
        final long availableInQueue = availableInQueue(pIndex, cIndex);
        RangeUtil.checkPositive(availableInQueue, "availableInQueue");
        //赋值给ProducerLimit
        soProducerLimit(pIndex + Math.min(newMask, availableInQueue));
        //生产者索引 + 2
        soProducerIndex(pIndex + 2);
        //将旧数组原位置添加jump占位
        //长这个样子[1,JUMP,[2](new)](old)
        soRefElement(oldBuffer, offsetInOld, JUMP);
    }

由于是单消费者模式,无并发问题所以poll方法就比较简单了

public E poll()
    {
        //就是初始的生产者数组
        final E[] buffer = consumerBuffer;
        //从0开始
        final long index = lpConsumerIndex();
        //和生产者的初始mask一样
        final long mask = consumerMask;
        //获取数组初始元素内存偏移量
        final long offset = modifiedCalcCircularRefElementOffset(index, mask);
        //获取数组index位置的元素
        Object e = lvRefElement(buffer, offset);
        if (e == null)
        {
            //可以拿到元素但是没有拿到 可能是因为cache line 还没有share本线程还看不到
            //自旋去获取
            if (index != lvProducerIndex())
            {
                do
                {
                    e = lvRefElement(buffer, offset);
                }
                while (e == null);
            }
            else
            {
                //读写索引一致则确实没有元素可以拿了
                return null;
            }
        }
        //如果是JUMP则下一个元素必定是一个新数组的引用
        if (e == JUMP)
        {
            final E[] nextBuffer = nextBuffer(buffer, mask);
            return newBufferPoll(nextBuffer, index);
        }
        //将该位置为空(容量不变)
        soRefElement(buffer, offset, null); // release element null
        //和pindex一样每次 + 2
        soConsumerIndex(index + 2); // release cIndex
        //返回命中的元素
        return (E) e;
    }

nextBuffer其实和resize差不多,将当前consumerBuffer改为下一个数组,并且将当前位置使用BUFFER_CONSUMED填充用来标记已经消费过了

 private E[] nextBuffer(final E[] buffer, final long mask)
    {
        final long offset = nextArrayOffset(mask);
        final E[] nextBuffer = (E[]) lvRefElement(buffer, offset);
        consumerBuffer = nextBuffer;
        consumerMask = (length(nextBuffer) - 2) << 1;
        soRefElement(buffer, offset, BUFFER_CONSUMED);
        return nextBuffer;
    }

总结

多生产者模型和NioEventLoop需要同时处理多个事件比较契合,而且做到无锁并发提高了性能,通过链表方式扩容减少了array copy,巧妙利用偶次方计算长度,临界值,大量使用内存偏移+unsafe cas极大提高了性能和原子性解决无锁处理并发