From d9406688a3974a52667e1f51a239accf480b846a Mon Sep 17 00:00:00 2001 From: Nicolas Grekas Date: Mon, 20 Jul 2026 17:23:13 +0200 Subject: [PATCH 1/2] feat: declared background workers + frankenphp_get_worker_handle() Background workers are long-lived non-HTTP PHP scripts declared via WithWorkerBackground() or the `background` flag in Caddyfile worker blocks. The script runs in a loop: it is re-run on cooperative exit (status 0) and restarted with a quadratic backoff on crash, failing hard on max_consecutive_failures during startup only. frankenphp_get_worker_handle() returns a stream over the read end of a per-thread stop pipe; draining the thread (shutdown, reboot, handler transition) closes the write end, so a script parked in stream_select wakes up and can exit gracefully. Background workers attach to a Server through the existing WithWorkerServerScope(); their name is mandatory (it is the script's identity, exposed as FRANKENPHP_WORKER) and lives in the global worker namespace, which the Caddy module already qualifies per server. --- bgworker_test.go | 149 +++++++++++++++++++++++++ caddy/config_test.go | 54 +++++++++ caddy/workerconfig.go | 19 +++- docs/config.md | 2 + frankenphp.c | 141 +++++++++++++++++++++++ frankenphp.go | 29 ++++- frankenphp.h | 4 + frankenphp.stub.php | 11 ++ frankenphp_arginfo.h | 9 +- options.go | 15 +++ phpmainthread.go | 4 + phpthread.go | 12 +- requestoptions.go | 11 +- testdata/bgworker/basic.php | 20 ++++ testdata/bgworker/crash.php | 29 +++++ testdata/bgworker/named.php | 19 ++++ threadbackgroundworker.go | 217 ++++++++++++++++++++++++++++++++++++ worker.go | 38 ++++++- 18 files changed, 768 insertions(+), 15 deletions(-) create mode 100644 bgworker_test.go create mode 100644 testdata/bgworker/basic.php create mode 100644 testdata/bgworker/crash.php create mode 100644 testdata/bgworker/named.php create mode 100644 threadbackgroundworker.go diff --git a/bgworker_test.go b/bgworker_test.go new file mode 100644 index 0000000000..04199dc0ee --- /dev/null +++ b/bgworker_test.go @@ -0,0 +1,149 @@ +package frankenphp_test + +import ( + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "github.com/dunglas/frankenphp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// requireFileEventually asserts that `path` appears on disk before the +// deadline. Wraps require.Eventually so call sites stay short. +func requireFileEventually(t testing.TB, path string, msgAndArgs ...any) { + t.Helper() + require.Eventually(t, func() bool { + _, err := os.Stat(path) + return err == nil + }, 5*time.Second, 25*time.Millisecond, msgAndArgs...) +} + +// TestBackgroundWorkerLifecycle boots a background worker that touches a +// sentinel file then parks on the stop pipe. It proves the bg worker runs +// (sentinel appears) and that Shutdown returns within a reasonable time. +// The test asserts on Shutdown timing, so it manages Shutdown itself +// instead of using initServers' t.Cleanup hook. +func TestBackgroundWorkerLifecycle(t *testing.T) { + tmp := t.TempDir() + sentinel := filepath.Join(tmp, "bg-lifecycle.sentinel") + + require.NoError(t, frankenphp.Init( + frankenphp.WithWorkers("bg-lifecycle", "testdata/bgworker/basic.php", 1, + frankenphp.WithWorkerBackground(), + frankenphp.WithWorkerEnv(map[string]string{"BG_SENTINEL": sentinel}), + ), + frankenphp.WithNumThreads(2), + )) + + requireFileEventually(t, sentinel, "background worker did not touch sentinel") + + done := make(chan struct{}) + go func() { + frankenphp.Shutdown() + close(done) + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatalf("Shutdown did not return within 10s") + } +} + +// TestBackgroundWorkerCrashRestarts boots a worker that exit(1)s on its +// first run and touches a "restarted" sentinel on its second run. The +// sentinel proves the crash-restart loop fired. +func TestBackgroundWorkerCrashRestarts(t *testing.T) { + tmp := t.TempDir() + crashMarker := filepath.Join(tmp, "bg-crash.marker") + restarted := filepath.Join(tmp, "bg-crash.restarted") + + initServers(t, + frankenphp.WithWorkers("bg-crash", "testdata/bgworker/crash.php", 1, + frankenphp.WithWorkerBackground(), + frankenphp.WithWorkerEnv(map[string]string{ + "BG_CRASH_MARKER": crashMarker, + "BG_RESTARTED_SENTINEL": restarted, + }), + ), + frankenphp.WithNumThreads(2), + ) + + requireFileEventually(t, restarted, "background worker did not restart after crash") +} + +// TestBackgroundWorkerOnServer scopes a background worker to a Server. It +// proves that the worker inherits the server env (the sentinel directory is +// declared on the server, not on the worker), that FRANKENPHP_WORKER holds +// the worker name, and that the worker does not intercept HTTP requests +// served by the same server. +func TestBackgroundWorkerOnServer(t *testing.T) { + tmp := t.TempDir() + + server, err := frankenphp.NewServer( + testDataDir, + frankenphp.WithServerName("sidekick-server"), + frankenphp.WithServerEnv(map[string]string{"BG_SENTINEL_DIR": tmp}), + ) + require.NoError(t, err) + + initServers(t, + frankenphp.WithServer(server), + frankenphp.WithWorkers("jobs", "testdata/bgworker/named.php", 1, + frankenphp.WithWorkerBackground(), + frankenphp.WithWorkerServerScope(server), + ), + frankenphp.WithNumThreads(2), + ) + + // named.php touches "/" + requireFileEventually(t, filepath.Join(tmp, "jobs"), "background worker did not touch its per-name sentinel") + + body := serverGet(t, server, "http://example.com/index.php") + assert.Contains(t, body, "I am by birth a Genevese", "the server must still serve regular requests") +} + +// TestBackgroundWorkerValidation covers the declaration-time errors. +func TestBackgroundWorkerValidation(t *testing.T) { + t.Cleanup(frankenphp.Shutdown) + + t.Run("name is required", func(t *testing.T) { + err := frankenphp.Init( + frankenphp.WithWorkers("", "testdata/bgworker/basic.php", 1, frankenphp.WithWorkerBackground()), + frankenphp.WithNumThreads(2), + ) + require.ErrorContains(t, err, "must have an explicit name") + }) + + t.Run("num must be >= 1", func(t *testing.T) { + err := frankenphp.Init( + frankenphp.WithWorkers("bg-zero", "testdata/bgworker/basic.php", 0, frankenphp.WithWorkerBackground()), + frankenphp.WithNumThreads(2), + ) + require.ErrorContains(t, err, "must declare num >= 1") + }) + + t.Run("names are global", func(t *testing.T) { + err := frankenphp.Init( + frankenphp.WithWorkers("bg-shared", "testdata/bgworker/basic.php", 1, frankenphp.WithWorkerBackground()), + frankenphp.WithWorkers("bg-shared", "testdata/bgworker/named.php", 1, frankenphp.WithWorkerBackground()), + frankenphp.WithNumThreads(3), + ) + require.ErrorContains(t, err, "two workers cannot have the same name") + }) + + t.Run("request matchers are rejected", func(t *testing.T) { + err := frankenphp.Init( + frankenphp.WithWorkers("bg-matched", "testdata/bgworker/basic.php", 1, + frankenphp.WithWorkerBackground(), + frankenphp.WithWorkerMatcher(func(*http.Request) bool { return true }), + ), + frankenphp.WithNumThreads(2), + ) + require.ErrorContains(t, err, "cannot match requests") + }) +} diff --git a/caddy/config_test.go b/caddy/config_test.go index 4540ece26b..d46b14753c 100644 --- a/caddy/config_test.go +++ b/caddy/config_test.go @@ -284,3 +284,57 @@ func TestCreateUniqueWorkerNamesQualifiedByServer(t *testing.T) { // workers without a server keep the numeric postfix behavior require.Equal(t, "queue_2", app.createUniqueWorkerName(wc, "")) } + +func TestWorkerBackgroundConfig(t *testing.T) { + d := caddyfile.NewTestDispenser(` + { + php_server { + worker { + name jobs + file ../testdata/worker-with-env.php + num 2 + background + } + } + }`) + module := &FrankenPHPModule{} + + require.NoError(t, module.UnmarshalCaddyfile(d)) + require.Len(t, module.Workers, 1) + require.True(t, module.Workers[0].Background) + require.Equal(t, "jobs", module.Workers[0].Name) +} + +func TestWorkerBackgroundRequiresName(t *testing.T) { + d := caddyfile.NewTestDispenser(` + { + php_server { + worker { + file ../testdata/worker-with-env.php + background + } + } + }`) + module := &FrankenPHPModule{} + + err := module.UnmarshalCaddyfile(d) + require.ErrorContains(t, err, `background workers must have an explicit "name"`) +} + +func TestWorkerBackgroundRejectsMatch(t *testing.T) { + d := caddyfile.NewTestDispenser(` + { + php_server { + worker { + name jobs + file ../testdata/worker-with-env.php + match /jobs/* + background + } + } + }`) + module := &FrankenPHPModule{} + + err := module.UnmarshalCaddyfile(d) + require.ErrorContains(t, err, `"match" is not supported for background workers`) +} diff --git a/caddy/workerconfig.go b/caddy/workerconfig.go index b39eb731e0..3f6f8fb3be 100644 --- a/caddy/workerconfig.go +++ b/caddy/workerconfig.go @@ -38,6 +38,8 @@ type workerConfig struct { MatchPath []string `json:"match_path,omitempty"` // MaxConsecutiveFailures sets the maximum number of consecutive failures before panicking (defaults to 6, set to -1 to never panick) MaxConsecutiveFailures int `json:"max_consecutive_failures,omitempty"` + // Background marks this worker as a background (non-HTTP) worker. + Background bool `json:"background,omitempty"` options []frankenphp.WorkerOption } @@ -139,8 +141,10 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) { } wc.MaxConsecutiveFailures = v + case "background": + wc.Background = true default: - return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, max_consecutive_failures, max_threads", v) + return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, max_consecutive_failures, max_threads, background", v) } } @@ -148,6 +152,15 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) { return wc, d.Err(`the "file" argument must be specified`) } + if wc.Background { + if wc.Name == "" { + return wc, d.Err(`background workers must have an explicit "name"`) + } + if len(wc.MatchPath) != 0 { + return wc, d.Err(`"match" is not supported for background workers`) + } + } + if frankenphp.EmbeddedAppPath != "" && filepath.IsLocal(wc.FileName) { wc.FileName = filepath.Join(frankenphp.EmbeddedAppPath, wc.FileName) } @@ -166,6 +179,10 @@ func (wc *workerConfig) toWorkerOptions() ([]frankenphp.WorkerOption, error) { // options collected while provisioning the module, e.g. the Mercure hub opts = append(opts, wc.options...) + if wc.Background { + opts = append(opts, frankenphp.WithWorkerBackground()) + } + // copy the caddy match logic and create a unique matcher function for this worker // inject the matcher into frankenphp if len(wc.MatchPath) > 0 { diff --git a/docs/config.md b/docs/config.md index 281f05dc75..a951a58e9d 100644 --- a/docs/config.md +++ b/docs/config.md @@ -111,6 +111,7 @@ You can also explicitly configure FrankenPHP using the [global option](https://c watch # Sets the path to watch for file changes. Can be specified more than once for multiple paths. name # Sets the name of the worker, used in logs and metrics. Default: absolute path of worker file max_consecutive_failures # Sets the maximum number of consecutive failures before the worker is considered unhealthy, -1 means the worker will always restart. Default: 6. + background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker; it runs the script in a loop without serving requests, "name" is required, "match" is not allowed. } } } @@ -198,6 +199,7 @@ php_server [] { watch # Sets the path to watch for file changes. Can be specified more than once for multiple paths. env # Sets an extra environment variable to the given value. Can be specified more than once for multiple environment variables. Environment variables for this worker are also inherited from the php_server parent, but can be overwritten here. match # match the worker to a path pattern. Overrides try_files and can only be used in the php_server directive. + background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker; it runs the script in a loop without serving requests, "name" is required, "match" is not allowed. } worker # Can also use the short form like in the global frankenphp block. } diff --git a/frankenphp.c b/frankenphp.c index b15507f69d..22fd5ed899 100644 --- a/frankenphp.c +++ b/frankenphp.c @@ -16,6 +16,7 @@ #else #include #endif +#include
#include #include #include @@ -126,6 +127,11 @@ HashTable *main_thread_env = NULL; static THREAD_LOCAL uintptr_t thread_index; static THREAD_LOCAL bool is_worker_thread = false; +static THREAD_LOCAL bool is_background_worker = false; +/* Stop pipe of a background worker thread: [0] is the read end, exposed to + * the PHP script via frankenphp_get_worker_handle(); [1] is the write end, + * transferred to the Go side, which closes it to signal a drain. */ +static THREAD_LOCAL int worker_stop_fds[2] = {-1, -1}; static THREAD_LOCAL HashTable *sandboxed_env = NULL; /* prepared_env holds entries from php(_server)'s `env KEY VAL`, exposed to * getenv() and merged into $_ENV when 'E' is in variables_order. Separate from @@ -342,7 +348,87 @@ void frankenphp_release_thread_for_kill(force_kill_slot slot) { #endif } +/* Stop pipe of background workers: an anonymous pipe whose read end is + * exposed to the PHP script via frankenphp_get_worker_handle(). When the Go + * side closes the write end (on drain), the read end reaches EOF so the + * script can return from stream_select() and exit its loop gracefully. */ +static int frankenphp_worker_open_stop_pipe(void) { +#ifdef PHP_WIN32 + return _pipe(worker_stop_fds, 4096, _O_BINARY); +#else + return pipe(worker_stop_fds); +#endif +} + +static void frankenphp_worker_close_stop_fds(void) { + for (int i = 0; i < 2; i++) { + if (worker_stop_fds[i] >= 0) { +#ifdef PHP_WIN32 + _close(worker_stop_fds[i]); +#else + close(worker_stop_fds[i]); +#endif + worker_stop_fds[i] = -1; + } + } +} + +static int frankenphp_worker_dup_fd(int fd) { +#ifdef PHP_WIN32 + return _dup(fd); +#else + return dup(fd); +#endif +} + +/* Marks the calling thread as a background worker, opens its stop pipe, + * transfers the write fd to the caller (clearing the TLS slot so a later + * recycle won't double-close it), and disarms max_execution_time. Returns + * -1 if the pipe could not be opened. */ +int frankenphp_set_background_worker_and_get_stop_fd_write(void) { + is_background_worker = true; + /* Bg workers don't enforce max_execution_time; disarm any lingering timer. */ + zend_unset_timeout(); + + frankenphp_worker_close_stop_fds(); + if (frankenphp_worker_open_stop_pipe() != 0) { + worker_stop_fds[0] = -1; + worker_stop_fds[1] = -1; + + return -1; + } + + int fd = worker_stop_fds[1]; + worker_stop_fds[1] = -1; + + return fd; +} + +void frankenphp_worker_close_fd(int fd) { + if (fd < 0) { + return; + } + /* Closing the write end of the stop pipe lands as EOF on the read end, + * so the PHP side's stream_select returns promptly. */ +#ifdef PHP_WIN32 + _close(fd); +#else + close(fd); +#endif +} + void frankenphp_update_local_thread_context(bool is_worker) { + /* A thread that ran a background worker can be recycled into an HTTP + * worker or a regular request thread: reset the bg TLS so + * frankenphp_get_worker_handle() rejects callers again, and release the + * stop pipe's read end. Streams handed out by + * frankenphp_get_worker_handle() own dup'ed fds and were already + * destroyed by request shutdown. */ + if (is_background_worker) { + is_background_worker = false; + frankenphp_worker_close_stop_fds(); + } + is_worker_thread = is_worker; /* workers should keep running if the user aborts the connection */ @@ -857,6 +943,15 @@ PHP_FUNCTION(frankenphp_handle_request) { RETURN_THROWS(); } + if (is_background_worker) { + /* background workers never receive HTTP requests */ + zend_throw_exception( + spl_ce_RuntimeException, + "frankenphp_handle_request() cannot be called from a background worker", + 0); + RETURN_THROWS(); + } + #ifdef ZEND_MAX_EXECUTION_TIMERS /* Disable timeouts while waiting for a request to handle */ zend_unset_timeout(); @@ -1017,6 +1112,46 @@ PHP_FUNCTION(frankenphp_log) { } } +PHP_FUNCTION(frankenphp_get_worker_handle) { + ZEND_PARSE_PARAMETERS_NONE(); + + if (!is_background_worker) { + zend_throw_exception(spl_ce_RuntimeException, + "frankenphp_get_worker_handle() can only be called " + "from a background worker", + 0); + RETURN_THROWS(); + } + + if (worker_stop_fds[0] < 0) { + zend_throw_exception(spl_ce_RuntimeException, + "the background worker stop pipe is not available", 0); + RETURN_THROWS(); + } + + /* Dup so the returned stream owns its fd: closing the stream (or request + * shutdown destroying it) never touches worker_stop_fds[0], which stays + * owned by the C side until the next setup or thread recycle. Each call + * returns a fresh stream over the same pipe; the EOF triggered by a drain + * reaches all of them. */ + int fd = frankenphp_worker_dup_fd(worker_stop_fds[0]); + if (fd < 0) { + zend_throw_exception(spl_ce_RuntimeException, + "failed to dup the background worker stop fd", 0); + RETURN_THROWS(); + } + + php_stream *stream = php_stream_fopen_from_fd(fd, "rb", NULL); + if (stream == NULL) { + frankenphp_worker_close_fd(fd); + zend_throw_exception(spl_ce_RuntimeException, + "failed to create a stream over the stop fd", 0); + RETURN_THROWS(); + } + + php_stream_to_zval(stream, return_value); +} + /* {{{ thread-safe opcache reset */ PHP_FUNCTION(frankenphp_opcache_reset) { go_schedule_opcache_reset(frankenphp_thread_index()); @@ -1526,6 +1661,12 @@ static void *php_thread(void *arg) { frankenphp_override_opcache_reset(); #endif + /* php_request_startup re-arms max_execution_time; background workers + * never enforce it, disarm again. */ + if (is_background_worker) { + zend_unset_timeout(); + } + zend_file_handle file_handle; zend_stream_init_filename(&file_handle, scriptName); diff --git a/frankenphp.go b/frankenphp.go index 79b135b808..87de27f9b4 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -156,11 +156,38 @@ func Config() PHPConfig { } } -func calculateMaxThreads(opt *opt) (numWorkers int, _ error) { +func calculateMaxThreads(opt *opt) (numWorkers int, err error) { maxProcs := runtime.GOMAXPROCS(0) * 2 maxThreadsFromWorkers := 0 + // background workers reserve their thread budget separately so they + // don't count against the HTTP-oriented admission checks below; the + // bump is applied on top of the calculated totals at the end + reservedThreads := 0 + defer func() { + if err != nil { + return + } + opt.numThreads += reservedThreads + if opt.maxThreads > 0 { + // in auto mode (maxThreads < 0), the resolved value is floored + // to numThreads later, which already includes the reservation + opt.maxThreads += reservedThreads + } + numWorkers += reservedThreads + }() + for i, w := range opt.workers { + if w.isBackgroundWorker { + if w.num < 1 { + return 0, fmt.Errorf("background worker %q must declare num >= 1", w.name) + } + metrics.TotalWorkers(w.name, w.num) + reservedThreads += w.num + + continue + } + if w.num <= 0 { // https://github.com/php/frankenphp/issues/126 opt.workers[i].num = maxProcs diff --git a/frankenphp.h b/frankenphp.h index 99ac0ab7ec..8a0ccef685 100644 --- a/frankenphp.h +++ b/frankenphp.h @@ -201,6 +201,10 @@ size_t frankenphp_get_thread_memory_usage(uintptr_t thread_index); void frankenphp_force_kill_thread(force_kill_slot slot); void frankenphp_release_thread_for_kill(force_kill_slot slot); +/* Background worker primitives. */ +int frankenphp_set_background_worker_and_get_stop_fd_write(void); +void frankenphp_worker_close_fd(int fd); + void register_extensions(zend_module_entry **m, int len); #endif diff --git a/frankenphp.stub.php b/frankenphp.stub.php index d6c85aa05f..ecad8b3120 100644 --- a/frankenphp.stub.php +++ b/frankenphp.stub.php @@ -54,3 +54,14 @@ function mercure_publish(string|array $topics, string $data = '', bool $private * array $context Values of the array will be converted to the corresponding Go type (if supported by FrankenPHP) and added to the context of the structured logs using https://pkg.go.dev/log/slog#Attr */ function frankenphp_log(string $message, int $level = 0, array $context = []): void {} + +/** + * Returns a stop-signal stream for the current background worker. The + * stream reaches EOF when FrankenPHP drains the worker, so the script can + * park on stream_select() and exit its loop gracefully. Each call returns + * a fresh stream over the same underlying pipe. Only callable from inside + * a background worker. + * + * @return resource + */ +function frankenphp_get_worker_handle(): mixed {} diff --git a/frankenphp_arginfo.h b/frankenphp_arginfo.h index 4f2707cbca..cb55b70137 100644 --- a/frankenphp_arginfo.h +++ b/frankenphp_arginfo.h @@ -1,5 +1,5 @@ -/* This is a generated file, edit the .stub.php file instead. - * Stub hash: 60f0d27c04f94d7b24c052e91ef294595a2bc421 */ +/* This is a generated file, edit frankenphp.stub.php instead. + * Stub hash: 105fcd19c04e2652e78ebf3abc5ab1f7d724a5d9 */ ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_frankenphp_handle_request, 0, 1, _IS_BOOL, 0) ZEND_ARG_TYPE_INFO(0, callback, IS_CALLABLE, 0) @@ -41,6 +41,8 @@ ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_frankenphp_log, 0, 1, IS_VOID, 0 ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, context, IS_ARRAY, 0, "[]") ZEND_END_ARG_INFO() +ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_frankenphp_get_worker_handle, 0, 0, IS_MIXED, 0) +ZEND_END_ARG_INFO() ZEND_FUNCTION(frankenphp_handle_request); ZEND_FUNCTION(headers_send); @@ -49,7 +51,7 @@ ZEND_FUNCTION(frankenphp_request_headers); ZEND_FUNCTION(frankenphp_response_headers); ZEND_FUNCTION(mercure_publish); ZEND_FUNCTION(frankenphp_log); - +ZEND_FUNCTION(frankenphp_get_worker_handle); static const zend_function_entry ext_functions[] = { ZEND_FE(frankenphp_handle_request, arginfo_frankenphp_handle_request) @@ -63,6 +65,7 @@ static const zend_function_entry ext_functions[] = { ZEND_FALIAS(apache_response_headers, frankenphp_response_headers, arginfo_apache_response_headers) ZEND_FE(mercure_publish, arginfo_mercure_publish) ZEND_FE(frankenphp_log, arginfo_frankenphp_log) + ZEND_FE(frankenphp_get_worker_handle, arginfo_frankenphp_get_worker_handle) ZEND_FE_END }; diff --git a/options.go b/options.go index e1eaeb7b55..930eb5d182 100644 --- a/options.go +++ b/options.go @@ -57,6 +57,7 @@ type workerOpt struct { onServerStartup func() onServerShutdown func() server *Server + isBackgroundWorker bool } // WithContext sets the main context to use. @@ -239,6 +240,20 @@ func WithWorkerServerScope(s *Server) WorkerOption { } } +// EXPERIMENTAL: WithWorkerBackground marks this worker as a background +// (non-HTTP) worker. Background workers run outside the request cycle: +// they share the PHP runtime with HTTP threads but never receive HTTP +// requests. The script can park on the stream returned by +// frankenphp_get_worker_handle(), which reaches EOF when FrankenPHP +// drains the worker, to exit gracefully on shutdown or restart. +func WithWorkerBackground() WorkerOption { + return func(w *workerOpt) error { + w.isBackgroundWorker = true + + return nil + } +} + // WithWorkerMaxFailures sets the maximum number of consecutive failures before panicking func WithWorkerMaxFailures(maxFailures int) WorkerOption { return func(w *workerOpt) error { diff --git a/phpmainthread.go b/phpmainthread.go index 64c782319b..5b92d491ca 100644 --- a/phpmainthread.go +++ b/phpmainthread.go @@ -161,6 +161,10 @@ func (mainThread *phpMainThread) rebootAllThreads() bool { for _, thread := range rebootingThreads { rebootWg.Go(func() { + // wake up handlers parked in a blocking C call (background + // workers' stream_select on the stop pipe) so they can yield + // for the reboot without waiting for the force-kill below + thread.handler.drain() close(thread.drainChan) if thread.state.WaitForStateWithTimeout(rebootGracePeriod, state.YieldingForReboot) { return diff --git a/phpthread.go b/phpthread.go index 39ee24b5d4..63c4545d67 100644 --- a/phpthread.go +++ b/phpthread.go @@ -39,11 +39,10 @@ type threadHandler interface { beforeScriptExecution() string afterScriptExecution(exitStatus int) frankenPHPContext() *frankenPHPContext - // drain is a hook called by drainWorkerThreads right before drainChan is - // closed. Handlers that need to wake up a thread parked in a blocking C - // call (e.g. by closing a stop pipe) plug their signal in here. All - // current handlers are no-ops; this is the seam later handler types use - // without having to modify drainWorkerThreads. + // drain is a hook called right before drainChan is closed on shutdown + // and reboot. Handlers that need to wake up a thread parked in a + // blocking C call (background workers' stream_select on the stop pipe) + // plug their signal in here; the other handlers are no-ops. drain() } @@ -119,6 +118,9 @@ func (thread *phpThread) shutdown() { return } + // wake up handlers parked in a blocking C call (background workers' + // stream_select on the stop pipe); no-op for the other handlers + thread.handler.drain() close(thread.drainChan) // Arm force-kill after the grace period to wake any thread stuck in diff --git a/requestoptions.go b/requestoptions.go index 8414fb557d..0d408ad998 100644 --- a/requestoptions.go +++ b/requestoptions.go @@ -5,6 +5,7 @@ import ( "log/slog" "net/http" "path/filepath" + "strconv" "strings" "sync" "sync/atomic" @@ -208,10 +209,16 @@ 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 == "" { + return nil } + w := workersByName[name] + if w != nil && w.isBackgroundWorker { + return errors.New("background worker " + strconv.Quote(name) + " cannot handle requests") + } + o.worker = w + return nil } } diff --git a/testdata/bgworker/basic.php b/testdata/bgworker/basic.php new file mode 100644 index 0000000000..48a2305be8 --- /dev/null +++ b/testdata/bgworker/basic.php @@ -0,0 +1,20 @@ += 0 { + C.frankenphp_worker_close_fd(C.int(fd)) + } +} + +// beforeScriptExecution returns the name of the script or an empty string on shutdown +func (handler *backgroundWorkerThread) beforeScriptExecution() string { + switch handler.state.Get() { + case state.TransitionRequested: + if handler.worker.onThreadShutdown != nil { + handler.worker.onThreadShutdown(handler.thread.threadIndex) + } + handler.worker.detachThread(handler.thread) + return handler.thread.transitionToNewHandler() + case state.Ready, state.TransitionComplete: + handler.thread.updateContext(true) + if handler.worker.onThreadReady != nil { + handler.worker.onThreadReady(handler.thread.threadIndex) + } + + for { + err := handler.setupScript() + if err == nil { + return handler.worker.fileName + } + + if globalLogger.Enabled(globalCtx, slog.LevelError) { + globalLogger.LogAttrs(globalCtx, slog.LevelError, "failed to start background worker", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex), slog.Any("error", err)) + } + + // fail fast during startup so Init() surfaces the error to the + // operator; past startup, back off and retry like a crash + if startupFailChan != nil { + startupFailChan <- err + handler.thread.state.Set(state.ShuttingDown) + return handler.beforeScriptExecution() + } + + handler.backoff() + if !handler.state.Is(state.Ready) { + // drained during the backoff (shutdown, reboot, transition) + return handler.beforeScriptExecution() + } + } + case state.Rebooting, state.ForceRebooting: + return "" + case state.RebootReady: + handler.state.Set(state.Ready) + return handler.beforeScriptExecution() + case state.ShuttingDown: + if handler.worker.onThreadShutdown != nil { + handler.worker.onThreadShutdown(handler.thread.threadIndex) + } + handler.worker.detachThread(handler.thread) + + // signal to stop + return "" + default: + panic("unexpected state: " + handler.state.Name()) + } +} + +// setupScript marks the thread as a background worker on the C side and +// takes ownership of the stop pipe's write end. +func (handler *backgroundWorkerThread) setupScript() error { + fd := int32(C.frankenphp_set_background_worker_and_get_stop_fd_write()) + if fd < 0 { + return fmt.Errorf("failed to create the stop pipe of background worker %q", handler.worker.name) + } + handler.stopFdWrite.Store(fd) + + switch handler.state.Get() { + case state.ShuttingDown, state.Rebooting, state.ForceRebooting: + // a concurrent drain may have run before the fd was published; + // close it now so the script observes EOF immediately + handler.drain() + } + + fc, err := newWorkerDummyContext(handler.worker) + if err != nil { + handler.drain() + return err + } + handler.dummyFrankenPHPContext = fc + + metrics.StartWorker(handler.worker.name) + // background workers have no later "reached steady state" marker like + // frankenphp_handle_request; they count as ready once the script starts + metrics.ReadyWorker(handler.worker.name) + + if fc.logger.Enabled(fc.ctx, slog.LevelDebug) { + fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "starting background worker", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex)) + } + + if handler.state.Is(state.TransitionComplete) { + handler.state.Set(state.Ready) + } + + return nil +} + +func (handler *backgroundWorkerThread) afterScriptExecution(exitStatus int) { + // the write end of the stop pipe belongs to this thread; release it on + // every exit path so the next run gets a fresh pipe (drain() already + // took it when the exit was drain-triggered) + handler.drain() + worker := handler.worker + handler.dummyFrankenPHPContext = nil + + // cooperative exit: the script is re-run with a reset backoff, unless + // the thread is being drained (beforeScriptExecution checks the state) + if exitStatus == 0 { + metrics.StopWorker(worker.name, StopReasonRestart) + + if globalLogger.Enabled(globalCtx, slog.LevelDebug) { + globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "restarting background worker", slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex)) + } + + handler.failureCount = 0 + return + } + + metrics.StopWorker(worker.name, StopReasonCrash) + + // max_consecutive_failures only fails hard during startup, where it + // surfaces on startupFailChan so Init() returns the error to the + // operator. Past startup, a crashing background worker keeps + // restarting with a louder log line: silently giving up would leave + // the server in a broken half-state with no clear way to recover. + pastCap := worker.maxConsecutiveFailures >= 0 && handler.failureCount >= worker.maxConsecutiveFailures + if pastCap && startupFailChan != nil && !watcherIsEnabled { + startupFailChan <- fmt.Errorf("too many consecutive failures: background worker %s keeps crashing", worker.fileName) + handler.thread.state.Set(state.ShuttingDown) + return + } + + logLevel := slog.LevelWarn + logMsg := "background worker crashed, restarting" + if pastCap { + logLevel = slog.LevelError + logMsg = "background worker exceeded max_consecutive_failures, still restarting" + } + if globalLogger.Enabled(globalCtx, logLevel) { + globalLogger.LogAttrs(globalCtx, logLevel, logMsg, slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex), slog.Int("failures", handler.failureCount), slog.Int("exit_status", exitStatus)) + } + + handler.backoff() +} + +// backoff waits before the next run of a crashed script: quadratic in the +// number of consecutive failures, capped at 1 second. +func (handler *backgroundWorkerThread) backoff() { + backoffDuration := time.Duration(handler.failureCount*handler.failureCount*100) * time.Millisecond + if backoffDuration > time.Second { + backoffDuration = time.Second + } + handler.failureCount++ + time.Sleep(backoffDuration) +} diff --git a/worker.go b/worker.go index ecbed9e392..26b9a5ed82 100644 --- a/worker.go +++ b/worker.go @@ -34,6 +34,8 @@ type worker struct { onThreadShutdown func(int) queuedRequests atomic.Int32 server *Server + // isBackgroundWorker marks this as a background (non-HTTP) worker + isBackgroundWorker bool } var ( @@ -67,6 +69,11 @@ func initWorkers(opts []workerOpt) error { totalThreadsToStart += w.num workers = append(workers, w) workersByName[w.name] = w + // background workers never serve HTTP requests, so they stay out + // of the path-matching maps used to route requests to workers + if w.isBackgroundWorker { + continue + } if w.server == nil { globalWorkersByPath[w.fileName] = w } else if err := w.server.addWorker(w); err != nil { @@ -79,7 +86,11 @@ func initWorkers(opts []workerOpt) error { for _, w := range workers { for i := 0; i < w.num; i++ { thread := getInactivePHPThread() - convertToWorkerThread(thread, w) + if w.isBackgroundWorker { + convertToBackgroundWorkerThread(thread, w) + } else { + convertToWorkerThread(thread, w) + } workersReady.Go(func() { thread.state.WaitFor(state.Ready, state.ShuttingDown, state.Done) @@ -119,12 +130,25 @@ func newWorker(o workerOpt) (*worker, error) { return nil, fmt.Errorf("worker file not found %q: %w", absFileName, err) } + if o.isBackgroundWorker { + // the name is the script's identity (exposed via FRANKENPHP_WORKER); + // empty names are reserved for the catch-all workers of a future build + if o.name == "" { + return nil, fmt.Errorf("background worker %q must have an explicit name", o.fileName) + } + if o.matchRequest != nil { + return nil, fmt.Errorf("background worker %q cannot match requests", o.name) + } + } + if o.name == "" { o.name = absFileName } if o.server == nil { - if globalWorkersByPath[absFileName] != nil { + // background workers are keyed by name and stay out of globalWorkersByPath, + // so several of them may share one script + if !o.isBackgroundWorker && globalWorkersByPath[absFileName] != nil { return nil, fmt.Errorf("two global workers cannot have the same filename: %q", absFileName) } @@ -152,7 +176,14 @@ func newWorker(o workerOpt) (*worker, error) { } } - o.env["FRANKENPHP_WORKER\x00"] = "1" + // $_SERVER['FRANKENPHP_WORKER'] is "1" for HTTP workers (existing + // behavior) and the worker name for background workers, so a bg + // script can know which declaration it runs for + if o.isBackgroundWorker { + o.env["FRANKENPHP_WORKER\x00"] = o.name + } else { + o.env["FRANKENPHP_WORKER\x00"] = "1" + } w := &worker{ name: o.name, @@ -167,6 +198,7 @@ func newWorker(o workerOpt) (*worker, error) { onThreadReady: o.onThreadReady, onThreadShutdown: o.onThreadShutdown, server: o.server, + isBackgroundWorker: o.isBackgroundWorker, } w.configureMercure(&o) From 3618c7af9f0da88c09e22ab2062d4924333e2ec3 Mon Sep 17 00:00:00 2001 From: Nicolas Grekas Date: Sun, 23 Aug 2026 15:02:53 +0200 Subject: [PATCH 2/2] address review: drain on transitions, timer guards, cooperative-exit guard - setHandler() closed drainChan without calling the old handler's drain(), so a background script parked in stream_select slept through handler transitions (autoscaling, thread recycling) until the force-kill grace period. Drain first, guarding against the nil handler of a fresh thread. - Wrap the two zend_unset_timeout() calls in #ifdef ZEND_MAX_EXECUTION_TIMERS, matching every other timer call site; on builds without max-execution-timers (macOS) the setitimer path must stay untouched. - A background script that exited 0 without ever fetching its handle was treated as a cooperative exit and respawned immediately: an early return spun in a tight loop. Fetching the handle via frankenphp_get_worker_handle() is now the "reached steady state" marker, the background analog of HTTP workers reaching frankenphp_handle_request(): exits without it count as boot failures (backoff, max_consecutive_failures fails Init during startup). ReadyWorker moves from script start to handle fetch, so the ready gauge only counts scripts that actually reached their park point and stays balanced with StopReasonBootFailure. - Spell out the fd lifecycle at the dup site: dup'ed fds die with their streams at request shutdown, the read end is closed by the next setup or thread recycle, the write end by the Go side on every exit path. --- bgworker_test.go | 23 ++++++++++++-- docs/config.md | 4 +-- frankenphp.c | 14 +++++++- phpthread.go | 6 ++++ testdata/bgworker/early-return.php | 5 +++ threadbackgroundworker.go | 51 ++++++++++++++++++++++++------ 6 files changed, 88 insertions(+), 15 deletions(-) create mode 100644 testdata/bgworker/early-return.php diff --git a/bgworker_test.go b/bgworker_test.go index 04199dc0ee..a4affa0958 100644 --- a/bgworker_test.go +++ b/bgworker_test.go @@ -128,14 +128,33 @@ func TestBackgroundWorkerValidation(t *testing.T) { }) t.Run("names are global", func(t *testing.T) { - err := frankenphp.Init( + // the second worker being scoped to a server changes nothing: the + // name namespace is global, so Init refuses the duplicate + server, err := frankenphp.NewServer(testDataDir) + require.NoError(t, err) + err = frankenphp.Init( + frankenphp.WithServer(server), frankenphp.WithWorkers("bg-shared", "testdata/bgworker/basic.php", 1, frankenphp.WithWorkerBackground()), - frankenphp.WithWorkers("bg-shared", "testdata/bgworker/named.php", 1, frankenphp.WithWorkerBackground()), + frankenphp.WithWorkers("bg-shared", "testdata/bgworker/named.php", 1, + frankenphp.WithWorkerBackground(), + frankenphp.WithWorkerServerScope(server), + ), frankenphp.WithNumThreads(3), ) require.ErrorContains(t, err, "two workers cannot have the same name") }) + t.Run("early return without the handle fails startup", func(t *testing.T) { + err := frankenphp.Init( + frankenphp.WithWorkers("bg-early", "testdata/bgworker/early-return.php", 1, + frankenphp.WithWorkerBackground(), + frankenphp.WithWorkerMaxFailures(2), + ), + frankenphp.WithNumThreads(2), + ) + require.ErrorContains(t, err, "frankenphp_get_worker_handle") + }) + t.Run("request matchers are rejected", func(t *testing.T) { err := frankenphp.Init( frankenphp.WithWorkers("bg-matched", "testdata/bgworker/basic.php", 1, diff --git a/docs/config.md b/docs/config.md index a951a58e9d..5d2655e843 100644 --- a/docs/config.md +++ b/docs/config.md @@ -111,7 +111,7 @@ You can also explicitly configure FrankenPHP using the [global option](https://c watch # Sets the path to watch for file changes. Can be specified more than once for multiple paths. name # Sets the name of the worker, used in logs and metrics. Default: absolute path of worker file max_consecutive_failures # Sets the maximum number of consecutive failures before the worker is considered unhealthy, -1 means the worker will always restart. Default: 6. - background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker; it runs the script in a loop without serving requests, "name" is required, "match" is not allowed. + background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker; it runs the script in a loop without serving requests, "name" is required, "match" is not allowed. The script must fetch its stop stream via frankenphp_get_worker_handle() before returning, an exit without it counts as a failure. } } } @@ -199,7 +199,7 @@ php_server [] { watch # Sets the path to watch for file changes. Can be specified more than once for multiple paths. env # Sets an extra environment variable to the given value. Can be specified more than once for multiple environment variables. Environment variables for this worker are also inherited from the php_server parent, but can be overwritten here. match # match the worker to a path pattern. Overrides try_files and can only be used in the php_server directive. - background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker; it runs the script in a loop without serving requests, "name" is required, "match" is not allowed. + background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker; it runs the script in a loop without serving requests, "name" is required, "match" is not allowed. The script must fetch its stop stream via frankenphp_get_worker_handle() before returning, an exit without it counts as a failure. } worker # Can also use the short form like in the global frankenphp block. } diff --git a/frankenphp.c b/frankenphp.c index 22fd5ed899..5f8bd05b99 100644 --- a/frankenphp.c +++ b/frankenphp.c @@ -387,8 +387,10 @@ static int frankenphp_worker_dup_fd(int fd) { * -1 if the pipe could not be opened. */ int frankenphp_set_background_worker_and_get_stop_fd_write(void) { is_background_worker = true; +#ifdef ZEND_MAX_EXECUTION_TIMERS /* Bg workers don't enforce max_execution_time; disarm any lingering timer. */ zend_unset_timeout(); +#endif frankenphp_worker_close_stop_fds(); if (frankenphp_worker_open_stop_pipe() != 0) { @@ -1133,7 +1135,11 @@ PHP_FUNCTION(frankenphp_get_worker_handle) { * shutdown destroying it) never touches worker_stop_fds[0], which stays * owned by the C side until the next setup or thread recycle. Each call * returns a fresh stream over the same pipe; the EOF triggered by a drain - * reaches all of them. */ + * reaches all of them. No fd outlives a script run: the dup'ed fds die + * with their streams at request shutdown, the read end is closed by the + * next frankenphp_set_background_worker_and_get_stop_fd_write() or by + * frankenphp_update_local_thread_context() on recycle, and the write end + * belongs to the Go side, which closes it on every exit path (drain). */ int fd = frankenphp_worker_dup_fd(worker_stop_fds[0]); if (fd < 0) { zend_throw_exception(spl_ce_RuntimeException, @@ -1149,6 +1155,10 @@ PHP_FUNCTION(frankenphp_get_worker_handle) { RETURN_THROWS(); } + /* the script reached its park point: report it ready so a clean exit + * counts as cooperative and the ready metric reflects reality */ + go_frankenphp_background_worker_ready(frankenphp_thread_index()); + php_stream_to_zval(stream, return_value); } @@ -1661,11 +1671,13 @@ static void *php_thread(void *arg) { frankenphp_override_opcache_reset(); #endif +#ifdef ZEND_MAX_EXECUTION_TIMERS /* php_request_startup re-arms max_execution_time; background workers * never enforce it, disarm again. */ if (is_background_worker) { zend_unset_timeout(); } +#endif zend_file_handle file_handle; zend_stream_init_filename(&file_handle, scriptName); diff --git a/phpthread.go b/phpthread.go index 63c4545d67..b176b26e7b 100644 --- a/phpthread.go +++ b/phpthread.go @@ -159,6 +159,12 @@ func (thread *phpThread) setHandler(handler threadHandler) { return } + // wake up a handler parked in a blocking C call (background workers' + // stream_select on the stop pipe) so it can yield for the transition; + // nil on the very first conversion of a fresh thread + if thread.handler != nil { + thread.handler.drain() + } close(thread.drainChan) thread.state.WaitFor(state.TransitionInProgress) diff --git a/testdata/bgworker/early-return.php b/testdata/bgworker/early-return.php new file mode 100644 index 0000000000..13727cf25f --- /dev/null +++ b/testdata/bgworker/early-return.php @@ -0,0 +1,5 @@ += 0 && handler.failureCount >= worker.maxConsecutiveFailures if pastCap && startupFailChan != nil && !watcherIsEnabled { - startupFailChan <- fmt.Errorf("too many consecutive failures: background worker %s keeps crashing", worker.fileName) + if exitStatus == 0 { + startupFailChan <- fmt.Errorf("background worker %s exits without calling frankenphp_get_worker_handle()", worker.fileName) + } else { + startupFailChan <- fmt.Errorf("too many consecutive failures: background worker %s keeps crashing", worker.fileName) + } handler.thread.state.Set(state.ShuttingDown) return } logLevel := slog.LevelWarn logMsg := "background worker crashed, restarting" + if exitStatus == 0 { + logMsg = "background worker exited without fetching its handle, restarting" + } if pastCap { logLevel = slog.LevelError logMsg = "background worker exceeded max_consecutive_failures, still restarting" @@ -205,6 +225,17 @@ func (handler *backgroundWorkerThread) afterScriptExecution(exitStatus int) { handler.backoff() } +//export go_frankenphp_background_worker_ready +func go_frankenphp_background_worker_ready(threadIndex C.uintptr_t) { + // called on the PHP thread while the script executes; the handler is a + // backgroundWorkerThread because frankenphp_get_worker_handle() throws + // on every other thread kind + if handler, ok := phpThreads[threadIndex].handler.(*backgroundWorkerThread); ok && !handler.reachedHandle { + handler.reachedHandle = true + metrics.ReadyWorker(handler.worker.name) + } +} + // backoff waits before the next run of a crashed script: quadratic in the // number of consecutive failures, capped at 1 second. func (handler *backgroundWorkerThread) backoff() {