From a483bd9e38333b153df1d4df95276cca38f99ff5 Mon Sep 17 00:00:00 2001
From: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com>
Date: Thu, 11 Apr 2024 10:45:35 +0100
Subject: [performance] massively improved ActivityPub delivery worker
 efficiency (#2812)

* add delivery worker type that pulls from queue to httpclient package

* finish up some code commenting, bodge a vendored activity library change, integrate the deliverypool changes into transportcontroller

* hook up queue deletion logic

* support deleting queued http requests by target ID

* don't index APRequest by hostname in the queue

* use gorun

* use the original context's values when wrapping msg type as delivery{}

* actually log in the AP delivery worker ...

* add uncommitted changes

* use errors.AsV2()

* use errorsv2.AsV2()

* finish adding some code comments, add bad host handling to delivery workers

* slightly tweak deliveryworkerpool API, use advanced sender multiplier

* remove PopCtx() method, let others instead rely on Wait()

* shuffle things around to move delivery stuff into transport/ subpkg

* remove dead code

* formatting

* validate request before queueing for delivery

* finish adding code comments, fix up backoff code

* finish adding more code comments

* clamp minimum no. senders to 1

* add start/stop logging to delivery worker, some slight changes

* remove double logging

* use worker ptrs

* expose the embedded log fields in httpclient.Request{}

* ensure request context values are preserved when updating ctx

* add delivery worker tests

* fix linter issues

* ensure delivery worker gets inited in testrig

* fix tests to delivering messages to check worker delivery queue

* update error type to use ptr instead of value receiver

* fix test calling Workers{}.Start() instead of testrig.StartWorkers()

* update docs for advanced-sender-multiplier

* update to the latest activity library version

* add comment about not using httptest.Server{}
---
 internal/workers/workers.go | 21 ++++++++++++++++++++-
 1 file changed, 20 insertions(+), 1 deletion(-)

(limited to 'internal/workers/workers.go')

diff --git a/internal/workers/workers.go b/internal/workers/workers.go
index 3617ce333..17728c255 100644
--- a/internal/workers/workers.go
+++ b/internal/workers/workers.go
@@ -23,14 +23,22 @@ import (
 	"runtime"
 
 	"codeberg.org/gruf/go-runners"
+	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/scheduler"
+	"github.com/superseriousbusiness/gotosocial/internal/transport/delivery"
 )
 
 type Workers struct {
 	// Main task scheduler instance.
 	Scheduler scheduler.Scheduler
 
+	// Delivery provides a worker pool that
+	// handles outgoing ActivityPub deliveries.
+	// It contains an embedded (but accessible)
+	// indexed queue of Delivery{} objects.
+	Delivery delivery.WorkerPool
+
 	// ClientAPI provides a worker pool that handles both
 	// incoming client actions, and our own side-effects.
 	ClientAPI runners.WorkerPool
@@ -65,13 +73,23 @@ type Workers struct {
 	_ nocopy
 }
 
-// Start will start all of the contained worker pools (and global scheduler).
+// Start will start all of the contained
+// worker pools (and global scheduler).
 func (w *Workers) Start() {
 	// Get currently set GOMAXPROCS.
 	maxprocs := runtime.GOMAXPROCS(0)
 
 	tryUntil("starting scheduler", 5, w.Scheduler.Start)
 
+	tryUntil("start delivery workerpool", 5, func() bool {
+		n := config.GetAdvancedSenderMultiplier()
+		if n < 1 {
+			// clamp min senders to 1.
+			return w.Delivery.Start(1)
+		}
+		return w.Delivery.Start(n * maxprocs)
+	})
+
 	tryUntil("starting client API workerpool", 5, func() bool {
 		return w.ClientAPI.Start(4*maxprocs, 400*maxprocs)
 	})
@@ -88,6 +106,7 @@ func (w *Workers) Start() {
 // Stop will stop all of the contained worker pools (and global scheduler).
 func (w *Workers) Stop() {
 	tryUntil("stopping scheduler", 5, w.Scheduler.Stop)
+	tryUntil("stopping delivery workerpool", 5, w.Delivery.Stop)
 	tryUntil("stopping client API workerpool", 5, w.ClientAPI.Stop)
 	tryUntil("stopping federator workerpool", 5, w.Federator.Stop)
 	tryUntil("stopping media workerpool", 5, w.Media.Stop)
-- 
cgit v1.2.3