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/struct.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/struct.go')
-rw-r--r-- | internal/queue/struct.go | 78 |
1 files changed, 78 insertions, 0 deletions
diff --git a/internal/queue/struct.go b/internal/queue/struct.go new file mode 100644 index 000000000..27139daec --- /dev/null +++ b/internal/queue/struct.go @@ -0,0 +1,78 @@ +// 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" + + "codeberg.org/gruf/go-structr" +) + +// StructQueue wraps a structr.Queue{} to +// provide simple index caching by name. +type StructQueue[StructType any] struct { + queue structr.QueueCtx[StructType] + index map[string]*structr.Index +} + +// 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.QueueCtx[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) { + values := q.queue.PopFrontN(1) + if ok = (len(values) > 0); !ok { + return + } + value = values[0] + return +} + +// PopCtx: see structr.QueueCtx{}.PopFront(). +func (q *StructQueue[T]) PopCtx(ctx context.Context) (value T, ok bool) { + return q.queue.PopFront(ctx) +} + +// Push: see structr.Queue.PushBack(). +func (q *StructQueue[T]) Push(values ...T) { + q.queue.PushBack(values...) +} + +// 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{} { + return q.queue.Wait() +} |