README
¶
concurrent
Concurrent is a library of lock-free and wait-free algorithms
Environment Requirements
- amd64 or arm64 CPU architecture
- Go 1.25 or later
Installation
To install the concurrent library, run the following command:
go get code.hybscloud.com/concurrent
Basic Usages
Queue (Convenience Constructor)
c, p := concurrent.NewQueue[string](256)
message := "Hello, Concurrent!"
err := concurrent.EnqueueWait(p, &message)
if err != nil {
return err
}
result, err := concurrent.DequeueWait(c)
if err != nil {
return err
}
println(*result)
Multi-Producer Multi-Consumer Queue
c, p := concurrent.NewMPMCQueue[int](256)
i := 100
err := p.Enqueue(&i)
if err != nil {
return err
}
res, err := c.Dequeue()
if err != nil {
return err
}
println(*res)
Multi-Producer Multi-Consumer Queue (Indirect)
c, p := concurrent.NewMPMCQueueIndirect(256)
index := uintptr(42)
err := p.Enqueue(index)
if err != nil {
return err
}
value, err := c.Dequeue()
if err != nil {
return err
}
println(value)
Spin Lock
lock := concurrent.SpinLock{} // the zero value is ready to use
lock.Lock()
...
lock.Unlock()
Spin Wait
sw := concurrent.SpinWait{} // the zero value is ready to use
sw.Once()
Next Step
Implement the sCQ lock-free FIFO queue
References
- A. Morrison and Y. Afek, "Fast concurrent queues for x86 processors," in Proc. 18th ACM SIGPLAN Symposium on Principles and Practice of Parallel Programming (PPoPP), 2013.
- R. Nikolaev, "A scalable, portable, and memory-efficient lock-free FIFO queue," in Proc. 33rd International Symposium on Distributed Computing (DISC), 2019. LIPIcs.
- N. Koval and V. Aksenov, "POSTER: Restricted memory-friendly lock-free bounded queues," in Proc. 25th ACM SIGPLAN Symposium on Principles and Practice of Parallel Programming (PPoPP), 2020, pp. 433–434.
- R. Nikolaev and B. Ravindran, "wCQ: A fast wait-free queue with bounded memory usage," arXiv preprint arXiv:2201.02179, Jan. 2022.
- V. Aksenov, N. Koval, P. Kuznetsov, and A. Paramonov, "Memory bounds for concurrent bounded queues," arXiv preprint arXiv:2104.15003v5, Jan. 2024.
- A. Denis and C. Goedefroit, "NBLFQ: A lock-free MPMC queue optimized for low contention," in Proc. 39th IEEE International Parallel and Distributed Processing Symposium (IPDPS), 2025, pp. 962–973.
- Intel Corporation, "Combined Volume Set of Intel 64 and IA-32 Architectures Software Developer’s Manuals."
- Arm Limited, "Arm Architecture Reference Manual for A-profile architecture," DDI 0596, latest revision.
License
©2023 Hayabusa Cloud Co., Ltd.
#5F Eclat BLDG, 3-6-2 Shibuya, Shibuya City, Tokyo 150-0002, Japan
Released under the MIT license
Documentation
¶
Index ¶
- Variables
- func DequeueWait[T any](c Consumer[T]) (elem *T, err error)
- func DoubleUint64(first, second uint64) (u128 []uint64)
- func DoubleUintPtr(first, second uintptr) (dw []uintptr)
- func EnqueueWait[T any](p Producer[T], elem *T) error
- func NewMPMCQueue[T any](capacity int) (Consumer[T], Producer[T])
- func NewMPMCQueueIndirect(capacity int) (ConsumerIndirect, ProducerIndirect)
- func NewMPSCQueue[T any](capacity int) (Consumer[T], Producer[T])
- func NewQueue[T any](capacity int, opts ...func(opts *QueueOptions)) (Consumer[T], Producer[T])
- func NewSPMCQueue[T any](capacity int) (Consumer[T], Producer[T])
- func NewSPSCQueue[T any](capacity int) (Consumer[T], Producer[T])
- func Pause(cycles ...int)
- func SetYieldDuration(d time.Duration)
- func Yield(lv ...int)
- type Closer
- type Consumer
- type ConsumerIndirect
- type MPMCQueue
- type MPMCQueueIndirect
- type MPSCQueue
- type Producer
- type ProducerIndirect
- type QueueOptions
- type SPMCQueue
- type SPSCQueue
- type SpinLock
- type SpinWait
Constants ¶
This section is empty.
Variables ¶
var AndUint64 func(u64 *uint64, val uint64) = and64
AndUint64 takes val and atomically performs a bit-wise "and" operation with the value of u64, storing the result into u64
var CompareAndSwapUint128 func(u128 *uint64, old, new [2]uint64) bool = cas128
CompareAndSwapUint128 atomically compares u128 with old, and if they're equal, swaps value of u128 with new. It reports whether the swap ran.
var ( // on a fulled queue or Dequeue operations on an empty queue ErrTemporaryUnavailable = errors.New("temporary unavailable") )
var OrUint64 func(u64 *uint64, val uint64) = or64
OrUint64 takes val and atomically performs a bit-wise "or" operation with the value of u64, storing the result into u64
Functions ¶
func DequeueWait ¶
DequeueWait pops items from fifo queue. the operation will block until a success or error occurred
func DoubleUint64 ¶
DoubleUint64 creates and returns []uint64 with size 2 and the given values the address of u128 will be 16-bytes aligned
func DoubleUintPtr ¶
DoubleUintPtr creates and returns []uintptr with size 2 and the given values the address of dw will be 16-bytes aligned
func EnqueueWait ¶
EnqueueWait pushes the given item to a fifo queue. the operation will block until a success or error occurred
func NewMPMCQueue ¶
NewMPMCQueue creates a new multiple producers multiple consumers FIFO queue with the given capacity
func NewMPMCQueueIndirect ¶
func NewMPMCQueueIndirect(capacity int) (ConsumerIndirect, ProducerIndirect)
NewMPMCQueueIndirect creates a new multiple producers multiple consumers FIFO queue with the given capacity
func NewMPSCQueue ¶
NewMPSCQueue creates a new multiple producers single consumer FIFO queue with the given capacity
func NewQueue ¶
func NewQueue[T any](capacity int, opts ...func(opts *QueueOptions)) (Consumer[T], Producer[T])
NewQueue creates a new queue with the given capacity and options.
func NewSPMCQueue ¶
NewSPMCQueue creates a new single producer multiple consumers FIFO queue with the given capacity
func NewSPSCQueue ¶
NewSPSCQueue creates a new simple producer single consumer FIFO queue with the given capacity
func Pause ¶
func Pause(cycles ...int)
Pause executes CPU pause instructions to reduce energy consumption in spin-wait loops.
Defaults to 20 cycles if not specified. Uses optimized assembly on amd64/arm64.
Usage:
Pause() // 20 cycles (default) Pause(1) // 1 cycle Pause(40) // 40 cycles
func SetYieldDuration ¶
SetYieldDuration sets the base duration unit for Yield(). Recommended: 50-250 microseconds for real-time systems, 1-4 ms for general workloads.
Types ¶
type Closer ¶
type Closer interface {
// Close closes the queue.
// Enqueue and Dequeue operations on a closed queue are undefined
Close() error
}
Closer is the interface that wraps the Close method
type Consumer ¶
type Consumer[T any] interface { // Dequeue pops items from the FIFO queue. // if the queue is empty, ErrTemporaryUnavailable will be returned Dequeue() (elem *T, err error) }
Consumer is the interface that wraps the Dequeue method
type ConsumerIndirect ¶
ConsumerIndirect is a non-generic consumer interface that dequeues uintptr values. Use this for indirect references such as indices, handles, or other integer-like identifiers.
type MPMCQueue ¶
type MPMCQueue[T any] struct { // contains filtered or unexported fields }
MPMCQueue represents multiple producers multiple consumers FIFO queue
type MPMCQueueIndirect ¶
type MPMCQueueIndirect struct {
// contains filtered or unexported fields
}
MPMCQueueIndirect represents multiple producers multiple consumers FIFO queue with indirect references
func (*MPMCQueueIndirect) Dequeue ¶
func (q *MPMCQueueIndirect) Dequeue() (elem uintptr, err error)
func (*MPMCQueueIndirect) Enqueue ¶
func (q *MPMCQueueIndirect) Enqueue(elem uintptr) error
type MPSCQueue ¶
type MPSCQueue[T any] struct{}
MPSCQueue represents multiple producers single consumer FIFO queue
type Producer ¶
type Producer[T any] interface { // Enqueue pushes item to FIFO queue. // if the queue is fulled, ErrTemporaryUnavailable will be returned Enqueue(elem *T) error }
Producer is the interface that wraps the Enqueue method
type ProducerIndirect ¶
ProducerIndirect is a non-generic producer interface that enqueues uintptr values. Use this for indirect references such as indices, handles, or other integer-like identifiers.
type QueueOptions ¶
type QueueOptions struct {
SingleProducer bool // TODO: implement
SingleConsumer bool // TODO: implement
LowContention bool // TODO: implement
DistinctValues bool
}
QueueOptions is a struct that contains options for creating a queue.
type SPMCQueue ¶
type SPMCQueue[T any] struct{}
SPMCQueue represents single producer multiple consumers FIFO queue
type SPSCQueue ¶
type SPSCQueue[T any] struct{}
SPSCQueue represents simple producer single consumer FIFO queue
Source Files