add: lock-free FIFO queue

Author: changkun

queue.go

@@ -0,0 +1,86 @@
+package lockfree
+
+import (
+	"sync/atomic"
+	"unsafe"
+)
+
+// Queue implements lock-free FIFO freelist based queue.
+// ref: https://dl.acm.org/citation.cfm?doid=248052.248106
+type Queue struct {
+	head unsafe.Pointer
+	tail unsafe.Pointer
+	len  uint64
+}
+
+// NewQueue creates a new lock-free queue.
+func NewQueue() *Queue {
+	head := queueitem{next: nil, v: nil} // allocate a free item
+	return &Queue{
+		tail: unsafe.Pointer(&head), // both head and tail points
+		head: unsafe.Pointer(&head), // to the free item
+	}
+}
+
+// Enqueue puts the given value v at the tail of the queue.
+func (q *Queue) Enqueue(v interface{}) {
+	item := &queueitem{next: nil, v: v} // allocate new item
+	var last, lastnext *queueitem
+	for {
+		last = loadqitem(&q.tail)
+		lastnext = loadqitem(&last.next)
+		if loadqitem(&q.tail) == last { // are tail and next consistent?
+			if lastnext == nil { // was tail pointing to the last node?
+				if casqitem(&last.next, lastnext, item) { // try to link item at the end of linked list
+					casqitem(&q.tail, last, item) // enqueue is done. try swing tail to the inserted node
+					atomic.AddUint64(&q.len, 1)
+					return
+				}
+			} else { // tail was not pointing to the last node
+				casqitem(&q.tail, last, lastnext) // try swing tail to the next node
+			}
+		}
+	}
+}
+
+// Dequeue removes and returns the value at the head of the queue.
+// It returns nil if the queue is empty.
+func (q *Queue) Dequeue() interface{} {
+	var first, last, firstnext *queueitem
+	for {
+		first = loadqitem(&q.head)
+		last = loadqitem(&q.tail)
+		firstnext = loadqitem(&first.next)
+		if first == loadqitem(&q.head) { // are head, tail and next consistent?
+			if first == last { // is queue empty?
+				if firstnext == nil { // queue is empty, couldn't dequeue
+					return nil
+				}
+				casqitem(&q.tail, last, firstnext) // tail is falling behind, try to advance it
+			} else { // read value before cas, otherwise another dequeue might free the next node
+				v := firstnext.v
+				if casqitem(&q.head, first, firstnext) { // try to swing head to the next node
+					atomic.AddUint64(&q.len, ^uint64(0))
+					return v // queue was not empty and dequeue finished.
+				}
+			}
+		}
+	}
+}
+
+// Length returns the length of the queue.
+func (q *Queue) Length() uint64 {
+	return atomic.LoadUint64(&q.len)
+}
+
+type queueitem struct {
+	next unsafe.Pointer
+	v    interface{}
+}
+
+func loadqitem(p *unsafe.Pointer) *queueitem {
+	return (*queueitem)(atomic.LoadPointer(p))
+}
+func casqitem(p *unsafe.Pointer, old, new *queueitem) bool {
+	return atomic.CompareAndSwapPointer(p, unsafe.Pointer(old), unsafe.Pointer(new))
+}

queue_test.go

@@ -0,0 +1,111 @@
+package lockfree_test
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/changkun/lockfree"
+)
+
+func TestQueueDequeueEmpty(t *testing.T) {
+	q := lockfree.NewQueue()
+	if q.Dequeue() != nil {
+		t.Fatalf("dequeue empty queue returns non-nil")
+	}
+}
+
+func TestQueue_Length(t *testing.T) {
+	q := lockfree.NewQueue()
+	if q.Length() != 0 {
+		t.Fatalf("empty queue has non-zero length")
+	}
+
+	q.Enqueue(1)
+	if q.Length() != 1 {
+		t.Fatalf("count of enqueue wrong, want %d, got %d.", 1, q.Length())
+	}
+
+	q.Dequeue()
+	if q.Length() != 0 {
+		t.Fatalf("count of dequeue wrong, want %d, got %d", 0, q.Length())
+	}
+}
+
+func ExampleQueue() {
+	q := lockfree.NewQueue()
+
+	q.Enqueue("1st item")
+	q.Enqueue("2nd item")
+	q.Enqueue("3rd item")
+
+	fmt.Println(q.Dequeue())
+	fmt.Println(q.Dequeue())
+	fmt.Println(q.Dequeue())
+
+	// Output:
+	// 1st item
+	// 2nd item
+	// 3rd item
+}
+
+type queueInterface interface {
+	Enqueue(interface{})
+	Dequeue() interface{}
+}
+
+type mutexQueue struct {
+	v  []interface{}
+	mu sync.Mutex
+}
+
+func newMutexQueue() *mutexQueue {
+	return &mutexQueue{v: make([]interface{}, 0)}
+}
+
+func (q *mutexQueue) Enqueue(v interface{}) {
+	q.mu.Lock()
+	q.v = append(q.v, v)
+	q.mu.Unlock()
+}
+
+func (q *mutexQueue) Dequeue() interface{} {
+	q.mu.Lock()
+	if len(q.v) == 0 {
+		q.mu.Unlock()
+		return nil
+	}
+	v := q.v[0]
+	q.v = q.v[1:]
+	q.mu.Unlock()
+	return v
+}
+
+func BenchmarkQueue(b *testing.B) {
+	length := 1 << 12
+	inputs := make([]int, length)
+	for i := 0; i < length; i++ {
+		inputs = append(inputs, rand.Int())
+	}
+	q, mq := lockfree.NewQueue(), newMutexQueue()
+	b.ResetTimer()
+
+	for _, q := range [...]queueInterface{q, mq} {
+		b.Run(fmt.Sprintf("%T", q), func(b *testing.B) {
+			var c int64
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					i := int(atomic.AddInt64(&c, 1)-1) % length
+					v := inputs[i]
+					if v >= 0 {
+						q.Enqueue(v)
+					} else {
+						q.Dequeue()
+					}
+				}
+			})
+		})
+	}
+}