mirror of
1
Fork 0

[bugfix] function queue memory pools limitlessly grow (#2882)

* updates the simple queue memory pool to actually self-clean + limit growth

* update memory pool cleaning frequency
This commit is contained in:
kim 2024-05-01 12:30:43 +01:00 committed by GitHub
parent eb61c783ed
commit 2300d5e73b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 75 additions and 15 deletions

View File

@ -24,38 +24,67 @@ import (
"codeberg.org/gruf/go-list" "codeberg.org/gruf/go-list"
) )
// frequency of GC cycles
// per no. unlocks. i.e.
// every 'gcfreq' unlocks.
const gcfreq = 1024
// SimpleQueue provides a simple concurrency safe // SimpleQueue provides a simple concurrency safe
// queue using generics and a memory pool of list // queue using generics and a memory pool of list
// elements to reduce overall memory usage. // elements to reduce overall memory usage.
type SimpleQueue[T any] struct { type SimpleQueue[T any] struct {
l list.List[T] l list.List[T]
p []*list.Elem[T] p elemPool[T]
w chan struct{} w chan struct{}
m sync.Mutex m sync.Mutex
n uint32 // pop counter (safely wraps around)
} }
// Push will push given value to the queue. // Push will push given value to the queue.
func (q *SimpleQueue[T]) Push(value T) { func (q *SimpleQueue[T]) Push(value T) {
q.m.Lock() q.m.Lock()
elem := q.alloc()
// Wrap in element.
elem := q.p.alloc()
elem.Value = value elem.Value = value
// Push new elem to queue.
q.l.PushElemFront(elem) q.l.PushElemFront(elem)
if q.w != nil { if q.w != nil {
// Notify any goroutines
// blocking on q.Wait(),
// or on PopCtx(...).
close(q.w) close(q.w)
q.w = nil q.w = nil
} }
q.m.Unlock() q.m.Unlock()
} }
// Pop will attempt to pop value from the queue. // Pop will attempt to pop value from the queue.
func (q *SimpleQueue[T]) Pop() (value T, ok bool) { func (q *SimpleQueue[T]) Pop() (value T, ok bool) {
q.m.Lock() q.m.Lock()
// Check for a tail (i.e. not empty).
if ok = (q.l.Tail != nil); ok { if ok = (q.l.Tail != nil); ok {
// Extract value.
tail := q.l.Tail tail := q.l.Tail
value = tail.Value value = tail.Value
// Remove tail.
q.l.Remove(tail) q.l.Remove(tail)
q.free(tail) q.p.free(tail)
// Every 'gcfreq' pops perform
// a garbage collection to keep
// us squeaky clean :]
if q.n++; q.n%gcfreq == 0 {
q.p.GC()
}
} }
q.m.Unlock() q.m.Unlock()
return return
} }
@ -105,7 +134,14 @@ func (q *SimpleQueue[T]) PopCtx(ctx context.Context) (value T, ok bool) {
// Remove element. // Remove element.
q.l.Remove(elem) q.l.Remove(elem)
q.free(elem) q.p.free(elem)
// Every 'gcfreq' pops perform
// a garbage collection to keep
// us squeaky clean :]
if q.n++; q.n%gcfreq == 0 {
q.p.GC()
}
// Done with lock. // Done with lock.
q.m.Unlock() q.m.Unlock()
@ -121,21 +157,45 @@ func (q *SimpleQueue[T]) Len() int {
return l return l
} }
// alloc will allocate new list element (relying on memory pool). // elemPool is a very simple
func (q *SimpleQueue[T]) alloc() *list.Elem[T] { // list.Elem[T] memory pool.
if len(q.p) > 0 { type elemPool[T any] struct {
elem := q.p[len(q.p)-1] current []*list.Elem[T]
q.p = q.p[:len(q.p)-1] victim []*list.Elem[T]
return elem
}
return new(list.Elem[T])
} }
// free will free list element and release to pool. func (p *elemPool[T]) alloc() *list.Elem[T] {
func (q *SimpleQueue[T]) free(elem *list.Elem[T]) { // First try the current queue
if l := len(p.current) - 1; l >= 0 {
mu := p.current[l]
p.current = p.current[:l]
return mu
}
// Next try the victim queue.
if l := len(p.victim) - 1; l >= 0 {
mu := p.victim[l]
p.victim = p.victim[:l]
return mu
}
// Lastly, alloc new.
mu := new(list.Elem[T])
return mu
}
// free will release given element to pool.
func (p *elemPool[T]) free(elem *list.Elem[T]) {
var zero T var zero T
elem.Next = nil elem.Next = nil
elem.Prev = nil elem.Prev = nil
elem.Value = zero elem.Value = zero
q.p = append(q.p, elem) p.current = append(p.current, elem)
}
// GC will clear out unused entries from the elemPool.
func (p *elemPool[T]) GC() {
current := p.current
p.current = nil
p.victim = current
} }