From 2eec8f4e3aae427a9c49ea3679b6599ab6c2e63a Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 21 Aug 2026 12:09:38 +0200 Subject: [PATCH 1/7] adds ping implementation --- caddy/config_test.go | 93 +++++++++++++++++++++++++++ caddy/workerconfig.go | 70 ++++++++++++++++++++- docs/config.md | 34 ++++++++++ docs/worker.md | 43 +++++++++++++ frankenphp.go | 5 +- frankenphp_test.go | 24 +++++++ options.go | 15 +++++ ping.go | 143 ++++++++++++++++++++++++++++++++++++++++++ ping_test.go | 18 ++++++ worker.go | 21 +++++++ 10 files changed, 464 insertions(+), 2 deletions(-) create mode 100644 ping.go create mode 100644 ping_test.go diff --git a/caddy/config_test.go b/caddy/config_test.go index 4540ece26b..ce88598c35 100644 --- a/caddy/config_test.go +++ b/caddy/config_test.go @@ -284,3 +284,96 @@ func TestCreateUniqueWorkerNamesQualifiedByServer(t *testing.T) { // workers without a server keep the numeric postfix behavior require.Equal(t, "queue_2", app.createUniqueWorkerName(wc, "")) } + +func TestModuleWorkerWithPingConfiguration(t *testing.T) { + configWithPing := ` + { + php { + worker ../testdata/worker-with-counter.php { + ping 60s health + ping each 1m aligned message + ping overlap 1h aligned message + ping idle 3s "HELLO THERE!!!!" + } + } + }` + + d := caddyfile.NewTestDispenser(configWithPing) + module := &FrankenPHPModule{} + + err := module.UnmarshalCaddyfile(d) + require.NoError(t, err) + require.Len(t, module.Workers, 1) + + pings := module.Workers[0].Pings + require.Len(t, pings, 4) + require.Equal(t, 60*time.Second, pings[0].Interval) + require.Equal(t, "health", pings[0].Message) + require.False(t, pings[0].Aligned) + require.Equal(t, frankenphp.PingModeSynchronous, pings[0].Mode) + + require.Equal(t, time.Minute, pings[1].Interval) + require.Equal(t, "message", pings[1].Message) + require.True(t, pings[1].Aligned) + require.Equal(t, frankenphp.PingModeEach, pings[1].Mode) + + require.Equal(t, time.Hour, pings[2].Interval) + require.Equal(t, "message", pings[2].Message) + require.True(t, pings[2].Aligned) + require.Equal(t, frankenphp.PingModeOverlapping, pings[2].Mode) + + require.Equal(t, "HELLO THERE!!!!", pings[3].Message) + require.False(t, pings[3].Aligned) + require.Equal(t, frankenphp.PingModeIdle, pings[3].Mode) + require.Equal(t, 3*time.Second, pings[3].Interval) +} + +func TestModuleWorkerWithInvalidPingConfiguration(t *testing.T) { + tests := []struct { + name string + config string + }{ + { + name: "missing message", + config: `{ + php { + worker { + file ../testdata/worker-with-counter.php + ping 60s + } + } + }`, + }, + { + name: "invalid interval", + config: `{ + php { + worker { + file ../testdata/worker-with-counter.php + ping not-a-duration health + } + } + }`, + }, + { + name: "each must come first", + config: `{ + php { + worker { + file ../testdata/worker-with-counter.php + ping 60s health each + } + } + }`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + d := caddyfile.NewTestDispenser(tt.config) + module := &FrankenPHPModule{} + err := module.UnmarshalCaddyfile(d) + require.Error(t, err) + }) + } +} diff --git a/caddy/workerconfig.go b/caddy/workerconfig.go index b39eb731e0..ba13ffc010 100644 --- a/caddy/workerconfig.go +++ b/caddy/workerconfig.go @@ -1,8 +1,10 @@ package caddy import ( + "fmt" "path/filepath" "strconv" + "time" "github.com/caddyserver/caddy/v2" "github.com/caddyserver/caddy/v2/caddyconfig/caddyfile" @@ -22,6 +24,7 @@ import ( type workerConfig struct { mercureContext + // Name for the worker. Default: the absolute path of the worker file, postfixed with a number if the name is already used. // Name for the worker. Default: the absolute path of the worker file, postfixed with a number if the name is already used. Name string `json:"name,omitempty"` // FileName sets the path to the worker script. @@ -38,10 +41,26 @@ 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"` + // Pings configures periodic internal messages sent to the worker. + Pings []*pingConfig `json:"pings,omitempty"` options []frankenphp.WorkerOption } +type pingConfig struct { + Interval time.Duration `json:"interval"` + Message string `json:"message"` + Aligned bool `json:"aligned,omitempty"` + Mode frankenphp.PingMode `json:"mode,omitempty"` +} + +var pingModes = map[string]frankenphp.PingMode{ + "sync": frankenphp.PingModeSynchronous, + "overlap": frankenphp.PingModeOverlapping, + "each": frankenphp.PingModeEach, + "idle": frankenphp.PingModeIdle, +} + func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) { wc := workerConfig{} if d.NextArg() { @@ -139,8 +158,15 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) { } wc.MaxConsecutiveFailures = v + case "ping": + ping, err := parsePingConfig(d) + if err != nil { + return wc, d.WrapErr(err) + } + + wc.Pings = append(wc.Pings, ping) 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, ping, max_consecutive_failures, max_threads", v) } } @@ -175,5 +201,47 @@ func (wc *workerConfig) toWorkerOptions() ([]frankenphp.WorkerOption, error) { } opts = append(opts, frankenphp.WithWorkerMatcher(matchFunc.Match)) } + + if len(wc.Pings) > 0 { + for _, p := range wc.Pings { + opts = append(opts, frankenphp.WithWorkerPings(p.Mode, p.Interval, p.Message, p.Aligned)) + } + } + return opts, nil } + +// parse the configuration for recurring pings to the worker +// ping 1s "Hello, world!" +// ping overlap aligned 1m "Hello, world!" +func parsePingConfig(d *caddyfile.Dispenser) (*pingConfig, error) { + args := d.RemainingArgs() + if len(args) < 2 { + return nil, d.ArgErr() + } + + mode := frankenphp.PingModeSynchronous + if m, ok := pingModes[args[0]]; ok { + mode, args = m, args[1:] + } + + aligned := false + if len(args) > 2 && args[len(args)-2] == "aligned" { + aligned = true + args = append(args[:len(args)-2], args[len(args)-1]) + } + + if len(args) != 2 { + return nil, d.ArgErr() + } + + interval, err := time.ParseDuration(args[0]) + if err != nil { + return nil, fmt.Errorf("ping interval must be a valid duration, received: %s (%s)", args[0], err) + } + if interval <= 0 { + return nil, fmt.Errorf("ping interval must be positive, received: %s (%s)", args[0], interval) + } + + return &pingConfig{Interval: interval, Message: args[1], Aligned: aligned, Mode: mode}, nil +} diff --git a/docs/config.md b/docs/config.md index 281f05dc75..3b9bf84953 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. + ping [aligned] # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align pings to the start of each interval. Mode can be sync, overlap, each and idle. Can be specified more than once. } } } @@ -198,6 +199,8 @@ 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. + ping [aligned] # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align pings to the start of each interval. Can be specified more than once. + ping [aligned] # Like ping, with an explicit mode (sync, overlap, each, idle). } worker # Can also use the short form like in the global frankenphp block. } @@ -248,6 +251,37 @@ where the FrankenPHP process was started. You can instead also specify one or mo - If you have multiple workers defined, all of them will be restarted when a file changes - Be wary about watching files that are created at runtime (like logs) since they might cause unwanted worker restarts. +### Pinging workers + +Workers can receive periodic messages via the `ping` directive. +The message is passed as the first argument to the closure given to `frankenphp_handle_request()`, like [extension workers](extension-workers.md#headless-mode-sendmessage). +This is useful to run scheduled tasks inside your worker script (for example, to flush queues or run cron jobs). + +```caddyfile +worker /path/to/worker.php { + ping 60s run-tasks + ping each 1m aligned minutely +} +``` + +```php +while (frankenphp_handle_request(function ($message) { + match ($message) { + 'run-tasks' => runTasks(), + 'minutely' => runMinutelyTasks(), + default => handleRequest(), + }; +})) { +} +``` + +The interval must be a [Go duration](https://pkg.go.dev/time#ParseDuration) such as `60s`, `1m`, or `5m`. +Add the `aligned` keyword to align pings to the start of each interval (e.g. `ping 1m aligned minutely` runs at the start of every minute). + +By default, each ping is handled by a single worker thread. Prefix the interval with `each` to send the ping to every thread of the worker. This is useful when each thread maintains its own state and must run the scheduled task independently. + +You can specify multiple `ping` lines to schedule different messages at different intervals. + The file watcher is based on [e-dant/watcher](https://github.com/e-dant/watcher). ## Matching the worker to a path diff --git a/docs/worker.md b/docs/worker.md index 466c7cf684..0418c955d1 100644 --- a/docs/worker.md +++ b/docs/worker.md @@ -256,3 +256,46 @@ while (\frankenphp_handle_request($handler)) { When writing worker scripts, make sure to reset any request-specific state between requests. Frameworks like [Symfony](symfony.md) and [Laravel Octane](laravel.md) take care of resetting most state for you, but you may still need to reset your own services. With Symfony, services that hold request-specific state should implement [`Symfony\Contracts\Service\ResetInterface`](https://github.com/symfony/contracts/blob/main/Service/ResetInterface.php) so they're reset by the kernel between requests. + +## Pinging + +Workers can also be pinged repeatedly with a message. + +```caddyfile +worker /path/to/worker.php { + ping 1s "Hello Worker" # send a single ping with message "Hello Worker" each second +} +``` + +In the worker script, the function passed to `frankenphp_handle_request()` will receive the message directly as an argument: + +```php +$handler = function(string $message = "") { + echo $message; # "Hello Worker" +} + +while(frankenphp_handle_request($handler)){} +``` + +### Ping modes + +Available modes for pinging are: "sync", "overlap", "each" and "idle". + +```caddyfile +worker /path/to/worker { + ping sync 10s "message" # send a single ping each 10s, wait for completion in-between pings + ping overlap 10s "message" # send a single ping each 10s, don't wait for completion + ping each 10s "message" # send pings to each active thread every 10s, don't wait for completion + ping idle 10s "message" # send pings to each active thread every 10s if the thread has been idle for more than 10s +} +``` + +### Aligned pings + +A ping can also be aligned to the start of each interval with the `aligned` keyword (cron-like). + +```caddyfile +worker /path/to/worker { + ping overlap 1m aligned "message" # send the ping at the exact start of each minute +} +``` diff --git a/frankenphp.go b/frankenphp.go index 79b135b808..7af6f9e6dd 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -359,6 +359,8 @@ func Init(options ...Option) error { } } + initPings() + return nil } @@ -380,6 +382,7 @@ func shutdown() { fn() } + shutdownPings() drainWatchers() drainPHPThreads() unregisterServers() @@ -629,7 +632,7 @@ func go_sapi_flush(threadIndex C.uintptr_t) bool { func go_read_post(threadIndex C.uintptr_t, cBuf *C.char, countBytes C.size_t) (readBytes C.size_t) { fc := phpThreads[threadIndex].handler.frankenPHPContext() - if fc.responseWriter == nil { + if fc.responseWriter == nil || fc.request == nil { return 0 } diff --git a/frankenphp_test.go b/frankenphp_test.go index 19fb3dd06f..e3e89d677a 100644 --- a/frankenphp_test.go +++ b/frankenphp_test.go @@ -1425,3 +1425,27 @@ func testOpcachePreload(t *testing.T, opts *testOptions) { assert.Equal(t, "I am preloaded", body) }, opts) } + +func TestPings(t *testing.T) { + logger, buf := newTestLogger(t) + require.NoError(t, frankenphp.Init( + frankenphp.WithLogger(logger), + frankenphp.WithWorkers("ping-worker", "testdata/worker-with-counter.php", 1, + frankenphp.WithWorkerPings(frankenphp.PingModeSynchronous, 100*time.Microsecond, "ping", false), + ), + )) + t.Cleanup(frankenphp.Shutdown) + + i := 0 + for { + output := buf.String() + if strings.Contains(output, "requests:1") { + break + } + time.Sleep(500 * time.Microsecond) + i++ + if i > 10000 { // 5s timeout + t.Fatal("timed out without recording a worker ping") + } + } +} diff --git a/options.go b/options.go index e1eaeb7b55..b65d6c0b18 100644 --- a/options.go +++ b/options.go @@ -50,6 +50,7 @@ type workerOpt struct { requestOptions []RequestOption watch []string matchRequest func(*http.Request) bool + pings []*ping maxConsecutiveFailures int extensionWorkers *extensionWorkers onThreadReady func(int) @@ -239,6 +240,20 @@ func WithWorkerServerScope(s *Server) WorkerOption { } } +// WithWorkerPings configures a periodic message sent to the worker via frankenphp_handle_request(). +func WithWorkerPings(mode PingMode, interval time.Duration, message string, aligned bool) WorkerOption { + return func(w *workerOpt) error { + w.pings = append(w.pings, &ping{ + interval: interval, + message: message, + aligned: aligned, + mode: mode, + }) + + 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/ping.go b/ping.go new file mode 100644 index 0000000000..f3c46b2f03 --- /dev/null +++ b/ping.go @@ -0,0 +1,143 @@ +package frankenphp + +import ( + "context" + "log/slog" + "time" +) + +type PingMode int + +const ( + // PingModeSynchronous sends the ping to the worker and waits for completion before sending the next ping + PingModeSynchronous PingMode = iota + // PingModeOverlapping sends the ping to the worker without waiting for completion + PingModeOverlapping + // PingModeEach sends the ping to each active worker thread without waiting for completion + PingModeEach + // PingModeIdle sends the ping to each thread that has been idle for at least the interval + PingModeIdle +) + +// pings are periodic internal messages sent to the worker. +// they are received via frankenphp_handle_request(fn(string $message) => ...). +type ping struct { + interval time.Duration + message string + aligned bool + mode PingMode + worker *worker +} + +func initPings() { + for _, w := range workers { + w.initPings() + } +} + +func shutdownPings() { + for _, w := range workers { + w.stopPings() + } +} + +func (w *worker) initPings() { + if len(w.pings) == 0 { + return + } + + ctx, cancel := context.WithCancel(globalCtx) + w.pingCancel = cancel + + for _, p := range w.pings { + p.worker = w + if p.aligned { + go p.startAlignedLoop(ctx) + } else { + go p.startLoop(ctx) + } + } +} + +func (w *worker) stopPings() { + if w.pingCancel != nil { + w.pingCancel() + w.pingCancel = nil + } +} + +func (p *ping) startLoop(ctx context.Context) { + ticker := time.NewTicker(p.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.send() + } + } +} + +func (p *ping) startAlignedLoop(ctx context.Context) { + timer := time.NewTimer(time.Until(nextAlignedPing(p.interval, time.Now()))) + defer timer.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + p.send() + timer.Reset(time.Until(nextAlignedPing(p.interval, time.Now()))) + } + } +} + +func nextAlignedPing(interval time.Duration, now time.Time) time.Time { + switch interval { + case time.Minute: + return now.Truncate(time.Minute).Add(time.Minute) + case time.Hour: + return now.Truncate(time.Hour).Add(time.Hour) + default: + return now.Truncate(interval).Add(interval) + } +} + +func (p *ping) send() { + switch p.mode { + case PingModeEach, PingModeIdle: + p.sendToEachThread() + case PingModeOverlapping: + go p.sendOnce() + case PingModeSynchronous: + p.sendOnce() + } +} + +func (p *ping) sendOnce() { + fc := newContextFromMessage(p.message, nil, globalCtx, p.worker) + + if err := p.worker.handleRequest(fc); err != nil && globalLogger.Enabled(globalCtx, slog.LevelWarn) { + globalLogger.LogAttrs(globalCtx, slog.LevelWarn, "worker ping failed", slog.String("worker", p.worker.name), slog.String("message", p.message), slog.Any("error", err)) + } +} + +func (p *ping) sendToEachThread() { + w := p.worker + w.threadMutex.RLock() + for _, thread := range w.threads { + if p.mode == PingModeIdle && thread.state.WaitTime() < p.interval.Milliseconds() { + continue + } + go func(thread *phpThread) { + fc := newContextFromMessage(p.message, nil, globalCtx, w) + if err := w.handleRequestOnThread(thread, fc); err != nil && globalLogger.Enabled(globalCtx, slog.LevelWarn) { + globalLogger.LogAttrs(globalCtx, slog.LevelWarn, "worker ping failed", slog.String("worker", w.name), slog.String("message", p.message), slog.Any("error", err)) + } + }(thread) + } + w.threadMutex.RUnlock() +} diff --git a/ping_test.go b/ping_test.go new file mode 100644 index 0000000000..c78d08a43c --- /dev/null +++ b/ping_test.go @@ -0,0 +1,18 @@ +package frankenphp + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestNextAlignedPing(t *testing.T) { + now := time.Date(2026, 7, 10, 12, 34, 56, 0, time.UTC) + + next := nextAlignedPing(time.Minute, now) + assert.Equal(t, time.Date(2026, 7, 10, 12, 35, 0, 0, time.UTC), next) + + next = nextAlignedPing(time.Hour, now) + assert.Equal(t, time.Date(2026, 7, 10, 13, 0, 0, 0, time.UTC), next) +} diff --git a/worker.go b/worker.go index ecbed9e392..f6c3bd7952 100644 --- a/worker.go +++ b/worker.go @@ -3,6 +3,7 @@ package frankenphp // #include "frankenphp.h" import "C" import ( + "context" "fmt" "net/http" "os" @@ -32,6 +33,8 @@ type worker struct { maxConsecutiveFailures int onThreadReady func(int) onThreadShutdown func(int) + pings []*ping + pingCancel context.CancelFunc queuedRequests atomic.Int32 server *Server } @@ -167,6 +170,7 @@ func newWorker(o workerOpt) (*worker, error) { onThreadReady: o.onThreadReady, onThreadShutdown: o.onThreadShutdown, server: o.server, + pings: o.pings, } w.configureMercure(&o) @@ -234,6 +238,23 @@ func (worker *worker) isAtThreadLimit() bool { return atMaxThreads } +func (worker *worker) handleRequestOnThread(thread *phpThread, fc *frankenPHPContext) error { + metrics.StartWorkerRequest(worker.name) + + select { + case thread.requestChan <- fc: + <-fc.done + metrics.StopWorkerRequest(worker.name, time.Since(fc.startedAt)) + + return nil + case <-timeoutChan(time.Duration(maxWaitTime.Load())): + metrics.StopWorkerRequest(worker.name, time.Since(fc.startedAt)) + fc.reject(ErrMaxWaitTimeExceeded) + + return ErrMaxWaitTimeExceeded + } +} + func (worker *worker) handleRequest(fc *frankenPHPContext) error { metrics.StartWorkerRequest(worker.name) From 916fd3acd5f89e716ddbb7c1445e38adf1d0ae1a Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 21 Aug 2026 18:46:42 +0200 Subject: [PATCH 2/7] improves alignment --- ping.go | 26 ++++++++++++++++++++------ 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/ping.go b/ping.go index f3c46b2f03..de5cdf85fa 100644 --- a/ping.go +++ b/ping.go @@ -67,7 +67,13 @@ func (w *worker) stopPings() { } func (p *ping) startLoop(ctx context.Context) { - ticker := time.NewTicker(p.interval) + interval := p.interval + if p.mode == PingModeIdle { + // reduce the interval when pinging for idle threads + // this way threads will be idle for at most 4/3 of the original interval + interval = p.interval / 3 + } + ticker := time.NewTicker(interval) defer ticker.Stop() for { @@ -95,15 +101,23 @@ func (p *ping) startAlignedLoop(ctx context.Context) { } } +// nextAlignedPing returns the next time that is a multiple of the given interval +// e.g. interval=15m aligns to :00, :15, :30, :45 func nextAlignedPing(interval time.Duration, now time.Time) time.Time { - switch interval { - case time.Minute: - return now.Truncate(time.Minute).Add(time.Minute) - case time.Hour: - return now.Truncate(time.Hour).Add(time.Hour) + var periodStart time.Time + switch { + case interval <= time.Minute: + periodStart = time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), 0, 0, now.Location()) + case interval <= time.Hour: + periodStart = time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location()) + case interval <= 24*time.Hour: + periodStart = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) default: + // multi-day intervals: fall back to epoch-based truncation return now.Truncate(interval).Add(interval) } + + return periodStart.Add((now.Sub(periodStart)/interval + 1) * interval) } func (p *ping) send() { From 1836172b75230c3ffce68a8feb1150d7c3ecb921 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Fri, 21 Aug 2026 18:51:07 +0200 Subject: [PATCH 3/7] mentions polling staleness --- docs/worker.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/worker.md b/docs/worker.md index 0418c955d1..66dbd88685 100644 --- a/docs/worker.md +++ b/docs/worker.md @@ -286,7 +286,7 @@ worker /path/to/worker { ping sync 10s "message" # send a single ping each 10s, wait for completion in-between pings ping overlap 10s "message" # send a single ping each 10s, don't wait for completion ping each 10s "message" # send pings to each active thread every 10s, don't wait for completion - ping idle 10s "message" # send pings to each active thread every 10s if the thread has been idle for more than 10s + ping idle 10s "message" # send pings to each active thread that has been idle for more than 10s (worst case staleness up to 33% higher due to polling) } ``` From 314fd20392e8c2699ee9c182fd0e8d4da84d3f38 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Mon, 24 Aug 2026 13:58:23 +0200 Subject: [PATCH 4/7] cleans up docs --- docs/config.md | 31 ------------------------------- docs/worker.md | 28 +++++++++++----------------- 2 files changed, 11 insertions(+), 48 deletions(-) diff --git a/docs/config.md b/docs/config.md index 3b9bf84953..c073d4de68 100644 --- a/docs/config.md +++ b/docs/config.md @@ -251,37 +251,6 @@ where the FrankenPHP process was started. You can instead also specify one or mo - If you have multiple workers defined, all of them will be restarted when a file changes - Be wary about watching files that are created at runtime (like logs) since they might cause unwanted worker restarts. -### Pinging workers - -Workers can receive periodic messages via the `ping` directive. -The message is passed as the first argument to the closure given to `frankenphp_handle_request()`, like [extension workers](extension-workers.md#headless-mode-sendmessage). -This is useful to run scheduled tasks inside your worker script (for example, to flush queues or run cron jobs). - -```caddyfile -worker /path/to/worker.php { - ping 60s run-tasks - ping each 1m aligned minutely -} -``` - -```php -while (frankenphp_handle_request(function ($message) { - match ($message) { - 'run-tasks' => runTasks(), - 'minutely' => runMinutelyTasks(), - default => handleRequest(), - }; -})) { -} -``` - -The interval must be a [Go duration](https://pkg.go.dev/time#ParseDuration) such as `60s`, `1m`, or `5m`. -Add the `aligned` keyword to align pings to the start of each interval (e.g. `ping 1m aligned minutely` runs at the start of every minute). - -By default, each ping is handled by a single worker thread. Prefix the interval with `each` to send the ping to every thread of the worker. This is useful when each thread maintains its own state and must run the scheduled task independently. - -You can specify multiple `ping` lines to schedule different messages at different intervals. - The file watcher is based on [e-dant/watcher](https://github.com/e-dant/watcher). ## Matching the worker to a path diff --git a/docs/worker.md b/docs/worker.md index 66dbd88685..b140a7cbd7 100644 --- a/docs/worker.md +++ b/docs/worker.md @@ -263,20 +263,24 @@ Workers can also be pinged repeatedly with a message. ```caddyfile worker /path/to/worker.php { - ping 1s "Hello Worker" # send a single ping with message "Hello Worker" each second + ping 10s "Hello Worker" # send "Hello Worker" every 10s } ``` In the worker script, the function passed to `frankenphp_handle_request()` will receive the message directly as an argument: ```php -$handler = function(string $message = "") { - echo $message; # "Hello Worker" -} - -while(frankenphp_handle_request($handler)){} +while(frankenphp_handle_request(function(string $message = "") { + match($message){ + 'Hello Worker' => handleMessage() + default => handleRequest() # if the worker also handles HTTP requests + } +})){} ``` +The interval must be a [Go duration](https://pkg.go.dev/time#ParseDuration) such as `60s`, `1m`, or `5m`. +Add the `aligned` keyword to align pings to the start of each interval (e.g. `ping 1m aligned minutely` runs at the start of every minute). + ### Ping modes Available modes for pinging are: "sync", "overlap", "each" and "idle". @@ -286,16 +290,6 @@ worker /path/to/worker { ping sync 10s "message" # send a single ping each 10s, wait for completion in-between pings ping overlap 10s "message" # send a single ping each 10s, don't wait for completion ping each 10s "message" # send pings to each active thread every 10s, don't wait for completion - ping idle 10s "message" # send pings to each active thread that has been idle for more than 10s (worst case staleness up to 33% higher due to polling) -} -``` - -### Aligned pings - -A ping can also be aligned to the start of each interval with the `aligned` keyword (cron-like). - -```caddyfile -worker /path/to/worker { - ping overlap 1m aligned "message" # send the ping at the exact start of each minute + ping idle 10s "message" # send pings to each active thread that has been idle for more than 10s (worst case staleness up to 33% higher) } ``` From 5fe1e769f966d0b8ad62320f5323976b590f7e99 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Tue, 25 Aug 2026 17:30:49 +0200 Subject: [PATCH 5/7] fixes races on shutdown --- ping.go | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/ping.go b/ping.go index de5cdf85fa..e19481210e 100644 --- a/ping.go +++ b/ping.go @@ -81,7 +81,7 @@ func (p *ping) startLoop(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - p.send() + p.send(ctx) } } } @@ -95,7 +95,7 @@ func (p *ping) startAlignedLoop(ctx context.Context) { case <-ctx.Done(): return case <-timer.C: - p.send() + p.send(ctx) timer.Reset(time.Until(nextAlignedPing(p.interval, time.Now()))) } } @@ -120,26 +120,26 @@ func nextAlignedPing(interval time.Duration, now time.Time) time.Time { return periodStart.Add((now.Sub(periodStart)/interval + 1) * interval) } -func (p *ping) send() { +func (p *ping) send(ctx context.Context) { switch p.mode { case PingModeEach, PingModeIdle: - p.sendToEachThread() + p.sendToEachThread(ctx) case PingModeOverlapping: - go p.sendOnce() + go p.sendOnce(ctx) case PingModeSynchronous: - p.sendOnce() + p.sendOnce(ctx) } } -func (p *ping) sendOnce() { - fc := newContextFromMessage(p.message, nil, globalCtx, p.worker) +func (p *ping) sendOnce(ctx context.Context) { + fc := newContextFromMessage(p.message, nil, ctx, p.worker) - if err := p.worker.handleRequest(fc); err != nil && globalLogger.Enabled(globalCtx, slog.LevelWarn) { - globalLogger.LogAttrs(globalCtx, slog.LevelWarn, "worker ping failed", slog.String("worker", p.worker.name), slog.String("message", p.message), slog.Any("error", err)) + if err := p.worker.handleRequest(fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { + globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker ping failed", slog.String("worker", p.worker.name), slog.String("message", p.message), slog.Any("error", err)) } } -func (p *ping) sendToEachThread() { +func (p *ping) sendToEachThread(ctx context.Context) { w := p.worker w.threadMutex.RLock() for _, thread := range w.threads { @@ -147,9 +147,9 @@ func (p *ping) sendToEachThread() { continue } go func(thread *phpThread) { - fc := newContextFromMessage(p.message, nil, globalCtx, w) - if err := w.handleRequestOnThread(thread, fc); err != nil && globalLogger.Enabled(globalCtx, slog.LevelWarn) { - globalLogger.LogAttrs(globalCtx, slog.LevelWarn, "worker ping failed", slog.String("worker", w.name), slog.String("message", p.message), slog.Any("error", err)) + fc := newContextFromMessage(p.message, nil, ctx, w) + if err := w.handleRequestOnThread(thread, fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { + globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker ping failed", slog.String("worker", w.name), slog.String("message", p.message), slog.Any("error", err)) } }(thread) } From 47ceeb38507c771cc7daf0148faa984a15c1cc18 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Tue, 25 Aug 2026 22:54:22 +0200 Subject: [PATCH 6/7] waits for all in-flight pings --- ping.go | 64 ++++++++++++++++++++++++++++--------------------------- worker.go | 2 -- 2 files changed, 33 insertions(+), 33 deletions(-) diff --git a/ping.go b/ping.go index e19481210e..ffc355d30f 100644 --- a/ping.go +++ b/ping.go @@ -3,6 +3,7 @@ package frankenphp import ( "context" "log/slog" + "sync" "time" ) @@ -19,6 +20,11 @@ const ( PingModeIdle ) +var ( + pingCancel chan any + pingWg sync.WaitGroup +) + // pings are periodic internal messages sent to the worker. // they are received via frankenphp_handle_request(fn(string $message) => ...). type ping struct { @@ -31,39 +37,29 @@ type ping struct { func initPings() { for _, w := range workers { - w.initPings() + for _, p := range w.pings { + if pingCancel == nil { + pingCancel = make(chan any) + pingWg = sync.WaitGroup{} + } + pingWg.Add(1) + p.worker = w + if p.aligned { + go p.startAlignedLoop(globalCtx) + } else { + go p.startLoop(globalCtx) + } + } } } func shutdownPings() { - for _, w := range workers { - w.stopPings() - } -} - -func (w *worker) initPings() { - if len(w.pings) == 0 { + if pingCancel == nil { return } - - ctx, cancel := context.WithCancel(globalCtx) - w.pingCancel = cancel - - for _, p := range w.pings { - p.worker = w - if p.aligned { - go p.startAlignedLoop(ctx) - } else { - go p.startLoop(ctx) - } - } -} - -func (w *worker) stopPings() { - if w.pingCancel != nil { - w.pingCancel() - w.pingCancel = nil - } + close(pingCancel) + pingWg.Wait() + pingCancel = nil } func (p *ping) startLoop(ctx context.Context) { @@ -74,11 +70,12 @@ func (p *ping) startLoop(ctx context.Context) { interval = p.interval / 3 } ticker := time.NewTicker(interval) - defer ticker.Stop() for { select { - case <-ctx.Done(): + case <-pingCancel: + ticker.Stop() + pingWg.Done() return case <-ticker.C: p.send(ctx) @@ -88,11 +85,12 @@ func (p *ping) startLoop(ctx context.Context) { func (p *ping) startAlignedLoop(ctx context.Context) { timer := time.NewTimer(time.Until(nextAlignedPing(p.interval, time.Now()))) - defer timer.Stop() for { select { - case <-ctx.Done(): + case <-pingCancel: + timer.Stop() + pingWg.Done() return case <-timer.C: p.send(ctx) @@ -132,11 +130,13 @@ func (p *ping) send(ctx context.Context) { } func (p *ping) sendOnce(ctx context.Context) { + pingWg.Add(1) fc := newContextFromMessage(p.message, nil, ctx, p.worker) if err := p.worker.handleRequest(fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker ping failed", slog.String("worker", p.worker.name), slog.String("message", p.message), slog.Any("error", err)) } + pingWg.Done() } func (p *ping) sendToEachThread(ctx context.Context) { @@ -146,11 +146,13 @@ func (p *ping) sendToEachThread(ctx context.Context) { if p.mode == PingModeIdle && thread.state.WaitTime() < p.interval.Milliseconds() { continue } + pingWg.Add(1) go func(thread *phpThread) { fc := newContextFromMessage(p.message, nil, ctx, w) if err := w.handleRequestOnThread(thread, fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker ping failed", slog.String("worker", w.name), slog.String("message", p.message), slog.Any("error", err)) } + pingWg.Done() }(thread) } w.threadMutex.RUnlock() diff --git a/worker.go b/worker.go index f6c3bd7952..7197a26d00 100644 --- a/worker.go +++ b/worker.go @@ -3,7 +3,6 @@ package frankenphp // #include "frankenphp.h" import "C" import ( - "context" "fmt" "net/http" "os" @@ -34,7 +33,6 @@ type worker struct { onThreadReady func(int) onThreadShutdown func(int) pings []*ping - pingCancel context.CancelFunc queuedRequests atomic.Int32 server *Server } From fedeed6dc144433ab9aedb45a8c98bb49202c086 Mon Sep 17 00:00:00 2001 From: Alliballibaba Date: Tue, 25 Aug 2026 23:03:51 +0200 Subject: [PATCH 7/7] ping -> tick --- caddy/config_test.go | 64 +++++++------- caddy/workerconfig.go | 50 +++++------ docs/config.md | 6 +- docs/worker.md | 24 +++--- frankenphp.go | 4 +- frankenphp_test.go | 8 +- options.go | 8 +- ping.go | 159 ----------------------------------- tick.go | 159 +++++++++++++++++++++++++++++++++++ ping_test.go => tick_test.go | 6 +- worker.go | 4 +- 11 files changed, 244 insertions(+), 248 deletions(-) delete mode 100644 ping.go create mode 100644 tick.go rename ping_test.go => tick_test.go (70%) diff --git a/caddy/config_test.go b/caddy/config_test.go index ce88598c35..ad1190d04b 100644 --- a/caddy/config_test.go +++ b/caddy/config_test.go @@ -285,50 +285,50 @@ func TestCreateUniqueWorkerNamesQualifiedByServer(t *testing.T) { require.Equal(t, "queue_2", app.createUniqueWorkerName(wc, "")) } -func TestModuleWorkerWithPingConfiguration(t *testing.T) { - configWithPing := ` +func TestModuleWorkerWithTickConfiguration(t *testing.T) { + configWithTick := ` { php { worker ../testdata/worker-with-counter.php { - ping 60s health - ping each 1m aligned message - ping overlap 1h aligned message - ping idle 3s "HELLO THERE!!!!" + tick 60s health + tick each 1m aligned message + tick overlap 1h aligned message + tick idle 3s "HELLO THERE!!!!" } } }` - d := caddyfile.NewTestDispenser(configWithPing) + d := caddyfile.NewTestDispenser(configWithTick) module := &FrankenPHPModule{} err := module.UnmarshalCaddyfile(d) require.NoError(t, err) require.Len(t, module.Workers, 1) - pings := module.Workers[0].Pings - require.Len(t, pings, 4) - require.Equal(t, 60*time.Second, pings[0].Interval) - require.Equal(t, "health", pings[0].Message) - require.False(t, pings[0].Aligned) - require.Equal(t, frankenphp.PingModeSynchronous, pings[0].Mode) - - require.Equal(t, time.Minute, pings[1].Interval) - require.Equal(t, "message", pings[1].Message) - require.True(t, pings[1].Aligned) - require.Equal(t, frankenphp.PingModeEach, pings[1].Mode) - - require.Equal(t, time.Hour, pings[2].Interval) - require.Equal(t, "message", pings[2].Message) - require.True(t, pings[2].Aligned) - require.Equal(t, frankenphp.PingModeOverlapping, pings[2].Mode) - - require.Equal(t, "HELLO THERE!!!!", pings[3].Message) - require.False(t, pings[3].Aligned) - require.Equal(t, frankenphp.PingModeIdle, pings[3].Mode) - require.Equal(t, 3*time.Second, pings[3].Interval) + ticks := module.Workers[0].Ticks + require.Len(t, ticks, 4) + require.Equal(t, 60*time.Second, ticks[0].Interval) + require.Equal(t, "health", ticks[0].Message) + require.False(t, ticks[0].Aligned) + require.Equal(t, frankenphp.TickModeSynchronous, ticks[0].Mode) + + require.Equal(t, time.Minute, ticks[1].Interval) + require.Equal(t, "message", ticks[1].Message) + require.True(t, ticks[1].Aligned) + require.Equal(t, frankenphp.TickModeEach, ticks[1].Mode) + + require.Equal(t, time.Hour, ticks[2].Interval) + require.Equal(t, "message", ticks[2].Message) + require.True(t, ticks[2].Aligned) + require.Equal(t, frankenphp.TickModeOverlapping, ticks[2].Mode) + + require.Equal(t, "HELLO THERE!!!!", ticks[3].Message) + require.False(t, ticks[3].Aligned) + require.Equal(t, frankenphp.TickModeIdle, ticks[3].Mode) + require.Equal(t, 3*time.Second, ticks[3].Interval) } -func TestModuleWorkerWithInvalidPingConfiguration(t *testing.T) { +func TestModuleWorkerWithInvalidTickConfiguration(t *testing.T) { tests := []struct { name string config string @@ -339,7 +339,7 @@ func TestModuleWorkerWithInvalidPingConfiguration(t *testing.T) { php { worker { file ../testdata/worker-with-counter.php - ping 60s + tick 60s } } }`, @@ -350,7 +350,7 @@ func TestModuleWorkerWithInvalidPingConfiguration(t *testing.T) { php { worker { file ../testdata/worker-with-counter.php - ping not-a-duration health + tick not-a-duration health } } }`, @@ -361,7 +361,7 @@ func TestModuleWorkerWithInvalidPingConfiguration(t *testing.T) { php { worker { file ../testdata/worker-with-counter.php - ping 60s health each + tick 60s health each } } }`, diff --git a/caddy/workerconfig.go b/caddy/workerconfig.go index ba13ffc010..cec2d47e06 100644 --- a/caddy/workerconfig.go +++ b/caddy/workerconfig.go @@ -41,24 +41,24 @@ 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"` - // Pings configures periodic internal messages sent to the worker. - Pings []*pingConfig `json:"pings,omitempty"` + // Ticks configures periodic internal messages sent to the worker. + Ticks []*tickConfig `json:"ticks,omitempty"` options []frankenphp.WorkerOption } -type pingConfig struct { +type tickConfig struct { Interval time.Duration `json:"interval"` Message string `json:"message"` Aligned bool `json:"aligned,omitempty"` - Mode frankenphp.PingMode `json:"mode,omitempty"` + Mode frankenphp.TickMode `json:"mode,omitempty"` } -var pingModes = map[string]frankenphp.PingMode{ - "sync": frankenphp.PingModeSynchronous, - "overlap": frankenphp.PingModeOverlapping, - "each": frankenphp.PingModeEach, - "idle": frankenphp.PingModeIdle, +var tickModes = map[string]frankenphp.TickMode{ + "sync": frankenphp.TickModeSynchronous, + "overlap": frankenphp.TickModeOverlapping, + "each": frankenphp.TickModeEach, + "idle": frankenphp.TickModeIdle, } func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) { @@ -158,15 +158,15 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) { } wc.MaxConsecutiveFailures = v - case "ping": - ping, err := parsePingConfig(d) + case "tick": + tick, err := parseTickConfig(d) if err != nil { return wc, d.WrapErr(err) } - wc.Pings = append(wc.Pings, ping) + wc.Ticks = append(wc.Ticks, tick) default: - return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, ping, max_consecutive_failures, max_threads", v) + return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, tick, max_consecutive_failures, max_threads", v) } } @@ -202,26 +202,26 @@ func (wc *workerConfig) toWorkerOptions() ([]frankenphp.WorkerOption, error) { opts = append(opts, frankenphp.WithWorkerMatcher(matchFunc.Match)) } - if len(wc.Pings) > 0 { - for _, p := range wc.Pings { - opts = append(opts, frankenphp.WithWorkerPings(p.Mode, p.Interval, p.Message, p.Aligned)) + if len(wc.Ticks) > 0 { + for _, t := range wc.Ticks { + opts = append(opts, frankenphp.WithWorkerTicks(t.Mode, t.Interval, t.Message, t.Aligned)) } } return opts, nil } -// parse the configuration for recurring pings to the worker -// ping 1s "Hello, world!" -// ping overlap aligned 1m "Hello, world!" -func parsePingConfig(d *caddyfile.Dispenser) (*pingConfig, error) { +// parse the configuration for recurring ticks to the worker +// tick 1s "Hello, world!" +// tick overlap aligned 1m "Hello, world!" +func parseTickConfig(d *caddyfile.Dispenser) (*tickConfig, error) { args := d.RemainingArgs() if len(args) < 2 { return nil, d.ArgErr() } - mode := frankenphp.PingModeSynchronous - if m, ok := pingModes[args[0]]; ok { + mode := frankenphp.TickModeSynchronous + if m, ok := tickModes[args[0]]; ok { mode, args = m, args[1:] } @@ -237,11 +237,11 @@ func parsePingConfig(d *caddyfile.Dispenser) (*pingConfig, error) { interval, err := time.ParseDuration(args[0]) if err != nil { - return nil, fmt.Errorf("ping interval must be a valid duration, received: %s (%s)", args[0], err) + return nil, fmt.Errorf("tick interval must be a valid duration, received: %s (%s)", args[0], err) } if interval <= 0 { - return nil, fmt.Errorf("ping interval must be positive, received: %s (%s)", args[0], interval) + return nil, fmt.Errorf("tick interval must be positive, received: %s (%s)", args[0], interval) } - return &pingConfig{Interval: interval, Message: args[1], Aligned: aligned, Mode: mode}, nil + return &tickConfig{Interval: interval, Message: args[1], Aligned: aligned, Mode: mode}, nil } diff --git a/docs/config.md b/docs/config.md index c073d4de68..752a99dc4f 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. - ping [aligned] # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align pings to the start of each interval. Mode can be sync, overlap, each and idle. Can be specified more than once. + tick [aligned] # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align ticks to the start of each interval. Mode can be sync, overlap, each and idle. Can be specified more than once. } } } @@ -199,8 +199,8 @@ 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. - ping [aligned] # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align pings to the start of each interval. Can be specified more than once. - ping [aligned] # Like ping, with an explicit mode (sync, overlap, each, idle). + tick [aligned] # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align ticks to the start of each interval. Can be specified more than once. + tick [aligned] # Like tick, with an explicit mode (sync, overlap, each, idle). } worker # Can also use the short form like in the global frankenphp block. } diff --git a/docs/worker.md b/docs/worker.md index b140a7cbd7..ad49145f5c 100644 --- a/docs/worker.md +++ b/docs/worker.md @@ -257,39 +257,35 @@ while (\frankenphp_handle_request($handler)) { When writing worker scripts, make sure to reset any request-specific state between requests. Frameworks like [Symfony](symfony.md) and [Laravel Octane](laravel.md) take care of resetting most state for you, but you may still need to reset your own services. With Symfony, services that hold request-specific state should implement [`Symfony\Contracts\Service\ResetInterface`](https://github.com/symfony/contracts/blob/main/Service/ResetInterface.php) so they're reset by the kernel between requests. -## Pinging +## Ticking -Workers can also be pinged repeatedly with a message. +Workers can also be triggered repeatedly with a message. ```caddyfile worker /path/to/worker.php { - ping 10s "Hello Worker" # send "Hello Worker" every 10s + tick 10s "Hello Worker" # send "Hello Worker" every 10s } ``` -In the worker script, the function passed to `frankenphp_handle_request()` will receive the message directly as an argument: +In the worker script, the function passed to `frankenphp_handle_request()` will receive the message directly as an argument every 10s: ```php while(frankenphp_handle_request(function(string $message = "") { match($message){ 'Hello Worker' => handleMessage() - default => handleRequest() # if the worker also handles HTTP requests + default => handleRequest() # if the worker also handles regular HTTP requests } })){} ``` The interval must be a [Go duration](https://pkg.go.dev/time#ParseDuration) such as `60s`, `1m`, or `5m`. -Add the `aligned` keyword to align pings to the start of each interval (e.g. `ping 1m aligned minutely` runs at the start of every minute). - -### Ping modes - -Available modes for pinging are: "sync", "overlap", "each" and "idle". +Add the `aligned` keyword to align ticks to the start of each interval (e.g. `tick 1m aligned minutely` runs at the start of every minute). Available modes for ticking are: "sync", "overlap", "each" and "idle". ```caddyfile worker /path/to/worker { - ping sync 10s "message" # send a single ping each 10s, wait for completion in-between pings - ping overlap 10s "message" # send a single ping each 10s, don't wait for completion - ping each 10s "message" # send pings to each active thread every 10s, don't wait for completion - ping idle 10s "message" # send pings to each active thread that has been idle for more than 10s (worst case staleness up to 33% higher) + tick sync 10s "message" # send a single tick each 10s, wait for completion in-between ticks + tick overlap 10s "message" # send a single tick each 10s, don't wait for completion + tick each 10s "message" # send ticks to each active thread every 10s, don't wait for completion + tick idle 10s "message" # send ticks to each active thread that has been idle for more than 10-13.3s } ``` diff --git a/frankenphp.go b/frankenphp.go index 7af6f9e6dd..6baf19cd92 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -359,7 +359,7 @@ func Init(options ...Option) error { } } - initPings() + initTicks() return nil } @@ -382,7 +382,7 @@ func shutdown() { fn() } - shutdownPings() + shutdownTicks() drainWatchers() drainPHPThreads() unregisterServers() diff --git a/frankenphp_test.go b/frankenphp_test.go index e3e89d677a..99a529b050 100644 --- a/frankenphp_test.go +++ b/frankenphp_test.go @@ -1426,12 +1426,12 @@ func testOpcachePreload(t *testing.T, opts *testOptions) { }, opts) } -func TestPings(t *testing.T) { +func TestTicks(t *testing.T) { logger, buf := newTestLogger(t) require.NoError(t, frankenphp.Init( frankenphp.WithLogger(logger), - frankenphp.WithWorkers("ping-worker", "testdata/worker-with-counter.php", 1, - frankenphp.WithWorkerPings(frankenphp.PingModeSynchronous, 100*time.Microsecond, "ping", false), + frankenphp.WithWorkers("tick-worker", "testdata/worker-with-counter.php", 1, + frankenphp.WithWorkerTicks(frankenphp.TickModeSynchronous, 100*time.Microsecond, "tick", false), ), )) t.Cleanup(frankenphp.Shutdown) @@ -1445,7 +1445,7 @@ func TestPings(t *testing.T) { time.Sleep(500 * time.Microsecond) i++ if i > 10000 { // 5s timeout - t.Fatal("timed out without recording a worker ping") + t.Fatal("timed out without recording a worker tick") } } } diff --git a/options.go b/options.go index b65d6c0b18..1a0b5463b2 100644 --- a/options.go +++ b/options.go @@ -50,7 +50,7 @@ type workerOpt struct { requestOptions []RequestOption watch []string matchRequest func(*http.Request) bool - pings []*ping + ticks []*tick maxConsecutiveFailures int extensionWorkers *extensionWorkers onThreadReady func(int) @@ -240,10 +240,10 @@ func WithWorkerServerScope(s *Server) WorkerOption { } } -// WithWorkerPings configures a periodic message sent to the worker via frankenphp_handle_request(). -func WithWorkerPings(mode PingMode, interval time.Duration, message string, aligned bool) WorkerOption { +// WithWorkerTicks configures a periodic message sent to the worker via frankenphp_handle_request(). +func WithWorkerTicks(mode TickMode, interval time.Duration, message string, aligned bool) WorkerOption { return func(w *workerOpt) error { - w.pings = append(w.pings, &ping{ + w.ticks = append(w.ticks, &tick{ interval: interval, message: message, aligned: aligned, diff --git a/ping.go b/ping.go deleted file mode 100644 index ffc355d30f..0000000000 --- a/ping.go +++ /dev/null @@ -1,159 +0,0 @@ -package frankenphp - -import ( - "context" - "log/slog" - "sync" - "time" -) - -type PingMode int - -const ( - // PingModeSynchronous sends the ping to the worker and waits for completion before sending the next ping - PingModeSynchronous PingMode = iota - // PingModeOverlapping sends the ping to the worker without waiting for completion - PingModeOverlapping - // PingModeEach sends the ping to each active worker thread without waiting for completion - PingModeEach - // PingModeIdle sends the ping to each thread that has been idle for at least the interval - PingModeIdle -) - -var ( - pingCancel chan any - pingWg sync.WaitGroup -) - -// pings are periodic internal messages sent to the worker. -// they are received via frankenphp_handle_request(fn(string $message) => ...). -type ping struct { - interval time.Duration - message string - aligned bool - mode PingMode - worker *worker -} - -func initPings() { - for _, w := range workers { - for _, p := range w.pings { - if pingCancel == nil { - pingCancel = make(chan any) - pingWg = sync.WaitGroup{} - } - pingWg.Add(1) - p.worker = w - if p.aligned { - go p.startAlignedLoop(globalCtx) - } else { - go p.startLoop(globalCtx) - } - } - } -} - -func shutdownPings() { - if pingCancel == nil { - return - } - close(pingCancel) - pingWg.Wait() - pingCancel = nil -} - -func (p *ping) startLoop(ctx context.Context) { - interval := p.interval - if p.mode == PingModeIdle { - // reduce the interval when pinging for idle threads - // this way threads will be idle for at most 4/3 of the original interval - interval = p.interval / 3 - } - ticker := time.NewTicker(interval) - - for { - select { - case <-pingCancel: - ticker.Stop() - pingWg.Done() - return - case <-ticker.C: - p.send(ctx) - } - } -} - -func (p *ping) startAlignedLoop(ctx context.Context) { - timer := time.NewTimer(time.Until(nextAlignedPing(p.interval, time.Now()))) - - for { - select { - case <-pingCancel: - timer.Stop() - pingWg.Done() - return - case <-timer.C: - p.send(ctx) - timer.Reset(time.Until(nextAlignedPing(p.interval, time.Now()))) - } - } -} - -// nextAlignedPing returns the next time that is a multiple of the given interval -// e.g. interval=15m aligns to :00, :15, :30, :45 -func nextAlignedPing(interval time.Duration, now time.Time) time.Time { - var periodStart time.Time - switch { - case interval <= time.Minute: - periodStart = time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), 0, 0, now.Location()) - case interval <= time.Hour: - periodStart = time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location()) - case interval <= 24*time.Hour: - periodStart = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) - default: - // multi-day intervals: fall back to epoch-based truncation - return now.Truncate(interval).Add(interval) - } - - return periodStart.Add((now.Sub(periodStart)/interval + 1) * interval) -} - -func (p *ping) send(ctx context.Context) { - switch p.mode { - case PingModeEach, PingModeIdle: - p.sendToEachThread(ctx) - case PingModeOverlapping: - go p.sendOnce(ctx) - case PingModeSynchronous: - p.sendOnce(ctx) - } -} - -func (p *ping) sendOnce(ctx context.Context) { - pingWg.Add(1) - fc := newContextFromMessage(p.message, nil, ctx, p.worker) - - if err := p.worker.handleRequest(fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { - globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker ping failed", slog.String("worker", p.worker.name), slog.String("message", p.message), slog.Any("error", err)) - } - pingWg.Done() -} - -func (p *ping) sendToEachThread(ctx context.Context) { - w := p.worker - w.threadMutex.RLock() - for _, thread := range w.threads { - if p.mode == PingModeIdle && thread.state.WaitTime() < p.interval.Milliseconds() { - continue - } - pingWg.Add(1) - go func(thread *phpThread) { - fc := newContextFromMessage(p.message, nil, ctx, w) - if err := w.handleRequestOnThread(thread, fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { - globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker ping failed", slog.String("worker", w.name), slog.String("message", p.message), slog.Any("error", err)) - } - pingWg.Done() - }(thread) - } - w.threadMutex.RUnlock() -} diff --git a/tick.go b/tick.go new file mode 100644 index 0000000000..a6f284aa46 --- /dev/null +++ b/tick.go @@ -0,0 +1,159 @@ +package frankenphp + +import ( + "context" + "log/slog" + "sync" + "time" +) + +type TickMode int + +const ( + // TickModeSynchronous sends the tick to the worker and waits for completion before sending the next tick + TickModeSynchronous TickMode = iota + // TickModeOverlapping sends the tick to the worker without waiting for completion + TickModeOverlapping + // TickModeEach sends the tick to each active worker thread without waiting for completion + TickModeEach + // TickModeIdle sends the tick to each thread that has been idle for at least the interval + TickModeIdle +) + +var ( + tickCancel chan any + tickWg sync.WaitGroup +) + +// ticks are periodic internal messages sent to the worker. +// they are received via frankenphp_handle_request(fn(string $message) => ...). +type tick struct { + interval time.Duration + message string + aligned bool + mode TickMode + worker *worker +} + +func initTicks() { + for _, w := range workers { + for _, t := range w.ticks { + if tickCancel == nil { + tickCancel = make(chan any) + tickWg = sync.WaitGroup{} + } + tickWg.Add(1) + t.worker = w + if t.aligned { + go t.startAlignedLoop(globalCtx) + } else { + go t.startLoop(globalCtx) + } + } + } +} + +func shutdownTicks() { + if tickCancel == nil { + return + } + close(tickCancel) + tickWg.Wait() + tickCancel = nil +} + +func (t *tick) startLoop(ctx context.Context) { + interval := t.interval + if t.mode == TickModeIdle { + // reduce the interval when ticking for idle threads + // this way threads will be idle for at most 4/3 of the original interval + interval = t.interval / 3 + } + ticker := time.NewTicker(interval) + + for { + select { + case <-tickCancel: + ticker.Stop() + tickWg.Done() + return + case <-ticker.C: + t.send(ctx) + } + } +} + +func (t *tick) startAlignedLoop(ctx context.Context) { + timer := time.NewTimer(time.Until(nextAlignedTick(t.interval, time.Now()))) + + for { + select { + case <-tickCancel: + timer.Stop() + tickWg.Done() + return + case <-timer.C: + t.send(ctx) + timer.Reset(time.Until(nextAlignedTick(t.interval, time.Now()))) + } + } +} + +// nextAlignedTick returns the next time that is a multiple of the given interval +// e.g. interval=15m aligns to :00, :15, :30, :45 +func nextAlignedTick(interval time.Duration, now time.Time) time.Time { + var periodStart time.Time + switch { + case interval <= time.Minute: + periodStart = time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), 0, 0, now.Location()) + case interval <= time.Hour: + periodStart = time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location()) + case interval <= 24*time.Hour: + periodStart = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) + default: + // multi-day intervals: fall back to epoch-based truncation + return now.Truncate(interval).Add(interval) + } + + return periodStart.Add((now.Sub(periodStart)/interval + 1) * interval) +} + +func (t *tick) send(ctx context.Context) { + switch t.mode { + case TickModeEach, TickModeIdle: + t.sendToEachThread(ctx) + case TickModeOverlapping: + go t.sendOnce(ctx) + case TickModeSynchronous: + t.sendOnce(ctx) + } +} + +func (t *tick) sendOnce(ctx context.Context) { + tickWg.Add(1) + fc := newContextFromMessage(t.message, nil, ctx, t.worker) + + if err := t.worker.handleRequest(fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { + globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker tick failed", slog.String("worker", t.worker.name), slog.String("message", t.message), slog.Any("error", err)) + } + tickWg.Done() +} + +func (t *tick) sendToEachThread(ctx context.Context) { + w := t.worker + w.threadMutex.RLock() + for _, thread := range w.threads { + if t.mode == TickModeIdle && thread.state.WaitTime() < t.interval.Milliseconds() { + continue + } + tickWg.Add(1) + go func(thread *phpThread) { + fc := newContextFromMessage(t.message, nil, ctx, w) + if err := w.handleRequestOnThread(thread, fc); err != nil && globalLogger.Enabled(ctx, slog.LevelWarn) { + globalLogger.LogAttrs(ctx, slog.LevelWarn, "worker tick failed", slog.String("worker", w.name), slog.String("message", t.message), slog.Any("error", err)) + } + tickWg.Done() + }(thread) + } + w.threadMutex.RUnlock() +} diff --git a/ping_test.go b/tick_test.go similarity index 70% rename from ping_test.go rename to tick_test.go index c78d08a43c..5ca3189956 100644 --- a/ping_test.go +++ b/tick_test.go @@ -7,12 +7,12 @@ import ( "github.com/stretchr/testify/assert" ) -func TestNextAlignedPing(t *testing.T) { +func TestNextAlignedTick(t *testing.T) { now := time.Date(2026, 7, 10, 12, 34, 56, 0, time.UTC) - next := nextAlignedPing(time.Minute, now) + next := nextAlignedTick(time.Minute, now) assert.Equal(t, time.Date(2026, 7, 10, 12, 35, 0, 0, time.UTC), next) - next = nextAlignedPing(time.Hour, now) + next = nextAlignedTick(time.Hour, now) assert.Equal(t, time.Date(2026, 7, 10, 13, 0, 0, 0, time.UTC), next) } diff --git a/worker.go b/worker.go index 7197a26d00..01fee4d02f 100644 --- a/worker.go +++ b/worker.go @@ -32,7 +32,7 @@ type worker struct { maxConsecutiveFailures int onThreadReady func(int) onThreadShutdown func(int) - pings []*ping + ticks []*tick queuedRequests atomic.Int32 server *Server } @@ -168,7 +168,7 @@ func newWorker(o workerOpt) (*worker, error) { onThreadReady: o.onThreadReady, onThreadShutdown: o.onThreadShutdown, server: o.server, - pings: o.pings, + ticks: o.ticks, } w.configureMercure(&o)