Структура данных Queue: от наивного алгоритма к lock-free реализации
В прошлом посте мы разобрались что такое lock-free алгоритм на примере задачи TransferMoney. В заметке я обмолвился что идеи из предложенного алгоритма имеют место быть в реальном мире.
Сегодня хочу это продемонстрировать на примере реализации структуры данных Queue. Пройдем путь от наивной реализации до очереди Майкла-Скотта.
Классические реализации
Очередь — это линейная структура данных, в которой элементы обрабатываются по принципу FIFO (First In, First Out — «первым пришел — первым вышел»)
Классическая реализация на языке Go:
type Node struct {
val int
next *Node
}
type Queue struct {
head *Node
tail *Node
}
func NewQueue() *Queue {
sentinel := &Node{}
return &Queue{head: sentinel, tail: sentinel}
}
func (q *Queue) Push(val int) {
node := &Node{val: val}
q.tail.next = node
q.tail = node
}
func (q *Queue) Pop() *int {
next := q.head.next
if next == nil {
return nil
}
q.head = next
val := next.val
return &val
}Если нам нужно работать из нескольких потоков:
type Node struct {
val int
next *Node
}
type MutexQueue struct {
mu sync.Mutex
head *Node
tail *Node
}
func NewMutexQueue() *MutexQueue {
sentinel := &Node{}
return &MutexQueue{head: sentinel, tail: sentinel}
}
func (q *MutexQueue) Push(val int) {
q.mu.Lock()
defer q.mu.Unlock()
node := &Node{val: val}
q.tail.next = node
q.tail = node
}
func (q *MutexQueue) Pop() *int {
q.mu.Lock()
defer q.mu.Unlock()
next := q.head.next
if next == nil {
return nil
}
q.head = next
val := next.val
return &val
}Реализация простая и понятная. Но у нее есть важный минус - она сериализует чтения и записи в один поток исполнения. А хочется чтобы читатели сами по себе и писатели тоже.
Two Lock Queue
В статье Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms от Maged M. Michael и Michael L. Scott был предложен вариант реализации очереди не на одном мьютексе а на двух - для головы и хвоста соответственно.
type Node struct {
val int
next atomic.Pointer[Node]
}
type TwoLockQueue struct {
head *Node
headLock sync.Mutex
tail *Node
tailLock sync.Mutex
}
func NewTwoLockQueue() *TwoLockQueue {
sentinel := &Node{}
return &TwoLockQueue{head: sentinel, tail: sentinel}
}
func (q *TwoLockQueue) Push(val int) {
node := &Node{val: val}
q.tailLock.Lock()
defer q.tailLock.Unlock()
q.tail.next.Store(node)
q.tail = node
}
func (q *TwoLockQueue) Pop() *int {
q.headLock.Lock()
defer q.headLock.Unlock()
next := q.head.next.Load()
if next == nil {
return nil
}
q.head = next
val := next.val
return &val
}Код совсем не отличается от того что мы рассмотрели ранее. Кроме одной вещи - в структуре Node атрибут next теперь не просто Node, а atomic.Pointer. Нужно это для того чтобы получить синхронизацию когда в очереди 0 и 1 элемент. Без атомика одновременные push и pop будут бить в одну и ту же область памяти.
Такую реализацию очереди можно встретить в Java - класс LinkedBlockingQueue
Итог - мы добились того что чтения и записи у нас не зависят друг от друга.
Отказываемся от Mutex совсем
А что насчет Lock-free реализации? Она существует и ее предложили в той же статье что я упомянул выше.
type AtomicNode struct {
val int
next atomic.Pointer[AtomicNode]
}
type LockFreeQueue struct {
head atomic.Pointer[AtomicNode]
tail atomic.Pointer[AtomicNode]
}
func NewLockFreeQueue() *LockFreeQueue {
q := &LockFreeQueue{}
q.head = atomic.Pointer[AtomicNode]{}
node := &AtomicNode{}
q.head.Store(node)
q.tail.Store(node)
return q
}
func (q *LockFreeQueue) Push(val int) {
node := AtomicNode{val: val}
for {
curr := q.tail.Load()
currNext := curr.next.Load()
// поймали незавершенный push,
// пытаемся завершить операцию "за того парня"
if currNext != nil {
q.tail.CompareAndSwap(curr, currNext)
continue
}
// подвешиваемся на то место где раньше был nil
if curr.next.CompareAndSwap(nil, &node) {
q.tail.CompareAndSwap(curr, &node)
return
}
}
}
func (q *LockFreeQueue) Pop() *int {
for {
curr := q.head.Load()
currNext := curr.next.Load()
if currNext == nil {
return nil
}
val := currNext.val
if ok := q.head.CompareAndSwap(curr, currNext); ok {
return &val
}
}
}- Каждый узел очереди - обязательно спрятан за atomic.
- Операции реализованы через бесконечный цикл (классика lock-free).
- У нас есть dummy узел. Если для классической реализации и версии с мьютексом это скорее украшательство для удобства и простоты кода. То для lock-free очереди это обязательно.
- В функции Push спрятана логика взаимопомощи потоков. Если при вставке мы видим что у tail есть ненулевой tail.next то в первую очередь мы протолкнем его, и уйдем на следующую итерацию цикла.
- В случае когда у нас честный tail и tail.next нам нужно обновить 2 атомика. И как вы помните, тут нужно быть осторожными. Поэтому мы в обязательном порядке линкуем к хвосту наш новый узел и Если операция прошла успешно, то уже по принципу best effort пытаемся подвинуть и сам tail, но уже не проверяем успешность - ведь если что, за нас это сделает другой поток.
Как итог - мы снова добились того что чтения и записи у нас не зависят друг от друга. На этот раз без мьютексов вообще.
P.S. Реализация подобная этой есть в Java - ConcurrentLinkedQueue.
Где можно встретить подобный код?
Код который мы рассмотрели в заметке корректный, пару десятилетий назад только такие реализации и можно было встретить в реальном коде. Но время идет и не стоит на месте и сейчас уже такой код встретить практически невозможно.
Почему спросите вы? Дело в том что для настоящего хайлоад прода мало взять lock-free, нужно еще и оптимизироваться под железо (которое стало и мощнее и сложнее). А это отдельный вид трюков, для каждого можно писать отдельную заметку. В своем же цикле заметок я стараюсь идти поступательно, от простого (насколько это возможно) к сложному.
В любом случае начало положено, мы уже научились писать код без мьютексов, в следующей заметке добьем последний кусочек теории - wait free алгоритмы и структуры данных.