[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 <grufwub@gmail.com>

1 files changed, 38 insertions, 14 deletions

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
 }