From ba70f14636b2ec3586c0092988491f91019ba4d7 Mon Sep 17 00:00:00 2001 From: Nicolas Grekas Date: Mon, 20 Jul 2026 17:23:13 +0200 Subject: [PATCH] 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 | 145 ++++++++++++++++++++++++ caddy/config_test.go | 54 +++++++++ caddy/workerconfig.go | 19 +++- docs/config.md | 2 + frankenphp.c | 142 +++++++++++++++++++++++ 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 | 36 +++++- 18 files changed, 763 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..60f02f8013 --- /dev/null +++ b/bgworker_test.go @@ -0,0 +1,145 @@ +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("sidekick-server", testDataDir, nil, map[string]string{"BG_SENTINEL_DIR": tmp}, nil) + 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 04498ecf05..b1746a5998 100644 --- a/caddy/config_test.go +++ b/caddy/config_test.go @@ -269,3 +269,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 88e7bc9aa5..30c8d69817 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 requestOptions []frankenphp.RequestOption @@ -140,8 +142,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) } } @@ -149,6 +153,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) } @@ -165,6 +178,10 @@ func (wc *workerConfig) toWorkerOptions() []frankenphp.WorkerOption { frankenphp.WithWorkerRequestOptions(wc.requestOptions...), } + 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 1923aa58d1..8d1ab5a046 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. } } } @@ -197,6 +198,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 823ba11c1c..b633002b15 100644 --- a/frankenphp.c +++ b/frankenphp.c @@ -16,6 +16,7 @@ #else #include #endif +#include
#include #include #include @@ -120,6 +121,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 @@ -336,7 +342,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 */ @@ -843,6 +929,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(); @@ -1003,6 +1098,47 @@ 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); +} + static void frankenphp_opcache_restart_hook(int reason) { (void)reason; go_schedule_opcache_reset(frankenphp_thread_index()); @@ -1526,6 +1662,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 478ddb4de2..5c215b8623 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 db32a82fe0..b2bf157ab6 100644 --- a/frankenphp.h +++ b/frankenphp.h @@ -228,6 +228,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 6a66d85dc4..6d6899fbd7 100644 --- a/options.go +++ b/options.go @@ -54,6 +54,7 @@ type workerOpt struct { onServerStartup func() onServerShutdown func() server *Server + isBackgroundWorker bool } // WithContext sets the main context to use. @@ -236,6 +237,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 175460ac1e..f6bc436265 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 cf684e8f7c..9119eea066 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,11 +130,22 @@ 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 && globalWorkersByPath[absFileName] != nil { + if o.server == nil && !o.isBackgroundWorker && globalWorkersByPath[absFileName] != nil { return nil, fmt.Errorf("two global workers cannot have the same filename: %q", absFileName) } @@ -145,7 +167,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, @@ -160,6 +189,7 @@ func newWorker(o workerOpt) (*worker, error) { onThreadReady: o.onThreadReady, onThreadShutdown: o.onThreadShutdown, server: o.server, + isBackgroundWorker: o.isBackgroundWorker, } w.configureMercure(&o)