perf(audiobooks): parallelize enricher sweep
Was: serial loop over claimed batch of 50. Now: N workers fan out over the batch (default 4, configurable via SILO_AUDIOBOOK_ENRICH_WORKERS). Adds a runBatch seam used by Run so tests can stub the per-item enrich function and assert worker fan-out without hitting providers.
This commit is contained in:
@@ -20,6 +20,8 @@ import (
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
@@ -148,23 +150,12 @@ func (e *Enricher) Run(ctx context.Context) (int, error) {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
slog.Info("audiobook enrichment: sweep started", "count", len(items))
|
||||
slog.Info("audiobook enrichment: sweep started",
|
||||
"count", len(items),
|
||||
"workers", e.workers,
|
||||
)
|
||||
|
||||
enriched := 0
|
||||
for _, item := range items {
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
if err := e.enrichItem(ctx, item); err != nil {
|
||||
slog.Warn("audiobook enrichment: item failed",
|
||||
"content_id", item.ContentID,
|
||||
"title", item.Title,
|
||||
"error", err,
|
||||
)
|
||||
continue
|
||||
}
|
||||
enriched++
|
||||
}
|
||||
enriched := e.runBatch(ctx, items, e.enrichItem)
|
||||
|
||||
slog.Info("audiobook enrichment: sweep complete",
|
||||
"attempted", len(items),
|
||||
@@ -173,6 +164,54 @@ func (e *Enricher) Run(ctx context.Context) (int, error) {
|
||||
return enriched, nil
|
||||
}
|
||||
|
||||
// runBatch processes items in parallel using e.workers goroutines. The
|
||||
// enrichFn parameter exists for testing: production calls pass e.enrichItem.
|
||||
// Returns the count of items where enrichFn returned nil.
|
||||
func (e *Enricher) runBatch(ctx context.Context, items []enrichmentItemRow, enrichFn func(context.Context, enrichmentItemRow) error) int {
|
||||
workers := e.workers
|
||||
if workers <= 0 {
|
||||
workers = 1
|
||||
}
|
||||
if workers > len(items) {
|
||||
workers = len(items)
|
||||
}
|
||||
|
||||
ch := make(chan enrichmentItemRow, workers)
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
enriched int64
|
||||
)
|
||||
for i := 0; i < workers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for item := range ch {
|
||||
if ctx.Err() != nil {
|
||||
continue // drain
|
||||
}
|
||||
if err := enrichFn(ctx, item); err != nil {
|
||||
slog.Warn("audiobook enrichment: item failed",
|
||||
"content_id", item.ContentID,
|
||||
"title", item.Title,
|
||||
"error", err,
|
||||
)
|
||||
continue
|
||||
}
|
||||
atomic.AddInt64(&enriched, 1)
|
||||
}
|
||||
}()
|
||||
}
|
||||
for _, item := range items {
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
ch <- item
|
||||
}
|
||||
close(ch)
|
||||
wg.Wait()
|
||||
return int(enriched)
|
||||
}
|
||||
|
||||
// claimBatch returns up to batchSize audiobook items that need enrichment.
|
||||
// "Needs enrichment" means poster_path IS NULL or empty AND last_refreshed IS NULL.
|
||||
// We skip items where last_refreshed IS NOT NULL — those have already had at
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package audiobooks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestEnricherRunFansOut verifies that runBatch processes a claimed batch with
|
||||
// multiple workers in flight at once. The test installs a fake enrichItem path
|
||||
// that blocks until N concurrent calls are observed; if runBatch is still
|
||||
// serial, the test will time out and fail the max-in-flight assertion.
|
||||
func TestEnricherRunFansOut(t *testing.T) {
|
||||
const wantWorkers = 4
|
||||
const itemCount = 16
|
||||
|
||||
items := make([]enrichmentItemRow, itemCount)
|
||||
for i := range items {
|
||||
items[i] = enrichmentItemRow{ContentID: "test", Title: "t"}
|
||||
}
|
||||
|
||||
var inFlight int32
|
||||
var maxInFlight int32
|
||||
var signaled int32
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(wantWorkers)
|
||||
gate := make(chan struct{})
|
||||
|
||||
enrich := func(ctx context.Context, item enrichmentItemRow) error {
|
||||
cur := atomic.AddInt32(&inFlight, 1)
|
||||
defer atomic.AddInt32(&inFlight, -1)
|
||||
for {
|
||||
prev := atomic.LoadInt32(&maxInFlight)
|
||||
if cur <= prev || atomic.CompareAndSwapInt32(&maxInFlight, prev, cur) {
|
||||
break
|
||||
}
|
||||
}
|
||||
// First wantWorkers callers release the gate; later ones don't block.
|
||||
if atomic.AddInt32(&signaled, 1) <= wantWorkers {
|
||||
wg.Done()
|
||||
}
|
||||
select {
|
||||
case <-gate:
|
||||
case <-time.After(2 * time.Second):
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(gate)
|
||||
}()
|
||||
|
||||
e := &Enricher{workers: wantWorkers, batchSize: itemCount}
|
||||
e.runBatch(context.Background(), items, enrich)
|
||||
|
||||
if got := atomic.LoadInt32(&maxInFlight); got < wantWorkers {
|
||||
t.Errorf("max in-flight = %d, want >= %d", got, wantWorkers)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user