golang_deque.md

golang 双端队列

虽然使用 golang 标准库 container/list 很容易实现双端队列，但经过 benchmark 压测发现数据差点意思. 在 github 中找到一个叫 edwingeng/deque 的项目，该队列的实现很不错，另外经过线上使用后，高频读写下 cpu 有下降，另外在几百万条数据堆积场景下， gc duration 也有下降. 下面分析下 deque 的设计实现.

代码地址:

https://github.com/edwingeng/deque

deque 的设计实现

优缺点

优点

• 双端队列设计
• 使用二维数组实现队列，比 list 更快，数据结构 gc 更友好
• sync.pool 优化 chunk 对象
• 更好的扩缩容机制
• 代码生成器，避免 interface{} 装箱拆箱操作.
• v2 版直接支持泛型

缺点

没锁，需要自己加一把锁.

主体设计

使用了二维数组实现了队列结构，第二层切片是一个 chunk，存放真实数据，而第一层切片关联索引第二层的 chunk.

type chunk struct {
	s    int
	e    int
	data [chunkSize]Elem
}

type deque struct {
	chunks []*chunk

	chunkPitch []*chunk
	sFree      int
	eFree      int
}

func (dq *deque) PushFront(v Elem) {
	var c *chunk
	n := len(dq.chunks)
	if n == 0 {
		dq.expandStart()
		c = dq.chunks[0]
	} else {
		c = dq.chunks[0]  // 获取 chunks 头节点.
		if c.s == 0 {    // 如果该 chunk 满了， 则获取一个 chunk.
			dq.expandStart() // 把 start, end 设置 243
			c = dq.chunks[0]
		}
	}
	c.s--
	c.data[c.s] = v // 也把数据放到 chunk 的尾部.
}

func (dq *deque) PushBack(v Elem) {
	var c *chunk
	n := len(dq.chunks)
	if n == 0 {
		dq.expandEnd()
		c = dq.chunks[n]
	} else {
		c = dq.chunks[n-1]  // 获取尾部节点，如果该节点满了，则创建一个新的 chunk, start 和 end 为 0.
		if c.e == chunkSize { 
			dq.expandEnd()
			c = dq.chunks[n]
		}
	}

	// 把数据放到 chunk 对应的 index 里.
	c.data[c.e] = v

	// 自增，求下次 index 的位置.
	c.e++
}

func (dq *deque) PopBack() Elem {
	n := len(dq.chunks)
	if n == 0 {
		return elemDefValue
	}

	// 获取尾部的 chunk
	c := dq.chunks[n-1]

	// 相等说明为空
	if c.e == c.s {
		return elemDefValue
	}

	// 获取 chunk 尾部的元素, 并向前滚动索引
	c.e--
	r := c.data[c.e]

	// 置空
	c.data[c.e] = elemDefValue

	// 如果没有数据了， 则进行缩容
	if c.e == 0 {
		dq.shrinkEnd()
	}
	return r
}

func (dq *deque) Len() int {
	n := len(dq.chunks)
	switch n {
	case 0:
		return 0
	case 1:
		// 只有单 chunk
		return dq.chunks[0].e - dq.chunks[0].s
	default:
		// 头部 chunk + 尾部 chunk + 中间 chunk
		return chunkSize - dq.chunks[0].s + dq.chunks[n-1].e + (n-2)*chunkSize
	}
}

chunk 对象缓存

使用 sync.pool 缓存 chunk 数据结构, 用来减轻 gc 开销.

var (
	sharedChunkPool = newChunkPool(func() interface{} {
		return &chunk{}
	})
)

// 缩容
func (dq *deque) shrinkEnd() {
    ...
	sharedChunkPool.Put(c)
    ...
}

// 扩容
func (dq *deque) expandEnd() {
    ...
	c := sharedChunkPool.Get().(*chunk)
    ...
}

取巧的扩缩容设计

chunk 最大为 254 个元素，当超过阈值时会实例化一个新的 chunk. 每次新增的 chunk 也都会放到 chunks 切片里. deque 没有使用 list.List 链表来关联 chunk，则使用切片.

deque 对 chunks 切片做了一些优化，由于双端队列还有个向前插入 pushFront() 的场景，如果单纯用 slice，那么会不断实例化一个新切片，把新的chunk 放到 chunks[0]，以前的切片 append 后面.

不断的实例化新切片和搬迁旧切片，该操作还是有开销的. 所以抽象了一个比 chunks 更大的切片做引用占位符.

func (dq *deque) expandEnd() {
	if f := dq.eFree; f == 0 {
		dq.realloc()
	}

	// 从 sync.pool 获取新的 chunk 数组
	c := sharedChunkPool.Get().(*chunk)

	// 重置 start, end 偏移量
	c.s, c.e = 0, 0
	dq.eFree--
	newEnd := len(dq.chunkPitch) - dq.eFree
	dq.chunkPitch[newEnd-1] = c
    
    // 把 chunkPitch 的数据引用到 chunks.
	dq.chunks = dq.chunkPitch[dq.sFree:newEnd]
}

benchmark

PushBack/Deque<harden>       100000000       12.0 ns/op       8 B/op      0 allocs/op
PushBack/Deque                20000000       55.5 ns/op      24 B/op      1 allocs/op
PushBack/list.List             5000000      158.7 ns/op      56 B/op      1 allocs/op

PushFront/Deque<harden>      195840157        9.2 ns/op       8 B/op      0 allocs/op
PushFront/Deque               30000000       49.2 ns/op      24 B/op      1 allocs/op
PushFront/list.List            5000000      159.2 ns/op      56 B/op      1 allocs/op

Random/Deque<harden>          65623633       15.1 ns/op       0 B/op      0 allocs/op
Random/Deque                  50000000       24.7 ns/op       4 B/op      0 allocs/op
Random/list.List              30000000       46.9 ns/op      28 B/op      1 allocs/op