mirrors/gotosocial
1
0
Fork 0
mirror of https://github.com/superseriousbusiness/gotosocial.git synced 2025-10-29 04:22:24 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

[feature] log worker startup counts (#2958)

Browse source 
* log number of each worker kinds started, and log when stopped

* remove worker debug logging

* whoops, fix the count of media workers
This commit is contained in:
kim 2024-06-03 21:55:50 +00:00 • committed by GitHub
commit f17dd62ff5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 34 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

3
internal/transport/delivery/delivery.go
View file

@ -26,7 +26,6 @@ import (
"codeberg.org/gruf/go-structr"
"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
"github.com/superseriousbusiness/gotosocial/internal/httpclient"
"github.com/superseriousbusiness/gotosocial/internal/log"
"github.com/superseriousbusiness/gotosocial/internal/queue"
"github.com/superseriousbusiness/gotosocial/internal/util"
)
@ -181,8 +180,6 @@ func (w *Worker) run(ctx context.Context) {
if w.Client == nil || w.Queue == nil {
panic("not yet initialized")
}
log.Debugf(ctx, "%p: starting worker", w)
defer log.Debugf(ctx, "%p: stopped worker", w)
util.Must(func() { w.process(ctx) })
}

2
internal/workers/worker_msg.go
View file

@ -127,8 +127,6 @@ func (w *MsgWorker[T]) run(ctx context.Context) {
if w.Process == nil || w.Queue == nil {
panic("not yet initialized")
}
log.Debugf(ctx, "%p: starting worker", w)
defer log.Debugf(ctx, "%p: stopped worker", w)
util.Must(func() { w.process(ctx) })
}

39
internal/workers/workers.go
View file

@ -21,6 +21,7 @@ import (
"runtime"
"github.com/superseriousbusiness/gotosocial/internal/config"
"github.com/superseriousbusiness/gotosocial/internal/log"
"github.com/superseriousbusiness/gotosocial/internal/messages"
"github.com/superseriousbusiness/gotosocial/internal/scheduler"
"github.com/superseriousbusiness/gotosocial/internal/transport/delivery"
@ -59,26 +60,54 @@ type Workers struct {
// StartScheduler starts the job scheduler.
func (w *Workers) StartScheduler() {
_ = w.Scheduler.Start() // false = already running
log.Info(nil, "started scheduler")
}
// Start will start contained worker pools.
func (w *Workers) Start() {
var n int
maxprocs := runtime.GOMAXPROCS(0)
w.Delivery.Start(deliveryWorkers(maxprocs))
w.Client.Start(4 * maxprocs)
w.Federator.Start(4 * maxprocs)
w.Dereference.Start(4 * maxprocs)
w.Media.Start(8 * maxprocs)
n = deliveryWorkers(maxprocs)
w.Delivery.Start(n)
log.Infof(nil, "started %d delivery workers", n)
n = 4 * maxprocs
w.Client.Start(n)
log.Infof(nil, "started %d client workers", n)
n = 4 * maxprocs
w.Federator.Start(n)
log.Infof(nil, "started %d federator workers", n)
n = 4 * maxprocs
w.Dereference.Start(n)
log.Infof(nil, "started %d dereference workers", n)
n = 8 * maxprocs
w.Media.Start(n)
log.Infof(nil, "started %d media workers", n)
}
// Stop will stop all of the contained worker pools (and global scheduler).
func (w *Workers) Stop() {
_ = w.Scheduler.Stop() // false = not running
w.Delivery.Stop()
log.Info(nil, "stopped delivery workers")
w.Client.Stop()
log.Info(nil, "stopped client workers")
w.Federator.Stop()
log.Info(nil, "stopped federator workers")
w.Dereference.Stop()
log.Info(nil, "stopped dereference workers")
w.Media.Stop()
log.Info(nil, "stopped media workers")
}
// nocopy when embedded will signal linter to