From 180c301c2cdc08471bed3b3ca014fc5379540f1b Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Thu, 6 Nov 2025 23:17:25 +0100 Subject: [PATCH 01/14] adds worker max_threads --- caddy/app.go | 1 + caddy/workerconfig.go | 13 +++++++++++++ frankenphp.go | 6 ++++++ options.go | 10 ++++++++++ scaling.go | 13 ++++++++++--- testdata/worker-with-counter.php | 1 + worker.go | 8 +++++++- 7 files changed, 48 insertions(+), 4 deletions(-) diff --git a/caddy/app.go b/caddy/app.go index ad648efd75..c09617a555 100644 --- a/caddy/app.go +++ b/caddy/app.go @@ -123,6 +123,7 @@ func (f *FrankenPHPApp) Start() error { frankenphp.WithWorkerEnv(w.Env), frankenphp.WithWorkerWatchMode(w.Watch), frankenphp.WithWorkerMaxFailures(w.MaxConsecutiveFailures), + frankenphp.WithWorkerMaxThreads(w.MaxThreads), } opts = append(opts, frankenphp.WithWorkers(w.Name, repl.ReplaceKnown(w.FileName, ""), w.Num, workerOpts...)) diff --git a/caddy/workerconfig.go b/caddy/workerconfig.go index 716bf3f91f..7391ee1c76 100644 --- a/caddy/workerconfig.go +++ b/caddy/workerconfig.go @@ -29,6 +29,8 @@ type workerConfig struct { FileName string `json:"file_name,omitempty"` // Num sets the number of workers to start. Num int `json:"num,omitempty"` + // MaxThreads sets the maximum number of threads for this worker. + MaxThreads int `json:"max_threads,omitempty"` // Env sets an extra environment variable to the given value. Can be specified more than once for multiple environment variables. Env map[string]string `json:"env,omitempty"` // Directories to watch for file changes @@ -86,6 +88,17 @@ func parseWorkerConfig(d *caddyfile.Dispenser) (workerConfig, error) { } wc.Num = int(v) + case "max_threads": + if !d.NextArg() { + return wc, d.ArgErr() + } + + v, err := strconv.ParseUint(d.Val(), 10, 32) + if err != nil { + return wc, err + } + + wc.MaxThreads = int(v) case "env": args := d.RemainingArgs() if len(args) != 2 { diff --git a/frankenphp.go b/frankenphp.go index ae11ec1027..fc6832866c 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -140,6 +140,7 @@ func Config() PHPConfig { func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { maxProcs := runtime.GOMAXPROCS(0) * 2 + maxThreadsFromWorkerOpts := 0 for i, w := range opt.workers { if w.num <= 0 { @@ -149,6 +150,11 @@ func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { metrics.TotalWorkers(w.name, w.num) numWorkers += opt.workers[i].num + maxThreadsFromWorkerOpts += w.maxThreads + } + + if opt.maxThreads == 0 && maxThreadsFromWorkerOpts > 0 { + opt.maxThreads = maxThreadsFromWorkerOpts } numThreadsIsSet := opt.numThreads > 0 diff --git a/options.go b/options.go index befe3a7fb8..c2bb8581c8 100644 --- a/options.go +++ b/options.go @@ -32,6 +32,7 @@ type workerOpt struct { name string fileName string num int + maxThreads int env PreparedEnv watch []string maxConsecutiveFailures int @@ -99,6 +100,15 @@ func WithWorkerEnv(env map[string]string) WorkerOption { } } +// WithWorkerEnv sets environment variables for the worker +func WithWorkerMaxThreads(num int) WorkerOption { + return func(w *workerOpt) error { + w.maxThreads = num + + return nil + } +} + // WithWorkerWatchMode sets directories to watch for file changes func WithWorkerWatchMode(watch []string) WorkerOption { return func(w *workerOpt) error { diff --git a/scaling.go b/scaling.go index 57e6c598b9..95eaa962fc 100644 --- a/scaling.go +++ b/scaling.go @@ -158,11 +158,18 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, } // if the request has been stalled long enough, scale - if fc.worker != nil { - scaleWorkerThread(fc.worker) - } else { + if fc.worker == nil { scaleRegularThread() + return } + + if fc.worker.maxThreads > 0 && fc.worker.countThreads() >= fc.worker.maxThreads { + logger.Debug("cannot scale worker thread, max threads reached for worker " + fc.worker.name) + logger.Debug("max_threads", fc.worker.maxThreads) + continue + } + + scaleWorkerThread(fc.worker) case <-done: return } diff --git a/testdata/worker-with-counter.php b/testdata/worker-with-counter.php index 248cf469f8..0406b63feb 100644 --- a/testdata/worker-with-counter.php +++ b/testdata/worker-with-counter.php @@ -4,6 +4,7 @@ $printNumberOfRequests = function () use (&$numberOfRequests) { $numberOfRequests++; echo "requests:$numberOfRequests"; + sleep(1); }; while (frankenphp_handle_request($printNumberOfRequests)) { diff --git a/worker.go b/worker.go index ee468bd6cd..a76b2f9fa2 100644 --- a/worker.go +++ b/worker.go @@ -17,6 +17,7 @@ type worker struct { name string fileName string num int + maxThreads int env PreparedEnv requestChan chan *frankenPHPContext threads []*phpThread @@ -122,6 +123,7 @@ func newWorker(o workerOpt) (*worker, error) { name: o.name, fileName: absFileName, num: o.num, + maxThreads: o.maxThreads, env: o.env, requestChan: make(chan *frankenPHPContext), threads: make([]*phpThread, 0, o.num), @@ -240,13 +242,17 @@ func (worker *worker) handleRequest(fc *frankenPHPContext) { // if no thread was available, mark the request as queued and apply the scaling strategy metrics.QueuedWorkerRequest(worker.name) for { + workerScaleChan := scaleChan + if worker.maxThreads > 0 && worker.countThreads() >= worker.maxThreads { + workerScaleChan = nil // max_threads for this worker reached, do not attempt scaling + } select { case worker.requestChan <- fc: metrics.DequeuedWorkerRequest(worker.name) <-fc.done metrics.StopWorkerRequest(worker.name, time.Since(fc.startedAt)) return - case scaleChan <- fc: + case workerScaleChan <- fc: // the request has triggered scaling, continue to wait for a thread case <-timeoutChan(maxWaitTime): metrics.DequeuedWorkerRequest(worker.name) From d322b232c6f26083c4e60a96ad93545c9e163b50 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 22:41:58 +0100 Subject: [PATCH 02/14] Adds tests for all calculation cases. --- frankenphp.go | 12 +++++++++++- phpmainthread_test.go | 16 ++++++++++++++++ scaling.go | 1 - 3 files changed, 27 insertions(+), 2 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index fc6832866c..7501d3da3a 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -150,9 +150,19 @@ func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { metrics.TotalWorkers(w.name, w.num) numWorkers += opt.workers[i].num - maxThreadsFromWorkerOpts += w.maxThreads + + if w.maxThreads > 0 { + if w.maxThreads < w.num { + return 0, fmt.Errorf("worker max_threads (%d) must be greater or equal to worker num (%d) (%q)", w.maxThreads, w.num, w.fileName) + } + if w.maxThreads > opt.maxThreads && opt.maxThreads > 0 { + return 0, fmt.Errorf("worker max_threads (%d) cannot be greater than total max_threads (%d) (%q)", w.maxThreads, opt.maxThreads, w.fileName) + } + maxThreadsFromWorkerOpts += w.maxThreads + } } + // if no max_threads is defined, use the sum of worker max_threads if opt.maxThreads == 0 && maxThreadsFromWorkerOpts > 0 { opt.maxThreads = maxThreadsFromWorkerOpts } diff --git a/phpmainthread_test.go b/phpmainthread_test.go index a49484b61f..6ca033f808 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -272,6 +272,14 @@ func TestCorrectThreadCalculation(t *testing.T) { testThreadCalculation(t, 2, -1, &opt{maxThreads: -1, workers: oneWorkerThread}) testThreadCalculation(t, 2, -1, &opt{numThreads: 2, maxThreads: -1}) + // max_threads should be sum of worker max_threads + testThreadCalculation(t, 2, 5, &opt{workers: []workerOpt{{num: 1, maxThreads: 5}}}) + testThreadCalculation(t, 6, 8, &opt{workers: []workerOpt{{num: 1, maxThreads: 4}, {num: 4, maxThreads: 4}}}) + + // max_threads should remain equal to overall max_threads + testThreadCalculation(t, 2, 5, &opt{maxThreads: 5, workers: []workerOpt{{num: 1, maxThreads: 3}}}) + testThreadCalculation(t, 3, 5, &opt{maxThreads: 5, workers: []workerOpt{{num: 1, maxThreads: 4}, {num: 1, maxThreads: 4}}}) + // not enough num threads testThreadCalculationError(t, &opt{numThreads: 1, workers: oneWorkerThread}) testThreadCalculationError(t, &opt{numThreads: 1, maxThreads: 1, workers: oneWorkerThread}) @@ -279,9 +287,16 @@ func TestCorrectThreadCalculation(t *testing.T) { // not enough max_threads testThreadCalculationError(t, &opt{numThreads: 2, maxThreads: 1}) testThreadCalculationError(t, &opt{maxThreads: 1, workers: oneWorkerThread}) + + // worker max_threads is bigger than overall max_threads + testThreadCalculationError(t, &opt{maxThreads: 5, workers: []workerOpt{{num: 1, maxThreads: 10}}}) + + // worker max_threads is smaller than num_threads + testThreadCalculationError(t, &opt{workers: []workerOpt{{num: 3, maxThreads: 2}}}) } func testThreadCalculation(t *testing.T, expectedNumThreads int, expectedMaxThreads int, o *opt) { + t.Helper() _, err := calculateMaxThreads(o) assert.NoError(t, err, "no error should be returned") assert.Equal(t, expectedNumThreads, o.numThreads, "num_threads must be correct") @@ -289,6 +304,7 @@ func testThreadCalculation(t *testing.T, expectedNumThreads int, expectedMaxThre } func testThreadCalculationError(t *testing.T, o *opt) { + t.Helper() _, err := calculateMaxThreads(o) assert.Error(t, err, "configuration must error") } diff --git a/scaling.go b/scaling.go index 95eaa962fc..0b26c861f3 100644 --- a/scaling.go +++ b/scaling.go @@ -165,7 +165,6 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, if fc.worker.maxThreads > 0 && fc.worker.countThreads() >= fc.worker.maxThreads { logger.Debug("cannot scale worker thread, max threads reached for worker " + fc.worker.name) - logger.Debug("max_threads", fc.worker.maxThreads) continue } From 808ee89d9c505c08181ce53a3c856fb82a1c1736 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 22:49:15 +0100 Subject: [PATCH 03/14] Adds max_threads limitation to test. --- caddy/admin_test.go | 9 ++++++--- caddy/workerconfig.go | 16 ++++++++-------- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/caddy/admin_test.go b/caddy/admin_test.go index 345ceff806..4f145a9f16 100644 --- a/caddy/admin_test.go +++ b/caddy/admin_test.go @@ -94,7 +94,10 @@ func TestAutoScaleWorkerThreads(t *testing.T) { frankenphp { max_threads 10 num_threads 2 - worker ../testdata/sleep.php 1 + worker ../testdata/sleep.php { + num 1 + max_threads 3 + } } } @@ -128,8 +131,8 @@ func TestAutoScaleWorkerThreads(t *testing.T) { } } - // assert that there are now more threads than before - assert.NotEqual(t, amountOfThreads, 2) + assert.NotEqual(t, amountOfThreads, 2, "at least one thread should have been auto-scaled") + assert.LessOrEqual(t, amountOfThreads, 4, "not more than 3 max_threads + 1 regular thread should be present") } // Note this test requires at least 2x40MB available memory for the process diff --git a/caddy/workerconfig.go b/caddy/workerconfig.go index 7391ee1c76..4e3d198e03 100644 --- a/caddy/workerconfig.go +++ b/caddy/workerconfig.go @@ -89,16 +89,16 @@ func parseWorkerConfig(d *caddyfile.Dispenser) (workerConfig, error) { wc.Num = int(v) case "max_threads": - if !d.NextArg() { - return wc, d.ArgErr() - } + if !d.NextArg() { + return wc, d.ArgErr() + } - v, err := strconv.ParseUint(d.Val(), 10, 32) - if err != nil { - return wc, err - } + v, err := strconv.ParseUint(d.Val(), 10, 32) + if err != nil { + return wc, err + } - wc.MaxThreads = int(v) + wc.MaxThreads = int(v) case "env": args := d.RemainingArgs() if len(args) != 2 { From e2b066cc2278b75e63b0b161b70d5f02b0265610 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 22:50:29 +0100 Subject: [PATCH 04/14] Removes the test sleep. --- testdata/worker-with-counter.php | 1 - 1 file changed, 1 deletion(-) diff --git a/testdata/worker-with-counter.php b/testdata/worker-with-counter.php index 0406b63feb..248cf469f8 100644 --- a/testdata/worker-with-counter.php +++ b/testdata/worker-with-counter.php @@ -4,7 +4,6 @@ $printNumberOfRequests = function () use (&$numberOfRequests) { $numberOfRequests++; echo "requests:$numberOfRequests"; - sleep(1); }; while (frankenphp_handle_request($printNumberOfRequests)) { From 2c2ce254bb2b572b277afd7701366927f51cafb2 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 22:53:41 +0100 Subject: [PATCH 05/14] Adds max_threads to error message. --- caddy/workerconfig.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/caddy/workerconfig.go b/caddy/workerconfig.go index 4e3d198e03..bf94d4bfd5 100644 --- a/caddy/workerconfig.go +++ b/caddy/workerconfig.go @@ -136,7 +136,7 @@ func parseWorkerConfig(d *caddyfile.Dispenser) (workerConfig, error) { wc.MaxConsecutiveFailures = int(v) default: - allowedDirectives := "name, file, num, env, watch, match, max_consecutive_failures" + allowedDirectives := "name, file, num, env, watch, match, max_consecutive_failures, max_threads" return wc, wrongSubDirectiveError("worker", allowedDirectives, v) } } From 7da6d07bca0a02ba8bdbfbc0249174e2e44a07eb Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 23:00:19 +0100 Subject: [PATCH 06/14] correctly uses continue. --- scaling.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scaling.go b/scaling.go index 0b26c861f3..265484cf7c 100644 --- a/scaling.go +++ b/scaling.go @@ -160,7 +160,7 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, // if the request has been stalled long enough, scale if fc.worker == nil { scaleRegularThread() - return + continue } if fc.worker.maxThreads > 0 && fc.worker.countThreads() >= fc.worker.maxThreads { From 6fb6d9f60a2e22f0a6807f98833302260c9fd96c Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 23:27:45 +0100 Subject: [PATCH 07/14] Fixes logic with only worker max_threads set. --- frankenphp.go | 23 ++++++++++++++--------- phpmainthread_test.go | 7 ++++--- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index 7501d3da3a..49d3aef129 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -140,7 +140,7 @@ func Config() PHPConfig { func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { maxProcs := runtime.GOMAXPROCS(0) * 2 - maxThreadsFromWorkerOpts := 0 + maxThreadsFromWorkers := 0 for i, w := range opt.workers { if w.num <= 0 { @@ -158,21 +158,26 @@ func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { if w.maxThreads > opt.maxThreads && opt.maxThreads > 0 { return 0, fmt.Errorf("worker max_threads (%d) cannot be greater than total max_threads (%d) (%q)", w.maxThreads, opt.maxThreads, w.fileName) } - maxThreadsFromWorkerOpts += w.maxThreads + maxThreadsFromWorkers += w.maxThreads - w.num } } - // if no max_threads is defined, use the sum of worker max_threads - if opt.maxThreads == 0 && maxThreadsFromWorkerOpts > 0 { - opt.maxThreads = maxThreadsFromWorkerOpts - } - numThreadsIsSet := opt.numThreads > 0 maxThreadsIsSet := opt.maxThreads != 0 maxThreadsIsAuto := opt.maxThreads < 0 // maxthreads < 0 signifies auto mode (see phpmaintread.go) + // consider the case where max_threads is only defined in workers + if !maxThreadsIsSet && maxThreadsFromWorkers > 0 { + maxThreadsIsSet = true + if numThreadsIsSet { + opt.maxThreads = opt.numThreads + maxThreadsFromWorkers + } else { + opt.maxThreads = numWorkers + 1 + maxThreadsFromWorkers + } + } + if numThreadsIsSet && !maxThreadsIsSet { - opt.maxThreads = opt.numThreads + opt.maxThreads = opt.numThreads + maxThreadsFromWorkers if opt.numThreads <= numWorkers { return 0, fmt.Errorf("num_threads (%d) must be greater than the number of worker threads (%d)", opt.numThreads, numWorkers) } @@ -189,7 +194,7 @@ func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { return numWorkers, nil } - if !numThreadsIsSet { + if !maxThreadsIsSet && !numThreadsIsSet { if numWorkers >= maxProcs { // Start at least as many threads as workers, and keep a free thread to handle requests in non-worker mode opt.numThreads = numWorkers + 1 diff --git a/phpmainthread_test.go b/phpmainthread_test.go index 6ca033f808..deca8ccdd5 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -272,9 +272,10 @@ func TestCorrectThreadCalculation(t *testing.T) { testThreadCalculation(t, 2, -1, &opt{maxThreads: -1, workers: oneWorkerThread}) testThreadCalculation(t, 2, -1, &opt{numThreads: 2, maxThreads: -1}) - // max_threads should be sum of worker max_threads - testThreadCalculation(t, 2, 5, &opt{workers: []workerOpt{{num: 1, maxThreads: 5}}}) - testThreadCalculation(t, 6, 8, &opt{workers: []workerOpt{{num: 1, maxThreads: 4}, {num: 4, maxThreads: 4}}}) + // max_threads should be thread minimum + sum of worker max_threads + testThreadCalculation(t, 2, 6, &opt{workers: []workerOpt{{num: 1, maxThreads: 5}}}) + testThreadCalculation(t, 6, 9, &opt{workers: []workerOpt{{num: 1, maxThreads: 4}, {num: 4, maxThreads: 4}}}) + testThreadCalculation(t, 10, 14, &opt{numThreads: 10, workers: []workerOpt{{num: 1, maxThreads: 4}, {num: 3, maxThreads: 4}}}) // max_threads should remain equal to overall max_threads testThreadCalculation(t, 2, 5, &opt{maxThreads: 5, workers: []workerOpt{{num: 1, maxThreads: 3}}}) From 9b0855c1586855cd3daf45250973715c0bdff4c2 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 23:33:27 +0100 Subject: [PATCH 08/14] Adjust comments. --- frankenphp.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frankenphp.go b/frankenphp.go index 49d3aef129..300a554918 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -166,7 +166,7 @@ func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { maxThreadsIsSet := opt.maxThreads != 0 maxThreadsIsAuto := opt.maxThreads < 0 // maxthreads < 0 signifies auto mode (see phpmaintread.go) - // consider the case where max_threads is only defined in workers + // if max_threads is only defined in workers, scale up to the sum of all worker max_threads if !maxThreadsIsSet && maxThreadsFromWorkers > 0 { maxThreadsIsSet = true if numThreadsIsSet { From be9c492a56d7fe529557fc070457756a39bb50bc Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 23:36:06 +0100 Subject: [PATCH 09/14] Removes unnecessary check. --- frankenphp.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frankenphp.go b/frankenphp.go index 300a554918..d0a0a2bd6f 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -177,7 +177,7 @@ func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { } if numThreadsIsSet && !maxThreadsIsSet { - opt.maxThreads = opt.numThreads + maxThreadsFromWorkers + opt.maxThreads = opt.numThreads if opt.numThreads <= numWorkers { return 0, fmt.Errorf("num_threads (%d) must be greater than the number of worker threads (%d)", opt.numThreads, numWorkers) } From 6c3e93089491fadc3fc874d3989ddc1ab8b27752 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 7 Nov 2025 23:41:03 +0100 Subject: [PATCH 10/14] Fixes comment. --- options.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/options.go b/options.go index c2bb8581c8..de20a82e9b 100644 --- a/options.go +++ b/options.go @@ -100,7 +100,7 @@ func WithWorkerEnv(env map[string]string) WorkerOption { } } -// WithWorkerEnv sets environment variables for the worker +// WithWorkerMaxThreads sets the max number of threads for this specific worker func WithWorkerMaxThreads(num int) WorkerOption { return func(w *workerOpt) error { w.maxThreads = num From 5d29092d1f993f0cbc99fb491b75a28ccc21da5c Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Mon, 10 Nov 2025 18:37:37 +0100 Subject: [PATCH 11/14] suggestions by @dunlgas. --- caddy/workerconfig.go | 4 ++-- scaling.go | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/caddy/workerconfig.go b/caddy/workerconfig.go index bf94d4bfd5..fd728f7c60 100644 --- a/caddy/workerconfig.go +++ b/caddy/workerconfig.go @@ -84,7 +84,7 @@ func parseWorkerConfig(d *caddyfile.Dispenser) (workerConfig, error) { v, err := strconv.ParseUint(d.Val(), 10, 32) if err != nil { - return wc, err + return wc, d.WrapErr(err) } wc.Num = int(v) @@ -95,7 +95,7 @@ func parseWorkerConfig(d *caddyfile.Dispenser) (workerConfig, error) { v, err := strconv.ParseUint(d.Val(), 10, 32) if err != nil { - return wc, err + return wc, d.WrapErr(err) } wc.MaxThreads = int(v) diff --git a/scaling.go b/scaling.go index 265484cf7c..b60a9021be 100644 --- a/scaling.go +++ b/scaling.go @@ -163,8 +163,9 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, continue } + // check for max worker threads here again in case requests overflowed while waiting if fc.worker.maxThreads > 0 && fc.worker.countThreads() >= fc.worker.maxThreads { - logger.Debug("cannot scale worker thread, max threads reached for worker " + fc.worker.name) + logger.Debug("cannot scale worker thread, max threads reached for worker", slog.String("worker", fc.worker.name)) continue } From ab0a40b5e3ead8a21369c20083b217a86e33591f Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Wed, 12 Nov 2025 18:52:39 +0100 Subject: [PATCH 12/14] copilot suggestions. --- caddy/admin_test.go | 2 +- scaling.go | 2 +- worker.go | 15 ++++++++++++++- 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/caddy/admin_test.go b/caddy/admin_test.go index 4f145a9f16..17f211cd22 100644 --- a/caddy/admin_test.go +++ b/caddy/admin_test.go @@ -132,7 +132,7 @@ func TestAutoScaleWorkerThreads(t *testing.T) { } assert.NotEqual(t, amountOfThreads, 2, "at least one thread should have been auto-scaled") - assert.LessOrEqual(t, amountOfThreads, 4, "not more than 3 max_threads + 1 regular thread should be present") + assert.LessOrEqual(t, amountOfThreads, 4, "at most 3 max_threads + 1 regular thread should be present") } // Note this test requires at least 2x40MB available memory for the process diff --git a/scaling.go b/scaling.go index b60a9021be..d9e0ba89cc 100644 --- a/scaling.go +++ b/scaling.go @@ -164,7 +164,7 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, } // check for max worker threads here again in case requests overflowed while waiting - if fc.worker.maxThreads > 0 && fc.worker.countThreads() >= fc.worker.maxThreads { + if fc.worker.isAtThreadLimit() { logger.Debug("cannot scale worker thread, max threads reached for worker", slog.String("worker", fc.worker.name)) continue } diff --git a/worker.go b/worker.go index 91b53cfa8f..94bced7c31 100644 --- a/worker.go +++ b/worker.go @@ -230,6 +230,19 @@ func (worker *worker) countThreads() int { return l } +// check if max_threads has been reached +func (worker *worker) isAtThreadLimit() bool { + if worker.maxThreads <= 0 { + return false + } + + worker.threadMutex.RLock() + atMaxThreads := len(worker.threads) >= worker.maxThreads + worker.threadMutex.RUnlock() + + return atMaxThreads +} + func (worker *worker) handleRequest(fc *frankenPHPContext) error { metrics.StartWorkerRequest(worker.name) @@ -253,7 +266,7 @@ func (worker *worker) handleRequest(fc *frankenPHPContext) error { metrics.QueuedWorkerRequest(worker.name) for { workerScaleChan := scaleChan - if worker.maxThreads > 0 && worker.countThreads() >= worker.maxThreads { + if worker.isAtThreadLimit() { workerScaleChan = nil // max_threads for this worker reached, do not attempt scaling } select { From b733e66228f4f73e1caddb616994ca63c3cd08c2 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Mon, 17 Nov 2025 19:04:02 +0100 Subject: [PATCH 13/14] Renames logger. --- scaling.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scaling.go b/scaling.go index fab08879b4..e2b9b9d4eb 100644 --- a/scaling.go +++ b/scaling.go @@ -178,7 +178,7 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, // check for max worker threads here again in case requests overflowed while waiting if fc.worker.isAtThreadLimit() { - logger.Debug("cannot scale worker thread, max threads reached for worker", slog.String("worker", fc.worker.name)) + globalLogger.Debug("cannot scale worker thread, max threads reached for worker", slog.String("worker", fc.worker.name)) continue } From e62103d7c581ad730966ffd98ba31d5e65c64601 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?K=C3=A9vin=20Dunglas?= Date: Tue, 18 Nov 2025 11:31:05 +0100 Subject: [PATCH 14/14] review --- frankenphp.go | 2 ++ phpmainthread_test.go | 2 ++ scaling.go | 5 ++++- worker.go | 1 + 4 files changed, 9 insertions(+), 1 deletion(-) diff --git a/frankenphp.go b/frankenphp.go index 80610770e3..3a66fcaffc 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -170,9 +170,11 @@ func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { if w.maxThreads < w.num { return 0, fmt.Errorf("worker max_threads (%d) must be greater or equal to worker num (%d) (%q)", w.maxThreads, w.num, w.fileName) } + if w.maxThreads > opt.maxThreads && opt.maxThreads > 0 { return 0, fmt.Errorf("worker max_threads (%d) cannot be greater than total max_threads (%d) (%q)", w.maxThreads, opt.maxThreads, w.fileName) } + maxThreadsFromWorkers += w.maxThreads - w.num } } diff --git a/phpmainthread_test.go b/phpmainthread_test.go index fe5b2bd0f6..4ad502fa97 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -310,6 +310,7 @@ func TestCorrectThreadCalculation(t *testing.T) { func testThreadCalculation(t *testing.T, expectedNumThreads int, expectedMaxThreads int, o *opt) { t.Helper() + _, err := calculateMaxThreads(o) assert.NoError(t, err, "no error should be returned") assert.Equal(t, expectedNumThreads, o.numThreads, "num_threads must be correct") @@ -318,6 +319,7 @@ func testThreadCalculation(t *testing.T, expectedNumThreads int, expectedMaxThre func testThreadCalculationError(t *testing.T, o *opt) { t.Helper() + _, err := calculateMaxThreads(o) assert.Error(t, err, "configuration must error") } diff --git a/scaling.go b/scaling.go index e2b9b9d4eb..af7c6c29d3 100644 --- a/scaling.go +++ b/scaling.go @@ -178,7 +178,10 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, // check for max worker threads here again in case requests overflowed while waiting if fc.worker.isAtThreadLimit() { - globalLogger.Debug("cannot scale worker thread, max threads reached for worker", slog.String("worker", fc.worker.name)) + if globalLogger.Enabled(globalCtx, slog.LevelInfo) { + globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "cannot scale worker thread, max threads reached for worker", slog.String("worker", fc.worker.name)) + } + continue } diff --git a/worker.go b/worker.go index 499f30c7ed..cf6ea9dbf1 100644 --- a/worker.go +++ b/worker.go @@ -269,6 +269,7 @@ func (worker *worker) handleRequest(ch contextHolder) error { if worker.isAtThreadLimit() { workerScaleChan = nil // max_threads for this worker reached, do not attempt scaling } + select { case worker.requestChan <- ch: metrics.DequeuedWorkerRequest(worker.name)