diff options
author | 2024-04-26 13:50:46 +0100 | |
---|---|---|
committer | 2024-04-26 13:50:46 +0100 | |
commit | c9c0773f2c2363dcfa37e675b83ec3f0b49bd0d9 (patch) | |
tree | dbd3409070765d5ca81448a574ccd32b4da1ffe6 /internal/queue/simple.go | |
parent | [chore] update Docker container to use new go swagger hash (#2872) (diff) | |
download | gotosocial-c9c0773f2c2363dcfa37e675b83ec3f0b49bd0d9.tar.xz |
[performance] update remaining worker pools to use queues (#2865)
* start replacing client + federator + media workers with new worker + queue types
* refactor federatingDB.Delete(), drop queued messages when deleting account / status
* move all queue purging to the processor workers
* undo toolchain updates
* code comments, ensure dereferencer worker pool gets started
* update gruf libraries in readme
* start the job scheduler separately to the worker pools
* reshuffle ordering or server.go + remove duplicate worker start / stop
* update go-list version
* fix vendoring
* move queue invalidation to before wipeing / deletion, to ensure queued work not dropped
* add logging to worker processing functions in testrig, don't start workers in unexpected places
* update go-structr to add (+then rely on) QueueCtx{} type
* ensure more worker pools get started properly in tests
* fix remaining broken tests relying on worker queue logic
* fix account test suite queue popping logic, ensure noop workers do not pull from queue
* move back accidentally shuffled account deletion order
* ensure error (non nil!!) gets passed in refactored federatingDB{}.Delete()
* silently drop deletes from accounts not permitted to
* don't warn log on forwarded deletes
* make if else clauses easier to parse
* use getFederatorMsg()
* improved code comment
* improved code comment re: requesting account delete checks
* remove boolean result from worker start / stop since false = already running or already stopped
* remove optional passed-in http.client
* remove worker starting from the admin CLI commands (we don't need to handle side-effects)
* update prune cli to start scheduler but not all of the workers
* fix rebase issues
* remove redundant return statements
* i'm sorry sir linter
Diffstat (limited to 'internal/queue/simple.go')
-rw-r--r-- | internal/queue/simple.go | 141 |
1 files changed, 141 insertions, 0 deletions
diff --git a/internal/queue/simple.go b/internal/queue/simple.go new file mode 100644 index 000000000..9307183f2 --- /dev/null +++ b/internal/queue/simple.go @@ -0,0 +1,141 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + +package queue + +import ( + "context" + "sync" + + "codeberg.org/gruf/go-list" +) + +// SimpleQueue provides a simple concurrency safe +// queue using generics and a memory pool of list +// elements to reduce overall memory usage. +type SimpleQueue[T any] struct { + l list.List[T] + p []*list.Elem[T] + w chan struct{} + m sync.Mutex +} + +// Push will push given value to the queue. +func (q *SimpleQueue[T]) Push(value T) { + q.m.Lock() + elem := q.alloc() + elem.Value = value + q.l.PushElemFront(elem) + if q.w != nil { + close(q.w) + q.w = nil + } + q.m.Unlock() +} + +// Pop will attempt to pop value from the queue. +func (q *SimpleQueue[T]) Pop() (value T, ok bool) { + q.m.Lock() + if ok = (q.l.Tail != nil); ok { + tail := q.l.Tail + value = tail.Value + q.l.Remove(tail) + q.free(tail) + } + q.m.Unlock() + return +} + +// PopCtx will attempt to pop value from queue, else blocking on context. +func (q *SimpleQueue[T]) PopCtx(ctx context.Context) (value T, ok bool) { + + // Acquire lock. + q.m.Lock() + + var elem *list.Elem[T] + + for { + // Get next elem. + elem = q.l.Tail + if ok = (elem != nil); ok { + break + } + + if q.w == nil { + // Create new wait channel. + q.w = make(chan struct{}) + } + + // Get current + // ch pointer. + ch := q.w + + // Done with lock. + q.m.Unlock() + + select { + // Context canceled. + case <-ctx.Done(): + return + + // Pushed! + case <-ch: + } + + // Relock queue. + q.m.Lock() + } + + // Extract value. + value = elem.Value + + // Remove element. + q.l.Remove(elem) + q.free(elem) + + // Done with lock. + q.m.Unlock() + + return +} + +// Len returns the current length of the queue. +func (q *SimpleQueue[T]) Len() int { + q.m.Lock() + l := q.l.Len() + q.m.Unlock() + return l +} + +// alloc will allocate new list element (relying on memory pool). +func (q *SimpleQueue[T]) alloc() *list.Elem[T] { + if len(q.p) > 0 { + elem := q.p[len(q.p)-1] + q.p = q.p[:len(q.p)-1] + return elem + } + return new(list.Elem[T]) +} + +// free will free list element and release to pool. +func (q *SimpleQueue[T]) free(elem *list.Elem[T]) { + var zero T + elem.Next = nil + elem.Prev = nil + elem.Value = zero + q.p = append(q.p, elem) +} |