From c9c0773f2c2363dcfa37e675b83ec3f0b49bd0d9 Mon Sep 17 00:00:00 2001 From: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com> Date: Fri, 26 Apr 2024 13:50:46 +0100 Subject: [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 --- testrig/util.go | 59 ++++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 42 insertions(+), 17 deletions(-) (limited to 'testrig/util.go') diff --git a/testrig/util.go b/testrig/util.go index f6f139e79..d5eaedcd5 100644 --- a/testrig/util.go +++ b/testrig/util.go @@ -27,7 +27,10 @@ import ( "os" "time" + "codeberg.org/gruf/go-byteutil" + "codeberg.org/gruf/go-kv/format" "github.com/superseriousbusiness/gotosocial/internal/filter/visibility" + "github.com/superseriousbusiness/gotosocial/internal/log" "github.com/superseriousbusiness/gotosocial/internal/messages" tlprocessor "github.com/superseriousbusiness/gotosocial/internal/processing/timeline" "github.com/superseriousbusiness/gotosocial/internal/processing/workers" @@ -39,40 +42,55 @@ import ( // Starts workers on the provided state using noop processing functions. // Useful when you *don't* want to trigger side effects in a test. func StartNoopWorkers(state *state.State) { - state.Workers.EnqueueClientAPI = func(context.Context, ...messages.FromClientAPI) {} - state.Workers.EnqueueFediAPI = func(context.Context, ...messages.FromFediAPI) {} - state.Workers.ProcessFromClientAPI = func(context.Context, messages.FromClientAPI) error { return nil } - state.Workers.ProcessFromFediAPI = func(context.Context, messages.FromFediAPI) error { return nil } + state.Workers.Client.Process = func(ctx context.Context, msg *messages.FromClientAPI) error { return nil } + state.Workers.Federator.Process = func(ctx context.Context, msg *messages.FromFediAPI) error { return nil } + state.Workers.Client.Init(messages.ClientMsgIndices()) + state.Workers.Federator.Init(messages.FederatorMsgIndices()) state.Workers.Delivery.Init(nil) + // Specifically do NOT start the workers + // as caller may require queue contents. + // (i.e. don't want workers pulling) + // _ = state.Workers.Client.Start(1) + // _ = state.Workers.Federator.Start(1) + // _ = state.Workers.Dereference.Start(1) + // _ = state.Workers.Media.Start(1) + // + // (except for the scheduler, that's fine) _ = state.Workers.Scheduler.Start() - _ = state.Workers.ClientAPI.Start(1, 10) - _ = state.Workers.Federator.Start(1, 10) - _ = state.Workers.Media.Start(1, 10) } // Starts workers on the provided state using processing functions from the given // workers processor. Useful when you *do* want to trigger side effects in a test. func StartWorkers(state *state.State, processor *workers.Processor) { - state.Workers.EnqueueClientAPI = processor.EnqueueClientAPI - state.Workers.EnqueueFediAPI = processor.EnqueueFediAPI - state.Workers.ProcessFromClientAPI = processor.ProcessFromClientAPI - state.Workers.ProcessFromFediAPI = processor.ProcessFromFediAPI + state.Workers.Client.Process = func(ctx context.Context, msg *messages.FromClientAPI) error { + log.Debugf(ctx, "Workers{}.Client{}.Process(%s)", dump(msg)) + return processor.ProcessFromClientAPI(ctx, msg) + } + + state.Workers.Federator.Process = func(ctx context.Context, msg *messages.FromFediAPI) error { + log.Debugf(ctx, "Workers{}.Federator{}.Process(%s)", dump(msg)) + return processor.ProcessFromFediAPI(ctx, msg) + } + state.Workers.Client.Init(messages.ClientMsgIndices()) + state.Workers.Federator.Init(messages.FederatorMsgIndices()) state.Workers.Delivery.Init(nil) _ = state.Workers.Scheduler.Start() - _ = state.Workers.ClientAPI.Start(1, 10) - _ = state.Workers.Federator.Start(1, 10) - _ = state.Workers.Media.Start(1, 10) + state.Workers.Client.Start(1) + state.Workers.Federator.Start(1) + state.Workers.Dereference.Start(1) + state.Workers.Media.Start(1) } func StopWorkers(state *state.State) { _ = state.Workers.Scheduler.Stop() - _ = state.Workers.ClientAPI.Stop() - _ = state.Workers.Federator.Stop() - _ = state.Workers.Media.Stop() + state.Workers.Client.Stop() + state.Workers.Federator.Stop() + state.Workers.Dereference.Stop() + state.Workers.Media.Stop() } func StartTimelines(state *state.State, filter *visibility.Filter, converter *typeutils.Converter) { @@ -241,3 +259,10 @@ func WaitFor(condition func() bool) bool { } } } + +// dump returns debug output of 'v'. +func dump(v any) string { + var buf byteutil.Buffer + format.Append(&buf, v) + return buf.String() +} -- cgit v1.2.3