实现一个无锁的无界队列，使用新的atomic.Pointer类型。

huangapple
go评论108阅读模式
英文:

Implementing an lock-free unbounded queue with new atomic.Pointer types

问题

我正在尝试实现Michael和Scott的这个非阻塞队列。

我尝试使用Go 1.19中引入的新的atomic.Pointer类型，但是我的应用程序中出现了数据竞争。

这是我的实现代码：

  1. package queue
  3. import (
  4. 	"errors"
  5. 	"sync/atomic"
  6. )
  8. // LockfreeQueue表示具有入队和出队通用值操作的FIFO结构。
  9. // 参考：https://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html
  10. type LockFreeQueue[T any] struct {
  11. 	head atomic.Pointer[node[T]]
  12. 	tail atomic.Pointer[node[T]]
  13. }
  15. // node表示队列中的一个节点
  16. type node[T any] struct {
  17. 	value T
  18. 	next  atomic.Pointer[node[T]]
  19. }
  21. // newNode创建并初始化一个节点
  22. func newNode[T any](v T) *node[T] {
  23. 	return &node[T]{value: v}
  24. }
  26. // NewQueue创建并初始化一个LockFreeQueue
  27. func NewLockFreeQueue[T any]() *LockFreeQueue[T] {
  28. 	var head atomic.Pointer[node[T]]
  29. 	var tail atomic.Pointer[node[T]]
  30. 	n := &node[T]{}
  31. 	head.Store(n)
  32. 	tail.Store(n)
  33. 	return &LockFreeQueue[T]{
  34. 		head: head,
  35. 		tail: tail,
  36. 	}
  37. }
  39. // Enqueue向队列中添加一系列请求
  40. func (q *LockFreeQueue[T]) Enqueue(v T) {
  41. 	n := newNode(v)
  42. 	for {
  43. 		tail := q.tail.Load()
  44. 		next := tail.next.Load()
  45. 		if tail == q.tail.Load() {
  46. 			if next == nil {
  47. 				if tail.next.CompareAndSwap(next, n) {
  48. 					q.tail.CompareAndSwap(tail, n)
  49. 					return
  50. 				}
  51. 			} else {
  52. 				q.tail.CompareAndSwap(tail, next)
  53. 			}
  54. 		}
  55. 	}
  56. }
  58. // Dequeue从队列中移除一个请求
  59. func (q *LockFreeQueue[T]) Dequeue() (T, error) {
  60. 	for {
  61. 		head := q.head.Load()
  62. 		tail := q.tail.Load()
  63. 		next := head.next.Load()
  64. 		if head == q.head.Load() {
  65. 			if head == tail {
  66. 				if next == nil {
  67. 					return head.value, errors.New("队列为空")
  68. 				}
  69. 				q.tail.CompareAndSwap(tail, next)
  70. 			} else {
  71. 				v := next.value
  72. 				if q.head.CompareAndSwap(head, next) {
  73. 					return v, nil
  74. 				}
  75. 			}
  76. 		}
  77. 	}
  78. }
  80. // 检查队列是否为空。
  81. func (q *LockFreeQueue[T]) IsEmpty() bool {
  82. 	return q.head.Load() == q.tail.Load()
  83. }

我在这里找到了另一种实现，它在我的应用程序中没有数据竞争，但我似乎无法确定两者之间的确切区别。

任何帮助或反馈将不胜感激！

英文:

I am trying to implement this non-blocking queue from Michael and Scott.

I am trying to use the new atomic.Pointer types introduced in Go 1.19 but I am getting a data race in my application.

