From acc95923da555b2bf17a5638e62e533218c5840a Mon Sep 17 00:00:00 2001 From: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com> Date: Mon, 13 Feb 2023 18:40:48 +0000 Subject: [performance] processing media and scheduled jobs improvements (#1482) * replace media workers with just runners.WorkerPool, move to state structure, use go-sched for global task scheduling * improved code comment * fix worker tryUntil function, update go-runners/go-sched * make preprocess functions package public, use these where possible to stop doubled up processing * remove separate emoji worker pool * limit calls to time.Now() during media preprocessing * use Processor{} to manage singular runtime of processing media * ensure workers get started when media manager is used * improved error setting in processing media, fix media test * port changes from processingmedia to processing emoji * finish code commenting * finish code commenting and comment-out client API + federator worker pools until concurrency worker pools replaced * linterrrrrrrrrrrrrrrr --------- Signed-off-by: kim --- vendor/codeberg.org/gruf/go-runners/service.go | 52 +++++++++++++++++++------- 1 file changed, 38 insertions(+), 14 deletions(-) (limited to 'vendor/codeberg.org/gruf/go-runners/service.go') diff --git a/vendor/codeberg.org/gruf/go-runners/service.go b/vendor/codeberg.org/gruf/go-runners/service.go index c019a10f6..8a7c0051a 100644 --- a/vendor/codeberg.org/gruf/go-runners/service.go +++ b/vendor/codeberg.org/gruf/go-runners/service.go @@ -9,7 +9,7 @@ import ( // changes and preventing multiple instances running. Also providing service state information. type Service struct { state uint32 // 0=stopped, 1=running, 2=stopping - mutex sync.Mutex // mutext protects overall state changes + mutex sync.Mutex // mutex protects overall state changes wait sync.Mutex // wait is used as a single-entity wait-group, only ever locked within 'mutex' ctx chan struct{} // ctx is the current context for running function (or nil if not running) } @@ -62,6 +62,29 @@ func (svc *Service) GoRun(fn func(context.Context)) bool { return true } +// RunWait is functionally the same as .Run(), but blocks until the first instance of .Run() returns. +func (svc *Service) RunWait(fn func(context.Context)) bool { + // Attempt to start the svc + ctx, ok := svc.doStart() + if !ok { + <-ctx // block + return false + } + + defer func() { + // unlock single wait + svc.wait.Unlock() + + // ensure stopped + _ = svc.Stop() + }() + + // Run with context. + fn(CancelCtx(ctx)) + + return true +} + // Stop will attempt to stop the service, cancelling the running function's context. Immediately // returns false if not running, and true only after Service is fully stopped. func (svc *Service) Stop() bool { @@ -108,28 +131,29 @@ func (svc *Service) doStart() (chan struct{}, bool) { // Protect startup svc.mutex.Lock() - if svc.state != 0 /* not stopped */ { - svc.mutex.Unlock() - return nil, false - } - - // state started - svc.state = 1 - if svc.ctx == nil { // this will only have been allocated // if svc.Done() was already called. svc.ctx = make(chan struct{}) } - // Start the waiter - svc.wait.Lock() - // Take our own ptr - // and unlock state ctx := svc.ctx - svc.mutex.Unlock() + if svc.state != 0 { + // State was not stopped. + svc.mutex.Unlock() + return ctx, false + } + + // Set started. + svc.state = 1 + + // Start waiter. + svc.wait.Lock() + + // Unlock and return + svc.mutex.Unlock() return ctx, true } -- cgit v1.2.3