diff --git a/caddy/config_test.go b/caddy/config_test.go index 4540ece26b..ad1190d04b 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 TestModuleWorkerWithTickConfiguration(t *testing.T) { + configWithTick := ` + { + php { + worker ../testdata/worker-with-counter.php { + tick 60s health + tick each 1m aligned message + tick overlap 1h aligned message + tick idle 3s "HELLO THERE!!!!" + } + } + }` + + d := caddyfile.NewTestDispenser(configWithTick) + module := &FrankenPHPModule{} + + err := module.UnmarshalCaddyfile(d) + require.NoError(t, err) + require.Len(t, module.Workers, 1) + + 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 TestModuleWorkerWithInvalidTickConfiguration(t *testing.T) { + tests := []struct { + name string + config string + }{ + { + name: "missing message", + config: `{ + php { + worker { + file ../testdata/worker-with-counter.php + tick 60s + } + } + }`, + }, + { + name: "invalid interval", + config: `{ + php { + worker { + file ../testdata/worker-with-counter.php + tick not-a-duration health + } + } + }`, + }, + { + name: "each must come first", + config: `{ + php { + worker { + file ../testdata/worker-with-counter.php + tick 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..cec2d47e06 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"` + // Ticks configures periodic internal messages sent to the worker. + Ticks []*tickConfig `json:"ticks,omitempty"` options []frankenphp.WorkerOption } +type tickConfig struct { + Interval time.Duration `json:"interval"` + Message string `json:"message"` + Aligned bool `json:"aligned,omitempty"` + Mode frankenphp.TickMode `json:"mode,omitempty"` +} + +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) { wc := workerConfig{} if d.NextArg() { @@ -139,8 +158,15 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) { } wc.MaxConsecutiveFailures = v + case "tick": + tick, err := parseTickConfig(d) + if err != nil { + return wc, d.WrapErr(err) + } + + wc.Ticks = append(wc.Ticks, tick) 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, tick, 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.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 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.TickModeSynchronous + if m, ok := tickModes[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("tick interval must be a valid duration, received: %s (%s)", args[0], err) + } + if interval <= 0 { + return nil, fmt.Errorf("tick interval must be positive, received: %s (%s)", args[0], interval) + } + + return &tickConfig{Interval: interval, Message: args[1], Aligned: aligned, Mode: mode}, nil +} diff --git a/docs/config.md b/docs/config.md index 281f05dc75..752a99dc4f 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. + 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. } } } @@ -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. + 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 466c7cf684..ad49145f5c 100644 --- a/docs/worker.md +++ b/docs/worker.md @@ -256,3 +256,36 @@ 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. + +## Ticking + +Workers can also be triggered repeatedly with a message. + +```caddyfile +worker /path/to/worker.php { + 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 every 10s: + +```php +while(frankenphp_handle_request(function(string $message = "") { + match($message){ + 'Hello Worker' => handleMessage() + 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 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 { + 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 79b135b808..6baf19cd92 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -359,6 +359,8 @@ func Init(options ...Option) error { } } + initTicks() + return nil } @@ -380,6 +382,7 @@ func shutdown() { fn() } + shutdownTicks() 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..99a529b050 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 TestTicks(t *testing.T) { + logger, buf := newTestLogger(t) + require.NoError(t, frankenphp.Init( + frankenphp.WithLogger(logger), + frankenphp.WithWorkers("tick-worker", "testdata/worker-with-counter.php", 1, + frankenphp.WithWorkerTicks(frankenphp.TickModeSynchronous, 100*time.Microsecond, "tick", 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 tick") + } + } +} diff --git a/options.go b/options.go index e1eaeb7b55..1a0b5463b2 100644 --- a/options.go +++ b/options.go @@ -50,6 +50,7 @@ type workerOpt struct { requestOptions []RequestOption watch []string matchRequest func(*http.Request) bool + ticks []*tick maxConsecutiveFailures int extensionWorkers *extensionWorkers onThreadReady func(int) @@ -239,6 +240,20 @@ func WithWorkerServerScope(s *Server) 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.ticks = append(w.ticks, &tick{ + 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/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/tick_test.go b/tick_test.go new file mode 100644 index 0000000000..5ca3189956 --- /dev/null +++ b/tick_test.go @@ -0,0 +1,18 @@ +package frankenphp + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestNextAlignedTick(t *testing.T) { + now := time.Date(2026, 7, 10, 12, 34, 56, 0, time.UTC) + + next := nextAlignedTick(time.Minute, now) + assert.Equal(t, time.Date(2026, 7, 10, 12, 35, 0, 0, time.UTC), next) + + 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 ecbed9e392..01fee4d02f 100644 --- a/worker.go +++ b/worker.go @@ -32,6 +32,7 @@ type worker struct { maxConsecutiveFailures int onThreadReady func(int) onThreadShutdown func(int) + ticks []*tick queuedRequests atomic.Int32 server *Server } @@ -167,6 +168,7 @@ func newWorker(o workerOpt) (*worker, error) { onThreadReady: o.onThreadReady, onThreadShutdown: o.onThreadShutdown, server: o.server, + ticks: o.ticks, } w.configureMercure(&o) @@ -234,6 +236,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)