Files
pixa/internal/config/concurrency_limits_internal_test.go
T
sneak 1c8f3376ad Default max_concurrent_processing to half the CPUs pixa sees (closes #242)
The default was one image per CPU Go uses, and a container with no CPU
limit sees every CPU of the host. It is now half of runtime.GOMAXPROCS(0),
rounded down and at least 1, from imageprocessor.DefaultMaxConcurrentProcessing,
which both the config and imageprocessor.New use. The two existing tests of
the default expect the new one, as the owner allowed on the issue.

"Running under upaas" gives the worst-case memory, with a per-image peak of
about 1.6 GiB measured on pixad for an 8192x8192 source in each output
format, and tells the operator to set PIXA_MAX_CONCURRENT_PROCESSING and
PIXA_UPSTREAM_CONNECTIONS to fit the container's memory.

Model: opus-5-5
2026-10-15 17:19:47 +00:00

210 lines
6.4 KiB
Go

package config
import (
"fmt"
"runtime"
"testing"
)
// The variables that set the two concurrency limits.
const (
testMaxConcurrentProcessingVar = "PIXA_MAX_CONCURRENT_PROCESSING"
testUpstreamConnectionsVar = "PIXA_UPSTREAM_CONNECTIONS"
)
// TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
// max_concurrent_processing is half the CPUs Go uses, and at least 1, and
// an omitted upstream_connections is 64.
func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
t.Parallel()
c, err := configFromYAML(t, signingKeyLine)
if err != nil {
t.Fatalf("minimal config should be valid, got error: %v", err)
}
if c.MaxConcurrentProcessing != max(runtime.GOMAXPROCS(0)/2, 1) {
t.Errorf("MaxConcurrentProcessing = %d, want %d, half the CPUs",
c.MaxConcurrentProcessing, max(runtime.GOMAXPROCS(0)/2, 1))
}
if c.UpstreamConnections != 64 {
t.Errorf("UpstreamConnections = %d, want 64", c.UpstreamConnections)
}
}
// TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne checks
// that an omitted max_concurrent_processing is half the CPUs Go uses,
// rounded down, and 1 when Go uses one CPU.
//
//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
func TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne(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)
c, err := configFromYAML(t, signingKeyLine)
if err != nil {
t.Fatalf("minimal config should be valid, got error: %v", err)
}
t.Logf("GOMAXPROCS %d: MaxConcurrentProcessing %d",
tc.cpus, c.MaxConcurrentProcessing)
if c.MaxConcurrentProcessing != tc.want {
t.Errorf("GOMAXPROCS %d: MaxConcurrentProcessing = %d, want %d",
tc.cpus, c.MaxConcurrentProcessing, tc.want)
}
}
}
// TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
// two limits are used as given.
func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
t.Parallel()
c, err := configFromYAML(t, signingKeyLine+
"max_concurrent_processing: 3\nupstream_connections: 10\n")
if err != nil {
t.Fatalf("valid config should load, got error: %v", err)
}
if c.MaxConcurrentProcessing != 3 {
t.Errorf("MaxConcurrentProcessing = %d, want 3", c.MaxConcurrentProcessing)
}
if c.UpstreamConnections != 10 {
t.Errorf("UpstreamConnections = %d, want 10", c.UpstreamConnections)
}
}
// TestInvalidConcurrencyLimitAbortsStartup checks that a limit that is
// not a whole number of at least 1, or is null, aborts startup naming the
// key and the value, and the variable too where the value could have come
// from it.
func TestInvalidConcurrencyLimitAbortsStartup(t *testing.T) {
t.Parallel()
processing := keyMaxConcurrentProcessing
connections := keyUpstreamConnections
runAbortCases(t, []abortCase{
{
name: "max_concurrent_processing zero",
yaml: signingKeyLine + processing + ": 0\n",
wantErrSubstrings: []string{
processing, testMaxConcurrentProcessingVar, "value 0",
},
},
{
name: "max_concurrent_processing negative",
yaml: signingKeyLine + processing + ": -2\n",
wantErrSubstrings: []string{
processing, testMaxConcurrentProcessingVar, "value -2",
},
},
{
name: "max_concurrent_processing not a number",
yaml: signingKeyLine + processing + ": lots\n",
wantErrSubstrings: []string{
processing, testMaxConcurrentProcessingVar, "lots",
},
},
{
name: "max_concurrent_processing fractional",
yaml: signingKeyLine + processing + ": 1.5\n",
wantErrSubstrings: []string{processing, "1.5"},
},
{
name: "max_concurrent_processing null",
yaml: signingKeyLine + processing + ": null\n",
wantErrSubstrings: []string{processing, nullValueText},
},
{
name: "upstream_connections zero",
yaml: signingKeyLine + connections + ": 0\n",
wantErrSubstrings: []string{
connections, testUpstreamConnectionsVar, "value 0",
},
},
{
name: "upstream_connections negative",
yaml: signingKeyLine + connections + ": -5\n",
wantErrSubstrings: []string{
connections, testUpstreamConnectionsVar, "value -5",
},
},
{
name: "upstream_connections not a number",
yaml: signingKeyLine + connections + ": many\n",
wantErrSubstrings: []string{
connections, testUpstreamConnectionsVar, "many",
},
},
{
name: "upstream_connections null",
yaml: signingKeyLine + connections + ": null\n",
wantErrSubstrings: []string{connections, nullValueText},
},
})
}
// TestConcurrencyLimitsFromEnvironment checks that the two variables set
// the limits over the config file, and that an invalid value in either
// aborts startup naming the variable and the value.
func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
t.Setenv(testMaxConcurrentProcessingVar, "3")
t.Setenv(testUpstreamConnectionsVar, "10")
c, err := configFromYAML(t, signingKeyLine+
"max_concurrent_processing: 5\nupstream_connections: 50\n")
if err != nil {
t.Fatalf("limits from the environment should load: %v", err)
}
if c.MaxConcurrentProcessing != 3 || c.UpstreamConnections != 10 {
t.Errorf("limits = %d and %d, want 3 and 10 from the environment",
c.MaxConcurrentProcessing, c.UpstreamConnections)
}
cases := []struct {
variable string
value string
}{
{testMaxConcurrentProcessingVar, "lots"},
{testMaxConcurrentProcessingVar, "0"},
{testUpstreamConnectionsVar, "-1"},
{testUpstreamConnectionsVar, "ten"},
}
for _, tc := range cases {
t.Run(tc.variable+"="+tc.value, func(t *testing.T) {
t.Setenv(tc.variable, tc.value)
_, err := configFromYAML(t, signingKeyLine)
wantStartupError(t, err, tc.variable, tc.value)
})
}
}
// TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup checks
// that a PIXA_MAX_CONCURRENT_PROCESSING that is not a whole number of at
// least 1, an empty one included, aborts startup naming the variable and
// the key instead of taking the default.
func TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup(t *testing.T) {
for _, value := range []string{"", " ", "0.5", "half", "-1", "4 images"} {
t.Run(fmt.Sprintf("value %q", value), func(t *testing.T) {
t.Setenv(testMaxConcurrentProcessingVar, value)
_, err := configFromYAML(t, signingKeyLine)
wantStartupError(t, err,
testMaxConcurrentProcessingVar, keyMaxConcurrentProcessing)
})
}
}