8276232797

Committed 14 Mar 2024 06:05AM UTC coverage: 73.149% (+0.01%) from 73.139%

Build Type

push

github

Committed by chaoyuepan

Commit Message

add generic lock-free deque

Run Details

69 of 94 new or added lines in 1 file covered. (73.4%)

5 existing lines in 1 file now uncovered.

1828 of 2499 relevant lines covered (73.15%)

3781001.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.4

/container/queue/deque.go

package queue

import (
        "sync/atomic"
        "unsafe"
)

// RingBuffer is a basic ring buffer, the capacity must be a power of 2.
type RingBuffer[T any] struct {
        cap  uint64
        mask uint64
        buff []T
}

// NewRingBuffer creates a new ring buffer with the given capacity.
func NewRingBuffer[T any](cap uint64) *RingBuffer[T] {
        if cap == 0 || cap&(cap-1) != 0 {
                panic("capacity must be power of 2")
        }
        return &RingBuffer[T]{
                cap:  cap,
                mask: cap - 1,
                buff: make([]T, cap),
        }
}

// Capacity returns the capacity of the ring buffer.
func (r *RingBuffer[T]) Capacity() uint64 {
        return r.cap
}

// Store stores the value x at the given index i.
func (r *RingBuffer[T]) Store(i uint64, x T) {
        r.buff[i&r.mask] = x
}

// Load loads the value at the given index i.
func (r *RingBuffer[T]) Load(i uint64) T {
        return r.buff[i&r.mask]
}

// Resize resizes the ring buffer to the given bounds.
func (r *RingBuffer[T]) Resize(b, t uint64) *RingBuffer[T] {
        newRing := NewRingBuffer[T](2 * r.cap)
        for i := t; i != b; i++ {
                newRing.Store(i, r.Load(i))
        }
        return newRing
}

// Deque is a lock-free single-producer multi-consumer deque.
// Deque is a bleeding-edge lock-free, single-producer multi-consumer, Chase-Lev work stealing deque as presented
// in the paper ["Dynamic Circular Work-Stealing Deque"](https://dl.acm.org/doi/10.1145/1073970.1073974) and further improved in the follow up paper:
// ["Correct and Efficient Work-Stealing for Weak Memory Models"](https://dl.acm.org/doi/10.1145/2442516.2442524).
type Deque[T any] struct {
        top     uint64
        bottom  uint64
        buffer  unsafe.Pointer
        garbage []*RingBuffer[T]
}

// NewDeque creates a new deque with the given capacity.
func NewDeque[T any](cap uint64) *Deque[T] {
        return &Deque[T]{buffer: unsafe.Pointer(NewRingBuffer[T](cap))}
}

// Size returns the number of elements in the deque.
func (d *Deque[T]) Size() uint64 {
        b := atomic.LoadUint64(&d.bottom)
        t := atomic.LoadUint64(&d.top)
        if b >= t {
                return b - t
        }
        return 0
}

// Capacity returns the capacity of the deque.
func (d *Deque[T]) Capacity() uint64 {
        return (*RingBuffer[T])(d.buffer).Capacity()
}

// Empty returns true if the deque is empty.
func (d *Deque[T]) Empty() bool {
        return d.Size() == 0
}

// Push pushes the value x to the deque.
// The value x is pushed to the bottom of the deque.
func (d *Deque[T]) Push(x T) {
        b := atomic.LoadUint64(&d.bottom)
        t := atomic.LoadUint64(&d.top)
        buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))
        if buf.Capacity() < b-t+1 {
                d.garbage = append(d.garbage, buf)
                buf = buf.Resize(b, t)
                atomic.StorePointer(&d.buffer, unsafe.Pointer(buf))
        }
        buf.Store(b, x)
        atomic.StoreUint64(&d.bottom, b+1)
}

// Pop pops the value from the deque.
// The value x is popped from the bottom of the deque.
func (d *Deque[T]) Pop() (x T, ok bool) {
        b := atomic.AddUint64(&d.bottom, ^uint64(0))
        buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))
        t := atomic.LoadUint64(&d.top)
        if t <= b {
                if t == b {
                        if !atomic.CompareAndSwapUint64(&d.top, t, t+1) {
                                atomic.StoreUint64(&d.bottom, b+1)
                                return x, false
                        }
                        atomic.StoreUint64(&d.bottom, b+1)
                }
                x = buf.Load(b)
                return x, true
        }
        atomic.StoreUint64(&d.bottom, b+1)
        return x, false
}