Here is my implementation:

  1. package queue
  3. import (
  4. 	"errors"
  5. 	"sync/atomic"
  6. )
  8. // LockfreeQueue represents a FIFO structure with operations to enqueue
  9. // and dequeue generic values.
  10. // Reference: https://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html
  11. type LockFreeQueue[T any] struct {
  12. 	head atomic.Pointer[node[T]]
  13. 	tail atomic.Pointer[node[T]]
  14. }
  16. // node represents a node in the queue
  17. type node[T any] struct {
  18. 	value T
  19. 	next  atomic.Pointer[node[T]]
  20. }
  22. // newNode creates and initializes a node
  23. func newNode[T any](v T) *node[T] {
  24. 	return &node[T]{value: v}
  25. }
  27. // NewQueue creates and initializes a LockFreeQueue
  28. func NewLockFreeQueue[T any]() *LockFreeQueue[T] {
  29. 	var head atomic.Pointer[node[T]]
  30. 	var tail atomic.Pointer[node[T]]
  31. 	n := &node[T]{}
  32. 	head.Store(n)
  33. 	tail.Store(n)
  34. 	return &LockFreeQueue[T]{
  35. 		head: head,
  36. 		tail: tail,
  37. 	}
  38. }
  40. // Enqueue adds a series of Request to the queue
  41. func (q *LockFreeQueue[T]) Enqueue(v T) {
  42. 	n := newNode(v)
  43. 	for {
  44. 		tail := q.tail.Load()
  45. 		next := tail.next.Load()
  46. 		if tail == q.tail.Load() {
  47. 			if next == nil {
  48. 				if tail.next.CompareAndSwap(next, n) {
  49. 					q.tail.CompareAndSwap(tail, n)
  50. 					return
  51. 				}
  52. 			} else {
  53. 				q.tail.CompareAndSwap(tail, next)
  54. 			}
  55. 		}
  56. 	}
  57. }
  59. // Dequeue removes a Request from the queue
  60. func (q *LockFreeQueue[T]) Dequeue() (T, error) {
  61. 	for {
  62. 		head := q.head.Load()
  63. 		tail := q.tail.Load()
  64. 		next := head.next.Load()
  65. 		if head == q.head.Load() {
  66. 			if head == tail {
  67. 				if next == nil {
  68. 					return head.value, errors.New("queue is empty")
  69. 				}
  70. 				q.tail.CompareAndSwap(tail, next)
  71. 			} else {
  72. 				v := next.value
  73. 				if q.head.CompareAndSwap(head, next) {
  74. 					return v, nil
  75. 				}
  76. 			}
  77. 		}
  78. 	}
  79. }
  81. // Check if the queue is empty.
  82. func (q *LockFreeQueue[T]) IsEmpty() bool {
  83.         return q.head.Load() == q.tail.Load()
  84. }

I found a different implementation here which works in my application without a data race, but I can't seem to figure out what is exactly different between the two.

Any help or feedback is appreciated!

答案1

得分: 1

原来是改变了一些东西解决了问题。

第一个改变：

  1. var n = node[T]{}
  2. head.Store(&n)
  3. tail.Store(&n)

第二个改变是改变了Dequeue()的返回类型。

最终的文件如下所示：

  1. package queue
  3. import (
  4. 	"sync/atomic"
  5. )
  7. // LockfreeQueue 表示具有入队和出队通用操作的FIFO结构。
  8. // 参考：https://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html
  9. type LockFreeQueue[T any] struct {
  10. 	head atomic.Pointer[node[T]]
  11. 	tail atomic.Pointer[node[T]]
  12. }
  14. // node 表示队列中的一个节点
  15. type node[T any] struct {
  16. 	value T
  17. 	next  atomic.Pointer[node[T]]
  18. }
  20. // newNode 创建并初始化一个节点
  21. func newNode[T any](v T) *node[T] {
  22. 	return &node[T]{value: v}
  23. }
  25. // NewQueue 创建并初始化一个LockFreeQueue
  26. func NewLockFreeQueue[T any]() *LockFreeQueue[T] {
  27. 	var head atomic.Pointer[node[T]]
  28. 	var tail atomic.Pointer[node[T]]
  29. 	var n = node[T]{}
  30. 	head.Store(&n)
  31. 	tail.Store(&n)
  32. 	return &LockFreeQueue[T]{
  33. 		head: head,
  34. 		tail: tail,
  35. 	}
  36. }
  38. // Enqueue 向队列中添加一系列请求
  39. func (q *LockFreeQueue[T]) Enqueue(v T) {
  40. 	n := newNode(v)
  41. 	for {
  42. 		tail := q.tail.Load()
  43. 		next := tail.next.Load()
  44. 		if tail == q.tail.Load() {
  45. 			if next == nil {
  46. 				if tail.next.CompareAndSwap(next, n) {
  47. 					q.tail.CompareAndSwap(tail, n)
  48. 					return
  49. 				}
  50. 			} else {
  51. 				q.tail.CompareAndSwap(tail, next)
  52. 			}
  53. 		}
  54. 	}
  55. }
  57. // Dequeue 从队列中移除一个请求
  58. func (q *LockFreeQueue[T]) Dequeue() T {
  59. 	var t T
  60. 	for {
  61. 		head := q.head.Load()
  62. 		tail := q.tail.Load()
  63. 		next := head.next.Load()
  64. 		if head == q.head.Load() {
  65. 			if head == tail {
  66. 				if next == nil {
  67. 					return t
  68. 				}
  69. 				q.tail.CompareAndSwap(tail, next)
  70. 			} else {
  71. 				v := next.value
  72. 				if q.head.CompareAndSwap(head, next) {
  73. 					return v
  74. 				}
  75. 			}
  76. 		}
  77. 	}
  78. }
  80. // 检查队列是否为空。
  81. func (q *LockFreeQueue[T]) IsEmpty() bool {
  82. 	return q.head.Load() == q.tail.Load()
  83. }
