diff --git a/caddy/app.go b/caddy/app.go index b4f2d012b3..45fcf748e1 100644 --- a/caddy/app.go +++ b/caddy/app.go @@ -1,6 +1,7 @@ package caddy import ( + "bytes" "context" "errors" "fmt" @@ -69,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 "" ""`) @@ -104,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 } @@ -139,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)) @@ -159,7 +183,9 @@ func (f *FrankenPHPApp) collectOptions(repl *caddy.Replacer, keep bool) ([]frank return nil, err } - return append(opts, moduleOpts...), nil + opts = append(opts, moduleOpts...) + + 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..d1fd82c39f --- /dev/null +++ b/caddy/reload_queue_test.go @@ -0,0 +1,266 @@ +package caddy_test + +import ( + "bytes" + "io" + "net/http" + "strconv" + "strings" + "sync" + "sync/atomic" + "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 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 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() + + 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/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/context.go b/context.go index ace1e3dcdf..b58cf8b650 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{ @@ -168,7 +168,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 229eb6b2a0..0086a1ae51 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"} @@ -67,12 +68,6 @@ var ( globalLogger = slog.Default() metrics Metrics = nullMetrics{} - - // atomic: read by in-flight requests while a reload may rewrite it - maxWaitTime atomic.Int64 - maxRequestsPerThread int - - hotReloadEnabled bool ) type ErrRejected struct { @@ -387,14 +382,7 @@ func Init(options ...Option) error { metrics = opt.metrics } - maxWaitTime.Store(int64(opt.maxWaitTime)) - maxRequestsPerThread = opt.maxRequests - - if opt.maxIdleTime > 0 { - maxIdleTime = opt.maxIdleTime - } - - registerServers(opt.servers) + ensureServersHaveNames(opt.servers) workerThreadCount, err := calculateMaxThreads(opt) if err != nil { @@ -429,37 +417,35 @@ func Init(options ...Option) error { } } - hotReloadEnabled = opt.hasHotReload() - - mainThread, err := initPHPThreads(opt.numThreads, opt.maxThreads, opt.phpIni) + mThread, err := initPHPThreads(opt) if err != nil { shutdown() return err } - regularThreads = make([]*phpThread, 0, opt.numThreads-workerThreadCount) + mThread.regularThreads = make([]*phpThread, 0, mThread.numThreads-workerThreadCount) for range opt.numThreads - workerThreadCount { - convertToRegularThread(getInactivePHPThread()) + convertToRegularThread(getInactivePHPThread(), mThread) } - if err := initWorkers(opt.workers); err != nil { + if err := initWorkers(opt.workers, mThread); err != nil { shutdown() return err } - if err := initWatchers(opt); err != nil { + if err := initWatchers(opt, mThread); err != nil { shutdown() return err } - initAutoScaling(mainThread) + 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)) + 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)) @@ -477,6 +463,10 @@ func Init(options ...Option) error { } } + if oldMainThread != nil { + mThread.handleStaleRequests(oldMainThread, opt.shouldRequeue) + } + return nil } @@ -500,7 +490,6 @@ func shutdown() { drainWatchers() drainPHPThreads() - unregisterServers() metrics.Shutdown() @@ -526,7 +515,11 @@ func ServeHTTP(responseWriter http.ResponseWriter, request *http.Request) error return ErrInvalidRequest } - return fallbackServer.ServeHTTP(responseWriter, request, opts...) + if mainThread == nil { + return ErrNotRunning + } + + return mainThread.fallbackServer.ServeHTTP(responseWriter, request, opts...) } //export go_ub_write @@ -961,12 +954,4 @@ func timeoutChan(timeout time.Duration) <-chan time.Time { func resetGlobals() { globalCtx = context.Background() globalLogger = slog.Default() - workers = nil - workersByName = nil - globalWorkersByPath = nil - servers = nil - watcherIsEnabled = false - hotReloadEnabled = false - maxIdleTime = defaultMaxIdleTime - maxRequestsPerThread = 0 } diff --git a/options.go b/options.go index 5566ed41cf..77f12bd074 100644 --- a/options.go +++ b/options.go @@ -26,17 +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 + 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 { @@ -176,6 +177,15 @@ func WithMaxIdleTime(maxIdleTime time.Duration) Option { } } +// WithRequestRequeueOnReload sets if requests should be queued on reload. +func WithRequestRequeueOnReload(shouldRequeue bool) Option { + return func(o *opt) error { + o.shouldRequeue = shouldRequeue + + 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 { @@ -299,6 +309,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 23a544ced8..3f1ce86fd1 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -20,17 +20,36 @@ 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 + maxRequestsPerThread int + hotReloadEnabled bool + watcherIsEnabled bool + phpIni map[string]string + isRebooting atomic.Bool + maxWaitTime time.Duration + maxIdleTime time.Duration + + regularThreads []*phpThread + autoScaledThreads []*phpThread + + scaleChan chan *frankenPHPContext + regularRequestChan chan *frankenPHPContext + + workers []*worker + workersByName map[string]*worker + workersByPath map[string]*worker + + fallbackServer *Server + servers []*Server } var ( phpThreads []*phpThread mainThread *phpMainThread + oldMainThread *phpMainThread commonHeaders map[string]*C.zend_string // timeouts to wait for threads to yield before arming force-kill @@ -41,13 +60,30 @@ 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) { + idleTime := defaultMaxIdleTime + if opt.maxIdleTime > 0 { + idleTime = opt.maxIdleTime + } + + 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: 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), + autoScaledThreads: make([]*phpThread, 0, opt.numThreads), + maxWaitTime: opt.maxWaitTime, + maxIdleTime: idleTime, + maxRequestsPerThread: opt.maxRequests, + hotReloadEnabled: opt.hasHotReload(), } // initialize the first thread @@ -75,7 +111,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) } @@ -245,6 +281,66 @@ func getInactivePHPThread() *phpThread { return nil } +func (newMainThread *phpMainThread) handleStaleRequests(oldMainThread *phpMainThread, shouldRequeue bool) { + hardReloadTimeout := oldMainThread.maxWaitTime + if hardReloadTimeout == 0 { + hardReloadTimeout = 15 * time.Second + } + + go func() { + 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 { + select { + case newMainThread.regularRequestChan <- fc: + break + case <-time.After(hardReloadTimeout): + fc.reject(ErrConfigurationChangedOnReload) + } + } else { + fc.reject(ErrConfigurationChangedOnReload) + } + } + } + }() + + for _, oldWorker := range oldMainThread.workers { + 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 fc := <-oldWorker.requestChan: + if newWorker != nil && shouldRequeue { + select { + case newWorker.requestChan <- fc: + break + case <-time.After(hardReloadTimeout): + fc.reject(ErrConfigurationChangedOnReload) + } + } else { + fc.reject(ErrConfigurationChangedOnReload) + } + } + } + }() + } +} + //export go_frankenphp_main_thread_is_ready func go_frankenphp_main_thread_is_ready() { mainThread.setAutomaticMaxThreads() @@ -259,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)) } } @@ -287,7 +383,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/phpmainthread_test.go b/phpmainthread_test.go index d8a5b6742d..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,11 +41,11 @@ 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 - convertToRegularThread(phpThreads[0]) + convertToRegularThread(phpThreads[0], mainThread) assert.IsType(t, ®ularThread{}, phpThreads[0].handler) // transition to worker 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") @@ -166,69 +166,10 @@ 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(t *testing.T) { - t.Cleanup(Shutdown) - - initOpts := []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 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 @@ -250,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 } @@ -308,16 +254,16 @@ 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) { 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, } } @@ -398,7 +344,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/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/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.go b/scaling.go index dd21a7e37c..1cdd25510b 100644 --- a/scaling.go +++ b/scaling.go @@ -28,41 +28,27 @@ const ( var ( ErrMaxThreadsReached = errors.New("max amount of overall threads reached") - - maxIdleTime = defaultMaxIdleTime - scaleChan chan *frankenPHPContext - autoScaledThreads = []*phpThread{} - 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) - } - - done := mainThread.done - mstate := mainThread.state - - scalingMu.Lock() - maxScaledThreads := mainThread.maxThreads - mainThread.numThreads - autoScaledThreads = make([]*phpThread, 0, maxScaledThreads) - scalingMu.Unlock() + mThread.scaleChan = make(chan *frankenPHPContext) + maxScaledThreads := mThread.maxThreads - mThread.numThreads - go startUpscalingThreads(maxScaledThreads, scaleChan, done, mstate) - go startDownScalingThreads(done) + go startUpscalingThreads(maxScaledThreads, mThread.scaleChan, mThread) + go startDownScalingThreads(mThread) } -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 +66,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 } @@ -102,28 +88,28 @@ func scaleWorkerThread(worker *worker, done chan struct{}, mstate *state.ThreadS 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))) } } // 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)) @@ -132,22 +118,22 @@ func scaleRegularThread(done chan struct{}, mstate *state.ThreadState) { 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, done chan struct{}, mstate *state.ThreadState) { +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 select { - case <-done: + case <-mThread.done: return case <-time.After(downScaleCheckTime): continue @@ -161,7 +147,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 +156,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,33 +169,33 @@ 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() + 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() - 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 } @@ -219,13 +205,13 @@ 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) + 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 diff --git a/scaling_test.go b/scaling_test.go index 9a6d12da6d..c908b3251b 100644 --- a/scaling_test.go +++ b/scaling_test.go @@ -20,13 +20,13 @@ 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) // on down-scale, the thread will be marked as inactive setLongWaitTime(t, autoScaledThread) - deactivateThreads() + deactivateThreads(mainThread) assert.IsType(t, &inactiveThread{}, autoScaledThread.handler) } @@ -49,12 +49,12 @@ func TestScaleAWorkerThreadUpAndDown(t *testing.T) { autoScaledThread := phpThreads[2] // scale up - scaleWorkerThread(globalWorkersByPath[workerPath], mainThread.done, mainThread.state) + scaleWorkerThread(mainThread.workersByPath[workerPath], mainThread) assert.Equal(t, state.Ready, autoScaledThread.state.Get()) // on down-scale, the thread will be marked as inactive setLongWaitTime(t, autoScaledThread) - deactivateThreads() + deactivateThreads(mainThread) assert.IsType(t, &inactiveThread{}, autoScaledThread.handler) } @@ -70,17 +70,17 @@ 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) 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") } diff --git a/server.go b/server.go index 8274f23c54..2c96fc06a7 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 @@ -30,64 +31,40 @@ type Server struct { logger *slog.Logger } -var ( - servers []*Server - fallbackServer = newFallbackServer() -) - func newFallbackServer() *Server { s := &Server{ idx: -1, workersByPath: make(map[string]*worker), env: make(map[string]string), logger: globalLogger, + mainThread: mainThread, } 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 - fallbackServer.logger = globalLogger - fallbackServer.resetWorkers() - - 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 == "" { 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) - for _, s := range servers { +// registerServers registers the servers with the main thread and lets them accept requests +func registerServers(mthread *phpMainThread, newServers []*Server) { + mthread.servers = newServers + mthread.fallbackServer.isRegistered.Store(true) + mthread.fallbackServer.mainThread = mthread + for _, s := range mthread.servers { + s.mainThread = mthread s.isRegistered.Store(true) } } -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 8d88088754..b14e49017b 100644 --- a/threadregular.go +++ b/threadregular.go @@ -7,7 +7,6 @@ import ( "runtime" "sync" "sync/atomic" - "time" "github.com/dunglas/frankenphp/internal/state" ) @@ -20,19 +19,19 @@ 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, + thread: thread, + state: thread.state, + requestChan: mThread.regularRequestChan, }) attachRegularThread(thread) } @@ -84,11 +83,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), ) } @@ -105,7 +105,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: } @@ -130,9 +130,10 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { runtime.Gosched() + mThread := fc.server.mainThread if queuedRegularThreads.Load() == 0 { regularThreadMu.RLock() - for _, thread := range regularThreads { + for _, thread := range mThread.regularThreads { select { case thread.requestChan <- fc: regularThreadMu.RUnlock() @@ -153,7 +154,7 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { for { select { - case regularRequestChan <- fc: + case mThread.regularRequestChan <- fc: queuedRegularThreads.Add(-1) metrics.DequeuedRequest() @@ -161,9 +162,9 @@ func handleRequestWithRegularPHPThreads(fc *frankenPHPContext) error { metrics.StopRequest() return nil - case scaleChan <- fc: + case mThread.scaleChan <- fc: // the request has triggered scaling, continue to wait for a thread - case <-timeoutChan(time.Duration(maxWaitTime.Load())): + case <-timeoutChan(mThread.maxWaitTime): // the request has timed out stalling queuedRegularThreads.Add(-1) metrics.DequeuedRequest() @@ -178,15 +179,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 } } 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) - } - } -} diff --git a/threadworker.go b/threadworker.go index 8a308b4d87..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)) @@ -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), ) } 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/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 4a596c2fb6..64d96fce94 100644 --- a/worker.go +++ b/worker.go @@ -37,14 +37,10 @@ type worker struct { } var ( - workers []*worker - workersByName map[string]*worker - globalWorkersByPath map[string]*worker - watcherIsEnabled bool - startupFailChan chan error + startupFailChan chan error ) -func initWorkers(opts []workerOpt) error { +func initWorkers(opts []workerOpt, m *phpMainThread) error { if len(opts) == 0 { return nil } @@ -54,21 +50,17 @@ func initWorkers(opts []workerOpt) 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 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 err } @@ -76,7 +68,7 @@ func initWorkers(opts []workerOpt) error { startupFailChan = make(chan error, totalThreadsToStart) - for _, w := range workers { + for _, w := range m.workers { for range w.num { thread := getInactivePHPThread() convertToWorkerThread(thread, w) @@ -155,16 +147,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 @@ -230,7 +222,9 @@ func newWorker(o workerOpt) (*worker, error) { // full duration of the syscall. func RestartWorkers() { if mainThread != nil { + startupMu.Lock() mainThread.rebootAllThreads() + startupMu.Unlock() } } @@ -300,7 +294,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 } @@ -315,7 +309,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)