php小编草莓今天要为大家介绍一种新的技术——使用新的atomic.Pointer类型实现无锁无界队列。在并发编程中,队列是一种常见的数据结构,但传统的队列实现通常需要使用锁来保证线程安全,这会带来性能的损耗。而新的atomic.Pointer类型则提供了一种无锁的解决方案,可以实现高效的并发队列操作。下面我们将详细介绍这种新的实现方式,以及它的优势和使用方法。
我正在尝试实现 michael 和 scott 的这个非阻塞队列。
我正在尝试使用 go 1.19 中引入的新的atomic.pointer 类型,但我的应用程序中出现了数据争用。
这是我的实现:
package queue import ( "errors" "sync/atomic" ) // LockfreeQueue represents a FIFO structure with operations to enqueue // and dequeue generic values. // Reference: https://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html type LockFreeQueue[T any] struct { head atomic.Pointer[node[T]] tail atomic.Pointer[node[T]] } // node represents a node in the queue type node[T any] struct { value T next atomic.Pointer[node[T]] } // newNode creates and initializes a node func newNode[T any](v T) *node[T] { return &node[T]{value: v} } // NewQueue creates and initializes a LockFreeQueue func NewLockFreeQueue[T any]() *LockFreeQueue[T] { var head atomic.Pointer[node[T]] var tail atomic.Pointer[node[T]] n := &node[T]{} head.Store(n) tail.Store(n) return &LockFreeQueue[T]{ head: head, tail: tail, } } // Enqueue adds a series of Request to the queue func (q *LockFreeQueue[T]) Enqueue(v T) { n := newNode(v) for { tail := q.tail.Load() next := tail.next.Load() if tail == q.tail.Load() { if next == nil { if tail.next.CompareAndSwap(next, n) { q.tail.CompareAndSwap(tail, n) return } } else { q.tail.CompareAndSwap(tail, next) } } } } // Dequeue removes a Request from the queue func (q *LockFreeQueue[T]) Dequeue() (T, error) { for { head := q.head.Load() tail := q.tail.Load() next := head.next.Load() if head == q.head.Load() { if head == tail { if next == nil { return head.value, errors.New("queue is empty") } q.tail.CompareAndSwap(tail, next) } else { v := next.value if q.head.CompareAndSwap(head, next) { return v, nil } } } } } // Check if the queue is empty. func (q *LockFreeQueue[T]) IsEmpty() bool { return q.head.Load() == q.tail.Load() }
我在这里发现了一个不同的实现,它可以在我的应用程序中工作而无需数据竞争,但我似乎无法弄清楚两者之间到底有什么不同。
感谢任何帮助或反馈!
事实证明,更改一些内容可以解决问题。
第一个变化:
var n = node[t]{} head.store(&n) tail.store(&n)
第二个更改是更改 dequeue()
返回签名。
最终文件如下所示:
package queue import ( "sync/atomic" ) // LockfreeQueue represents a FIFO structure with operations to enqueue // and dequeue generic values. // Reference: https://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html type LockFreeQueue[T any] struct { head atomic.Pointer[node[T]] tail atomic.Pointer[node[T]] } // node represents a node in the queue type node[T any] struct { value T next atomic.Pointer[node[T]] } // newNode creates and initializes a node func newNode[T any](v T) *node[T] { return &node[T]{value: v} } // NewQueue creates and initializes a LockFreeQueue func NewLockFreeQueue[T any]() *LockFreeQueue[T] { var head atomic.Pointer[node[T]] var tail atomic.Pointer[node[T]] var n = node[T]{} head.Store(&n) tail.Store(&n) return &LockFreeQueue[T]{ head: head, tail: tail, } } // Enqueue adds a series of Request to the queue func (q *LockFreeQueue[T]) Enqueue(v T) { n := newNode(v) for { tail := q.tail.Load() next := tail.next.Load() if tail == q.tail.Load() { if next == nil { if tail.next.CompareAndSwap(next, n) { q.tail.CompareAndSwap(tail, n) return } } else { q.tail.CompareAndSwap(tail, next) } } } } // Dequeue removes a Request from the queue func (q *LockFreeQueue[T]) Dequeue() T { var t T for { head := q.head.Load() tail := q.tail.Load() next := head.next.Load() if head == q.head.Load() { if head == tail { if next == nil { return t } q.tail.CompareAndSwap(tail, next) } else { v := next.value if q.head.CompareAndSwap(head, next) { return v } } } } } // Check if the queue is empty. func (q *LockFreeQueue[T]) IsEmpty() bool { return q.head.Load() == q.tail.Load() }
以上是使用新的atomic.Pointer类型实现无锁无界队列的详细内容。更多信息请关注PHP中文网其他相关文章!