英文:

It turned out that changing some things fixed the issue.

First change:

  1. var n = node[T]{}
  2. head.Store(&n)
  3. tail.Store(&n)

Second change was changing the Dequeue() return signature.

The final file looks like this:

  1. package queue
  3. import (
  4. 	"sync/atomic"
  5. )
  7. // LockfreeQueue represents a FIFO structure with operations to enqueue
  8. // and dequeue generic values.
  9. // Reference: https://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html
  10. type LockFreeQueue[T any] struct {
  11. 	head atomic.Pointer[node[T]]
  12. 	tail atomic.Pointer[node[T]]
  13. }
  15. // node represents a node in the queue
  16. type node[T any] struct {
  17. 	value T
  18. 	next  atomic.Pointer[node[T]]
  19. }
  21. // newNode creates and initializes a node
  22. func newNode[T any](v T) *node[T] {
  23. 	return &node[T]{value: v}
  24. }
  26. // NewQueue creates and initializes a LockFreeQueue
  27. func NewLockFreeQueue[T any]() *LockFreeQueue[T] {
  28. 	var head atomic.Pointer[node[T]]
  29. 	var tail atomic.Pointer[node[T]]
  30. 	var n = node[T]{}
  31. 	head.Store(&n)
  32. 	tail.Store(&n)
  33. 	return &LockFreeQueue[T]{
  34. 		head: head,
  35. 		tail: tail,
  36. 	}
  37. }
  39. // Enqueue adds a series of Request to the queue
  40. func (q *LockFreeQueue[T]) Enqueue(v T) {
  41. 	n := newNode(v)
  42. 	for {
  43. 		tail := q.tail.Load()
  44. 		next := tail.next.Load()
  45. 		if tail == q.tail.Load() {
  46. 			if next == nil {
  47. 				if tail.next.CompareAndSwap(next, n) {
  48. 					q.tail.CompareAndSwap(tail, n)
  49. 					return
  50. 				}
  51. 			} else {
  52. 				q.tail.CompareAndSwap(tail, next)
  53. 			}
  54. 		}
  55. 	}
  56. }
  58. // Dequeue removes a Request from the queue
  59. func (q *LockFreeQueue[T]) Dequeue() T {
  60. 	var t T
  61. 	for {
  62. 		head := q.head.Load()
  63. 		tail := q.tail.Load()
  64. 		next := head.next.Load()
  65. 		if head == q.head.Load() {
  66. 			if head == tail {
  67. 				if next == nil {
  68. 					return t
  69. 				}
  70. 				q.tail.CompareAndSwap(tail, next)
  71. 			} else {
  72. 				v := next.value
  73. 				if q.head.CompareAndSwap(head, next) {
  74. 					return v
  75. 				}
  76. 			}
  77. 		}
  78. 	}
  79. }
  81. // Check if the queue is empty.
  82. func (q *LockFreeQueue[T]) IsEmpty() bool {
  83. 	return q.head.Load() == q.tail.Load()
  84. }

huangapple
  • 本文由 发表于 2023年4月22日 08:11:59
  • 转载请务必保留本文链接：https://go.coder-hub.com/76077440.html
  • atomic
  • go
  • nonblocking
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定