Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bbc354fa99 | ||
|
|
617f9b4a7b | ||
|
|
49799bc7b7 | ||
|
|
6b184ee2ff | ||
|
|
ed5d1ea2d1 | ||
|
|
137bc35faf |
@@ -126,6 +126,27 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
|
||||
default 75% of (free space + what the cache holds)
|
||||
- the rest are in the table under Configuration below
|
||||
- **Memory:** pixa can use up to `max_concurrent_processing` times 1.6 GiB for
|
||||
the images it processes at once, plus `upstream_connections` times
|
||||
`upstream_max_response_size` for the source images it has fetched that wait
|
||||
for processing. 1.6 GiB is the most pixad's resident memory was measured to
|
||||
grow while it processed one 8192x8192 source, the largest it accepts, into
|
||||
each output format in turn, one at a time; it grew most for a 40 MB JPEG of
|
||||
random pixels saved as AVIF. At the defaults that is half the CPUs pixa sees
|
||||
times 1.6 GiB, plus 64 times 50 MiB: on a host where pixa sees 48 CPUs, 24
|
||||
times 1.6 GiB plus 3.1 GiB, about 42 GiB. Set `PIXA_MAX_CONCURRENT_PROCESSING`
|
||||
and `PIXA_UPSTREAM_CONNECTIONS` so that this fits the memory the container may
|
||||
use.
|
||||
- **Temporary files:** while pixa processes a large source, libvips writes its
|
||||
decoded pixels to a file in `/tmp` and deletes the file when the image is
|
||||
done. For an 8192x8192 source the file was measured on pixad at 192 MiB for an
|
||||
8-bit RGB JPEG, 512 MiB for a 16-bit RGBA PNG or JPEG XL, and 1 GiB for a JPEG
|
||||
XL of 32-bit floating-point RGBA, the largest measured: up to 1 GiB per image
|
||||
processed at once, 24 GiB at the defaults on a host where pixa sees 48 CPUs.
|
||||
This `/tmp` is in the container's own filesystem, not on the volume at
|
||||
`/var/lib/pixa`, and `cache_max_bytes` does not count it. The 1.6 GiB above
|
||||
does not include these files; where `/tmp` is a tmpfs, they are memory on top
|
||||
of it.
|
||||
- **Health check:** the image's `HEALTHCHECK` requests
|
||||
`/.well-known/healthcheck.json`. upaas reads the container's health 60 seconds
|
||||
after a deploy and marks the deploy failed unless it is `healthy`. The probe
|
||||
@@ -524,7 +545,7 @@ tell. With no file, pixa uses the environment and the defaults.
|
||||
| `PIXA_ALLOW_HTTP` | `allow_http` | Allow plain-HTTP upstreams, for testing only; default `false` |
|
||||
| `PIXA_UPSTREAM_CONNECTIONS_PER_HOST` | `upstream_connections_per_host` | Concurrent connections per upstream host; default `20` |
|
||||
| `PIXA_UPSTREAM_CONNECTIONS` | `upstream_connections` | Concurrent connections to all upstream hosts together; default `64` |
|
||||
| `PIXA_MAX_CONCURRENT_PROCESSING` | `max_concurrent_processing` | Images processed at once; default the number of CPUs |
|
||||
| `PIXA_MAX_CONCURRENT_PROCESSING` | `max_concurrent_processing` | Images processed at once; default half the CPUs, at least 1 |
|
||||
| `PIXA_IMAGE_REQUESTS_PER_MINUTE` | `image_requests_per_minute` | Image requests per client per minute; `0` turns it off; default `600` |
|
||||
| `PIXA_UPSTREAM_FETCH_TIMEOUT` | `upstream_fetch_timeout` | Time allowed for one fetch from an upstream host; default `30s` |
|
||||
| `PIXA_UPSTREAM_MAX_RESPONSE_SIZE` | `upstream_max_response_size` | Largest upstream response accepted, in bytes; default 50 MiB |
|
||||
@@ -610,10 +631,12 @@ Key settings in more detail:
|
||||
does, and `downstream_timeout` has not ended first, the request is answered
|
||||
503 with the error `server busy, try again later`
|
||||
- `max_concurrent_processing` — the most images decoded and encoded at once;
|
||||
default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a
|
||||
container's CPU limit. A request that finds all of them in use waits up to 10
|
||||
seconds for one to free up; if none does, and `downstream_timeout` has not
|
||||
ended first, it is answered 503 the same way
|
||||
default half the CPUs pixa can use (`GOMAXPROCS`, which follows a container's
|
||||
CPU limit), rounded down, and at least 1. A value that is not a whole number
|
||||
of at least 1 aborts startup. A request that finds all of them in use waits up
|
||||
to 10 seconds for one to free up; if none does, and `downstream_timeout` has
|
||||
not ended first, it is answered 503 the same way. Each one can take a lot of
|
||||
memory: see "Running under upaas"
|
||||
- `image_requests_per_minute` — how many requests one client may send to the
|
||||
image routes per minute (see the limit on them under Routes); default `600`.
|
||||
`0` turns the limit off, and a negative or non-integer value aborts startup.
|
||||
|
||||
@@ -30,6 +30,18 @@ P2: security: per-origin rate limiting
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-15 `max_concurrent_processing` defaults to half the CPUs pixa sees,
|
||||
rounded down, and at least 1 (closes #242): it defaulted to one image per CPU,
|
||||
and a container with no CPU limit sees every CPU of the host, so on a 48-CPU
|
||||
host pixa processed 48 images at once, each of which can take about 1.6 GiB.
|
||||
`config.DefaultMaxConcurrentProcessing` gives the default to the config and to
|
||||
`imageprocessor.New`; `PIXA_MAX_CONCURRENT_PROCESSING` still overrides it.
|
||||
"Running under upaas" in `README.md` gives the worst-case memory, measured,
|
||||
and tells the operator to set `PIXA_MAX_CONCURRENT_PROCESSING` and
|
||||
`PIXA_UPSTREAM_CONNECTIONS` to fit the container's memory. It also gives the
|
||||
file libvips writes to `/tmp` for each large source, up to 1 GiB, measured,
|
||||
which is outside `/var/lib/pixa` and `cache_max_bytes` and is memory where
|
||||
`/tmp` is a tmpfs.
|
||||
- 2026-10-15 expired negative cache rows are removed on a schedule (closes
|
||||
#244): the row of a failed fetch, which makes pixa answer 502 for that source
|
||||
URL for 5 minutes, was deleted only when the same URL was requested again
|
||||
|
||||
@@ -108,10 +108,11 @@ upstream_connections_per_host: 20
|
||||
# downstream_timeout has ended first.
|
||||
upstream_connections: 64
|
||||
|
||||
# Maximum number of images decoded and encoded at once (default: the
|
||||
# number of CPUs pixa can use, which follows a container's CPU limit). A
|
||||
# request that finds none free waits up to 10 seconds for one, and if none
|
||||
# frees up it is answered 503, unless downstream_timeout has ended first.
|
||||
# Maximum number of images decoded and encoded at once (default: half the
|
||||
# number of CPUs pixa can use, at least 1; the number of CPUs follows a
|
||||
# container's CPU limit). A request that finds none free waits up to 10
|
||||
# seconds for one, and if none frees up it is answered 503, unless
|
||||
# downstream_timeout has ended first.
|
||||
# max_concurrent_processing: 4
|
||||
|
||||
# How many requests to the image routes (/v1/image/ and /v1/e/) one client
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
"testing"
|
||||
)
|
||||
@@ -12,8 +13,8 @@ const (
|
||||
)
|
||||
|
||||
// TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
|
||||
// max_concurrent_processing is the number of CPUs Go uses and an omitted
|
||||
// upstream_connections is 64.
|
||||
// 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()
|
||||
|
||||
@@ -22,9 +23,9 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
|
||||
t.Fatalf("minimal config should be valid, got error: %v", err)
|
||||
}
|
||||
|
||||
if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
|
||||
t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
|
||||
c.MaxConcurrentProcessing, runtime.GOMAXPROCS(0))
|
||||
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 {
|
||||
@@ -32,6 +33,36 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
@@ -160,3 +191,19 @@ func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,6 +37,16 @@ const (
|
||||
DefaultImageRequestsPerMinute = 600
|
||||
)
|
||||
|
||||
// DefaultMaxConcurrentProcessing returns the most images processed at once
|
||||
// when max_concurrent_processing is omitted: half the CPUs Go uses
|
||||
// (runtime.GOMAXPROCS(0), which follows a container's CPU limit), rounded
|
||||
// down, and at least 1.
|
||||
func DefaultMaxConcurrentProcessing() int {
|
||||
const cpusPerImage = 2
|
||||
|
||||
return max(runtime.GOMAXPROCS(0)/cpusPerImage, 1)
|
||||
}
|
||||
|
||||
// Configuration key names.
|
||||
const (
|
||||
keyDebug = "debug"
|
||||
@@ -297,10 +307,8 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
keyUpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost),
|
||||
UpstreamConnections: loader.intVal(
|
||||
keyUpstreamConnections, DefaultUpstreamConnections),
|
||||
// Decoding and encoding are CPU-bound, so the default is one image
|
||||
// per CPU Go uses, which follows a container's CPU limit.
|
||||
MaxConcurrentProcessing: loader.intVal(
|
||||
keyMaxConcurrentProcessing, runtime.GOMAXPROCS(0)),
|
||||
keyMaxConcurrentProcessing, DefaultMaxConcurrentProcessing()),
|
||||
UpstreamFetchTimeout: loader.durationVal(
|
||||
keyUpstreamFetchTimeout, DefaultUpstreamFetchTimeout),
|
||||
UpstreamMaxResponseSize: loader.int64Val(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -53,7 +53,7 @@ type ServiceConfig struct {
|
||||
// Allowlist is the list of hosts that don't require signatures
|
||||
Allowlist []string
|
||||
// MaxConcurrentProcessing is the most images processed at once; zero
|
||||
// uses the image processor's default, one per CPU
|
||||
// uses the image processor's default, half the CPUs
|
||||
MaxConcurrentProcessing int
|
||||
// Logger for logging
|
||||
Logger *slog.Logger
|
||||
|
||||
Reference in New Issue
Block a user