From 5a41ea6dbae5b4132adab3a1737ebfa3eaa9e549 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Thu, 8 Oct 2026 21:53:35 +0200 Subject: [PATCH 01/24] puts regular requests on main thread --- frankenphp.go | 6 ++--- phpmainthread.go | 2 ++ phpmainthread_test.go | 4 +-- scaling.go | 41 +++++++++++++--------------- scaling_test.go | 6 ++--- server.go | 4 +++ threadregular.go | 9 ++++--- threadregular_test.go | 62 ------------------------------------------- 8 files changed, 38 insertions(+), 96 deletions(-) delete mode 100644 threadregular_test.go diff --git a/frankenphp.go b/frankenphp.go index 48c1380b95..0a40b09456 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -427,7 +427,7 @@ func Init(options ...Option) error { } } - mainThread, err := initPHPThreads(opt.numThreads, opt.maxThreads, opt.phpIni) + mThread, err := initPHPThreads(opt.numThreads, opt.maxThreads, opt.phpIni) if err != nil { shutdown() return err @@ -435,7 +435,7 @@ func Init(options ...Option) error { regularThreads = make([]*phpThread, 0, opt.numThreads-workerThreadCount) for range opt.numThreads - workerThreadCount { - convertToRegularThread(getInactivePHPThread()) + convertToRegularThread(getInactivePHPThread(), mThread) } if err := initWorkers(opt.workers); err != nil { @@ -449,7 +449,7 @@ func Init(options ...Option) error { return err } - initAutoScaling(mainThread) + initAutoScaling(mThread) // only now that the workers and threads are up may requests reach a server activateServers() diff --git a/phpmainthread.go b/phpmainthread.go index 5e19c0fc75..7e0eda1cec 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -26,6 +26,7 @@ type phpMainThread struct { maxThreads int phpIni map[string]string isRebooting atomic.Bool + regularRequestChan chan *frankenPHPContext } var ( @@ -48,6 +49,7 @@ func initPHPThreads(numThreads int, numMaxThreads int, phpIni map[string]string) numThreads: numThreads, maxThreads: numMaxThreads, phpIni: phpIni, + regularRequestChan: make(chan *frankenPHPContext), } // initialize the first thread diff --git a/phpmainthread_test.go b/phpmainthread_test.go index d8a5b6742d..5a013a3ebf 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -45,7 +45,7 @@ func TestTransitionRegularThreadToWorkerThread(t *testing.T) { assert.NoError(t, err) // transition to regular thread - convertToRegularThread(phpThreads[0]) + convertToRegularThread(phpThreads[0], mainThread) assert.IsType(t, ®ularThread{}, phpThreads[0].handler) // transition to worker thread @@ -308,7 +308,7 @@ func assertRequestBody(t *testing.T, url string, expected string) { // create a mix of possible transitions of workers and regular threads func allPossibleTransitions(worker1Path string, worker2Path string) []func(*phpThread) { return []func(*phpThread){ - convertToRegularThread, + func(thread *phpThread) { convertToRegularThread(thread, mainThread) }, func(thread *phpThread) { thread.shutdown() }, func(thread *phpThread) { if thread.state.Is(state.Reserved) { diff --git a/scaling.go b/scaling.go index dd21a7e37c..2abe7fc765 100644 --- a/scaling.go +++ b/scaling.go @@ -45,24 +45,21 @@ func initAutoScaling(mainThread *phpMainThread) { scaleChan = make(chan *frankenPHPContext) } - done := mainThread.done - mstate := mainThread.state - scalingMu.Lock() maxScaledThreads := mainThread.maxThreads - mainThread.numThreads autoScaledThreads = make([]*phpThread, 0, maxScaledThreads) scalingMu.Unlock() - go startUpscalingThreads(maxScaledThreads, scaleChan, done, mstate) - go startDownScalingThreads(done) + go startUpscalingThreads(maxScaledThreads, scaleChan, mainThread) + go startDownScalingThreads(mainThread) } -func addRegularThread() (*phpThread, error) { +func addRegularThread(mThread *phpMainThread) (*phpThread, error) { thread := getInactivePHPThread() if thread == nil { return nil, ErrMaxThreadsReached } - convertToRegularThread(thread) + convertToRegularThread(thread, mThread) thread.state.WaitFor(state.Ready, state.Inactive, state.Reserved) // stable states return thread, nil @@ -80,16 +77,16 @@ func addWorkerThread(worker *worker) (*phpThread, error) { } // scaleWorkerThread adds a worker PHP thread automatically -func scaleWorkerThread(worker *worker, done chan struct{}, mstate *state.ThreadState) { +func scaleWorkerThread(worker *worker, mThread *phpMainThread) { // probe CPU usage before acquiring the lock (avoids holding lock during 120ms sleep) - if !cpu.ProbeCPUs(cpuProbeTime, maxCpuUsageForScaling, done) { + if !cpu.ProbeCPUs(cpuProbeTime, maxCpuUsageForScaling, mThread.done) { return } scalingMu.Lock() defer scalingMu.Unlock() - if !mstate.Is(state.Ready) { + if !mThread.state.Is(state.Ready) { return } @@ -110,20 +107,20 @@ func scaleWorkerThread(worker *worker, done chan struct{}, mstate *state.ThreadS } // scaleRegularThread adds a regular PHP thread automatically -func scaleRegularThread(done chan struct{}, mstate *state.ThreadState) { +func scaleRegularThread(mThread *phpMainThread) { // probe CPU usage before acquiring the lock (avoids holding lock during 120ms sleep) - if !cpu.ProbeCPUs(cpuProbeTime, maxCpuUsageForScaling, done) { + if !cpu.ProbeCPUs(cpuProbeTime, maxCpuUsageForScaling, mThread.done) { return } scalingMu.Lock() defer scalingMu.Unlock() - if !mstate.Is(state.Ready) { + if !mThread.state.Is(state.Ready) { return } - thread, err := addRegularThread() + thread, err := addRegularThread(mThread) if err != nil { if globalLogger.Enabled(globalCtx, slog.LevelWarn) { globalLogger.LogAttrs(globalCtx, slog.LevelWarn, "could not increase max_threads, consider raising this limit", slog.Any("error", err)) @@ -139,7 +136,7 @@ func scaleRegularThread(done chan struct{}, mstate *state.ThreadState) { } } -func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, done chan struct{}, mstate *state.ThreadState) { +func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, mThread *phpMainThread) { for { scalingMu.Lock() scaledThreadCount := len(autoScaledThreads) @@ -147,7 +144,7 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, if scaledThreadCount >= maxScaledThreads { // we have reached max_threads, check again later select { - case <-done: + case <-mThread.done: return case <-time.After(downScaleCheckTime): continue @@ -161,7 +158,7 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, // if the request has not been stalled long enough, wait and repeat if timeSinceStalled < minStallTime { select { - case <-done: + case <-mThread.done: return case <-time.After(minStallTime - timeSinceStalled): continue @@ -170,7 +167,7 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, // if the request has been stalled long enough, scale if fc.worker == nil { - scaleRegularThread(done, mstate) + scaleRegularThread(mThread) continue } @@ -183,17 +180,17 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, continue } - scaleWorkerThread(fc.worker, done, mstate) - case <-done: + scaleWorkerThread(fc.worker, mThread) + case <-mThread.done: return } } } -func startDownScalingThreads(done chan struct{}) { +func startDownScalingThreads(mThread *phpMainThread) { for { select { - case <-done: + case <-mThread.done: return case <-time.After(downScaleCheckTime): deactivateThreads() diff --git a/scaling_test.go b/scaling_test.go index 9a6d12da6d..07437c7163 100644 --- a/scaling_test.go +++ b/scaling_test.go @@ -20,7 +20,7 @@ func TestScaleARegularThreadUpAndDown(t *testing.T) { autoScaledThread := phpThreads[1] // scale up - scaleRegularThread(mainThread.done, mainThread.state) + scaleRegularThread(mainThread) assert.Equal(t, state.Ready, autoScaledThread.state.Get()) assert.IsType(t, ®ularThread{}, autoScaledThread.handler) @@ -49,7 +49,7 @@ func TestScaleAWorkerThreadUpAndDown(t *testing.T) { autoScaledThread := phpThreads[2] // scale up - scaleWorkerThread(globalWorkersByPath[workerPath], mainThread.done, mainThread.state) + scaleWorkerThread(globalWorkersByPath[workerPath], mainThread) assert.Equal(t, state.Ready, autoScaledThread.state.Get()) // on down-scale, the thread will be marked as inactive @@ -70,7 +70,7 @@ func TestMaxIdleTimePreventsEarlyDeactivation(t *testing.T) { autoScaledThread := phpThreads[1] // scale up - scaleRegularThread(mainThread.done, mainThread.state) + scaleRegularThread(mainThread) assert.Equal(t, state.Ready, autoScaledThread.state.Get()) // set wait time to 30 minutes (less than 1 hour max idle time) diff --git a/server.go b/server.go index 8274f23c54..856bea51e7 100644 --- a/server.go +++ b/server.go @@ -23,6 +23,7 @@ type Server struct { workers []*worker workersByPath map[string]*worker workersWithRequestMatcher []*worker + mainThread *phpMainThread // registered while FrankenPHP runs with this server; read by concurrent // ServeHTTP calls while Init()/Shutdown() flip it, hence atomic @@ -41,6 +42,7 @@ func newFallbackServer() *Server { workersByPath: make(map[string]*worker), env: make(map[string]string), logger: globalLogger, + mainThread: mainThread, } return s @@ -68,8 +70,10 @@ func registerServers(newServers []*Server) { // it runs once workers and threads are up, so a request cannot reach a server before them func activateServers() { fallbackServer.isRegistered.Store(true) + fallbackServer.mainThread = mainThread for _, s := range servers { s.isRegistered.Store(true) + s.mainThread = mainThread } } diff --git a/threadregular.go b/threadregular.go index 8d88088754..fc2030bb5d 100644 --- a/threadregular.go +++ b/threadregular.go @@ -20,19 +20,20 @@ type regularThread struct { state *state.ThreadState thread *phpThread requestCount int + requestChan chan *frankenPHPContext } var ( regularThreads []*phpThread regularThreadMu = &sync.RWMutex{} - regularRequestChan = make(chan *frankenPHPContext) queuedRegularThreads = atomic.Int32{} ) -func convertToRegularThread(thread *phpThread) { +func convertToRegularThread(thread *phpThread, mThread *phpMainThread) { thread.setHandler(®ularThread{ thread: thread, state: thread.state, + requestChan: mThread.regularRequestChan, }) attachRegularThread(thread) } @@ -105,7 +106,7 @@ func (handler *regularThread) waitForRequest() string { case <-handler.thread.drainChan: // go back to beforeScriptExecution return handler.beforeScriptExecution() - case fc = <-regularRequestChan: + case fc = <-handler.requestChan: case fc = <-handler.thread.requestChan: } @@ -153,7 +154,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { for { select { - case regularRequestChan <- fc: + case fc.server.mainThread.regularRequestChan <- fc: queuedRegularThreads.Add(-1) metrics.DequeuedRequest() diff --git a/threadregular_test.go b/threadregular_test.go deleted file mode 100644 index 7a1d0700f6..0000000000 --- a/threadregular_test.go +++ /dev/null @@ -1,62 +0,0 @@ -package frankenphp - -import ( - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestRegularRequestChanIsCreatedBeforeInitRuns(t *testing.T) { - require.NotNil(t, regularRequestChan, - "a request dispatched between the start of Init and the first thread being attached "+ - "must queue on a real channel: sending on a nil channel blocks forever, "+ - "even after the channel variable is later assigned") -} - -func TestRequestsQueuedBeforeThreadsAreReadyAreHandedOver(t *testing.T) { - regularThreadMu.Lock() - savedThreads := regularThreads - regularThreads = nil - regularThreadMu.Unlock() - savedMaxWaitTime := maxWaitTime.Swap(0) - savedScaleChan := scaleChan - scaleChan = nil - t.Cleanup(func() { - regularThreadMu.Lock() - regularThreads = savedThreads - regularThreadMu.Unlock() - maxWaitTime.Store(savedMaxWaitTime) - scaleChan = savedScaleChan - }) - - const requests = 5 - errChans := make([]chan error, requests) - for i := range errChans { - fc := &frankenPHPContext{done: make(chan any)} - errChan := make(chan error, 1) - go func() { - errChan <- handleRequestWithRegularPHPThreads(fc) - }() - errChans[i] = errChan - } - - for i := range requests { - select { - case fc := <-regularRequestChan: - fc.closeContext() - case <-time.After(5 * time.Second): - t.Fatalf("only %d of %d queued requests were handed over", i, requests) - } - } - - for i, errChan := range errChans { - select { - case err := <-errChan: - assert.NoError(t, err, "request %d", i) - case <-time.After(5 * time.Second): - t.Fatalf("dispatcher for request %d did not return after the request was handled", i) - } - } -} From 4256e99fa70667cd08d73f097b31f2bf756d9c46 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Thu, 8 Oct 2026 22:04:59 +0200 Subject: [PATCH 02/24] requeues requests --- frankenphp.go | 4 ++++ phpmainthread.go | 40 +++++++++++++++++++++++++++++----------- threadregular.go | 4 ++-- worker.go | 1 + 4 files changed, 36 insertions(+), 13 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index 0a40b09456..904432e651 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -473,6 +473,10 @@ func Init(options ...Option) error { } } + if oldMainThread != nil { + mainThread.reQueueRequests(oldMainThread) + } + return nil } diff --git a/phpmainthread.go b/phpmainthread.go index 7e0eda1cec..11693ca4bf 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -20,18 +20,20 @@ import ( // represents the main PHP thread // the thread needs to keep running as long as all other threads are running type phpMainThread struct { - state *state.ThreadState - done chan struct{} - numThreads int - maxThreads int - phpIni map[string]string - isRebooting atomic.Bool + state *state.ThreadState + done chan struct{} + numThreads int + maxThreads int + phpIni map[string]string + isRebooting atomic.Bool regularRequestChan chan *frankenPHPContext + workers []*worker } var ( phpThreads []*phpThread mainThread *phpMainThread + oldMainThread *phpMainThread commonHeaders map[string]*C.zend_string // timeouts to wait for threads to yield before arming force-kill @@ -43,12 +45,13 @@ var ( // a fixed number of inactive PHP threads // and reserves a fixed number of possible PHP threads func initPHPThreads(numThreads int, numMaxThreads int, phpIni map[string]string) (*phpMainThread, error) { + oldMainThread = mainThread mainThread = &phpMainThread{ - state: state.NewThreadState(), - done: make(chan struct{}), - numThreads: numThreads, - maxThreads: numMaxThreads, - phpIni: phpIni, + state: state.NewThreadState(), + done: make(chan struct{}), + numThreads: numThreads, + maxThreads: numMaxThreads, + phpIni: phpIni, regularRequestChan: make(chan *frankenPHPContext), } @@ -247,6 +250,21 @@ func getInactivePHPThread() *phpThread { return nil } +func (mainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread) { + go func(oldMainThread *phpMainThread) { + for { + select { + case <-mainThread.done: + return + case <-time.After(5 * time.Second): + return + case fc := <-oldMainThread.regularRequestChan: + mainThread.regularRequestChan <- fc + } + } + }(oldMainThread) +} + //export go_frankenphp_main_thread_is_ready func go_frankenphp_main_thread_is_ready() { mainThread.setAutomaticMaxThreads() diff --git a/threadregular.go b/threadregular.go index fc2030bb5d..3cca6a9daf 100644 --- a/threadregular.go +++ b/threadregular.go @@ -31,8 +31,8 @@ var ( func convertToRegularThread(thread *phpThread, mThread *phpMainThread) { thread.setHandler(®ularThread{ - thread: thread, - state: thread.state, + thread: thread, + state: thread.state, requestChan: mThread.regularRequestChan, }) attachRegularThread(thread) diff --git a/worker.go b/worker.go index 4a596c2fb6..b6e1409854 100644 --- a/worker.go +++ b/worker.go @@ -55,6 +55,7 @@ func initWorkers(opts []workerOpt) error { ) workers = make([]*worker, 0, len(opts)) + mainThread.workers = workers workersByName = make(map[string]*worker, len(opts)) globalWorkersByPath = make(map[string]*worker, len(opts)) From af4bd876c6516f64def3f6ed7f8185bbcc2c43ae Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Thu, 8 Oct 2026 22:45:05 +0200 Subject: [PATCH 03/24] Moves the fallbackserver to the main thread --- context.go | 4 ++-- frankenphp.go | 2 +- options.go | 3 +++ phpmainthread.go | 3 +++ phpmainthread_test.go | 2 +- server.go | 26 +++++--------------------- server_test.go | 4 +--- threadregular.go | 2 +- 8 files changed, 17 insertions(+), 29 deletions(-) diff --git a/context.go b/context.go index bed3b36e14..cbf3310e83 100644 --- a/context.go +++ b/context.go @@ -135,7 +135,7 @@ func newWorkerDummyContext(w *worker) (*frankenPHPContext, error) { server := w.server if server == nil { // global worker, not associated with a server - server = fallbackServer + server = mainThread.fallbackServer } fc := &frankenPHPContext{ @@ -164,7 +164,7 @@ func newWorkerDummyContext(w *worker) (*frankenPHPContext, error) { func newContextFromMessage(message any, rw http.ResponseWriter, ctx context.Context, w *worker) *frankenPHPContext { server := w.server if server == nil { - server = fallbackServer + server = mainThread.fallbackServer } if ctx == nil { diff --git a/frankenphp.go b/frankenphp.go index 904432e651..f48d659fd3 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -526,7 +526,7 @@ func ServeHTTP(responseWriter http.ResponseWriter, request *http.Request) error return ErrInvalidRequest } - return fallbackServer.ServeHTTP(responseWriter, request, opts...) + return mainThread.fallbackServer.ServeHTTP(responseWriter, request, opts...) } //export go_ub_write diff --git a/options.go b/options.go index 5566ed41cf..d0a0b03d13 100644 --- a/options.go +++ b/options.go @@ -299,6 +299,9 @@ func withExtensionWorkers(w *extensionWorkers) WorkerOption { // After registering, it will be possible to call Server.ServeHTTP() func WithServer(s *Server) Option { return func(o *opt) error { + if s.isRegistered.Load() { + return fmt.Errorf("cannot Init() with the same server twice: %s", s.name) + } o.servers = append(o.servers, s) return nil diff --git a/phpmainthread.go b/phpmainthread.go index 11693ca4bf..c8f1ccd18f 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -28,6 +28,8 @@ type phpMainThread struct { isRebooting atomic.Bool regularRequestChan chan *frankenPHPContext workers []*worker + + fallbackServer *Server } var ( @@ -53,6 +55,7 @@ func initPHPThreads(numThreads int, numMaxThreads int, phpIni map[string]string) maxThreads: numMaxThreads, phpIni: phpIni, regularRequestChan: make(chan *frankenPHPContext), + fallbackServer: newFallbackServer(), } // initialize the first thread diff --git a/phpmainthread_test.go b/phpmainthread_test.go index 5a013a3ebf..c982b7c026 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -398,7 +398,7 @@ func TestContextAndLoggerMustNotBeNil(t *testing.T) { assert.NotNil(t, fc.ctx, "context is defined for message context") r := httptest.NewRequest("GET", "http://localhost/index.php", nil) - fc, _ = newContextFromRequest(r, nil, fallbackServer) + fc, _ = newContextFromRequest(r, nil, mainThread.fallbackServer) assert.NotNil(t, fc.logger, "logger is defined for request context") assert.NotNil(t, fc.ctx, "context is defined for request context") diff --git a/server.go b/server.go index 856bea51e7..e4aa63056d 100644 --- a/server.go +++ b/server.go @@ -23,7 +23,7 @@ type Server struct { workers []*worker workersByPath map[string]*worker workersWithRequestMatcher []*worker - mainThread *phpMainThread + regularRequestChan chan *frankenPHPContext // registered while FrankenPHP runs with this server; read by concurrent // ServeHTTP calls while Init()/Shutdown() flip it, hence atomic @@ -32,8 +32,7 @@ type Server struct { } var ( - servers []*Server - fallbackServer = newFallbackServer() + servers []*Server ) func newFallbackServer() *Server { @@ -42,7 +41,6 @@ func newFallbackServer() *Server { workersByPath: make(map[string]*worker), env: make(map[string]string), logger: globalLogger, - mainThread: mainThread, } return s @@ -53,8 +51,6 @@ func newFallbackServer() *Server { // servers do not accept requests yet at this point, see activateServers() func registerServers(newServers []*Server) { servers = newServers - fallbackServer.logger = globalLogger - fallbackServer.resetWorkers() for i, s := range servers { s.idx = i @@ -62,36 +58,24 @@ func registerServers(newServers []*Server) { if s.name == "" { s.name = "server_" + strconv.Itoa(i) } - s.resetWorkers() } } // activateServers lets registered servers accept requests // it runs once workers and threads are up, so a request cannot reach a server before them func activateServers() { - fallbackServer.isRegistered.Store(true) - fallbackServer.mainThread = mainThread + mainThread.fallbackServer.regularRequestChan = mainThread.regularRequestChan + mainThread.fallbackServer.isRegistered.Store(true) for _, s := range servers { + s.regularRequestChan = mainThread.regularRequestChan s.isRegistered.Store(true) - s.mainThread = mainThread } } func unregisterServers() { - fallbackServer.isRegistered.Store(false) - for _, server := range servers { - server.isRegistered.Store(false) - } servers = nil } -// resetWorkers drops the workers of a previous run; initWorkers() adds them back -func (s *Server) resetWorkers() { - s.workers = nil - s.workersByPath = make(map[string]*worker) - s.workersWithRequestMatcher = nil -} - // NewServer creates a Server that can be registered via WithServer(). // name is a human-readable identifier used to attribute workers, metrics // and logs to this server; when empty, it defaults to the index the diff --git a/server_test.go b/server_test.go index 3ad2032bbe..9805d592c6 100644 --- a/server_test.go +++ b/server_test.go @@ -204,9 +204,7 @@ func TestServer(t *testing.T) { frankenphp.Shutdown() - initServers(t, opts...) - assert.Equal(t, "requests:1", serverGet(t, server, "http://example.com/worker-with-counter.php")) - assert.Equal(t, "server_0", server.Name()) + assert.Error(t, frankenphp.Init(opts...)) }) t.Run("error_on_missing_registration", func(t *testing.T) { diff --git a/threadregular.go b/threadregular.go index 3cca6a9daf..1ae186d4ec 100644 --- a/threadregular.go +++ b/threadregular.go @@ -154,7 +154,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { for { select { - case fc.server.mainThread.regularRequestChan <- fc: + case fc.server.regularRequestChan <- fc: queuedRegularThreads.Add(-1) metrics.DequeuedRequest() From b64b7782fe555e5869c164e20a406b274994054c Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Thu, 8 Oct 2026 22:58:09 +0200 Subject: [PATCH 04/24] moves servers to current mainthread --- frankenphp.go | 7 +++---- phpmainthread.go | 2 ++ server.go | 32 ++++++++++---------------------- 3 files changed, 15 insertions(+), 26 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index f48d659fd3..ed49b3867b 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -392,7 +392,7 @@ func Init(options ...Option) error { maxIdleTime = opt.maxIdleTime } - registerServers(opt.servers) + ensureServersHaveNames(opt.servers) workerThreadCount, err := calculateMaxThreads(opt) if err != nil { @@ -452,7 +452,7 @@ func Init(options ...Option) error { initAutoScaling(mThread) // only now that the workers and threads are up may requests reach a server - activateServers() + registerServers(mainThread, opt.servers) if globalLogger.Enabled(globalCtx, slog.LevelInfo) { globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "FrankenPHP started 🐘", slog.String("php_version", Version().Version), slog.Int("total_threads", mainThread.numThreads), slog.Int("worker_threads", workerThreadCount), slog.Int("max_threads", mainThread.maxThreads), slog.Int("max_requests", maxRequestsPerThread)) @@ -500,7 +500,6 @@ func shutdown() { drainWatchers() drainPHPThreads() - unregisterServers() metrics.Shutdown() @@ -968,7 +967,7 @@ func resetGlobals() { workers = nil workersByName = nil globalWorkersByPath = nil - servers = nil + mainThread.servers = nil watcherIsEnabled = false maxIdleTime = defaultMaxIdleTime maxRequestsPerThread = 0 diff --git a/phpmainthread.go b/phpmainthread.go index c8f1ccd18f..2d4981ab56 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -30,6 +30,7 @@ type phpMainThread struct { workers []*worker fallbackServer *Server + servers []*Server } var ( @@ -56,6 +57,7 @@ func initPHPThreads(numThreads int, numMaxThreads int, phpIni map[string]string) phpIni: phpIni, regularRequestChan: make(chan *frankenPHPContext), fallbackServer: newFallbackServer(), + servers: []*Server{}, } // initialize the first thread diff --git a/server.go b/server.go index e4aa63056d..2421de422d 100644 --- a/server.go +++ b/server.go @@ -31,10 +31,6 @@ type Server struct { logger *slog.Logger } -var ( - servers []*Server -) - func newFallbackServer() *Server { s := &Server{ idx: -1, @@ -46,13 +42,9 @@ func newFallbackServer() *Server { return s } -// registerServers assigns the identity of every server and clears the workers of a previous run, -// so the same *Server can be passed to Init() again after a Shutdown() -// servers do not accept requests yet at this point, see activateServers() -func registerServers(newServers []*Server) { - servers = newServers - - for i, s := range servers { +// ensure that all servers have a name +func ensureServersHaveNames(newServers []*Server) { + for i, s := range newServers { s.idx = i s.name = s.configuredName if s.name == "" { @@ -61,21 +53,17 @@ func registerServers(newServers []*Server) { } } -// activateServers lets registered servers accept requests -// it runs once workers and threads are up, so a request cannot reach a server before them -func activateServers() { - mainThread.fallbackServer.regularRequestChan = mainThread.regularRequestChan - mainThread.fallbackServer.isRegistered.Store(true) - for _, s := range servers { - s.regularRequestChan = mainThread.regularRequestChan +// registerServers registers the servers with the main thread and lets them accept requests +func registerServers(mthread *phpMainThread, newServers []*Server) { + mthread.servers = newServers + mthread.fallbackServer.regularRequestChan = mthread.regularRequestChan + mthread.fallbackServer.isRegistered.Store(true) + for _, s := range mthread.servers { + s.regularRequestChan = mthread.regularRequestChan s.isRegistered.Store(true) } } -func unregisterServers() { - servers = nil -} - // NewServer creates a Server that can be registered via WithServer(). // name is a human-readable identifier used to attribute workers, metrics // and logs to this server; when empty, it defaults to the index the From 13015cb01b7f3951fe8edc0895397067455d0bf5 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Thu, 8 Oct 2026 23:35:24 +0200 Subject: [PATCH 05/24] makes worker requests survive reloads --- frankenphp.go | 3 ++- phpmainthread.go | 30 +++++++++++++++++++---- phpmainthread_test.go | 57 +++++++++++++++++++++++++++++++++++++++++++ worker.go | 13 +++++----- 4 files changed, 90 insertions(+), 13 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index ed49b3867b..a5862063dd 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -438,7 +438,8 @@ func Init(options ...Option) error { convertToRegularThread(getInactivePHPThread(), mThread) } - if err := initWorkers(opt.workers); err != nil { + mThread.workers, err = initWorkers(opt.workers) + if err != nil { shutdown() return err diff --git a/phpmainthread.go b/phpmainthread.go index 2d4981ab56..402c891f46 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -255,19 +255,39 @@ func getInactivePHPThread() *phpThread { return nil } -func (mainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread) { - go func(oldMainThread *phpMainThread) { +func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread) { + go func() { for { select { - case <-mainThread.done: + case <-newMainThread.done: return case <-time.After(5 * time.Second): return case fc := <-oldMainThread.regularRequestChan: - mainThread.regularRequestChan <- fc + newMainThread.regularRequestChan <- fc } } - }(oldMainThread) + }() + + globalLogger.Info("re-queuing requests for workers", "oldWorkers", oldMainThread.workers, "newWorkers", newMainThread.workers) + for _, oldWorker := range oldMainThread.workers { + for _, newWorker := range newMainThread.workers { + if oldWorker.name == newWorker.name { + go func() { + for { + select { + case <-newMainThread.done: + return + case <-time.After(5 * time.Second): + return + case fc := <-oldWorker.requestChan: + newWorker.requestChan <- fc + } + } + }() + } + } + } } //export go_frankenphp_main_thread_is_ready diff --git a/phpmainthread_test.go b/phpmainthread_test.go index c982b7c026..c922326e90 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -225,6 +225,63 @@ func TestQueuedRequestSurvivesReload(t *testing.T) { } } +func TestQueuedWorkerRequestSurvivesReload(t *testing.T) { + t.Cleanup(Shutdown) + + initOpts := []Option{ + WithNumThreads(1), + WithMaxThreads(2), + WithMaxWaitTime(5 * time.Second), + WithWorkers("worker-1", filepath.Join(testDataPath, "sleep.php"), 1), + } + assert.NoError(t, Init(initOpts...)) + + doRequest := func(query string) error { + r := httptest.NewRequest("GET", "http://localhost/sleep.php?"+query, nil) + w := httptest.NewRecorder() + req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) + if err != nil { + return err + } + + return ServeHTTP(w, req) + } + + // Occupy the single PHP thread so the next request has to wait in the queue. + started := make(chan struct{}) + go func() { + close(started) + _ = doRequest("sleep=800") + }() + <-started + time.Sleep(100 * time.Millisecond) + + queued := make(chan error, 1) + go func() { + queued <- doRequest("sleep=0") + }() + + // Wait until the second request is genuinely queued waiting for a thread. + deadline := time.Now().Add(2 * time.Second) + for mainThread.workers[0].queuedRequests.Load() == 0 { + if !assert.True(t, time.Now().Before(deadline), "request was never queued") { + return + } + time.Sleep(time.Millisecond) + } + + // Reload while the request is queued, exactly like the Caddy module does. + Shutdown() + assert.NoError(t, Init(initOpts...)) + + select { + case err := <-queued: + assert.NoError(t, err, "a queued request must not be rejected by a reload") + case <-time.After(15 * time.Second): + t.Fatal("the queued request never completed after the reload") + } +} + func TestFinishBootingAWorkerScript(t *testing.T) { setupGlobals(t) diff --git a/worker.go b/worker.go index b6e1409854..7c8082c3af 100644 --- a/worker.go +++ b/worker.go @@ -44,9 +44,9 @@ var ( startupFailChan chan error ) -func initWorkers(opts []workerOpt) error { +func initWorkers(opts []workerOpt) ([]*worker, error) { if len(opts) == 0 { - return nil + return nil, nil } var ( @@ -55,14 +55,13 @@ func initWorkers(opts []workerOpt) error { ) workers = make([]*worker, 0, len(opts)) - mainThread.workers = workers workersByName = make(map[string]*worker, len(opts)) globalWorkersByPath = make(map[string]*worker, len(opts)) for _, o := range opts { w, err := newWorker(o) if err != nil { - return err + return nil, err } totalThreadsToStart += w.num @@ -71,7 +70,7 @@ func initWorkers(opts []workerOpt) error { if w.server == nil { globalWorkersByPath[w.fileName] = w } else if err := w.server.addWorker(w); err != nil { - return err + return nil, err } } @@ -93,13 +92,13 @@ func initWorkers(opts []workerOpt) error { select { case err := <-startupFailChan: // at least 1 worker has failed, return an error - return fmt.Errorf("failed to initialize workers: %w", err) + return nil, fmt.Errorf("failed to initialize workers: %w", err) default: // all workers started successfully startupFailChan = nil } - return nil + return workers, nil } // resolveWorkerFile turns a declared filename into the path a worker runs, From d1aae96ce92fd1d1614bb6b0ea0343e1a2152750 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 9 Oct 2026 00:06:19 +0200 Subject: [PATCH 06/24] optimize the queue test --- phpmainthread.go | 1 - phpmainthread_test.go | 122 ++++++++++++++---------------------------- 2 files changed, 40 insertions(+), 83 deletions(-) diff --git a/phpmainthread.go b/phpmainthread.go index 402c891f46..c33bd4e4a7 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -269,7 +269,6 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread } }() - globalLogger.Info("re-queuing requests for workers", "oldWorkers", oldMainThread.workers, "newWorkers", newMainThread.workers) for _, oldWorker := range oldMainThread.workers { for _, newWorker := range newMainThread.workers { if oldWorker.name == newWorker.name { diff --git a/phpmainthread_test.go b/phpmainthread_test.go index c922326e90..28fa238e7c 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -6,6 +6,7 @@ import ( "net/http/httptest" "path/filepath" "runtime" + "strconv" "sync" "sync/atomic" "testing" @@ -169,114 +170,71 @@ func TestTransitionThreadsWhileDoingRequests(t *testing.T) { // Test https://github.com/php/frankenphp/issues/2469 // A request queued for a thread when the server reloads must be served // by the new threads, not rejected with ErrMaxWaitTimeExceeded. -func TestQueuedRequestSurvivesReload(t *testing.T) { - t.Cleanup(Shutdown) - - initOpts := []Option{ +func TestQueuedRequestSurvivesReload_module(t *testing.T) { + testQueuedRequestSurvivesReload(t, []Option{ WithNumThreads(1), WithMaxThreads(1), WithMaxWaitTime(5 * time.Second), - } - assert.NoError(t, Init(initOpts...)) - - doRequest := func(query string) error { - r := httptest.NewRequest("GET", "http://localhost/sleep.php?"+query, nil) - w := httptest.NewRecorder() - req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) - if err != nil { - return err - } - - return ServeHTTP(w, req) - } - - // Occupy the single PHP thread so the next request has to wait in the queue. - started := make(chan struct{}) - go func() { - close(started) - _ = doRequest("sleep=800") - }() - <-started - time.Sleep(100 * time.Millisecond) - - queued := make(chan error, 1) - go func() { - queued <- doRequest("sleep=0") - }() - - // Wait until the second request is genuinely queued waiting for a thread. - deadline := time.Now().Add(2 * time.Second) - for queuedRegularThreads.Load() == 0 { - if !assert.True(t, time.Now().Before(deadline), "request was never queued") { - return - } - time.Sleep(time.Millisecond) - } - - // Reload while the request is queued, exactly like the Caddy module does. - Shutdown() - assert.NoError(t, Init(initOpts...)) - - select { - case err := <-queued: - assert.NoError(t, err, "a queued request must not be rejected by a reload") - case <-time.After(15 * time.Second): - t.Fatal("the queued request never completed after the reload") - } + }) } -func TestQueuedWorkerRequestSurvivesReload(t *testing.T) { - t.Cleanup(Shutdown) - - initOpts := []Option{ +func TestQueuedRequestSurvivesReload_worker(t *testing.T) { + testQueuedRequestSurvivesReload(t, []Option{ WithNumThreads(1), WithMaxThreads(2), WithMaxWaitTime(5 * time.Second), WithWorkers("worker-1", filepath.Join(testDataPath, "sleep.php"), 1), - } - assert.NoError(t, Init(initOpts...)) + }) +} - doRequest := func(query string) error { +func testQueuedRequestSurvivesReload(t *testing.T, opts []Option) { + t.Cleanup(Shutdown) + assert.NoError(t, Init(opts...)) + wg := sync.WaitGroup{} + done := make(chan struct{}) + + doRequest := func(query string) { + wg.Add(1) r := httptest.NewRequest("GET", "http://localhost/sleep.php?"+query, nil) w := httptest.NewRecorder() req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) - if err != nil { - return err - } - return ServeHTTP(w, req) + assert.NoError(t, err, "request must not be rejected") + assert.NoError(t, ServeHTTP(w, req), "request must not be rejected") + wg.Done() } - // Occupy the single PHP thread so the next request has to wait in the queue. - started := make(chan struct{}) - go func() { - close(started) - _ = doRequest("sleep=800") - }() - <-started - time.Sleep(100 * time.Millisecond) - - queued := make(chan error, 1) - go func() { - queued <- doRequest("sleep=0") - }() + // Occupy a single PHP thread so the next request has to wait in the queue. + go doRequest("sleep=10") // Wait until the second request is genuinely queued waiting for a thread. - deadline := time.Now().Add(2 * time.Second) - for mainThread.workers[0].queuedRequests.Load() == 0 { - if !assert.True(t, time.Now().Before(deadline), "request was never queued") { - return + hasWorker := len(mainThread.workers) != 0 + maxTries := 2000 + for i := 0; i <= maxTries; i++ { + if hasWorker && mainThread.workers[0].queuedRequests.Load() != 0 { + break + } else if !hasWorker && queuedRegularThreads.Load() != 0 { + break } + + go doRequest("sleep=" + strconv.Itoa(i)) time.Sleep(time.Millisecond) + if !assert.NotEqual(t, i, maxTries, "request was never queued") { + return + } } // Reload while the request is queued, exactly like the Caddy module does. Shutdown() - assert.NoError(t, Init(initOpts...)) + assert.NoError(t, Init(opts...)) + + go func() { + wg.Wait() + close(done) + }() select { - case err := <-queued: - assert.NoError(t, err, "a queued request must not be rejected by a reload") + case <-done: case <-time.After(15 * time.Second): t.Fatal("the queued request never completed after the reload") } From 03cc9de2780a39c1f541619ba68be9bbcba787d5 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Sun, 11 Oct 2026 11:52:09 +0200 Subject: [PATCH 07/24] requeues requests only on no config change --- caddy/app.go | 15 +- caddy/reload_queue_test.go | 281 +++++++++++++++++++++++++++++++++++++ frankenphp.go | 8 +- options.go | 11 ++ phpmainthread.go | 32 +++-- phpmainthread_test.go | 51 +++++++ 6 files changed, 383 insertions(+), 15 deletions(-) create mode 100644 caddy/reload_queue_test.go diff --git a/caddy/app.go b/caddy/app.go index 8dd40490d6..1b03f08107 100644 --- a/caddy/app.go +++ b/caddy/app.go @@ -2,6 +2,7 @@ package caddy import ( "context" + "crypto/sha256" "errors" "fmt" "log/slog" @@ -159,7 +160,19 @@ func (f *FrankenPHPApp) collectOptions(repl *caddy.Replacer, keep bool) ([]frank return nil, err } - return append(opts, moduleOpts...), nil + opts = append(opts, moduleOpts...) + + // add a fingerprint of the current configuration to the options to detect a reload + if keep { + fingerprint := sha256.New() + fingerprint.Write(caddyconfig.JSON(f, nil)) + for _, m := range f.modules { + fingerprint.Write(caddyconfig.JSON(m, nil)) + } + opts = append(opts, frankenphp.WithFingerprint(fmt.Sprintf("%x", fingerprint.Sum(nil)))) + } + + return opts, nil } func (f *FrankenPHPApp) Start() error { diff --git a/caddy/reload_queue_test.go b/caddy/reload_queue_test.go new file mode 100644 index 0000000000..2f915edc12 --- /dev/null +++ b/caddy/reload_queue_test.go @@ -0,0 +1,281 @@ +package caddy_test + +import ( + "bytes" + "io" + "net/http" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/caddyserver/caddy/v2/caddytest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// https://github.com/php/frankenphp/issues/2469 — queued requests must survive +// a Caddy reload that restarts FrankenPHP with the same configuration. +func TestQueuedRequestSurvivesReload_module(t *testing.T) { + testQueuedRequestSurvivesReload(t, ` + { + skip_install_trust + admin localhost:2999 + http_port `+testPort+` + metrics + + frankenphp { + num_threads 1 + max_threads 1 + max_wait_time 5s + } + } + + localhost:`+testPort+` { + route { + root ../testdata + rewrite sleep.php + php + } + } + `) +} + +func TestQueuedRequestRejectedWhenFrankenPHPConfigChanges(t *testing.T) { + initial := ` + { + skip_install_trust + admin localhost:2999 + http_port ` + testPort + ` + metrics + + frankenphp { + num_threads 1 + max_threads 1 + max_wait_time 5s + } + } + + localhost:` + testPort + ` { + route { + root ../testdata + rewrite sleep.php + php + } + } + ` + changed := strings.Replace(initial, "max_threads 1", "max_threads 2", 1) + + tester := caddytest.NewTester(t) + initServer(t, tester, initial, "caddyfile") + + client := &http.Client{Timeout: 30 * time.Second} + baseURL := "http://localhost:" + testPort + "/sleep.php" + + go func() { + req, err := http.NewRequest(http.MethodGet, baseURL+"?sleep=1000", nil) + require.NoError(t, err) + resp, err := client.Do(req) + if err == nil { + _ = resp.Body.Close() + } + }() + + queuedDone := make(chan *http.Response, 1) + go func() { + req, err := http.NewRequest(http.MethodGet, baseURL+"?sleep=1", nil) + require.NoError(t, err) + resp, err := client.Do(req) + if err != nil { + queuedDone <- nil + return + } + queuedDone <- resp + }() + + require.Eventually(t, func() bool { + depth, err := requestQueueDepth(false) + return err == nil && depth > 0 + }, 2*time.Second, time.Millisecond, "request was never queued") + + reloadCaddyConfig(t, changed) + + select { + case resp := <-queuedDone: + require.NotNil(t, resp) + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + require.Contains(t, string(body), "configuration has changed on reload") + case <-time.After(5 * time.Second): + t.Fatal("queued request never completed after reload") + } +} + +func TestQueuedRequestSurvivesReload_worker(t *testing.T) { + testQueuedRequestSurvivesReload(t, ` + { + skip_install_trust + admin localhost:2999 + http_port `+testPort+` + metrics + + frankenphp { + num_threads 1 + max_threads 2 + max_wait_time 5s + worker ../testdata/sleep.php 1 + } + } + + localhost:`+testPort+` { + route { + root ../testdata + rewrite sleep.php + php + } + } + `) +} + +func testQueuedRequestSurvivesReload(t *testing.T, config string) { + tester := caddytest.NewTester(t) + initServer(t, tester, config, "caddyfile") + + client := &http.Client{Timeout: 30 * time.Second} + baseURL := "http://localhost:" + testPort + "/sleep.php" + workerMode := strings.Contains(config, "worker ../testdata/sleep.php") + + var wg sync.WaitGroup + done := make(chan struct{}) + + doRequest := func(query string) { + wg.Add(1) + go func() { + defer wg.Done() + req, err := http.NewRequest(http.MethodGet, baseURL+"?"+query, nil) + require.NoError(t, err) + resp, err := client.Do(req) + require.NoError(t, err, "request must not be rejected") + require.NoError(t, resp.Body.Close()) + require.Equal(t, http.StatusOK, resp.StatusCode, "request must not be rejected") + }() + } + + // Occupy the only PHP thread so the next request has to wait in the queue. + doRequest("sleep=10") + doRequest("sleep=10") + + maxTries := 2000 + queued := false + for i := 0; i <= maxTries; i++ { + depth, err := requestQueueDepth(workerMode) + require.NoError(t, err) + if depth > 0 { + queued = true + break + } + + doRequest("sleep=" + strconv.Itoa(i)) + time.Sleep(time.Millisecond) + if !assert.NotEqual(t, i, maxTries, "request was never queued") { + return + } + } + require.True(t, queued) + + // Reload while the request is queued, exactly like the Caddy module does. + reloadCaddyConfig(t, config) + + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("the queued request never completed after the reload") + } +} + +func reloadCaddyConfig(t *testing.T, config string) { + t.Helper() + + r, err := http.NewRequest(http.MethodPost, "http://localhost:2999/load?force=true", bytes.NewBufferString(config)) + require.NoError(t, err) + r.Header.Set("Content-Type", "text/caddyfile") + r.Header.Set("Cache-Control", "must-revalidate") + + resp, err := http.DefaultClient.Do(r) + require.NoError(t, err) + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + require.Equal(t, http.StatusOK, resp.StatusCode, "reload failed: %s", body) +} + +func requestQueueDepth(workerMode bool) (float64, error) { + resp, err := http.Get("http://localhost:2999/metrics") + if err != nil { + return 0, err + } + body, err := io.ReadAll(resp.Body) + if err != nil { + _ = resp.Body.Close() + return 0, err + } + if err := resp.Body.Close(); err != nil { + return 0, err + } + + text := string(body) + if workerMode { + return sumPrometheusGauge(text, "frankenphp_worker_queue_depth"), nil + } + + return prometheusGaugeValue(text, "frankenphp_queue_depth"), nil +} + +func prometheusGaugeValue(body, name string) float64 { + for line := range strings.SplitSeq(body, "\n") { + if strings.HasPrefix(line, "#") { + continue + } + if strings.HasPrefix(line, name+" ") { + fields := strings.Fields(line) + if len(fields) < 2 { + return 0 + } + v, err := strconv.ParseFloat(fields[1], 64) + if err != nil { + return 0 + } + return v + } + } + + return 0 +} + +func sumPrometheusGauge(body, prefix string) float64 { + var total float64 + for line := range strings.SplitSeq(body, "\n") { + if strings.HasPrefix(line, "#") || !strings.HasPrefix(line, prefix) { + continue + } + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + v, err := strconv.ParseFloat(fields[len(fields)-1], 64) + if err != nil { + continue + } + total += v + } + + return total +} diff --git a/frankenphp.go b/frankenphp.go index a5862063dd..0771e635b7 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -51,9 +51,10 @@ var ( ErrScriptExecution = errors.New("error during PHP script execution") ErrNotRunning = errors.New("server is not registered, you must first call frankenphp.Init() with the WithServer() option") - ErrInvalidRequestPath = ErrRejected{"invalid request path", http.StatusBadRequest} - ErrInvalidContentLengthHeader = ErrRejected{"invalid Content-Length header", http.StatusBadRequest} - ErrMaxWaitTimeExceeded = ErrRejected{"maximum request handling time exceeded", http.StatusServiceUnavailable} + ErrInvalidRequestPath = ErrRejected{"invalid request path", http.StatusBadRequest} + ErrInvalidContentLengthHeader = ErrRejected{"invalid Content-Length header", http.StatusBadRequest} + ErrMaxWaitTimeExceeded = ErrRejected{"maximum request handling time exceeded", http.StatusServiceUnavailable} + ErrConfigurationChangedOnReload = ErrRejected{"configuration has changed on reload", http.StatusServiceUnavailable} contextKey = contextKeyStruct{} serverHeader = []string{"FrankenPHP"} @@ -474,6 +475,7 @@ func Init(options ...Option) error { } } + mainThread.fingerprint = opt.fingerprint if oldMainThread != nil { mainThread.reQueueRequests(oldMainThread) } diff --git a/options.go b/options.go index d0a0b03d13..b8c09fe7f6 100644 --- a/options.go +++ b/options.go @@ -37,6 +37,7 @@ type opt struct { maxIdleTime time.Duration maxRequests int servers []*Server + fingerprint string } type workerOpt struct { @@ -176,6 +177,16 @@ func WithMaxIdleTime(maxIdleTime time.Duration) Option { } } +// WithFingerprint sets a fingerprint of the FrankenPHP configuration. +// This determines if requests should be queued on reload. +func WithFingerprint(fingerprint string) Option { + return func(o *opt) error { + o.fingerprint = fingerprint + + return nil + } +} + // EXPERIMENTAL: WithMaxRequests sets the default max requests before restarting a PHP thread (0 = unlimited). Applies to regular and worker threads. func WithMaxRequests(maxRequests int) Option { return func(o *opt) error { diff --git a/phpmainthread.go b/phpmainthread.go index c33bd4e4a7..b75a58e5ea 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -20,17 +20,18 @@ import ( // represents the main PHP thread // the thread needs to keep running as long as all other threads are running type phpMainThread struct { - state *state.ThreadState - done chan struct{} - numThreads int - maxThreads int - phpIni map[string]string - isRebooting atomic.Bool + state *state.ThreadState + done chan struct{} + numThreads int + maxThreads int + phpIni map[string]string + isRebooting atomic.Bool + fingerprint string + regularRequestChan chan *frankenPHPContext workers []*worker - - fallbackServer *Server - servers []*Server + fallbackServer *Server + servers []*Server } var ( @@ -256,6 +257,7 @@ func getInactivePHPThread() *phpThread { } func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread) { + isSameConfiguration := oldMainThread.fingerprint != "" && oldMainThread.fingerprint == newMainThread.fingerprint go func() { for { select { @@ -264,7 +266,11 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread case <-time.After(5 * time.Second): return case fc := <-oldMainThread.regularRequestChan: - newMainThread.regularRequestChan <- fc + if isSameConfiguration { + newMainThread.regularRequestChan <- fc + } else { + fc.reject(ErrConfigurationChangedOnReload) + } } } }() @@ -280,7 +286,11 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread case <-time.After(5 * time.Second): return case fc := <-oldWorker.requestChan: - newWorker.requestChan <- fc + if isSameConfiguration { + newWorker.requestChan <- fc + } else { + fc.reject(ErrConfigurationChangedOnReload) + } } } }() diff --git a/phpmainthread_test.go b/phpmainthread_test.go index 28fa238e7c..de2988980f 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -3,6 +3,7 @@ package frankenphp import ( "io" "math/rand/v2" + "net/http" "net/http/httptest" "path/filepath" "runtime" @@ -14,6 +15,7 @@ import ( "github.com/dunglas/frankenphp/internal/state" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) var testDataPath, _ = filepath.Abs("./testdata") @@ -175,6 +177,7 @@ func TestQueuedRequestSurvivesReload_module(t *testing.T) { WithNumThreads(1), WithMaxThreads(1), WithMaxWaitTime(5 * time.Second), + WithFingerprint("test"), }) } @@ -184,6 +187,7 @@ func TestQueuedRequestSurvivesReload_worker(t *testing.T) { WithMaxThreads(2), WithMaxWaitTime(5 * time.Second), WithWorkers("worker-1", filepath.Join(testDataPath, "sleep.php"), 1), + WithFingerprint("test"), }) } @@ -240,6 +244,53 @@ func testQueuedRequestSurvivesReload(t *testing.T, opts []Option) { } } +func TestQueuedRequestRejectedWhenConfigChangesOnReload(t *testing.T) { + t.Cleanup(Shutdown) + + base := []Option{ + WithNumThreads(1), + WithMaxThreads(1), + WithMaxWaitTime(5 * time.Second), + } + assert.NoError(t, Init(append(base, WithFingerprint("before"))...)) + + go serveSleepRequest(t, "sleep=10000") + + queuedStatus := make(chan int, 1) + go func() { + r := httptest.NewRequest("GET", "http://localhost/sleep.php?sleep=1", nil) + w := httptest.NewRecorder() + req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) + require.NoError(t, err) + require.NoError(t, ServeHTTP(w, req)) + queuedStatus <- w.Result().StatusCode + }() + + require.Eventually(t, func() bool { + return queuedRegularThreads.Load() != 0 + }, 2*time.Second, time.Millisecond, "request was never queued") + + Shutdown() + assert.NoError(t, Init(append(base, WithFingerprint("after"))...)) + + select { + case status := <-queuedStatus: + assert.Equal(t, http.StatusServiceUnavailable, status) + case <-time.After(5 * time.Second): + t.Fatal("queued request never completed after reload") + } +} + +func serveSleepRequest(t *testing.T, query string) { + t.Helper() + + r := httptest.NewRequest("GET", "http://localhost/sleep.php?"+query, nil) + w := httptest.NewRecorder() + req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) + require.NoError(t, err) + require.NoError(t, ServeHTTP(w, req)) +} + func TestFinishBootingAWorkerScript(t *testing.T) { setupGlobals(t) From e8bcb9a1126f57f1315f747d05ba74902f608b7a Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Sun, 11 Oct 2026 18:44:53 +0200 Subject: [PATCH 08/24] also rejects if no worker is found --- phpmainthread.go | 42 +++++++++++++++++++++++------------------- 1 file changed, 23 insertions(+), 19 deletions(-) diff --git a/phpmainthread.go b/phpmainthread.go index b75a58e5ea..1ac04ef1c3 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -263,7 +263,7 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread select { case <-newMainThread.done: return - case <-time.After(5 * time.Second): + case <-time.After(15 * time.Second): return case fc := <-oldMainThread.regularRequestChan: if isSameConfiguration { @@ -276,26 +276,30 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread }() for _, oldWorker := range oldMainThread.workers { - for _, newWorker := range newMainThread.workers { - if oldWorker.name == newWorker.name { - go func() { - for { - select { - case <-newMainThread.done: - return - case <-time.After(5 * time.Second): - return - case fc := <-oldWorker.requestChan: - if isSameConfiguration { - newWorker.requestChan <- fc - } else { - fc.reject(ErrConfigurationChangedOnReload) - } - } - } - }() + var newWorker *worker + for _, w := range newMainThread.workers { + if oldWorker.name == w.name { + newWorker = w + break } } + + go func() { + for { + select { + case <-newMainThread.done: + return + case <-time.After(15 * time.Second): + return + case fc := <-oldWorker.requestChan: + if newWorker && isSameConfiguration { + newWorker.requestChan <- fc + } else { + fc.reject(ErrConfigurationChangedOnReload) + } + } + } + }() } } From 1fc1286c059635e61b329246ac84ef0de7e9bb63 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Sun, 11 Oct 2026 18:57:45 +0200 Subject: [PATCH 09/24] fixes tests --- phpmainthread.go | 25 +++++++++++++------ phpmainthread_test.go | 58 +++++++------------------------------------ 2 files changed, 27 insertions(+), 56 deletions(-) diff --git a/phpmainthread.go b/phpmainthread.go index 1ac04ef1c3..b920cb7562 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -258,16 +258,24 @@ func getInactivePHPThread() *phpThread { func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread) { isSameConfiguration := oldMainThread.fingerprint != "" && oldMainThread.fingerprint == newMainThread.fingerprint + hardReloadTimeout := time.Duration(maxWaitTime.Load()) * time.Second + if hardReloadTimeout == 0 { + hardReloadTimeout = 15 * time.Second + } + go func() { for { select { case <-newMainThread.done: return - case <-time.After(15 * time.Second): - return case fc := <-oldMainThread.regularRequestChan: if isSameConfiguration { - newMainThread.regularRequestChan <- fc + select { + case newMainThread.regularRequestChan <- fc: + break + case <-time.After(hardReloadTimeout): + fc.reject(ErrConfigurationChangedOnReload) + } } else { fc.reject(ErrConfigurationChangedOnReload) } @@ -289,11 +297,14 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread select { case <-newMainThread.done: return - case <-time.After(15 * time.Second): - return case fc := <-oldWorker.requestChan: - if newWorker && isSameConfiguration { - newWorker.requestChan <- fc + if newWorker != nil && isSameConfiguration { + select { + case newWorker.requestChan <- fc: + break + case <-time.After(hardReloadTimeout): + fc.reject(ErrConfigurationChangedOnReload) + } } else { fc.reject(ErrConfigurationChangedOnReload) } diff --git a/phpmainthread_test.go b/phpmainthread_test.go index de2988980f..aef490bd3a 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -3,7 +3,6 @@ package frankenphp import ( "io" "math/rand/v2" - "net/http" "net/http/httptest" "path/filepath" "runtime" @@ -213,20 +212,18 @@ func testQueuedRequestSurvivesReload(t *testing.T, opts []Option) { // Wait until the second request is genuinely queued waiting for a thread. hasWorker := len(mainThread.workers) != 0 - maxTries := 2000 - for i := 0; i <= maxTries; i++ { + sleepMilliseconds := 0 + require.Eventually(t, func() bool { if hasWorker && mainThread.workers[0].queuedRequests.Load() != 0 { - break - } else if !hasWorker && queuedRegularThreads.Load() != 0 { - break + return true } - - go doRequest("sleep=" + strconv.Itoa(i)) - time.Sleep(time.Millisecond) - if !assert.NotEqual(t, i, maxTries, "request was never queued") { - return + if !hasWorker && queuedRegularThreads.Load() != 0 { + return true } - } + go doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) + sleepMilliseconds++ + return false + }, 2*time.Second, time.Millisecond, "request was never queued") // Reload while the request is queued, exactly like the Caddy module does. Shutdown() @@ -244,43 +241,6 @@ func testQueuedRequestSurvivesReload(t *testing.T, opts []Option) { } } -func TestQueuedRequestRejectedWhenConfigChangesOnReload(t *testing.T) { - t.Cleanup(Shutdown) - - base := []Option{ - WithNumThreads(1), - WithMaxThreads(1), - WithMaxWaitTime(5 * time.Second), - } - assert.NoError(t, Init(append(base, WithFingerprint("before"))...)) - - go serveSleepRequest(t, "sleep=10000") - - queuedStatus := make(chan int, 1) - go func() { - r := httptest.NewRequest("GET", "http://localhost/sleep.php?sleep=1", nil) - w := httptest.NewRecorder() - req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) - require.NoError(t, err) - require.NoError(t, ServeHTTP(w, req)) - queuedStatus <- w.Result().StatusCode - }() - - require.Eventually(t, func() bool { - return queuedRegularThreads.Load() != 0 - }, 2*time.Second, time.Millisecond, "request was never queued") - - Shutdown() - assert.NoError(t, Init(append(base, WithFingerprint("after"))...)) - - select { - case status := <-queuedStatus: - assert.Equal(t, http.StatusServiceUnavailable, status) - case <-time.After(5 * time.Second): - t.Fatal("queued request never completed after reload") - } -} - func serveSleepRequest(t *testing.T, query string) { t.Helper() From 46e3f19a9f8a00f2788a862cb2351d1f5b1918f5 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Sun, 11 Oct 2026 19:00:03 +0200 Subject: [PATCH 10/24] removes reload test on the FrankenPHP side and only keeps them on the Caddy side --- phpmainthread_test.go | 83 ------------------------------------------- 1 file changed, 83 deletions(-) diff --git a/phpmainthread_test.go b/phpmainthread_test.go index aef490bd3a..a5967ad707 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -168,89 +168,6 @@ func TestTransitionThreadsWhileDoingRequests(t *testing.T) { transitionsWG.Wait() } -// Test https://github.com/php/frankenphp/issues/2469 -// A request queued for a thread when the server reloads must be served -// by the new threads, not rejected with ErrMaxWaitTimeExceeded. -func TestQueuedRequestSurvivesReload_module(t *testing.T) { - testQueuedRequestSurvivesReload(t, []Option{ - WithNumThreads(1), - WithMaxThreads(1), - WithMaxWaitTime(5 * time.Second), - WithFingerprint("test"), - }) -} - -func TestQueuedRequestSurvivesReload_worker(t *testing.T) { - testQueuedRequestSurvivesReload(t, []Option{ - WithNumThreads(1), - WithMaxThreads(2), - WithMaxWaitTime(5 * time.Second), - WithWorkers("worker-1", filepath.Join(testDataPath, "sleep.php"), 1), - WithFingerprint("test"), - }) -} - -func testQueuedRequestSurvivesReload(t *testing.T, opts []Option) { - t.Cleanup(Shutdown) - assert.NoError(t, Init(opts...)) - wg := sync.WaitGroup{} - done := make(chan struct{}) - - doRequest := func(query string) { - wg.Add(1) - r := httptest.NewRequest("GET", "http://localhost/sleep.php?"+query, nil) - w := httptest.NewRecorder() - req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) - - assert.NoError(t, err, "request must not be rejected") - assert.NoError(t, ServeHTTP(w, req), "request must not be rejected") - wg.Done() - } - - // Occupy a single PHP thread so the next request has to wait in the queue. - go doRequest("sleep=10") - - // Wait until the second request is genuinely queued waiting for a thread. - hasWorker := len(mainThread.workers) != 0 - sleepMilliseconds := 0 - require.Eventually(t, func() bool { - if hasWorker && mainThread.workers[0].queuedRequests.Load() != 0 { - return true - } - if !hasWorker && queuedRegularThreads.Load() != 0 { - return true - } - go doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) - sleepMilliseconds++ - return false - }, 2*time.Second, time.Millisecond, "request was never queued") - - // Reload while the request is queued, exactly like the Caddy module does. - Shutdown() - assert.NoError(t, Init(opts...)) - - go func() { - wg.Wait() - close(done) - }() - - select { - case <-done: - case <-time.After(15 * time.Second): - t.Fatal("the queued request never completed after the reload") - } -} - -func serveSleepRequest(t *testing.T, query string) { - t.Helper() - - r := httptest.NewRequest("GET", "http://localhost/sleep.php?"+query, nil) - w := httptest.NewRecorder() - req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false)) - require.NoError(t, err) - require.NoError(t, ServeHTTP(w, req)) -} - func TestFinishBootingAWorkerScript(t *testing.T) { setupGlobals(t) From 0dc8cf482faec53a65ffb5c75491ea4ebda22814 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Sun, 11 Oct 2026 19:05:51 +0200 Subject: [PATCH 11/24] fixes tests --- phpmainthread_test.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/phpmainthread_test.go b/phpmainthread_test.go index a5967ad707..377ed419d6 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -6,7 +6,6 @@ import ( "net/http/httptest" "path/filepath" "runtime" - "strconv" "sync" "sync/atomic" "testing" @@ -14,7 +13,6 @@ import ( "github.com/dunglas/frankenphp/internal/state" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) var testDataPath, _ = filepath.Abs("./testdata") From 88820ad93bc2d5103d43051b641a9b520ca637c6 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Sun, 11 Oct 2026 19:10:20 +0200 Subject: [PATCH 12/24] catches nil mainThreads --- frankenphp.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/frankenphp.go b/frankenphp.go index 416e6617ae..1869c7218d 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -532,6 +532,10 @@ func ServeHTTP(responseWriter http.ResponseWriter, request *http.Request) error return ErrInvalidRequest } + if mainThread == nil { + return ErrNotRunning + } + return mainThread.fallbackServer.ServeHTTP(responseWriter, request, opts...) } From ee4308706cff38d27fdee71444ec1b1938d00cf3 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Sun, 11 Oct 2026 19:31:22 +0200 Subject: [PATCH 13/24] makes reject test more consistent --- caddy/reload_queue_test.go | 63 +++++++++++++++----------------------- 1 file changed, 24 insertions(+), 39 deletions(-) diff --git a/caddy/reload_queue_test.go b/caddy/reload_queue_test.go index 2f915edc12..c70ade69c1 100644 --- a/caddy/reload_queue_test.go +++ b/caddy/reload_queue_test.go @@ -7,6 +7,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -43,6 +44,7 @@ func TestQueuedRequestSurvivesReload_module(t *testing.T) { } func TestQueuedRequestRejectedWhenFrankenPHPConfigChanges(t *testing.T) { + memoryLimit := 5000000 // ~5MB initial := ` { skip_install_trust @@ -53,7 +55,7 @@ func TestQueuedRequestRejectedWhenFrankenPHPConfigChanges(t *testing.T) { frankenphp { num_threads 1 max_threads 1 - max_wait_time 5s + php_ini memory_limit ` + strconv.Itoa(memoryLimit) + ` } } @@ -65,53 +67,36 @@ func TestQueuedRequestRejectedWhenFrankenPHPConfigChanges(t *testing.T) { } } ` - changed := strings.Replace(initial, "max_threads 1", "max_threads 2", 1) tester := caddytest.NewTester(t) initServer(t, tester, initial, "caddyfile") client := &http.Client{Timeout: 30 * time.Second} baseURL := "http://localhost:" + testPort + "/sleep.php" + requestWasRejected := atomic.Bool{} - go func() { - req, err := http.NewRequest(http.MethodGet, baseURL+"?sleep=1000", nil) - require.NoError(t, err) - resp, err := client.Do(req) - if err == nil { - _ = resp.Body.Close() - } - }() - - queuedDone := make(chan *http.Response, 1) - go func() { - req, err := http.NewRequest(http.MethodGet, baseURL+"?sleep=1", nil) - require.NoError(t, err) - resp, err := client.Do(req) - if err != nil { - queuedDone <- nil - return - } - queuedDone <- resp - }() + doRequest := func(query string) { + go func() { + req, err := http.NewRequest(http.MethodGet, baseURL+"?"+query, nil) + require.NoError(t, err) + resp, err := client.Do(req) + require.NoError(t, err, "request must not be rejected") + require.NoError(t, resp.Body.Close()) + if resp.StatusCode != http.StatusOK { + requestWasRejected.Store(true) + } + }() + } + sleepMilliseconds := 0 require.Eventually(t, func() bool { - depth, err := requestQueueDepth(false) - return err == nil && depth > 0 - }, 2*time.Second, time.Millisecond, "request was never queued") - - reloadCaddyConfig(t, changed) - - select { - case resp := <-queuedDone: - require.NotNil(t, resp) - body, err := io.ReadAll(resp.Body) - require.NoError(t, err) - require.NoError(t, resp.Body.Close()) - require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) - require.Contains(t, string(body), "configuration has changed on reload") - case <-time.After(5 * time.Second): - t.Fatal("queued request never completed after reload") - } + sleepMilliseconds++ + changedCaddyConfig := strings.Replace(initial, strconv.Itoa(memoryLimit), strconv.Itoa(memoryLimit+sleepMilliseconds), 1) + doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) + doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) + reloadCaddyConfig(t, changedCaddyConfig) + return requestWasRejected.Load() + }, 10*time.Second, time.Millisecond, "no queued request was ever rejected") } func TestQueuedRequestSurvivesReload_worker(t *testing.T) { From 58b13c8cdebe4fb26aad002babc4857d88e8256b Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 00:11:18 +0200 Subject: [PATCH 14/24] cleanup --- caddy/app.go | 35 ++++++++---- caddy/reload_queue_test.go | 112 ++++++++++++++++++------------------- frankenphp.go | 5 +- options.go | 31 +++++----- phpmainthread.go | 10 ++-- 5 files changed, 102 insertions(+), 91 deletions(-) diff --git a/caddy/app.go b/caddy/app.go index 1b347e7a26..45fcf748e1 100644 --- a/caddy/app.go +++ b/caddy/app.go @@ -1,8 +1,8 @@ package caddy import ( + "bytes" "context" - "crypto/sha256" "errors" "fmt" "log/slog" @@ -70,6 +70,7 @@ type FrankenPHPApp struct { httpApp *caddyhttp.App hasStarted atomic.Bool started chan any + shouldRequeue bool } var errIni = errors.New(`"php_ini" must be in the format: php_ini "" ""`) @@ -105,6 +106,27 @@ func (f *FrankenPHPApp) Provision(ctx caddy.Context) error { } } + // if FrankenPHP was running previously and the configuration has not changed, instruct request re-queueing + if old := caddy.ActiveContext(); old.Context != nil { + if prev, err := old.AppIfConfigured("frankenphp"); err == nil && prev != nil { + if p, ok := prev.(*FrankenPHPApp); ok { + hasChanged := false + if !bytes.Equal(caddyconfig.JSON(p, nil), caddyconfig.JSON(f, nil)) { + hasChanged = true + } + for i, oldModule := range p.modules { + if i >= len(f.modules) { + break + } + if !bytes.Equal(caddyconfig.JSON(oldModule, nil), caddyconfig.JSON(f.modules[i], nil)) { + break + } + } + f.shouldRequeue = !hasChanged + } + } + } + return nil } @@ -140,6 +162,7 @@ func (f *FrankenPHPApp) collectOptions(repl *caddy.Replacer, keep bool) ([]frank frankenphp.WithMaxWaitTime(f.MaxWaitTime), frankenphp.WithMaxIdleTime(f.MaxIdleTime), frankenphp.WithMaxRequests(f.MaxRequests), + frankenphp.WithRequestRequeueOnReload(f.shouldRequeue), ) usedWorkerNames := make(map[string]bool, len(f.Workers)) @@ -162,16 +185,6 @@ func (f *FrankenPHPApp) collectOptions(repl *caddy.Replacer, keep bool) ([]frank opts = append(opts, moduleOpts...) - // add a fingerprint of the current configuration to the options to detect a reload - if keep { - fingerprint := sha256.New() - fingerprint.Write(caddyconfig.JSON(f, nil)) - for _, m := range f.modules { - fingerprint.Write(caddyconfig.JSON(m, nil)) - } - opts = append(opts, frankenphp.WithFingerprint(fmt.Sprintf("%x", fingerprint.Sum(nil)))) - } - return opts, nil } diff --git a/caddy/reload_queue_test.go b/caddy/reload_queue_test.go index c70ade69c1..d1fd82c39f 100644 --- a/caddy/reload_queue_test.go +++ b/caddy/reload_queue_test.go @@ -43,62 +43,6 @@ func TestQueuedRequestSurvivesReload_module(t *testing.T) { `) } -func TestQueuedRequestRejectedWhenFrankenPHPConfigChanges(t *testing.T) { - memoryLimit := 5000000 // ~5MB - initial := ` - { - skip_install_trust - admin localhost:2999 - http_port ` + testPort + ` - metrics - - frankenphp { - num_threads 1 - max_threads 1 - php_ini memory_limit ` + strconv.Itoa(memoryLimit) + ` - } - } - - localhost:` + testPort + ` { - route { - root ../testdata - rewrite sleep.php - php - } - } - ` - - tester := caddytest.NewTester(t) - initServer(t, tester, initial, "caddyfile") - - client := &http.Client{Timeout: 30 * time.Second} - baseURL := "http://localhost:" + testPort + "/sleep.php" - requestWasRejected := atomic.Bool{} - - doRequest := func(query string) { - go func() { - req, err := http.NewRequest(http.MethodGet, baseURL+"?"+query, nil) - require.NoError(t, err) - resp, err := client.Do(req) - require.NoError(t, err, "request must not be rejected") - require.NoError(t, resp.Body.Close()) - if resp.StatusCode != http.StatusOK { - requestWasRejected.Store(true) - } - }() - } - - sleepMilliseconds := 0 - require.Eventually(t, func() bool { - sleepMilliseconds++ - changedCaddyConfig := strings.Replace(initial, strconv.Itoa(memoryLimit), strconv.Itoa(memoryLimit+sleepMilliseconds), 1) - doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) - doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) - reloadCaddyConfig(t, changedCaddyConfig) - return requestWasRejected.Load() - }, 10*time.Second, time.Millisecond, "no queued request was ever rejected") -} - func TestQueuedRequestSurvivesReload_worker(t *testing.T) { testQueuedRequestSurvivesReload(t, ` { @@ -186,6 +130,62 @@ func testQueuedRequestSurvivesReload(t *testing.T, config string) { } } +func TestQueuedRequestRejectedWhenFrankenPHPConfigChanges(t *testing.T) { + memoryLimit := 5000000 // ~5MB + initial := ` + { + skip_install_trust + admin localhost:2999 + http_port ` + testPort + ` + metrics + + frankenphp { + num_threads 1 + max_threads 1 + php_ini memory_limit ` + strconv.Itoa(memoryLimit) + ` + } + } + + localhost:` + testPort + ` { + route { + root ../testdata + rewrite sleep.php + php + } + } + ` + + tester := caddytest.NewTester(t) + initServer(t, tester, initial, "caddyfile") + + client := &http.Client{Timeout: 30 * time.Second} + baseURL := "http://localhost:" + testPort + "/sleep.php" + requestWasRejected := atomic.Bool{} + + doRequest := func(query string) { + go func() { + req, err := http.NewRequest(http.MethodGet, baseURL+"?"+query, nil) + require.NoError(t, err) + resp, err := client.Do(req) + require.NoError(t, err, "request must not be rejected") + require.NoError(t, resp.Body.Close()) + if resp.StatusCode != http.StatusOK { + requestWasRejected.Store(true) + } + }() + } + + sleepMilliseconds := 0 + require.Eventually(t, func() bool { + sleepMilliseconds++ + changedCaddyConfig := strings.Replace(initial, strconv.Itoa(memoryLimit), strconv.Itoa(memoryLimit+sleepMilliseconds), 1) + doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) + doRequest("sleep=" + strconv.Itoa(sleepMilliseconds)) + reloadCaddyConfig(t, changedCaddyConfig) + return requestWasRejected.Load() + }, 10*time.Second, time.Millisecond, "no queued request was ever rejected") +} + func reloadCaddyConfig(t *testing.T, config string) { t.Helper() diff --git a/frankenphp.go b/frankenphp.go index 1869c7218d..9802552679 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -479,9 +479,10 @@ func Init(options ...Option) error { } } - mainThread.fingerprint = opt.fingerprint + globalLogger.Warn("shouldRequeue", "shouldRequeue", opt.shouldRequeue) + if oldMainThread != nil { - mainThread.reQueueRequests(oldMainThread) + mainThread.handleOldRequests(oldMainThread, opt.shouldRequeue) } return nil diff --git a/options.go b/options.go index b8c09fe7f6..77f12bd074 100644 --- a/options.go +++ b/options.go @@ -26,18 +26,18 @@ type ServerOption func(*Server) error type opt struct { hotReloadOpt - ctx context.Context - numThreads int - maxThreads int - workers []workerOpt - logger *slog.Logger - metrics Metrics - phpIni map[string]string - maxWaitTime time.Duration - maxIdleTime time.Duration - maxRequests int - servers []*Server - fingerprint string + ctx context.Context + numThreads int + maxThreads int + workers []workerOpt + logger *slog.Logger + metrics Metrics + phpIni map[string]string + maxWaitTime time.Duration + maxIdleTime time.Duration + maxRequests int + servers []*Server + shouldRequeue bool } type workerOpt struct { @@ -177,11 +177,10 @@ func WithMaxIdleTime(maxIdleTime time.Duration) Option { } } -// WithFingerprint sets a fingerprint of the FrankenPHP configuration. -// This determines if requests should be queued on reload. -func WithFingerprint(fingerprint string) Option { +// WithRequestRequeueOnReload sets if requests should be queued on reload. +func WithRequestRequeueOnReload(shouldRequeue bool) Option { return func(o *opt) error { - o.fingerprint = fingerprint + o.shouldRequeue = shouldRequeue return nil } diff --git a/phpmainthread.go b/phpmainthread.go index 9a65cfbad2..9ac28fbc2d 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -26,7 +26,6 @@ type phpMainThread struct { maxThreads int phpIni map[string]string isRebooting atomic.Bool - fingerprint string regularRequestChan chan *frankenPHPContext workers []*worker @@ -256,9 +255,8 @@ func getInactivePHPThread() *phpThread { return nil } -func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread) { - isSameConfiguration := oldMainThread.fingerprint != "" && oldMainThread.fingerprint == newMainThread.fingerprint - hardReloadTimeout := time.Duration(maxWaitTime.Load()) * time.Second +func (newMainThread *phpMainThread) handleOldRequests(oldMainThread *phpMainThread, shouldRequeue bool) { + hardReloadTimeout := time.Duration(maxWaitTime.Load()) if hardReloadTimeout == 0 { hardReloadTimeout = 15 * time.Second } @@ -269,7 +267,7 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread case <-newMainThread.done: return case fc := <-oldMainThread.regularRequestChan: - if isSameConfiguration { + if shouldRequeue { select { case newMainThread.regularRequestChan <- fc: break @@ -298,7 +296,7 @@ func (newMainThread *phpMainThread) reQueueRequests(oldMainThread *phpMainThread case <-newMainThread.done: return case fc := <-oldWorker.requestChan: - if newWorker != nil && isSameConfiguration { + if newWorker != nil && shouldRequeue { select { case newWorker.requestChan <- fc: break From 354ac19fbb90897f61ca9975d9e5205d3707e310 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 00:15:45 +0200 Subject: [PATCH 15/24] unregister servers after new main thread is done --- phpmainthread.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/phpmainthread.go b/phpmainthread.go index 9ac28fbc2d..5ad8116225 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -265,6 +265,9 @@ func (newMainThread *phpMainThread) handleOldRequests(oldMainThread *phpMainThre for { select { case <-newMainThread.done: + for _, s := range oldMainThread.servers { + s.isRegistered.Store(false) // unregister servers from the previous generation if we have reloaded twice + } return case fc := <-oldMainThread.regularRequestChan: if shouldRequeue { From 24e9e0a3cda5aaf90aba2e3cf9574caadafd5a77 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 12:22:42 +0200 Subject: [PATCH 16/24] prevents races on workers --- cgi.go | 2 +- frankenphp.go | 7 +------ phpmainthread.go | 5 +++++ phpmainthread_test.go | 45 ++++++++++++++++++++++++------------------- requestoptions.go | 4 ++-- scaling_test.go | 2 +- server.go | 4 ++++ worker.go | 39 +++++++++++++++---------------------- 8 files changed, 55 insertions(+), 53 deletions(-) diff --git a/cgi.go b/cgi.go index 391174ceb4..5f7517a8e8 100644 --- a/cgi.go +++ b/cgi.go @@ -226,7 +226,7 @@ func splitCgiPath(fc *frankenPHPContext) { // see if a php_server worker or global worker matches the request path // aka: root + request path == worker.filename if fc.worker = fc.server.workersByPath[fc.scriptFilename]; fc.worker == nil { - fc.worker = globalWorkersByPath[fc.scriptFilename] + fc.worker = fc.server.mainThread.workersByPath[fc.scriptFilename] } } diff --git a/frankenphp.go b/frankenphp.go index 9802552679..b7cbf85169 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -443,8 +443,7 @@ func Init(options ...Option) error { convertToRegularThread(getInactivePHPThread(), mThread) } - mThread.workers, err = initWorkers(opt.workers) - if err != nil { + if err := initWorkers(opt.workers, mThread); err != nil { shutdown() return err @@ -972,10 +971,6 @@ func timeoutChan(timeout time.Duration) <-chan time.Time { func resetGlobals() { globalCtx = context.Background() globalLogger = slog.Default() - workers = nil - workersByName = nil - globalWorkersByPath = nil - mainThread.servers = nil watcherIsEnabled = false hotReloadEnabled = false maxIdleTime = defaultMaxIdleTime diff --git a/phpmainthread.go b/phpmainthread.go index 5ad8116225..4cfc61d337 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -29,6 +29,8 @@ type phpMainThread struct { regularRequestChan chan *frankenPHPContext workers []*worker + workersByName map[string]*worker + workersByPath map[string]*worker fallbackServer *Server servers []*Server } @@ -58,6 +60,9 @@ func initPHPThreads(numThreads int, numMaxThreads int, phpIni map[string]string) regularRequestChan: make(chan *frankenPHPContext), fallbackServer: newFallbackServer(), servers: []*Server{}, + workers: make([]*worker, 0, numThreads), + workersByName: make(map[string]*worker, numThreads), + workersByPath: make(map[string]*worker, numThreads), } // initialize the first thread diff --git a/phpmainthread_test.go b/phpmainthread_test.go index 377ed419d6..fa3d7b7b1e 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -191,44 +191,49 @@ func TestFinishBootingAWorkerScript(t *testing.T) { assert.Nil(t, phpThreads) } +func testWorkerMainThread() *phpMainThread { + return &phpMainThread{ + workers: []*worker{}, + workersByName: map[string]*worker{}, + workersByPath: map[string]*worker{}, + } +} + func TestReturnAnErrorIf2WorkersHaveTheSameFileName(t *testing.T) { resetGlobals() - workers = []*worker{} - workersByName = map[string]*worker{} - globalWorkersByPath = map[string]*worker{} - w, err1 := newWorker(workerOpt{fileName: testDataPath + "/index.php"}) + m := testWorkerMainThread() + w, err1 := newWorker(workerOpt{fileName: testDataPath + "/index.php"}, m) assert.NoError(t, err1) - workers = append(workers, w) - workersByName[w.name] = w - globalWorkersByPath[w.fileName] = w - _, err2 := newWorker(workerOpt{fileName: testDataPath + "/index.php"}) + m.workers = append(m.workers, w) + m.workersByName[w.name] = w + m.workersByPath[w.fileName] = w + _, err2 := newWorker(workerOpt{fileName: testDataPath + "/index.php"}, m) assert.Error(t, err2, "two workers cannot have the same filename") } func TestReturnAnErrorIf2ModuleWorkersHaveTheSameName(t *testing.T) { resetGlobals() - workers = []*worker{} - workersByName = map[string]*worker{} - w, err1 := newWorker(workerOpt{fileName: testDataPath + "/index.php", name: "workername"}) + m := testWorkerMainThread() + w, err1 := newWorker(workerOpt{fileName: testDataPath + "/index.php", name: "workername"}, m) assert.NoError(t, err1) - workers = append(workers, w) - workersByName[w.name] = w - _, err2 := newWorker(workerOpt{fileName: testDataPath + "/hello.php", name: "workername"}) + m.workers = append(m.workers, w) + m.workersByName[w.name] = w + _, err2 := newWorker(workerOpt{fileName: testDataPath + "/hello.php", name: "workername"}, m) assert.Error(t, err2, "two workers cannot have the same name") } func getDummyWorker(t *testing.T, fileName string) *worker { t.Helper() - if workers == nil { - workers = []*worker{} + if mainThread == nil || mainThread.workers == nil { + mainThread = testWorkerMainThread() } worker, _ := newWorker(workerOpt{ fileName: testDataPath + "/" + fileName, num: 1, - }) - workers = append(workers, worker) + }, mainThread) + mainThread.workers = append(mainThread.workers, worker) return worker } @@ -256,9 +261,9 @@ func allPossibleTransitions(worker1Path string, worker2Path string) []func(*phpT thread.boot() } }, - func(thread *phpThread) { convertToWorkerThread(thread, globalWorkersByPath[worker1Path]) }, + func(thread *phpThread) { convertToWorkerThread(thread, mainThread.workersByPath[worker1Path]) }, convertToInactiveThread, - func(thread *phpThread) { convertToWorkerThread(thread, globalWorkersByPath[worker2Path]) }, + func(thread *phpThread) { convertToWorkerThread(thread, mainThread.workersByPath[worker2Path]) }, convertToInactiveThread, } } diff --git a/requestoptions.go b/requestoptions.go index 962727562f..28c6205c16 100644 --- a/requestoptions.go +++ b/requestoptions.go @@ -208,8 +208,8 @@ func WithRequestBodyTimeout(timeout time.Duration) RequestOption { // WithWorkerName sets the worker that should handle the request func WithWorkerName(name string) RequestOption { return func(o *frankenPHPContext) error { - if name != "" { - o.worker = workersByName[name] + if name != "" && mainThread != nil { + o.worker = mainThread.workersByName[name] } return nil diff --git a/scaling_test.go b/scaling_test.go index 07437c7163..89bd378780 100644 --- a/scaling_test.go +++ b/scaling_test.go @@ -49,7 +49,7 @@ func TestScaleAWorkerThreadUpAndDown(t *testing.T) { autoScaledThread := phpThreads[2] // scale up - scaleWorkerThread(globalWorkersByPath[workerPath], mainThread) + scaleWorkerThread(mainThread.workersByPath[workerPath], mainThread) assert.Equal(t, state.Ready, autoScaledThread.state.Get()) // on down-scale, the thread will be marked as inactive diff --git a/server.go b/server.go index 2421de422d..0bf89af1a6 100644 --- a/server.go +++ b/server.go @@ -24,6 +24,7 @@ type Server struct { workersByPath map[string]*worker workersWithRequestMatcher []*worker regularRequestChan chan *frankenPHPContext + mainThread *phpMainThread // registered while FrankenPHP runs with this server; read by concurrent // ServeHTTP calls while Init()/Shutdown() flip it, hence atomic @@ -37,6 +38,7 @@ func newFallbackServer() *Server { workersByPath: make(map[string]*worker), env: make(map[string]string), logger: globalLogger, + mainThread: mainThread, } return s @@ -58,8 +60,10 @@ func registerServers(mthread *phpMainThread, newServers []*Server) { mthread.servers = newServers mthread.fallbackServer.regularRequestChan = mthread.regularRequestChan mthread.fallbackServer.isRegistered.Store(true) + mthread.fallbackServer.mainThread = mthread for _, s := range mthread.servers { s.regularRequestChan = mthread.regularRequestChan + s.mainThread = mthread s.isRegistered.Store(true) } } diff --git a/worker.go b/worker.go index 7c8082c3af..fd5ecbb473 100644 --- a/worker.go +++ b/worker.go @@ -37,16 +37,13 @@ type worker struct { } var ( - workers []*worker - workersByName map[string]*worker - globalWorkersByPath map[string]*worker - watcherIsEnabled bool - startupFailChan chan error + watcherIsEnabled bool + startupFailChan chan error ) -func initWorkers(opts []workerOpt) ([]*worker, error) { +func initWorkers(opts []workerOpt, m *phpMainThread) error { if len(opts) == 0 { - return nil, nil + return nil } var ( @@ -54,29 +51,25 @@ func initWorkers(opts []workerOpt) ([]*worker, error) { totalThreadsToStart int ) - workers = make([]*worker, 0, len(opts)) - workersByName = make(map[string]*worker, len(opts)) - globalWorkersByPath = make(map[string]*worker, len(opts)) - for _, o := range opts { - w, err := newWorker(o) + w, err := newWorker(o, m) if err != nil { - return nil, err + return err } totalThreadsToStart += w.num - workers = append(workers, w) - workersByName[w.name] = w + m.workers = append(m.workers, w) + m.workersByName[w.name] = w if w.server == nil { - globalWorkersByPath[w.fileName] = w + m.workersByPath[w.fileName] = w } else if err := w.server.addWorker(w); err != nil { - return nil, err + return err } } startupFailChan = make(chan error, totalThreadsToStart) - for _, w := range workers { + for _, w := range m.workers { for range w.num { thread := getInactivePHPThread() convertToWorkerThread(thread, w) @@ -92,13 +85,13 @@ func initWorkers(opts []workerOpt) ([]*worker, error) { select { case err := <-startupFailChan: // at least 1 worker has failed, return an error - return nil, fmt.Errorf("failed to initialize workers: %w", err) + return fmt.Errorf("failed to initialize workers: %w", err) default: // all workers started successfully startupFailChan = nil } - return workers, nil + return nil } // resolveWorkerFile turns a declared filename into the path a worker runs, @@ -155,16 +148,16 @@ func checkWorkerDeclaration(o workerOpt, nameTaken func(string) bool, pathTaken return nil } -func newWorker(o workerOpt) (*worker, error) { +func newWorker(o workerOpt, m *phpMainThread) (*worker, error) { o, err := resolveWorkerFile(o) if err != nil { return nil, err } - nameTaken := func(name string) bool { return workersByName[name] != nil } + nameTaken := func(name string) bool { return m.workersByName[name] != nil } pathTaken := func(server *Server, path string) bool { if server == nil { - return globalWorkersByPath[path] != nil + return m.workersByPath[path] != nil } return server.workersByPath[path] != nil From af6f0c63b45ae680cac8b6585c533a42c4638ced Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 12:35:11 +0200 Subject: [PATCH 17/24] moves more vars onto the main thread --- frankenphp.go | 8 ++------ phpmainthread.go | 22 ++++++++++++---------- phpmainthread_test.go | 8 ++++---- threadregular.go | 3 +-- types_test.go | 2 +- worker.go | 2 +- 6 files changed, 21 insertions(+), 24 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index b7cbf85169..d2ef01ca70 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -70,7 +70,6 @@ var ( metrics Metrics = nullMetrics{} // atomic: read by in-flight requests while a reload may rewrite it - maxWaitTime atomic.Int64 maxRequestsPerThread int hotReloadEnabled bool @@ -388,7 +387,6 @@ func Init(options ...Option) error { metrics = opt.metrics } - maxWaitTime.Store(int64(opt.maxWaitTime)) maxRequestsPerThread = opt.maxRequests if opt.maxIdleTime > 0 { @@ -432,7 +430,7 @@ func Init(options ...Option) error { hotReloadEnabled = opt.hasHotReload() - mThread, err := initPHPThreads(opt.numThreads, opt.maxThreads, opt.phpIni) + mThread, err := initPHPThreads(opt) if err != nil { shutdown() return err @@ -478,10 +476,8 @@ func Init(options ...Option) error { } } - globalLogger.Warn("shouldRequeue", "shouldRequeue", opt.shouldRequeue) - if oldMainThread != nil { - mainThread.handleOldRequests(oldMainThread, opt.shouldRequeue) + mThread.handleStaleRequests(oldMainThread, opt.shouldRequeue) } return nil diff --git a/phpmainthread.go b/phpmainthread.go index 4cfc61d337..d51fdda28a 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -26,6 +26,7 @@ type phpMainThread struct { maxThreads int phpIni map[string]string isRebooting atomic.Bool + maxWaitTime time.Duration regularRequestChan chan *frankenPHPContext workers []*worker @@ -49,20 +50,21 @@ var ( // initPHPThreads starts the main PHP thread, // a fixed number of inactive PHP threads // and reserves a fixed number of possible PHP threads -func initPHPThreads(numThreads int, numMaxThreads int, phpIni map[string]string) (*phpMainThread, error) { +func initPHPThreads(opt *opt) (*phpMainThread, error) { oldMainThread = mainThread mainThread = &phpMainThread{ state: state.NewThreadState(), done: make(chan struct{}), - numThreads: numThreads, - maxThreads: numMaxThreads, - phpIni: phpIni, + numThreads: opt.numThreads, + maxThreads: opt.maxThreads, + phpIni: opt.phpIni, regularRequestChan: make(chan *frankenPHPContext), fallbackServer: newFallbackServer(), servers: []*Server{}, - workers: make([]*worker, 0, numThreads), - workersByName: make(map[string]*worker, numThreads), - workersByPath: make(map[string]*worker, numThreads), + workers: make([]*worker, 0, opt.numThreads), + workersByName: make(map[string]*worker, opt.numThreads), + workersByPath: make(map[string]*worker, opt.numThreads), + maxWaitTime: opt.maxWaitTime, } // initialize the first thread @@ -90,7 +92,7 @@ func initPHPThreads(numThreads int, numMaxThreads int, phpIni map[string]string) // start the underlying C threads var ready sync.WaitGroup - for i := range numThreads { + for i := range mainThread.numThreads { ready.Go(phpThreads[i].boot) } @@ -260,8 +262,8 @@ func getInactivePHPThread() *phpThread { return nil } -func (newMainThread *phpMainThread) handleOldRequests(oldMainThread *phpMainThread, shouldRequeue bool) { - hardReloadTimeout := time.Duration(maxWaitTime.Load()) +func (newMainThread *phpMainThread) handleStaleRequests(oldMainThread *phpMainThread, shouldRequeue bool) { + hardReloadTimeout := oldMainThread.maxWaitTime if hardReloadTimeout == 0 { hardReloadTimeout = 15 * time.Second } diff --git a/phpmainthread_test.go b/phpmainthread_test.go index fa3d7b7b1e..396e75b92c 100644 --- a/phpmainthread_test.go +++ b/phpmainthread_test.go @@ -26,7 +26,7 @@ func setupGlobals(t *testing.T) { } func TestStartAndStopTheMainThreadWithOneInactiveThread(t *testing.T) { - _, err := initPHPThreads(1, 1, nil) // boot 1 thread + _, err := initPHPThreads(&opt{numThreads: 1, maxThreads: 1, phpIni: nil}) // boot 1 thread assert.NoError(t, err) assert.Len(t, phpThreads, 1) @@ -41,7 +41,7 @@ func TestStartAndStopTheMainThreadWithOneInactiveThread(t *testing.T) { func TestTransitionRegularThreadToWorkerThread(t *testing.T) { setupGlobals(t) - _, err := initPHPThreads(1, 1, nil) + _, err := initPHPThreads(&opt{numThreads: 1, maxThreads: 1, phpIni: nil}) assert.NoError(t, err) // transition to regular thread @@ -66,7 +66,7 @@ func TestTransitionRegularThreadToWorkerThread(t *testing.T) { func TestTransitionAThreadBetween2DifferentWorkers(t *testing.T) { setupGlobals(t) - _, err := initPHPThreads(1, 1, nil) + _, err := initPHPThreads(&opt{numThreads: 1, maxThreads: 1, phpIni: nil}) assert.NoError(t, err) firstWorker := getDummyWorker(t, "transition-worker-1.php") secondWorker := getDummyWorker(t, "transition-worker-2.php") @@ -169,7 +169,7 @@ func TestTransitionThreadsWhileDoingRequests(t *testing.T) { func TestFinishBootingAWorkerScript(t *testing.T) { setupGlobals(t) - _, err := initPHPThreads(1, 1, nil) + _, err := initPHPThreads(&opt{numThreads: 1, maxThreads: 1, phpIni: nil}) assert.NoError(t, err) // boot the worker diff --git a/threadregular.go b/threadregular.go index 1ae186d4ec..f5cd65644f 100644 --- a/threadregular.go +++ b/threadregular.go @@ -7,7 +7,6 @@ import ( "runtime" "sync" "sync/atomic" - "time" "github.com/dunglas/frankenphp/internal/state" ) @@ -164,7 +163,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { return nil case scaleChan <- fc: // the request has triggered scaling, continue to wait for a thread - case <-timeoutChan(time.Duration(maxWaitTime.Load())): + case <-timeoutChan(fc.server.mainThread.maxWaitTime): // the request has timed out stalling queuedRegularThreads.Add(-1) metrics.DequeuedRequest() diff --git a/types_test.go b/types_test.go index 1efe1a90d9..c6272bb2c5 100644 --- a/types_test.go +++ b/types_test.go @@ -18,7 +18,7 @@ func testOnDummyPHPThread(t *testing.T, test func()) { t.Helper() globalLogger = slog.Default() - _, err := initPHPThreads(1, 1, nil) // boot 1 thread + _, err := initPHPThreads(&opt{numThreads: 1, maxThreads: 1, phpIni: nil}) // boot 1 thread assert.NoError(t, err) handler := convertToTaskThread(phpThreads[0]) diff --git a/worker.go b/worker.go index fd5ecbb473..c3a73a83aa 100644 --- a/worker.go +++ b/worker.go @@ -308,7 +308,7 @@ func (worker *worker) handleRequest(fc *frankenPHPContext) error { return nil case workerScaleChan <- fc: // the request has triggered scaling, continue to wait for a thread - case <-timeoutChan(time.Duration(maxWaitTime.Load())): + case <-timeoutChan(fc.server.mainThread.maxWaitTime): // the request has timed out stalling worker.queuedRequests.Add(-1) metrics.DequeuedWorkerRequest(worker.name) From 02d295a8669b4ed162beb9f83618391a0c224625 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 12:43:19 +0200 Subject: [PATCH 18/24] moves more config onto the main thread --- frankenphp.go | 13 +------------ phpmainthread.go | 44 ++++++++++++++++++++++++-------------------- phpthread.go | 2 ++ threadregular.go | 5 +++-- threadworker.go | 4 ++-- 5 files changed, 32 insertions(+), 36 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index d2ef01ca70..e957754c4b 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -68,11 +68,6 @@ var ( globalLogger = slog.Default() metrics Metrics = nullMetrics{} - - // atomic: read by in-flight requests while a reload may rewrite it - maxRequestsPerThread int - - hotReloadEnabled bool ) type ErrRejected struct { @@ -387,8 +382,6 @@ func Init(options ...Option) error { metrics = opt.metrics } - maxRequestsPerThread = opt.maxRequests - if opt.maxIdleTime > 0 { maxIdleTime = opt.maxIdleTime } @@ -428,8 +421,6 @@ func Init(options ...Option) error { } } - hotReloadEnabled = opt.hasHotReload() - mThread, err := initPHPThreads(opt) if err != nil { shutdown() @@ -458,7 +449,7 @@ func Init(options ...Option) error { registerServers(mainThread, opt.servers) if globalLogger.Enabled(globalCtx, slog.LevelInfo) { - globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "FrankenPHP started 🐘", slog.String("php_version", Version().Version), slog.Int("total_threads", mainThread.numThreads), slog.Int("worker_threads", workerThreadCount), slog.Int("max_threads", mainThread.maxThreads), slog.Int("max_requests", maxRequestsPerThread)) + globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "FrankenPHP started 🐘", slog.String("php_version", Version().Version), slog.Int("total_threads", mainThread.numThreads), slog.Int("worker_threads", workerThreadCount), slog.Int("max_threads", mainThread.maxThreads), slog.Int("max_requests", mainThread.maxRequestsPerThread)) if EmbeddedAppPath != "" { globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "embedded PHP app 📦", slog.String("path", EmbeddedAppPath)) @@ -968,7 +959,5 @@ func resetGlobals() { globalCtx = context.Background() globalLogger = slog.Default() watcherIsEnabled = false - hotReloadEnabled = false maxIdleTime = defaultMaxIdleTime - maxRequestsPerThread = 0 } diff --git a/phpmainthread.go b/phpmainthread.go index d51fdda28a..ff903e8782 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -20,13 +20,15 @@ import ( // represents the main PHP thread // the thread needs to keep running as long as all other threads are running type phpMainThread struct { - state *state.ThreadState - done chan struct{} - numThreads int - maxThreads int - phpIni map[string]string - isRebooting atomic.Bool - maxWaitTime time.Duration + state *state.ThreadState + done chan struct{} + numThreads int + maxThreads int + phpIni map[string]string + isRebooting atomic.Bool + maxWaitTime time.Duration + maxRequestsPerThread int + hotReloadEnabled bool regularRequestChan chan *frankenPHPContext workers []*worker @@ -53,18 +55,20 @@ var ( func initPHPThreads(opt *opt) (*phpMainThread, error) { oldMainThread = mainThread mainThread = &phpMainThread{ - state: state.NewThreadState(), - done: make(chan struct{}), - numThreads: opt.numThreads, - maxThreads: opt.maxThreads, - phpIni: opt.phpIni, - regularRequestChan: make(chan *frankenPHPContext), - fallbackServer: newFallbackServer(), - servers: []*Server{}, - workers: make([]*worker, 0, opt.numThreads), - workersByName: make(map[string]*worker, opt.numThreads), - workersByPath: make(map[string]*worker, opt.numThreads), - maxWaitTime: opt.maxWaitTime, + state: state.NewThreadState(), + done: make(chan struct{}), + numThreads: opt.numThreads, + maxThreads: opt.maxThreads, + phpIni: opt.phpIni, + regularRequestChan: make(chan *frankenPHPContext), + fallbackServer: newFallbackServer(), + servers: []*Server{}, + workers: make([]*worker, 0, opt.numThreads), + workersByName: make(map[string]*worker, opt.numThreads), + workersByPath: make(map[string]*worker, opt.numThreads), + maxWaitTime: opt.maxWaitTime, + maxRequestsPerThread: opt.maxRequests, + hotReloadEnabled: opt.hasHotReload(), } // initialize the first thread @@ -364,7 +368,7 @@ func go_frankenphp_shutdown_main_thread() { //export go_is_hot_reload_enabled func go_is_hot_reload_enabled() C.bool { - return C.bool(hotReloadEnabled) + return C.bool(mainThread.hotReloadEnabled) } //export go_get_custom_php_ini diff --git a/phpthread.go b/phpthread.go index 6a285e0452..360924e8a5 100644 --- a/phpthread.go +++ b/phpthread.go @@ -25,6 +25,7 @@ type phpThread struct { contextMu sync.RWMutex state *state.ThreadState requestCount atomic.Int64 + mainThread *phpMainThread // forceKill holds &EG() pointers captured on the PHP thread itself. // forceKillMu pairs with go_frankenphp_clear_force_kill_slot's write // lock so a concurrent kill never dereferences pointers freed by @@ -52,6 +53,7 @@ func newPHPThread(threadIndex int) *phpThread { threadIndex: threadIndex, requestChan: make(chan *frankenPHPContext), state: state.NewThreadState(), + mainThread: mainThread, } } diff --git a/threadregular.go b/threadregular.go index f5cd65644f..2f547c101d 100644 --- a/threadregular.go +++ b/threadregular.go @@ -84,11 +84,12 @@ func (handler *regularThread) drain() {} func (handler *regularThread) waitForRequest() string { // max_requests reached: restart the thread to clean up all ZTS state - if maxRequestsPerThread > 0 && handler.requestCount >= maxRequestsPerThread { + maxRequestCount := handler.thread.mainThread.maxRequestsPerThread + if maxRequestCount > 0 && handler.requestCount >= maxRequestCount { if globalLogger.Enabled(globalCtx, slog.LevelDebug) { globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "max requests reached, restarting thread", slog.Int("thread", handler.thread.threadIndex), - slog.Int("max_requests", maxRequestsPerThread), + slog.Int("max_requests", maxRequestCount), ) } diff --git a/threadworker.go b/threadworker.go index 8a308b4d87..b0412d0e61 100644 --- a/threadworker.go +++ b/threadworker.go @@ -199,12 +199,12 @@ func (handler *workerThread) waitForWorkerRequest() (bool, any) { } // max_requests reached: signal reboot for full ZTS cleanup - if maxRequestsPerThread > 0 && handler.requestCount >= maxRequestsPerThread { + if mainThread.maxRequestsPerThread > 0 && handler.requestCount >= mainThread.maxRequestsPerThread { if globalLogger.Enabled(globalCtx, slog.LevelDebug) { globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "max requests reached, restarting", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex), - slog.Int("max_requests", maxRequestsPerThread), + slog.Int("max_requests", mainThread.maxRequestsPerThread), ) } From 36cec470bdb1c203c4fd6b0c368d06648a50ee16 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 12:49:45 +0200 Subject: [PATCH 19/24] moves maxidletime onto mainthread --- frankenphp.go | 5 ----- phpmainthread.go | 7 +++++++ scaling.go | 7 +++---- scaling_test.go | 8 ++++---- 4 files changed, 14 insertions(+), 13 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index e957754c4b..a749ada192 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -382,10 +382,6 @@ func Init(options ...Option) error { metrics = opt.metrics } - if opt.maxIdleTime > 0 { - maxIdleTime = opt.maxIdleTime - } - ensureServersHaveNames(opt.servers) workerThreadCount, err := calculateMaxThreads(opt) @@ -959,5 +955,4 @@ func resetGlobals() { globalCtx = context.Background() globalLogger = slog.Default() watcherIsEnabled = false - maxIdleTime = defaultMaxIdleTime } diff --git a/phpmainthread.go b/phpmainthread.go index ff903e8782..76d42d8065 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -27,6 +27,7 @@ type phpMainThread struct { phpIni map[string]string isRebooting atomic.Bool maxWaitTime time.Duration + maxIdleTime time.Duration maxRequestsPerThread int hotReloadEnabled bool @@ -53,6 +54,11 @@ var ( // a fixed number of inactive PHP threads // and reserves a fixed number of possible PHP threads func initPHPThreads(opt *opt) (*phpMainThread, error) { + idleTime := defaultMaxIdleTime + if opt.maxIdleTime > 0 { + idleTime = opt.maxIdleTime + } + oldMainThread = mainThread mainThread = &phpMainThread{ state: state.NewThreadState(), @@ -67,6 +73,7 @@ func initPHPThreads(opt *opt) (*phpMainThread, error) { workersByName: make(map[string]*worker, opt.numThreads), workersByPath: make(map[string]*worker, opt.numThreads), maxWaitTime: opt.maxWaitTime, + maxIdleTime: idleTime, maxRequestsPerThread: opt.maxRequests, hotReloadEnabled: opt.hasHotReload(), } diff --git a/scaling.go b/scaling.go index 2abe7fc765..32d0f3d78c 100644 --- a/scaling.go +++ b/scaling.go @@ -29,7 +29,6 @@ const ( var ( ErrMaxThreadsReached = errors.New("max amount of overall threads reached") - maxIdleTime = defaultMaxIdleTime scaleChan chan *frankenPHPContext autoScaledThreads = []*phpThread{} scalingMu = new(sync.RWMutex) @@ -193,13 +192,13 @@ func startDownScalingThreads(mThread *phpMainThread) { case <-mThread.done: return case <-time.After(downScaleCheckTime): - deactivateThreads() + deactivateThreads(mThread) } } } // deactivateThreads checks all threads and removes those that have been inactive for too long -func deactivateThreads() { +func deactivateThreads(mThread *phpMainThread) { stoppedThreadCount := 0 scalingMu.Lock() defer scalingMu.Unlock() @@ -216,7 +215,7 @@ func deactivateThreads() { } // convert threads to inactive if they have been idle for too long - if thread.state.Is(state.Ready) && waitTime > maxIdleTime.Milliseconds() { + if thread.state.Is(state.Ready) && waitTime > mThread.maxIdleTime.Milliseconds() { convertToInactiveThread(thread) stoppedThreadCount++ autoScaledThreads = slices.Delete(autoScaledThreads, i, i+1) diff --git a/scaling_test.go b/scaling_test.go index 89bd378780..c908b3251b 100644 --- a/scaling_test.go +++ b/scaling_test.go @@ -26,7 +26,7 @@ func TestScaleARegularThreadUpAndDown(t *testing.T) { // on down-scale, the thread will be marked as inactive setLongWaitTime(t, autoScaledThread) - deactivateThreads() + deactivateThreads(mainThread) assert.IsType(t, &inactiveThread{}, autoScaledThread.handler) } @@ -54,7 +54,7 @@ func TestScaleAWorkerThreadUpAndDown(t *testing.T) { // on down-scale, the thread will be marked as inactive setLongWaitTime(t, autoScaledThread) - deactivateThreads() + deactivateThreads(mainThread) assert.IsType(t, &inactiveThread{}, autoScaledThread.handler) } @@ -75,12 +75,12 @@ func TestMaxIdleTimePreventsEarlyDeactivation(t *testing.T) { // set wait time to 30 minutes (less than 1 hour max idle time) autoScaledThread.state.SetWaitTime(time.Now().Add(-30 * time.Minute)) - deactivateThreads() + deactivateThreads(mainThread) assert.IsType(t, ®ularThread{}, autoScaledThread.handler, "thread should still be active after 30min with 1h max idle time") // set wait time to over 1 hour (exceeds max idle time) autoScaledThread.state.SetWaitTime(time.Now().Add(-time.Hour - time.Minute)) - deactivateThreads() + deactivateThreads(mainThread) assert.IsType(t, &inactiveThread{}, autoScaledThread.handler, "thread should be deactivated after exceeding max idle time") } From c961c707f1d6f80055dce2797a67b4e5cd82dce7 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 13:00:53 +0200 Subject: [PATCH 20/24] moves regular threads onto mainthread --- frankenphp.go | 2 +- phpmainthread.go | 1 + server.go | 3 --- threadregular.go | 11 +++++------ 4 files changed, 7 insertions(+), 10 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index a749ada192..5432e97c3a 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -423,7 +423,7 @@ func Init(options ...Option) error { return err } - regularThreads = make([]*phpThread, 0, opt.numThreads-workerThreadCount) + mThread.regularThreads = make([]*phpThread, 0, mThread.numThreads-workerThreadCount) for range opt.numThreads - workerThreadCount { convertToRegularThread(getInactivePHPThread(), mThread) } diff --git a/phpmainthread.go b/phpmainthread.go index 76d42d8065..def605b76c 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -31,6 +31,7 @@ type phpMainThread struct { maxRequestsPerThread int hotReloadEnabled bool + regularThreads []*phpThread regularRequestChan chan *frankenPHPContext workers []*worker workersByName map[string]*worker diff --git a/server.go b/server.go index 0bf89af1a6..2c96fc06a7 100644 --- a/server.go +++ b/server.go @@ -23,7 +23,6 @@ type Server struct { workers []*worker workersByPath map[string]*worker workersWithRequestMatcher []*worker - regularRequestChan chan *frankenPHPContext mainThread *phpMainThread // registered while FrankenPHP runs with this server; read by concurrent @@ -58,11 +57,9 @@ func ensureServersHaveNames(newServers []*Server) { // registerServers registers the servers with the main thread and lets them accept requests func registerServers(mthread *phpMainThread, newServers []*Server) { mthread.servers = newServers - mthread.fallbackServer.regularRequestChan = mthread.regularRequestChan mthread.fallbackServer.isRegistered.Store(true) mthread.fallbackServer.mainThread = mthread for _, s := range mthread.servers { - s.regularRequestChan = mthread.regularRequestChan s.mainThread = mthread s.isRegistered.Store(true) } diff --git a/threadregular.go b/threadregular.go index 2f547c101d..5c2221b0ed 100644 --- a/threadregular.go +++ b/threadregular.go @@ -23,7 +23,6 @@ type regularThread struct { } var ( - regularThreads []*phpThread regularThreadMu = &sync.RWMutex{} queuedRegularThreads = atomic.Int32{} ) @@ -133,7 +132,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { if queuedRegularThreads.Load() == 0 { regularThreadMu.RLock() - for _, thread := range regularThreads { + for _, thread := range fc.server.mainThread.regularThreads { select { case thread.requestChan <- fc: regularThreadMu.RUnlock() @@ -154,7 +153,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { for { select { - case fc.server.regularRequestChan <- fc: + case fc.server.mainThread.regularRequestChan <- fc: queuedRegularThreads.Add(-1) metrics.DequeuedRequest() @@ -179,15 +178,15 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { func attachRegularThread(thread *phpThread) { regularThreadMu.Lock() - regularThreads = append(regularThreads, thread) + thread.mainThread.regularThreads = append(thread.mainThread.regularThreads, thread) regularThreadMu.Unlock() } func detachRegularThread(thread *phpThread) { regularThreadMu.Lock() - for i, t := range regularThreads { + for i, t := range thread.mainThread.regularThreads { if t == thread { - regularThreads = append(regularThreads[:i], regularThreads[i+1:]...) + thread.mainThread.regularThreads = append(thread.mainThread.regularThreads[:i], thread.mainThread.regularThreads[i+1:]...) break } } From 93a42a49b05bc73477778b428ade6ff972e3b1f1 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 13:06:08 +0200 Subject: [PATCH 21/24] moves scaled threads onto main thread --- phpmainthread.go | 2 ++ scaling.go | 27 +++++++++++---------------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/phpmainthread.go b/phpmainthread.go index def605b76c..39e73c2c7f 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -32,6 +32,7 @@ type phpMainThread struct { hotReloadEnabled bool regularThreads []*phpThread + autoScaledThreads []*phpThread regularRequestChan chan *frankenPHPContext workers []*worker workersByName map[string]*worker @@ -73,6 +74,7 @@ func initPHPThreads(opt *opt) (*phpMainThread, error) { workers: make([]*worker, 0, opt.numThreads), workersByName: make(map[string]*worker, opt.numThreads), workersByPath: make(map[string]*worker, opt.numThreads), + autoScaledThreads: make([]*phpThread, 0, opt.numThreads), maxWaitTime: opt.maxWaitTime, maxIdleTime: idleTime, maxRequestsPerThread: opt.maxRequests, diff --git a/scaling.go b/scaling.go index 32d0f3d78c..bc061c7528 100644 --- a/scaling.go +++ b/scaling.go @@ -29,9 +29,8 @@ const ( var ( ErrMaxThreadsReached = errors.New("max amount of overall threads reached") - scaleChan chan *frankenPHPContext - autoScaledThreads = []*phpThread{} - scalingMu = new(sync.RWMutex) + scaleChan chan *frankenPHPContext + scalingMu = new(sync.RWMutex) ) func initAutoScaling(mainThread *phpMainThread) { @@ -44,11 +43,7 @@ func initAutoScaling(mainThread *phpMainThread) { scaleChan = make(chan *frankenPHPContext) } - scalingMu.Lock() maxScaledThreads := mainThread.maxThreads - mainThread.numThreads - autoScaledThreads = make([]*phpThread, 0, maxScaledThreads) - scalingMu.Unlock() - go startUpscalingThreads(maxScaledThreads, scaleChan, mainThread) go startDownScalingThreads(mainThread) } @@ -98,10 +93,10 @@ func scaleWorkerThread(worker *worker, mThread *phpMainThread) { return } - autoScaledThreads = append(autoScaledThreads, thread) + mThread.autoScaledThreads = append(mThread.autoScaledThreads, thread) if globalLogger.Enabled(globalCtx, slog.LevelInfo) { - globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "upscaling worker thread", slog.String("worker", worker.name), slog.Int("thread", thread.threadIndex), slog.Int("num_threads", len(autoScaledThreads))) + globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "upscaling worker thread", slog.String("worker", worker.name), slog.Int("thread", thread.threadIndex), slog.Int("num_threads", len(mThread.autoScaledThreads))) } } @@ -128,17 +123,17 @@ func scaleRegularThread(mThread *phpMainThread) { return } - autoScaledThreads = append(autoScaledThreads, thread) + mThread.autoScaledThreads = append(mThread.autoScaledThreads, thread) if globalLogger.Enabled(globalCtx, slog.LevelInfo) { - globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "upscaling regular thread", slog.Int("thread", thread.threadIndex), slog.Int("num_threads", len(autoScaledThreads))) + globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "upscaling regular thread", slog.Int("thread", thread.threadIndex), slog.Int("num_threads", len(mThread.autoScaledThreads))) } } func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, mThread *phpMainThread) { for { scalingMu.Lock() - scaledThreadCount := len(autoScaledThreads) + scaledThreadCount := len(mThread.autoScaledThreads) scalingMu.Unlock() if scaledThreadCount >= maxScaledThreads { // we have reached max_threads, check again later @@ -202,10 +197,10 @@ func deactivateThreads(mThread *phpMainThread) { stoppedThreadCount := 0 scalingMu.Lock() defer scalingMu.Unlock() - for i, thread := range slices.Backward(autoScaledThreads) { + for i, thread := range slices.Backward(mThread.autoScaledThreads) { // the thread might have been stopped otherwise, remove it if thread.state.Is(state.Reserved) { - autoScaledThreads = slices.Delete(autoScaledThreads, i, i+1) + mThread.autoScaledThreads = slices.Delete(mThread.autoScaledThreads, i, i+1) continue } @@ -218,10 +213,10 @@ func deactivateThreads(mThread *phpMainThread) { if thread.state.Is(state.Ready) && waitTime > mThread.maxIdleTime.Milliseconds() { convertToInactiveThread(thread) stoppedThreadCount++ - autoScaledThreads = slices.Delete(autoScaledThreads, i, i+1) + mThread.autoScaledThreads = slices.Delete(mThread.autoScaledThreads, i, i+1) if globalLogger.Enabled(globalCtx, slog.LevelInfo) { - globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "downscaling thread", slog.Int("thread", thread.threadIndex), slog.Int64("wait_time", waitTime), slog.Int("num_threads", len(autoScaledThreads))) + globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "downscaling thread", slog.Int("thread", thread.threadIndex), slog.Int64("wait_time", waitTime), slog.Int("num_threads", len(mThread.autoScaledThreads))) } continue From dcb7726071e56bab1335527bded750d13dbe289d Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 13:11:05 +0200 Subject: [PATCH 22/24] moves scalechan onto main thread --- phpmainthread.go | 22 +++++++++++++--------- scaling.go | 19 +++++++------------ threadregular.go | 9 +++++---- worker.go | 2 +- 4 files changed, 26 insertions(+), 26 deletions(-) diff --git a/phpmainthread.go b/phpmainthread.go index 39e73c2c7f..8f63910736 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -24,21 +24,25 @@ type phpMainThread struct { done chan struct{} numThreads int maxThreads int + maxRequestsPerThread int + hotReloadEnabled bool phpIni map[string]string isRebooting atomic.Bool maxWaitTime time.Duration maxIdleTime time.Duration - maxRequestsPerThread int - hotReloadEnabled bool - regularThreads []*phpThread - autoScaledThreads []*phpThread + regularThreads []*phpThread + autoScaledThreads []*phpThread + + scaleChan chan *frankenPHPContext regularRequestChan chan *frankenPHPContext - workers []*worker - workersByName map[string]*worker - workersByPath map[string]*worker - fallbackServer *Server - servers []*Server + + workers []*worker + workersByName map[string]*worker + workersByPath map[string]*worker + + fallbackServer *Server + servers []*Server } var ( diff --git a/scaling.go b/scaling.go index bc061c7528..1cdd25510b 100644 --- a/scaling.go +++ b/scaling.go @@ -28,24 +28,19 @@ const ( var ( ErrMaxThreadsReached = errors.New("max amount of overall threads reached") - - scaleChan chan *frankenPHPContext - scalingMu = new(sync.RWMutex) + scalingMu = new(sync.RWMutex) ) -func initAutoScaling(mainThread *phpMainThread) { - if mainThread.maxThreads <= mainThread.numThreads { +func initAutoScaling(mThread *phpMainThread) { + if mThread.maxThreads <= mThread.numThreads { return } - // reused across reloads so queued requests aren't orphaned on a stale channel - if scaleChan == nil { - scaleChan = make(chan *frankenPHPContext) - } + mThread.scaleChan = make(chan *frankenPHPContext) + maxScaledThreads := mThread.maxThreads - mThread.numThreads - maxScaledThreads := mainThread.maxThreads - mainThread.numThreads - go startUpscalingThreads(maxScaledThreads, scaleChan, mainThread) - go startDownScalingThreads(mainThread) + go startUpscalingThreads(maxScaledThreads, mThread.scaleChan, mThread) + go startDownScalingThreads(mThread) } func addRegularThread(mThread *phpMainThread) (*phpThread, error) { diff --git a/threadregular.go b/threadregular.go index 5c2221b0ed..041187092d 100644 --- a/threadregular.go +++ b/threadregular.go @@ -130,9 +130,10 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { runtime.Gosched() + mainThread := fc.server.mainThread if queuedRegularThreads.Load() == 0 { regularThreadMu.RLock() - for _, thread := range fc.server.mainThread.regularThreads { + for _, thread := range mainThread.regularThreads { select { case thread.requestChan <- fc: regularThreadMu.RUnlock() @@ -153,7 +154,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { for { select { - case fc.server.mainThread.regularRequestChan <- fc: + case mainThread.regularRequestChan <- fc: queuedRegularThreads.Add(-1) metrics.DequeuedRequest() @@ -161,9 +162,9 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { metrics.StopRequest() return nil - case scaleChan <- fc: + case mainThread.scaleChan <- fc: // the request has triggered scaling, continue to wait for a thread - case <-timeoutChan(fc.server.mainThread.maxWaitTime): + case <-timeoutChan(mainThread.maxWaitTime): // the request has timed out stalling queuedRegularThreads.Add(-1) metrics.DequeuedRequest() diff --git a/worker.go b/worker.go index c3a73a83aa..09a9e21c43 100644 --- a/worker.go +++ b/worker.go @@ -293,7 +293,7 @@ func (worker *worker) handleRequest(fc *frankenPHPContext) error { metrics.QueuedWorkerRequest(worker.name) for { - workerScaleChan := scaleChan + workerScaleChan := fc.server.mainThread.scaleChan if worker.isAtThreadLimit() { workerScaleChan = nil // max_threads for this worker reached, do not attempt scaling } From e8638b6684dfd628335c23260380c4de49b3af2a Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 13:15:02 +0200 Subject: [PATCH 23/24] moves watcherIsEnabled onto main thread --- frankenphp.go | 3 +-- phpmainthread.go | 1 + threadworker.go | 4 ++-- watcher-skip.go | 2 +- watcher.go | 6 +++--- worker.go | 3 +-- 6 files changed, 9 insertions(+), 10 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index 5432e97c3a..0086a1ae51 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -434,7 +434,7 @@ func Init(options ...Option) error { return err } - if err := initWatchers(opt); err != nil { + if err := initWatchers(opt, mThread); err != nil { shutdown() return err } @@ -954,5 +954,4 @@ func timeoutChan(timeout time.Duration) <-chan time.Time { func resetGlobals() { globalCtx = context.Background() globalLogger = slog.Default() - watcherIsEnabled = false } diff --git a/phpmainthread.go b/phpmainthread.go index 8f63910736..fcfddd9acb 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -26,6 +26,7 @@ type phpMainThread struct { maxThreads int maxRequestsPerThread int hotReloadEnabled bool + watcherIsEnabled bool phpIni map[string]string isRebooting atomic.Bool maxWaitTime time.Duration diff --git a/threadworker.go b/threadworker.go index b0412d0e61..d4919e4810 100644 --- a/threadworker.go +++ b/threadworker.go @@ -149,13 +149,13 @@ func tearDownWorkerScript(handler *workerThread, exitStatus int) { return } - if worker.maxConsecutiveFailures >= 0 && startupFailChan != nil && !watcherIsEnabled && handler.failureCount >= worker.maxConsecutiveFailures { + if worker.maxConsecutiveFailures >= 0 && startupFailChan != nil && !handler.thread.mainThread.watcherIsEnabled && handler.failureCount >= worker.maxConsecutiveFailures { startupFailChan <- fmt.Errorf("too many consecutive failures: worker %s has not reached frankenphp_handle_request()", worker.fileName) handler.thread.state.Set(state.ShuttingDown) return } - if watcherIsEnabled { + if handler.thread.mainThread.watcherIsEnabled { // worker script has probably failed due to script changes while watcher is enabled if globalLogger.Enabled(globalCtx, slog.LevelError) { globalLogger.LogAttrs(globalCtx, slog.LevelError, "(watcher enabled) worker script has not reached frankenphp_handle_request()", slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex)) diff --git a/watcher-skip.go b/watcher-skip.go index 9f1db3c949..e7f24b97a2 100644 --- a/watcher-skip.go +++ b/watcher-skip.go @@ -25,7 +25,7 @@ func validateWatchers(o *opt) error { return nil } -func initWatchers(o *opt) error { +func initWatchers(o *opt, mThread *phpMainThread) error { return validateWatchers(o) } diff --git a/watcher.go b/watcher.go index 63864b9ef6..adba7ba187 100644 --- a/watcher.go +++ b/watcher.go @@ -25,7 +25,7 @@ func (o *hotReloadOpt) hasHotReload() bool { return len(o.hotReload) != 0 } -func initWatchers(o *opt) error { +func initWatchers(o *opt, mThread *phpMainThread) error { watchPatterns := make([]*watcher.PatternGroup, 0, len(o.hotReload)) for _, o := range o.workers { @@ -33,13 +33,13 @@ func initWatchers(o *opt) error { continue } - watcherIsEnabled = true + mThread.watcherIsEnabled = true watchPatterns = append(watchPatterns, &watcher.PatternGroup{Patterns: o.watch, Callback: func(_ []*watcherGo.Event) { restartWorkers.Store(true) }}) } - if watcherIsEnabled { + if mThread.watcherIsEnabled { watchPatterns = append(watchPatterns, &watcher.PatternGroup{ Callback: func(_ []*watcherGo.Event) { if restartWorkers.Swap(false) { diff --git a/worker.go b/worker.go index 09a9e21c43..3a46d50471 100644 --- a/worker.go +++ b/worker.go @@ -37,8 +37,7 @@ type worker struct { } var ( - watcherIsEnabled bool - startupFailChan chan error + startupFailChan chan error ) func initWorkers(opts []workerOpt, m *phpMainThread) error { From 9b60fccebc56011e9e56fc4f3279ad264cbf331e Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 16 Oct 2026 13:18:51 +0200 Subject: [PATCH 24/24] makes RestartWorkers reload-safe --- phpmainthread.go | 10 +++++----- threadregular.go | 10 +++++----- worker.go | 2 ++ 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/phpmainthread.go b/phpmainthread.go index fcfddd9acb..3f1ce86fd1 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -355,21 +355,21 @@ func go_frankenphp_main_thread_is_ready() { // max_threads = auto // setAutomaticMaxThreads estimates the amount of threads based on php.ini and system memory_limit // If unable to get the system's memory limit, simply double num_threads -func (mainThread *phpMainThread) setAutomaticMaxThreads() { - if mainThread.maxThreads >= 0 { +func (mThread *phpMainThread) setAutomaticMaxThreads() { + if mThread.maxThreads >= 0 { return } perThreadMemoryLimit := int64(C.frankenphp_get_current_memory_limit()) totalSysMemory := memory.TotalSysMemory() if perThreadMemoryLimit <= 0 || totalSysMemory == 0 { - mainThread.maxThreads = mainThread.numThreads * 2 + mThread.maxThreads = mThread.numThreads * 2 return } maxAllowedThreads := totalSysMemory / uint64(perThreadMemoryLimit) - mainThread.maxThreads = int(maxAllowedThreads) + mThread.maxThreads = int(maxAllowedThreads) if globalLogger.Enabled(globalCtx, slog.LevelDebug) { - globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "Automatic thread limit", slog.Int("perThreadMemoryLimitMB", int(perThreadMemoryLimit/1024/1024)), slog.Int("maxThreads", mainThread.maxThreads)) + globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "Automatic thread limit", slog.Int("perThreadMemoryLimitMB", int(perThreadMemoryLimit/1024/1024)), slog.Int("maxThreads", mThread.maxThreads)) } } diff --git a/threadregular.go b/threadregular.go index 041187092d..b14e49017b 100644 --- a/threadregular.go +++ b/threadregular.go @@ -130,10 +130,10 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { runtime.Gosched() - mainThread := fc.server.mainThread + mThread := fc.server.mainThread if queuedRegularThreads.Load() == 0 { regularThreadMu.RLock() - for _, thread := range mainThread.regularThreads { + for _, thread := range mThread.regularThreads { select { case thread.requestChan <- fc: regularThreadMu.RUnlock() @@ -154,7 +154,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { for { select { - case mainThread.regularRequestChan <- fc: + case mThread.regularRequestChan <- fc: queuedRegularThreads.Add(-1) metrics.DequeuedRequest() @@ -162,9 +162,9 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { metrics.StopRequest() return nil - case mainThread.scaleChan <- fc: + case mThread.scaleChan <- fc: // the request has triggered scaling, continue to wait for a thread - case <-timeoutChan(mainThread.maxWaitTime): + case <-timeoutChan(mThread.maxWaitTime): // the request has timed out stalling queuedRegularThreads.Add(-1) metrics.DequeuedRequest() diff --git a/worker.go b/worker.go index 3a46d50471..64d96fce94 100644 --- a/worker.go +++ b/worker.go @@ -222,7 +222,9 @@ func newWorker(o workerOpt, m *phpMainThread) (*worker, error) { // full duration of the syscall. func RestartWorkers() { if mainThread != nil { + startupMu.Lock() mainThread.rebootAllThreads() + startupMu.Unlock() } }