blob: 6e4f84bc0cc1f5a55b5bc61a5c8718359b033004 (
plain)
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
85
86
87
88
89
90
91
92
|
package xsync
import (
"sync/atomic"
)
// A SPSCQueue is a bounded single-producer single-consumer concurrent
// queue. This means that not more than a single goroutine must be
// publishing items to the queue while not more than a single goroutine
// must be consuming those items.
//
// SPSCQueue instances must be created with NewSPSCQueue function.
// A SPSCQueue must not be copied after first use.
//
// Based on the data structure from the following article:
// https://rigtorp.se/ringbuffer/
type SPSCQueue struct {
cap uint64
pidx uint64
//lint:ignore U1000 prevents false sharing
pad0 [cacheLineSize - 8]byte
pcachedIdx uint64
//lint:ignore U1000 prevents false sharing
pad1 [cacheLineSize - 8]byte
cidx uint64
//lint:ignore U1000 prevents false sharing
pad2 [cacheLineSize - 8]byte
ccachedIdx uint64
//lint:ignore U1000 prevents false sharing
pad3 [cacheLineSize - 8]byte
items []interface{}
}
// NewSPSCQueue creates a new SPSCQueue instance with the given
// capacity.
func NewSPSCQueue(capacity int) *SPSCQueue {
if capacity < 1 {
panic("capacity must be positive number")
}
return &SPSCQueue{
cap: uint64(capacity + 1),
items: make([]interface{}, capacity+1),
}
}
// TryEnqueue inserts the given item into the queue. Does not block
// and returns immediately. The result indicates that the queue isn't
// full and the item was inserted.
func (q *SPSCQueue) TryEnqueue(item interface{}) bool {
// relaxed memory order would be enough here
idx := atomic.LoadUint64(&q.pidx)
nextIdx := idx + 1
if nextIdx == q.cap {
nextIdx = 0
}
cachedIdx := q.ccachedIdx
if nextIdx == cachedIdx {
cachedIdx = atomic.LoadUint64(&q.cidx)
q.ccachedIdx = cachedIdx
if nextIdx == cachedIdx {
return false
}
}
q.items[idx] = item
atomic.StoreUint64(&q.pidx, nextIdx)
return true
}
// TryDequeue retrieves and removes the item from the head of the
// queue. Does not block and returns immediately. The ok result
// indicates that the queue isn't empty and an item was retrieved.
func (q *SPSCQueue) TryDequeue() (item interface{}, ok bool) {
// relaxed memory order would be enough here
idx := atomic.LoadUint64(&q.cidx)
cachedIdx := q.pcachedIdx
if idx == cachedIdx {
cachedIdx = atomic.LoadUint64(&q.pidx)
q.pcachedIdx = cachedIdx
if idx == cachedIdx {
return
}
}
item = q.items[idx]
q.items[idx] = nil
ok = true
nextIdx := idx + 1
if nextIdx == q.cap {
nextIdx = 0
}
atomic.StoreUint64(&q.cidx, nextIdx)
return
}
|
