-
Notifications
You must be signed in to change notification settings - Fork 0
/
lock-free-queue.go
84 lines (74 loc) · 1.87 KB
/
lock-free-queue.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
package queue
import (
"sync/atomic"
"unsafe"
)
// LKQueue 无边界的cas锁
type LKQueue struct {
head unsafe.Pointer
tail unsafe.Pointer
length int32
}
type node struct {
value interface{}
next unsafe.Pointer
}
// NewLKQueue 返回一个空的LKQueue
func NewLKQueue() *LKQueue {
n := unsafe.Pointer(&node{})
return &LKQueue{head: n, tail: n}
}
// Enqueue 使v入队
func (q *LKQueue) Enqueue(v interface{}) {
n := &node{value: v}
for {
tail := load(&q.tail)
next := load(&tail.next)
if tail == load(&q.tail) { // are tail and next consistent?
if next == nil {
if cas(&tail.next, next, n) {
cas(&q.tail, tail, n) // Enqueue is done. try to swing tail to the inserted node
atomic.AddInt32(&q.length, 1)
return
}
} else { // tail was not pointing to the last node
// try to swing Tail to the next node
cas(&q.tail, tail, next)
}
}
}
}
// Dequeue 出队,如果队列是空,则返回nil
func (q *LKQueue) Dequeue() interface{} {
for {
head := load(&q.head)
tail := load(&q.tail)
next := load(&head.next)
if head == load(&q.head) { // are head, tail, and next consistent?
if head == tail { // is queue empty or tail falling behind?
if next == nil { // is queue empty?
return nil
}
// tail is falling behind. try to advance it
cas(&q.tail, tail, next)
} else {
// read value before CAS otherwise another dequeue might free the next node
v := next.value
if cas(&q.head, head, next) {
atomic.AddInt32(&q.length, -1)
return v // Dequeue is done. return
}
}
}
}
}
func (q *LKQueue) IsEmpty() bool {
return atomic.LoadInt32(&q.length) == 0
}
func load(p *unsafe.Pointer) (n *node) {
return (*node)(atomic.LoadPointer(p))
}
func cas(p *unsafe.Pointer, old, new *node) (ok bool) {
return atomic.CompareAndSwapPointer(
p, unsafe.Pointer(old), unsafe.Pointer(new))
}