Documentation
¶
Index ¶
- type ProcessFunc
- type RingMPSC
- func (rb *RingMPSC[T]) Capacity() uint64
- func (rb *RingMPSC[T]) Close()
- func (rb *RingMPSC[T]) IsClosed() bool
- func (rb *RingMPSC[T]) IsEmpty() bool
- func (rb *RingMPSC[T]) IsFull() bool
- func (rb *RingMPSC[T]) Length() uint64
- func (rb *RingMPSC[T]) Pop() (T, bool)
- func (rb *RingMPSC[T]) Push(item T) bool
- func (rb *RingMPSC[T]) TryPop() (T, bool)
- func (rb *RingMPSC[T]) TryPush(item T) bool
- type RingMPSCConfig
- type RingSPSC
- func (rb *RingSPSC[T]) Capacity() uint64
- func (rb *RingSPSC[T]) Close()
- func (rb *RingSPSC[T]) IsClosed() bool
- func (rb *RingSPSC[T]) IsEmpty() bool
- func (rb *RingSPSC[T]) IsFull() bool
- func (rb *RingSPSC[T]) Length() uint64
- func (rb *RingSPSC[T]) Pop() (T, bool)
- func (rb *RingSPSC[T]) Push(item T) bool
- func (rb *RingSPSC[T]) TryPop() (T, bool)
- func (rb *RingSPSC[T]) TryPush(item T) bool
- type RingSPSCConfig
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ProcessFunc ¶
type ProcessFunc[T any] func(T)
ProcessFunc defines the function signature for processing messages T can be any type: []byte, string, struct, etc.
type RingMPSC ¶
type RingMPSC[T any] struct { // contains filtered or unexported fields }
RingMPSC is a high-performance ring buffer for multiple producers and a single consumer Uses a mutex to serialize concurrent writes from multiple producers
func NewRingMPSC ¶
func NewRingMPSC[T any](cfg RingMPSCConfig[T]) *RingMPSC[T]
NewRingMPSC creates a new MPSC ring buffer with the given configuration
Example (ManualPop) ¶
ExampleNewRingMPSC_manualPop shows manual consumer mode without ProcessFunc. Multiple producers push concurrently while the caller drives the Pop loop.
package main
import (
"fmt"
"sync"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
rb := ringbuf.NewRingMPSC(ringbuf.RingMPSCConfig[int]{
Capacity: 16,
// ProcessFunc is nil: no automatic consumer goroutine
})
const n = 5
var wg sync.WaitGroup
for i := range n {
wg.Add(1)
go func(v int) {
defer wg.Done()
rb.Push(v)
}(i)
}
// Close after all producers finish so the Pop loop can terminate.
go func() {
wg.Wait()
rb.Close()
}()
count := 0
for {
_, ok := rb.Pop()
if !ok {
break
}
count++
}
fmt.Println(count)
}
Output: 5
Example (ProcessFunc) ¶
ExampleNewRingMPSC_processFunc shows the typical MPSC pattern: multiple producer goroutines push concurrently while a single consumer goroutine (started automatically via ProcessFunc) processes each item sequentially.
package main
import (
"fmt"
"sync"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
const producers = 3
var (
producerWg sync.WaitGroup
consumerWg sync.WaitGroup
)
consumerWg.Add(producers)
rb := ringbuf.NewRingMPSC(ringbuf.RingMPSCConfig[string]{
Capacity: 16,
ProcessFunc: func(msg string) {
// consumer goroutine: called once per message, sequentially
_ = msg
consumerWg.Done()
},
})
// Launch multiple producer goroutines
for i := range producers {
producerWg.Add(1)
go func(id int) {
defer producerWg.Done()
rb.Push(fmt.Sprintf("producer-%d", id))
}(i)
}
producerWg.Wait() // wait for all producers to finish pushing
consumerWg.Wait() // wait for all items to be processed
rb.Close()
}
Output:
func (*RingMPSC[T]) Close ¶
func (rb *RingMPSC[T]) Close()
Close closes the ring buffer and wakes up blocked consumers After close, producers cannot push new data, but consumers will drain remaining data Safe to call multiple times
func (*RingMPSC[T]) IsEmpty ¶
IsEmpty reports whether the ring buffer contains no items. It suffers from a race condition where the result may be stale immediately after return.
func (*RingMPSC[T]) IsFull ¶
IsFull reports whether the ring buffer is at capacity. It suffers from a race condition where the result may be stale immediately after return.
func (*RingMPSC[T]) Length ¶
Length returns the approximate number of items currently in the buffer. The returned length is approximate and may be stale by the time it's used.
func (*RingMPSC[T]) Pop ¶
Pop reads data by consumer with blocking Blocks and waits until data arrives if buffer is empty Returns the data and success flag (false indicates closed AND buffer is empty) When closed, will drain all remaining data before returning false
func (*RingMPSC[T]) Push ¶
Push writes data by producer with blocking Blocks and waits until space is available if buffer is full Returns false if ring buffer is closed, true on success
Example ¶
ExampleRingMPSC_Push shows blocking push. Unlike TryPush, Push waits until space is available when the buffer is full, providing natural backpressure to the producer. It is safe to call from multiple goroutines concurrently.
package main
import (
"fmt"
"sync"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
var wg sync.WaitGroup
wg.Add(3)
rb := ringbuf.NewRingMPSC(ringbuf.RingMPSCConfig[string]{
Capacity: 4,
ProcessFunc: func(s string) {
fmt.Println(s)
wg.Done()
},
})
// Push from the current goroutine; blocks only when the buffer is full.
rb.Push("first")
rb.Push("second")
rb.Push("third")
wg.Wait()
rb.Close()
}
Output: first second third
func (*RingMPSC[T]) TryPop ¶
TryPop reads data by consumer (non-blocking) Returns the data and a flag indicating success
Example ¶
ExampleRingMPSC_TryPop shows non-blocking pop for the MPSC buffer. TryPop is intended for use by the single consumer only.
package main
import (
"fmt"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
rb := ringbuf.NewRingMPSC(ringbuf.RingMPSCConfig[int]{Capacity: 4})
defer rb.Close()
rb.TryPush(42)
v, ok := rb.TryPop()
fmt.Println(v, ok) // 42 true
_, ok = rb.TryPop()
fmt.Println(ok) // false – buffer now empty
}
Output: 42 true false
func (*RingMPSC[T]) TryPush ¶
TryPush writes data by producer (non-blocking) Returns true on success, false if buffer is full or closed
Example ¶
ExampleRingMPSC_TryPush shows non-blocking push for the MPSC buffer. Safe to call from multiple goroutines simultaneously.
package main
import (
"fmt"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
rb := ringbuf.NewRingMPSC(ringbuf.RingMPSCConfig[int]{Capacity: 2})
defer rb.Close()
fmt.Println(rb.TryPush(10)) // true – slot available
fmt.Println(rb.TryPush(20)) // true – slot available
fmt.Println(rb.TryPush(30)) // false – buffer full
}
Output: true true false
type RingMPSCConfig ¶
type RingMPSCConfig[T any] struct { // Capacity is the buffer size of the ring buffer Capacity uint64 // ProcessFunc is the function to process each message // Optional: if provided, a consumer goroutine will be automatically started // If nil, you need to manually call Pop() to consume messages ProcessFunc ProcessFunc[T] }
RingMPSCConfig holds configuration for RingMPSC T is the type of messages being processed
type RingSPSC ¶
type RingSPSC[T any] struct { // contains filtered or unexported fields }
RingSPSC is a lock-free ring buffer for single producer and single consumer Implements high-performance SPSC queue using atomic operations, ensuring strict FIFO ordering
func NewRingSPSC ¶
func NewRingSPSC[T any](cfg RingSPSCConfig[T]) *RingSPSC[T]
NewRingSPSC creates a new SPSC ring buffer with the given configuration
Example (ManualPop) ¶
ExampleNewRingSPSC_manualPop shows manual consumer mode: omit ProcessFunc and drive the Pop loop yourself. Useful when the consumer needs full control over scheduling or batching.
package main
import (
"fmt"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
rb := ringbuf.NewRingSPSC(ringbuf.RingSPSCConfig[string]{
Capacity: 4,
// ProcessFunc is nil: no automatic consumer goroutine
})
// Producer goroutine
go func() {
rb.Push("hello")
rb.Push("world")
rb.Close() // signal that no more items will be pushed
}()
// Consumer in the current goroutine: drain until closed and empty
for {
item, ok := rb.Pop()
if !ok {
break
}
fmt.Println(item)
}
}
Output: hello world
Example (ProcessFunc) ¶
ExampleNewRingSPSC_processFunc shows the most common usage pattern: supply a ProcessFunc so the ring buffer starts a consumer goroutine automatically. The producer calls Push; the consumer goroutine calls ProcessFunc for each item. Close drains all remaining items and waits for the consumer to finish.
package main
import (
"fmt"
"sync"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
var wg sync.WaitGroup
wg.Add(3)
rb := ringbuf.NewRingSPSC(ringbuf.RingSPSCConfig[int]{
Capacity: 8,
ProcessFunc: func(v int) {
fmt.Println(v)
wg.Done()
},
})
rb.Push(1)
rb.Push(2)
rb.Push(3)
wg.Wait() // wait until all items have been processed
rb.Close()
}
Output: 1 2 3
func (*RingSPSC[T]) Close ¶
func (rb *RingSPSC[T]) Close()
Close closes the ring buffer and wakes up blocked consumers After close, producers cannot push new data, but consumers will drain remaining data Safe to call multiple times
func (*RingSPSC[T]) IsEmpty ¶
IsEmpty reports whether the ring buffer contains no items. It suffers from a race condition where the result may be stale immediately after return.
func (*RingSPSC[T]) IsFull ¶
IsFull reports whether the ring buffer is at capacity. It suffers from a race condition where the result may be stale immediately after return.
func (*RingSPSC[T]) Length ¶
Length returns the approximate number of items currently in the buffer. The returned length is approximate and may be stale by the time it's used.
func (*RingSPSC[T]) Pop ¶
Pop reads data by consumer with blocking Blocks and waits until data arrives if buffer is empty Returns the data and success flag (false indicates closed AND buffer is empty) When closed, will drain all remaining data before returning false
func (*RingSPSC[T]) Push ¶
Push writes data by producer with blocking Blocks and waits until space is available if buffer is full Returns false if ring buffer is closed, true on success This approach ensures data is not lost, forming a backpressure mechanism
Example ¶
ExampleRingSPSC_Push shows blocking push. Unlike TryPush, Push waits until space is available when the buffer is full, providing natural backpressure to the single producer. Must only be called from one goroutine at a time.
package main
import (
"fmt"
"sync"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
var wg sync.WaitGroup
wg.Add(3)
rb := ringbuf.NewRingSPSC(ringbuf.RingSPSCConfig[string]{
Capacity: 4,
ProcessFunc: func(s string) {
fmt.Println(s)
wg.Done()
},
})
// Push from the single producer goroutine; blocks only when the buffer is full.
rb.Push("first")
rb.Push("second")
rb.Push("third")
wg.Wait()
rb.Close()
}
Output: first second third
func (*RingSPSC[T]) TryPop ¶
TryPop reads data by consumer (non-blocking) Returns the data and a flag indicating success
Example ¶
ExampleRingSPSC_TryPop shows non-blocking pop. TryPop returns false immediately when the buffer is empty instead of blocking the caller.
package main
import (
"fmt"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
rb := ringbuf.NewRingSPSC(ringbuf.RingSPSCConfig[int]{Capacity: 4})
defer rb.Close()
rb.TryPush(42)
v, ok := rb.TryPop()
fmt.Println(v, ok) // 42 true
_, ok = rb.TryPop()
fmt.Println(ok) // false – buffer now empty
}
Output: 42 true false
func (*RingSPSC[T]) TryPush ¶
TryPush writes data by producer (non-blocking) Returns true on success, false if buffer is full or closed
Example ¶
ExampleRingSPSC_TryPush shows non-blocking push. TryPush returns false immediately when the buffer is full instead of blocking the caller.
package main
import (
"fmt"
"github.com/phcp-tech/common-library-golang/ringbuf"
)
func main() {
rb := ringbuf.NewRingSPSC(ringbuf.RingSPSCConfig[int]{Capacity: 2})
defer rb.Close()
fmt.Println(rb.TryPush(10)) // true – slot available
fmt.Println(rb.TryPush(20)) // true – slot available
fmt.Println(rb.TryPush(30)) // false – buffer full (capacity 2)
}
Output: true true false
type RingSPSCConfig ¶
type RingSPSCConfig[T any] struct { // Capacity is the buffer size of the ring buffer Capacity uint64 // ProcessFunc is the function to process each message // Optional: if provided, a consumer goroutine will be automatically started // If nil, you need to manually call Pop() to consume messages ProcessFunc ProcessFunc[T] }
RingSPSCConfig holds configuration for RingSPSC T is the type of messages being processed
Source Files