// Peek peeks at the value from the deque without removing it.
// The value x is peeked from the top of the deque without removing it.
func (d *Deque[T]) Peek() (x T, ok bool) {
        b := atomic.AddUint64(&d.bottom, ^uint64(0))
        buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))
        t := atomic.LoadUint64(&d.top)
        if t <= b {
                x = buf.Load(b)
                return x, true
        }

        return x, false
}

// Steal steals a value from the deque.
// The value x is stolen from the top of the deque.
func (d *Deque[T]) Steal() (x T, ok bool) {
        t := atomic.LoadUint64(&d.top)
        b := atomic.LoadUint64(&d.bottom)
        if t < b {
                buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))
                x = buf.Load(t)
                if !atomic.CompareAndSwapUint64(&d.top, t, t+1) {
                        return x, false
                }
                return x, true
        }
        return x, false
}

// StealPeek peeks at a value from the deque without removing it.
// The value x is peeked from the top of the deque without removing it.
func (d *Deque[T]) StealPeek() (x T, ok bool) {
        t := atomic.LoadUint64(&d.top)
        b := atomic.LoadUint64(&d.bottom)
        if t < b {
                buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))
                x = buf.Load(t)
                return x, true
        }
        return x, false
}

1	package queue
2
3	import (
4	"sync/atomic"
5	"unsafe"
6	)
7
8	// RingBuffer is a basic ring buffer, the capacity must be a power of 2.
9	type RingBuffer[T any] struct {
10	cap uint64
11	mask uint64
12	buff []T
13	}
14
15	// NewRingBuffer creates a new ring buffer with the given capacity.
16	func NewRingBuffer[T any](cap uint64) *RingBuffer[T] {	7✔
17	if cap == 0 \|\| cap&(cap-1) != 0 {	7✔
NEW 18	panic("capacity must be power of 2")	×
19	}
20	return &RingBuffer[T]{	7✔
21	cap: cap,	7✔
22	mask: cap - 1,	7✔
23	buff: make([]T, cap),	7✔
24	}	7✔
25	}
26
27	// Capacity returns the capacity of the ring buffer.
28	func (r *RingBuffer[T]) Capacity() uint64 {	36✔
29	return r.cap	36✔
30	}	36✔
31
32	// Store stores the value x at the given index i.
33	func (r *RingBuffer[T]) Store(i uint64, x T) {	35✔
34	r.buff[i&r.mask] = x	35✔
35	}	35✔
36
37	// Load loads the value at the given index i.
38	func (r *RingBuffer[T]) Load(i uint64) T {	35✔
39	return r.buff[i&r.mask]	35✔
40	}	35✔
41
42	// Resize resizes the ring buffer to the given bounds.
43	func (r RingBuffer[T]) Resize(b, t uint64) RingBuffer[T] {	2✔
44	newRing := NewRingBuffer[T](2 * r.cap)	2✔
45	for i := t; i != b; i++ {	14✔
46	newRing.Store(i, r.Load(i))	12✔
47	}	12✔
48	return newRing	2✔
49	}
50
51	// Deque is a lock-free single-producer multi-consumer deque.
52	// Deque is a bleeding-edge lock-free, single-producer multi-consumer, Chase-Lev work stealing deque as presented
53	// in the paper ["Dynamic Circular Work-Stealing Deque"](https://dl.acm.org/doi/10.1145/1073970.1073974) and further improved in the follow up paper:
54	// ["Correct and Efficient Work-Stealing for Weak Memory Models"](https://dl.acm.org/doi/10.1145/2442516.2442524).
55	type Deque[T any] struct {
56	top uint64
57	bottom uint64
58	buffer unsafe.Pointer
59	garbage []*RingBuffer[T]
60	}
61
62	// NewDeque creates a new deque with the given capacity.
63	func NewDeque[T any](cap uint64) *Deque[T] {	5✔
64	return &Deque[T]{buffer: unsafe.Pointer(NewRingBuffer[T](cap))}	5✔
65	}	5✔
66
67	// Size returns the number of elements in the deque.
68	func (d *Deque[T]) Size() uint64 {	59✔
69	b := atomic.LoadUint64(&d.bottom)	59✔
70	t := atomic.LoadUint64(&d.top)	59✔
71	if b >= t {	118✔
72	return b - t	59✔
73	}	59✔
NEW 74	return 0	×
75	}
76
77	// Capacity returns the capacity of the deque.
78	func (d *Deque[T]) Capacity() uint64 {	13✔
79	return (*RingBuffer[T])(d.buffer).Capacity()	13✔
80	}	13✔
81
82	// Empty returns true if the deque is empty.
83	func (d *Deque[T]) Empty() bool {	29✔
84	return d.Size() == 0	29✔
85	}	29✔
86
87	// Push pushes the value x to the deque.
88	// The value x is pushed to the bottom of the deque.
89	func (d *Deque[T]) Push(x T) {	23✔
90	b := atomic.LoadUint64(&d.bottom)	23✔
91	t := atomic.LoadUint64(&d.top)	23✔
92	buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))	23✔
93	if buf.Capacity() < b-t+1 {	25✔
94	d.garbage = append(d.garbage, buf)	2✔
95	buf = buf.Resize(b, t)	2✔
96	atomic.StorePointer(&d.buffer, unsafe.Pointer(buf))	2✔
97	}	2✔
98	buf.Store(b, x)	23✔
99	atomic.StoreUint64(&d.bottom, b+1)	23✔
100	}
101
102	// Pop pops the value from the deque.
103	// The value x is popped from the bottom of the deque.
104	func (d *Deque[T]) Pop() (x T, ok bool) {	19✔
105	b := atomic.AddUint64(&d.bottom, ^uint64(0))	19✔
106	buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))	19✔
107	t := atomic.LoadUint64(&d.top)	19✔
108	if t <= b {	37✔
109	if t == b {	22✔
110	if !atomic.CompareAndSwapUint64(&d.top, t, t+1) {	4✔
NEW 111	atomic.StoreUint64(&d.bottom, b+1)	×
NEW 112	return x, false	×
NEW 113	}	×
114	atomic.StoreUint64(&d.bottom, b+1)	4✔
115	}
116	x = buf.Load(b)	18✔
117	return x, true	18✔
118	}
119	atomic.StoreUint64(&d.bottom, b+1)	1✔
120	return x, false	1✔
121	}
122
123	// Peek peeks at the value from the deque without removing it.
124	// The value x is peeked from the top of the deque without removing it.
NEW 125	func (d *Deque[T]) Peek() (x T, ok bool) {	×
NEW 126	b := atomic.AddUint64(&d.bottom, ^uint64(0))	×
NEW 127	buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))	×
NEW 128	t := atomic.LoadUint64(&d.top)	×
NEW 129	if t <= b {	×
NEW 130	x = buf.Load(b)	×
NEW 131	return x, true	×
NEW 132	}	×
133
NEW 134	return x, false	×
135	}
136
137	// Steal steals a value from the deque.
138	// The value x is stolen from the top of the deque.
139	func (d *Deque[T]) Steal() (x T, ok bool) {	6✔
140	t := atomic.LoadUint64(&d.top)	6✔
141	b := atomic.LoadUint64(&d.bottom)	6✔
142	if t < b {	11✔
143	buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))	5✔
144	x = buf.Load(t)	5✔
145	if !atomic.CompareAndSwapUint64(&d.top, t, t+1) {	5✔
NEW 146	return x, false	×
NEW 147	}	×
148	return x, true	5✔
149	}
150	return x, false	1✔
151	}
152
153	// StealPeek peeks at a value from the deque without removing it.
154	// The value x is peeked from the top of the deque without removing it.
NEW 155	func (d *Deque[T]) StealPeek() (x T, ok bool) {	×
NEW 156	t := atomic.LoadUint64(&d.top)	×
NEW 157	b := atomic.LoadUint64(&d.bottom)	×
NEW 158	if t < b {	×
NEW 159	buf := (*RingBuffer[T])(atomic.LoadPointer(&d.buffer))	×
NEW 160	x = buf.Load(t)	×
NEW 161	return x, true	×
NEW 162	}	×
NEW 163	return x, false	×
164	}

smallnest / exp / 8276232797

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous