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/wrappers.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/wrappers.go')
-rw-r--r-- | internal/queue/wrappers.go | 96 |
1 files changed, 0 insertions, 96 deletions
diff --git a/internal/queue/wrappers.go b/internal/queue/wrappers.go deleted file mode 100644 index e07984f84..000000000 --- a/internal/queue/wrappers.go +++ /dev/null @@ -1,96 +0,0 @@ -// 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 ( - "sync/atomic" - - "codeberg.org/gruf/go-structr" -) - -// StructQueue wraps a structr.Queue{} to -// provide simple index caching by name. -type StructQueue[StructType any] struct { - queue structr.Queue[StructType] - index map[string]*structr.Index - wait atomic.Pointer[chan struct{}] -} - -// Init initializes queue with structr.QueueConfig{}. -func (q *StructQueue[T]) Init(config structr.QueueConfig[T]) { - q.index = make(map[string]*structr.Index, len(config.Indices)) - q.queue = structr.Queue[T]{} - q.queue.Init(config) - for _, cfg := range config.Indices { - q.index[cfg.Fields] = q.queue.Index(cfg.Fields) - } -} - -// Pop: see structr.Queue{}.PopFront(). -func (q *StructQueue[T]) Pop() (value T, ok bool) { - return q.queue.PopFront() -} - -// Push wraps structr.Queue{}.PushBack() to awaken those blocking on <-.Wait(). -func (q *StructQueue[T]) Push(values ...T) { - q.queue.PushBack(values...) - q.broadcast() -} - -// Delete pops (and drops!) all queued entries under index with key. -func (q *StructQueue[T]) Delete(index string, key ...any) { - i := q.index[index] - _ = q.queue.Pop(i, i.Key(key...)) -} - -// Len: see structr.Queue{}.Len(). -func (q *StructQueue[T]) Len() int { - return q.queue.Len() -} - -// Wait returns current wait channel, which may be -// blocked on to awaken when new value pushed to queue. -func (q *StructQueue[T]) Wait() <-chan struct{} { - var ch chan struct{} - - for { - // Get channel ptr. - ptr := q.wait.Load() - if ptr != nil { - return *ptr - } - - if ch == nil { - // Allocate new channel. - ch = make(chan struct{}) - } - - // Try set the new wait channel ptr. - if q.wait.CompareAndSwap(ptr, &ch) { - return ch - } - } -} - -// broadcast safely closes wait channel if -// currently set, releasing waiting goroutines. -func (q *StructQueue[T]) broadcast() { - if ptr := q.wait.Swap(nil); ptr != nil { - close(*ptr) - } -} |