queue

README

queue

This project is a Go implementation of a high-performance priority queue with TTL (Time To Live) expiration functionality.

Features

• High Performance: Optimized with entry reuse and zero-allocation steady-state push/pop paths
• TTL Support: Automatic cleanup of expired entries with configurable intervals
• Thread-Safe: Concurrent operations with optimized read-write locking
• Generic Types: Support for any comparable type using Go generics

Usage

package main

import (
    "fmt"
    "time"

    "github.com/rleungx/queue"
)

func main() {
    // Create a new priority queue with a capacity of 10 and a cleanup interval of 1 second
    pq := queue.New[string](10, time.Second)
    defer pq.Close()

    // Add items to the priority queue
    pq.Push("v1", 1, 2*time.Second)
    pq.Push("v2", 2, time.Second)
    pq.Push("v3", 3, 3*time.Second)

    // Show all items
    fmt.Printf("All items: %v\n", pq.Elems())

    // Peek the highest priority item
    fmt.Printf("Highest priority item: %v\n", pq.Peek())

    // Remove the highest priority item
    fmt.Printf("Removed highest priority item: %v\n", pq.Pop())

    // Show remaining items
    fmt.Printf("Remaining items: %v\n", pq.Elems())

    // Remove a specific item
    pq.Remove("v1")
    fmt.Println("Attempted to remove item: v1")

    // Show remaining items after removal
    fmt.Printf("Remaining items after removal: %v\n", pq.Elems())

    // Get queue size
    fmt.Printf("Queue size: %d\n", pq.Size())

    // Check if queue is empty
    fmt.Printf("Is queue empty: %v\n", pq.Empty())

    // Get queue capacity
    fmt.Printf("Queue capacity: %d\n", pq.Capacity())

    // Add new items
    pq.Push("v4", 4, 500*time.Millisecond)
    pq.Push("v5", 5, 5*time.Second)

    // Get all elements as a slice
    elems := pq.Elems()
    fmt.Printf("All elements: %v\n", elems)

    // Wait for 600 milliseconds to let v4 expire
    time.Sleep(600 * time.Millisecond)
    
    // Manually call Cleanup
    pq.Cleanup()
    fmt.Println("Manually called Cleanup")

    // Show remaining items after manual cleanup
    fmt.Printf("Remaining items after manual cleanup: %v\n", pq.Elems())

    // Wait for 5 seconds to allow automatic cleanup
    time.Sleep(5 * time.Second)
    
    // Show remaining items after automatic cleanup
    fmt.Printf("Remaining items after automatic cleanup: %v\n", pq.Elems())
}

Contributing

Contributions are welcome! Please open an issue or submit a pull request.

License

This project is licensed under the Apache License 2.0. See the LICENSE file for details.

Documentation

Index

• type CleanupMetrics
• type Entry
• type PriorityQueue
  • func New[T comparable](capacity int, cleanupInterval time.Duration) *PriorityQueue[T]
  • func (pq *PriorityQueue[T]) Capacity() int
  • func (pq *PriorityQueue[T]) Cleanup()
  • func (pq *PriorityQueue[T]) Close()
  • func (pq *PriorityQueue[T]) Elems() []T
  • func (pq *PriorityQueue[T]) Empty() bool
  • func (pq *PriorityQueue[T]) GetCleanupMetrics() CleanupMetrics
  • func (pq *PriorityQueue[T]) Peek() (value T)
  • func (pq *PriorityQueue[T]) Pop() (value T)
  • func (pq *PriorityQueue[T]) Push(value T, priority int, ttl time.Duration)
  • func (pq *PriorityQueue[T]) Remove(value T)
  • func (pq *PriorityQueue[T]) Size() int

Types

type CleanupMetrics

type CleanupMetrics struct {
	AverageExpiredPerCleanup  float64
	TotalCleanupsPerformed    int
	ExpiredItemsInLastCleanup int
	CurrentCleanupBatchSize   int
	CurrentPopLimit           int
}

CleanupMetrics reports observed cleanup history and the queue's fixed cleanup/read-path policy.

type Entry

type Entry[T any] struct {
	Priority int
	Value    T
	// contains filtered or unexported fields
}

Entry is a value and its priority/expiration metadata.

type PriorityQueue

type PriorityQueue[T comparable] struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

PriorityQueue is a priority queue that supports adding, removing, and updating elements with priorities.

func New

func New[T comparable](capacity int, cleanupInterval time.Duration) *PriorityQueue[T]

New constructs a priority queue.

func (*PriorityQueue[T]) Capacity

func (pq *PriorityQueue[T]) Capacity() int

Capacity returns the capacity of the queue.

func (*PriorityQueue[T]) Cleanup

func (pq *PriorityQueue[T]) Cleanup()

Cleanup removes expired entries using the queue's fixed cleanup policy.

func (*PriorityQueue[T]) Close

func (pq *PriorityQueue[T]) Close()

Close stops the cleanup goroutine.

func (*PriorityQueue[T]) Elems

func (pq *PriorityQueue[T]) Elems() []T

Elems returns all elements in the queue sorted by priority from high to low.

func (*PriorityQueue[T]) Empty

func (pq *PriorityQueue[T]) Empty() bool

Empty returns true if the queue is empty.

func (*PriorityQueue[T]) GetCleanupMetrics

func (pq *PriorityQueue[T]) GetCleanupMetrics() CleanupMetrics

GetCleanupMetrics returns cleanup history and the current fixed policy.

func (*PriorityQueue[T]) Peek

func (pq *PriorityQueue[T]) Peek() (value T)

Peek returns the highest priority entry without removing it.

func (*PriorityQueue[T]) Pop

func (pq *PriorityQueue[T]) Pop() (value T)

Pop returns the highest priority entry and removes it from the queue.

func (*PriorityQueue[T]) Push

func (pq *PriorityQueue[T]) Push(value T, priority int, ttl time.Duration)

Push adds a value to the queue with a priority and TTL.

func (*PriorityQueue[T]) Remove

func (pq *PriorityQueue[T]) Remove(value T)

Remove removes a value from the queue.

func (*PriorityQueue[T]) Size

func (pq *PriorityQueue[T]) Size() int

Size returns the number of elements in the queue.