Default max_concurrent_processing to half the CPUs pixa sees (closes #242)
check / check (push) Canceled after 0s
check / check (push) Canceled after 0s
The default max_concurrent_processing was one image per CPU pixa sees, so in a container with no CPU limit on a 48-CPU host pixa processed 48 images at once. It is now half the CPUs, rounded down and at least 1, defined once in internal/config; PIXA_MAX_CONCURRENT_PROCESSING stays the override, and a value that is not a whole number of at least 1 stops startup. "Running under upaas" gives the worst-case memory, 4.2 GiB per image (the largest peak measured over every source format and kind at 8192x8192) plus the upstream buffers, and the temporary file libvips keeps under /tmp. The two existing tests expect the new default, as the owner allowed. Model: opus-5-5
This commit was merged in pull request #261.
This commit is contained in:
@@ -7,11 +7,11 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"runtime"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/davidbyttow/govips/v2/vips"
|
||||
"sneak.berlin/go/pixa/internal/config"
|
||||
)
|
||||
|
||||
// vipsOnce ensures vips is initialized exactly once.
|
||||
@@ -20,11 +20,10 @@ import (
|
||||
var vipsOnce sync.Once
|
||||
|
||||
// initVips initializes libvips with quiet logging, one worker thread per
|
||||
// image and no operation cache. Process already works on one image per CPU
|
||||
// by default, so more threads per image would only compete for the CPUs.
|
||||
// Each request decodes different source bytes, so the operation cache
|
||||
// would rarely be hit and would hold memory outside MaxConcurrentProcessing;
|
||||
// repeated requests are served from pixa's disk cache instead.
|
||||
// image and no operation cache. Each request decodes different source
|
||||
// bytes, so the operation cache would rarely be hit and would hold memory
|
||||
// outside MaxConcurrentProcessing; repeated requests are served from
|
||||
// pixa's disk cache instead.
|
||||
func initVips() {
|
||||
vipsOnce.Do(func() {
|
||||
vips.LoggingSettings(nil, vips.LogLevelError)
|
||||
@@ -179,7 +178,7 @@ type Params struct {
|
||||
// If <= 0, DefaultMaxInputBytes is used.
|
||||
MaxInputBytes int64
|
||||
// MaxConcurrentProcessing is the most images processed at once.
|
||||
// If <= 0, the number of CPUs Go uses (runtime.GOMAXPROCS(0)) is used.
|
||||
// If <= 0, config.DefaultMaxConcurrentProcessing() is used.
|
||||
MaxConcurrentProcessing int
|
||||
}
|
||||
|
||||
@@ -195,7 +194,7 @@ func New(params Params) *ImageProcessor {
|
||||
|
||||
maxConcurrentProcessing := params.MaxConcurrentProcessing
|
||||
if maxConcurrentProcessing <= 0 {
|
||||
maxConcurrentProcessing = runtime.GOMAXPROCS(0)
|
||||
maxConcurrentProcessing = config.DefaultMaxConcurrentProcessing()
|
||||
}
|
||||
|
||||
return &ImageProcessor{
|
||||
|
||||
@@ -115,14 +115,14 @@ func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
|
||||
func TestNewDefaultsMaxConcurrentProcessingToHalfTheCPUs(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, limit := range []int{0, -1} {
|
||||
proc := New(Params{MaxConcurrentProcessing: limit})
|
||||
if got := cap(proc.processingSemaphore); got != runtime.GOMAXPROCS(0) {
|
||||
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, one per CPU",
|
||||
limit, got, runtime.GOMAXPROCS(0))
|
||||
if got := cap(proc.processingSemaphore); got != max(runtime.GOMAXPROCS(0)/2, 1) {
|
||||
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, half the CPUs",
|
||||
limit, got, max(runtime.GOMAXPROCS(0)/2, 1))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -132,6 +132,30 @@ func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestNewDefaultIsHalfTheCPUsAndAtLeastOne checks that New, given no
|
||||
// MaxConcurrentProcessing, makes one slot for every two CPUs Go uses,
|
||||
// rounded down, and one slot when Go uses one CPU.
|
||||
//
|
||||
//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
|
||||
func TestNewDefaultIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
|
||||
cpusBefore := runtime.GOMAXPROCS(0)
|
||||
|
||||
t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
|
||||
|
||||
for _, tc := range []struct{ cpus, want int }{
|
||||
{1, 1}, {2, 1}, {3, 1}, {8, 4},
|
||||
} {
|
||||
runtime.GOMAXPROCS(tc.cpus)
|
||||
|
||||
got := cap(New(Params{}).processingSemaphore)
|
||||
t.Logf("GOMAXPROCS %d: %d slots", tc.cpus, got)
|
||||
|
||||
if got != tc.want {
|
||||
t.Errorf("GOMAXPROCS %d: %d slots, want %d", tc.cpus, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls
|
||||
// than MaxConcurrentProcessing allows and holds the first ones inside
|
||||
// Process until the test lets them go. No more than the limit may be
|
||||
|
||||
Reference in New Issue
Block a user