/
queue.go
74 lines (64 loc) · 1.81 KB
/
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
// Package mpsc provides an efficient implementation of a multi-producer, single-consumer lock-free queue.
//
// The Push function is safe to call from multiple goroutines. The Pop and Empty APIs must only be
// called from a single, consumer goroutine.
package task
// This implementation is based on http://www.1024cores.net/home/lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue
import (
"sync/atomic"
"unsafe"
)
type node struct {
next *node
val interface{}
}
type Queue struct {
head, tail *node
size int32
}
func NewQueue() *Queue {
q := &Queue{}
stub := &node{}
q.head = stub
q.tail = stub
q.size = 0
return q
}
// Push adds x to the back of the queue.
//
// Push can be safely called from multiple goroutines
func (q *Queue) Push(x interface{}) {
n := new(node)
n.val = x
// current producer acquires head node
prev := (*node)(atomic.SwapPointer((*unsafe.Pointer)(unsafe.Pointer(&q.head)), unsafe.Pointer(n)))
// release node to consumer
atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&prev.next)), unsafe.Pointer(n))
atomic.AddInt32(&q.size, 1)
}
// Pop removes the item from the front of the queue or nil if the queue is empty
//
// Pop must be called from a single, consumer goroutine
func (q *Queue) Pop() interface{} {
tail := q.tail
next := (*node)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&tail.next)))) // acquire
if next != nil {
q.tail = next
v := next.val
next.val = nil
atomic.AddInt32(&q.size, -1)
return v
}
return nil
}
// Empty returns true if the queue is empty
//
// Empty must be called from a single, consumer goroutine
func (q *Queue) Empty() bool {
tail := q.tail
next := (*node)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&tail.next))))
return next == nil
}
func (q *Queue) Size() int32 {
return atomic.LoadInt32(&q.size)
}