From 9a39cf97fb0946682d9c8b28d94c452eb832c4e6 Mon Sep 17 00:00:00 2001 From: "a.bolonin" Date: Fri, 23 Oct 2026 14:33:44 +0300 Subject: [PATCH 1/4] perf: avoid buffering worker startup output --- frankenphp.go | 21 ++---- ubwrite_test.go | 183 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 14 deletions(-) create mode 100644 ubwrite_test.go diff --git a/frankenphp.go b/frankenphp.go index 229eb6b2a0..7b2a0fad6e 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -20,11 +20,9 @@ package frankenphp // #include import "C" import ( - "bytes" "context" "errors" "fmt" - "io" "log/slog" "net/http" "os" @@ -554,25 +552,20 @@ func go_ub_write(threadIndex C.uintptr_t, cBuf *C.char, length C.size_t) (C.size return C.size_t(length), C.bool(fc.clientHadClosed) } - var writer io.Writer if fc.responseWriter == nil { - var b bytes.Buffer - // log the output of the worker - writer = &b - } else { - writer = fc.responseWriter + // Copy before Enabled: the C buffer may change during a logger callback. + message := string(unsafe.Slice((*byte)(unsafe.Pointer(cBuf)), length)) + if fc.logger.Enabled(fc.ctx, slog.LevelInfo) { + fc.logger.LogAttrs(fc.ctx, slog.LevelInfo, message) + } + return length, C.bool(fc.clientHasClosed()) } - i, e := writer.Write(unsafe.Slice((*byte)(unsafe.Pointer(cBuf)), length)) + i, e := fc.responseWriter.Write(unsafe.Slice((*byte)(unsafe.Pointer(cBuf)), length)) if e != nil && fc.logger.Enabled(fc.ctx, slog.LevelDebug) { fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "write error", slog.Any("error", e)) } - if fc.responseWriter == nil && fc.logger.Enabled(fc.ctx, slog.LevelInfo) { - // probably starting a worker script, log the output - fc.logger.LogAttrs(fc.ctx, slog.LevelInfo, writer.(*bytes.Buffer).String()) - } - return C.size_t(i), C.bool(fc.clientHasClosed()) } diff --git a/ubwrite_test.go b/ubwrite_test.go new file mode 100644 index 0000000000..07fc60ac96 --- /dev/null +++ b/ubwrite_test.go @@ -0,0 +1,183 @@ +package frankenphp + +import ( + "context" + "io" + "log/slog" + "net/http" + "reflect" + "strings" + "sync" + "testing" + "unsafe" +) + +func TestUBWrite(t *testing.T) { + for _, data := range []string{"", "a", "worker startup", "a\x00b\xff", strings.Repeat("x", 1024)} { + for _, mode := range []string{"log", "disabled", "http", "error", "short", "done", "done_closed", "canceled"} { + t.Run(mode+"/"+data, func(t *testing.T) { checkUBWrite(t, data, mode) }) + } + } +} + +func FuzzUBWrite(f *testing.F) { + f.Add("worker startup", uint8(0)) + f.Add("a\x00b\xff", uint8(1)) + f.Add("", uint8(2)) + f.Fuzz(func(t *testing.T, data string, mode uint8) { + modes := []string{"log", "disabled", "http", "error", "short", "done", "done_closed", "canceled"} + checkUBWrite(t, data, modes[int(mode)%len(modes)]) + }) +} + +func checkUBWrite(t *testing.T, message, mode string) { + t.Helper() + data := []byte(message) + capture := &ubWriteCapture{level: slog.LevelInfo} + fc := &frankenPHPContext{ctx: context.Background(), logger: slog.New(capture)} + writer := &ubWriteResponseWriter{n: -1} + expectedN, expectedClosed := len(data), false + switch mode { + case "disabled": + capture.level = slog.LevelWarn + case "http": + fc.responseWriter = writer + case "error": + writer.n = 0 + writer.err = io.ErrClosedPipe + fc.responseWriter = writer + capture.level = slog.LevelDebug + expectedN = 0 + case "short": + writer.n = 0 + fc.responseWriter = writer + expectedN = 0 + case "done": + fc.isDone = true + case "done_closed": + fc.isDone = true + fc.clientHadClosed = true + expectedClosed = true + case "canceled": + ctx, cancel := context.WithCancel(context.Background()) + cancel() + fc.ctx = ctx + fc.request = &http.Request{} + expectedClosed = true + } + oldThreads := phpThreads + phpThreads = []*phpThread{{handler: ®ularThread{fc: fc}}} + t.Cleanup(func() { phpThreads = oldThreads }) + n, closed := invokeUBWrite(data) + if n != expectedN || closed != expectedClosed { + t.Fatalf("write = (%d,%v), want (%d,%v)", n, closed, expectedN, expectedClosed) + } + // A handler is allowed to retain the record after the source buffer changes. + for i := range data { + data[i] = 'Z' + } + records := capture.snapshot() + switch mode { + case "log", "canceled": + if len(records) != 1 || records[0].Message != message || records[0].Level != slog.LevelInfo { + t.Fatalf("unexpected log records: %v", records) + } + case "error": + if len(records) != 1 || records[0].Message != "write error" || records[0].Level != slog.LevelDebug { + t.Fatalf("unexpected error records: %v", records) + } + var loggedError error + records[0].Attrs(func(a slog.Attr) bool { + if a.Key == "error" { + loggedError, _ = a.Value.Any().(error) + } + return true + }) + if loggedError != io.ErrClosedPipe { + t.Fatalf("logged error = %v", loggedError) + } + default: + if len(records) != 0 { + t.Fatalf("unexpected log records: %v", records) + } + } + if mode == "http" || mode == "error" || mode == "short" { + if string(writer.data) != message { + t.Fatalf("response bytes = %q, want %q", writer.data, message) + } + } +} + +func TestUBWriteCopiesBeforeLoggerEnabled(t *testing.T) { + data := []byte("original startup output") + capture := &ubWriteCapture{level: slog.LevelInfo, beforeEnabled: func() { + for i := range data { + data[i] = 'Z' + } + }} + fc := &frankenPHPContext{ctx: context.Background(), logger: slog.New(capture)} + oldThreads := phpThreads + phpThreads = []*phpThread{{handler: ®ularThread{fc: fc}}} + t.Cleanup(func() { phpThreads = oldThreads }) + invokeUBWrite(data) + records := capture.snapshot() + if len(records) != 1 || records[0].Message != "original startup output" { + t.Fatalf("source mutated before snapshot: %v", records) + } +} + +// Reflection bridges the C argument types without adding production test hooks +// or importing C from a _test.go file (which Go does not support). +func invokeUBWrite(data []byte) (int, bool) { + fn := reflect.ValueOf(go_ub_write) + signature := fn.Type() + values := fn.Call([]reflect.Value{ + reflect.Zero(signature.In(0)), + reflect.NewAt(signature.In(1).Elem(), unsafe.Pointer(unsafe.SliceData(data))), + reflect.ValueOf(len(data)).Convert(signature.In(2)), + }) + return int(values[0].Uint()), values[1].Bool() +} + +type ubWriteCapture struct { + level slog.Level + beforeEnabled func() + mu sync.Mutex + records []slog.Record +} + +func (h *ubWriteCapture) Enabled(_ context.Context, level slog.Level) bool { + if h.beforeEnabled != nil { + h.beforeEnabled() + } + return level >= h.level +} +func (h *ubWriteCapture) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + h.records = append(h.records, r.Clone()) + return nil +} +func (h *ubWriteCapture) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *ubWriteCapture) WithGroup(string) slog.Handler { return h } +func (h *ubWriteCapture) snapshot() []slog.Record { + h.mu.Lock() + defer h.mu.Unlock() + return append([]slog.Record(nil), h.records...) +} + +type ubWriteResponseWriter struct { + data []byte + n int + err error +} + +func (w *ubWriteResponseWriter) Header() http.Header { return nil } +func (w *ubWriteResponseWriter) WriteHeader(int) {} +func (w *ubWriteResponseWriter) Write(p []byte) (int, error) { + w.data = append(w.data, p...) + if w.n >= 0 { + return w.n, w.err + } + return len(p), w.err +} From ef6a806232e8fd895867bced3d8ecac5d36afafb Mon Sep 17 00:00:00 2001 From: "a.bolonin" Date: Fri, 23 Oct 2026 17:10:03 +0300 Subject: [PATCH 2/4] perf: skip worker output copy when info logging is disabled --- frankenphp.go | 3 +-- ubwrite_test.go | 28 +++------------------------- 2 files changed, 4 insertions(+), 27 deletions(-) diff --git a/frankenphp.go b/frankenphp.go index 7b2a0fad6e..e560031e62 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -553,9 +553,8 @@ func go_ub_write(threadIndex C.uintptr_t, cBuf *C.char, length C.size_t) (C.size } if fc.responseWriter == nil { - // Copy before Enabled: the C buffer may change during a logger callback. - message := string(unsafe.Slice((*byte)(unsafe.Pointer(cBuf)), length)) if fc.logger.Enabled(fc.ctx, slog.LevelInfo) { + message := string(unsafe.Slice((*byte)(unsafe.Pointer(cBuf)), length)) fc.logger.LogAttrs(fc.ctx, slog.LevelInfo, message) } return length, C.bool(fc.clientHasClosed()) diff --git a/ubwrite_test.go b/ubwrite_test.go index 07fc60ac96..73cb66eb24 100644 --- a/ubwrite_test.go +++ b/ubwrite_test.go @@ -108,24 +108,6 @@ func checkUBWrite(t *testing.T, message, mode string) { } } -func TestUBWriteCopiesBeforeLoggerEnabled(t *testing.T) { - data := []byte("original startup output") - capture := &ubWriteCapture{level: slog.LevelInfo, beforeEnabled: func() { - for i := range data { - data[i] = 'Z' - } - }} - fc := &frankenPHPContext{ctx: context.Background(), logger: slog.New(capture)} - oldThreads := phpThreads - phpThreads = []*phpThread{{handler: ®ularThread{fc: fc}}} - t.Cleanup(func() { phpThreads = oldThreads }) - invokeUBWrite(data) - records := capture.snapshot() - if len(records) != 1 || records[0].Message != "original startup output" { - t.Fatalf("source mutated before snapshot: %v", records) - } -} - // Reflection bridges the C argument types without adding production test hooks // or importing C from a _test.go file (which Go does not support). func invokeUBWrite(data []byte) (int, bool) { @@ -140,16 +122,12 @@ func invokeUBWrite(data []byte) (int, bool) { } type ubWriteCapture struct { - level slog.Level - beforeEnabled func() - mu sync.Mutex - records []slog.Record + level slog.Level + mu sync.Mutex + records []slog.Record } func (h *ubWriteCapture) Enabled(_ context.Context, level slog.Level) bool { - if h.beforeEnabled != nil { - h.beforeEnabled() - } return level >= h.level } func (h *ubWriteCapture) Handle(_ context.Context, r slog.Record) error { From a9cbe7056376cce904ef95d7bcc5db0101e2a6fc Mon Sep 17 00:00:00 2001 From: "a.bolonin" Date: Fri, 23 Oct 2026 17:30:28 +0300 Subject: [PATCH 3/4] test: abort worker request after dispatch to avoid cancellation race --- worker_test.go | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/worker_test.go b/worker_test.go index c5113c9da2..4460303ae8 100644 --- a/worker_test.go +++ b/worker_test.go @@ -186,17 +186,30 @@ func TestKeepRunningOnConnectionAbort(t *testing.T) { req := httptest.NewRequest("GET", "http://example.com/worker-with-counter.php", nil) ctx, cancel := context.WithCancel(req.Context()) + defer cancel() req = req.WithContext(ctx) - cancel() - body1, _ := testRequest(req, handler, t) + w := &cancelOnWriteResponseWriter{ResponseRecorder: httptest.NewRecorder(), cancel: cancel} + handler(w, req) - assert.Equal(t, "requests:1", body1, "should have handled exactly one request") + assert.ErrorIs(t, ctx.Err(), context.Canceled, "should have aborted during the response") + assert.Equal(t, "requests:1", w.Body.String(), "should have handled exactly one request") body2, _ := testGet("http://example.com/worker-with-counter.php", handler, t) assert.Equal(t, "requests:2", body2, "should not have stopped execution after the first request was aborted") }, &testOptions{workerScript: "worker-with-counter.php", nbWorkers: 1, nbParallelRequests: 1}) } +// Cancel only once the worker writes, so dispatch cannot discard the request first. +type cancelOnWriteResponseWriter struct { + *httptest.ResponseRecorder + cancel context.CancelFunc +} + +func (w *cancelOnWriteResponseWriter) Write(p []byte) (int, error) { + w.cancel() + return w.ResponseRecorder.Write(p) +} + // TestWorkerMaxRequests verifies that a worker restarts after reaching max_requests. func TestWorkerMaxRequests(t *testing.T) { const maxRequests = 5 From ced1d1ca88be478a75a5acb9c36fca9b99f7c117 Mon Sep 17 00:00:00 2001 From: "a.bolonin" Date: Sun, 25 Oct 2026 19:46:20 +0300 Subject: [PATCH 4/4] test: focus worker output coverage on log message ownership --- ubwrite_test.go | 161 ---------------------------------------- worker_internal_test.go | 44 +++++++++++ 2 files changed, 44 insertions(+), 161 deletions(-) delete mode 100644 ubwrite_test.go diff --git a/ubwrite_test.go b/ubwrite_test.go deleted file mode 100644 index 73cb66eb24..0000000000 --- a/ubwrite_test.go +++ /dev/null @@ -1,161 +0,0 @@ -package frankenphp - -import ( - "context" - "io" - "log/slog" - "net/http" - "reflect" - "strings" - "sync" - "testing" - "unsafe" -) - -func TestUBWrite(t *testing.T) { - for _, data := range []string{"", "a", "worker startup", "a\x00b\xff", strings.Repeat("x", 1024)} { - for _, mode := range []string{"log", "disabled", "http", "error", "short", "done", "done_closed", "canceled"} { - t.Run(mode+"/"+data, func(t *testing.T) { checkUBWrite(t, data, mode) }) - } - } -} - -func FuzzUBWrite(f *testing.F) { - f.Add("worker startup", uint8(0)) - f.Add("a\x00b\xff", uint8(1)) - f.Add("", uint8(2)) - f.Fuzz(func(t *testing.T, data string, mode uint8) { - modes := []string{"log", "disabled", "http", "error", "short", "done", "done_closed", "canceled"} - checkUBWrite(t, data, modes[int(mode)%len(modes)]) - }) -} - -func checkUBWrite(t *testing.T, message, mode string) { - t.Helper() - data := []byte(message) - capture := &ubWriteCapture{level: slog.LevelInfo} - fc := &frankenPHPContext{ctx: context.Background(), logger: slog.New(capture)} - writer := &ubWriteResponseWriter{n: -1} - expectedN, expectedClosed := len(data), false - switch mode { - case "disabled": - capture.level = slog.LevelWarn - case "http": - fc.responseWriter = writer - case "error": - writer.n = 0 - writer.err = io.ErrClosedPipe - fc.responseWriter = writer - capture.level = slog.LevelDebug - expectedN = 0 - case "short": - writer.n = 0 - fc.responseWriter = writer - expectedN = 0 - case "done": - fc.isDone = true - case "done_closed": - fc.isDone = true - fc.clientHadClosed = true - expectedClosed = true - case "canceled": - ctx, cancel := context.WithCancel(context.Background()) - cancel() - fc.ctx = ctx - fc.request = &http.Request{} - expectedClosed = true - } - oldThreads := phpThreads - phpThreads = []*phpThread{{handler: ®ularThread{fc: fc}}} - t.Cleanup(func() { phpThreads = oldThreads }) - n, closed := invokeUBWrite(data) - if n != expectedN || closed != expectedClosed { - t.Fatalf("write = (%d,%v), want (%d,%v)", n, closed, expectedN, expectedClosed) - } - // A handler is allowed to retain the record after the source buffer changes. - for i := range data { - data[i] = 'Z' - } - records := capture.snapshot() - switch mode { - case "log", "canceled": - if len(records) != 1 || records[0].Message != message || records[0].Level != slog.LevelInfo { - t.Fatalf("unexpected log records: %v", records) - } - case "error": - if len(records) != 1 || records[0].Message != "write error" || records[0].Level != slog.LevelDebug { - t.Fatalf("unexpected error records: %v", records) - } - var loggedError error - records[0].Attrs(func(a slog.Attr) bool { - if a.Key == "error" { - loggedError, _ = a.Value.Any().(error) - } - return true - }) - if loggedError != io.ErrClosedPipe { - t.Fatalf("logged error = %v", loggedError) - } - default: - if len(records) != 0 { - t.Fatalf("unexpected log records: %v", records) - } - } - if mode == "http" || mode == "error" || mode == "short" { - if string(writer.data) != message { - t.Fatalf("response bytes = %q, want %q", writer.data, message) - } - } -} - -// Reflection bridges the C argument types without adding production test hooks -// or importing C from a _test.go file (which Go does not support). -func invokeUBWrite(data []byte) (int, bool) { - fn := reflect.ValueOf(go_ub_write) - signature := fn.Type() - values := fn.Call([]reflect.Value{ - reflect.Zero(signature.In(0)), - reflect.NewAt(signature.In(1).Elem(), unsafe.Pointer(unsafe.SliceData(data))), - reflect.ValueOf(len(data)).Convert(signature.In(2)), - }) - return int(values[0].Uint()), values[1].Bool() -} - -type ubWriteCapture struct { - level slog.Level - mu sync.Mutex - records []slog.Record -} - -func (h *ubWriteCapture) Enabled(_ context.Context, level slog.Level) bool { - return level >= h.level -} -func (h *ubWriteCapture) Handle(_ context.Context, r slog.Record) error { - h.mu.Lock() - defer h.mu.Unlock() - h.records = append(h.records, r.Clone()) - return nil -} -func (h *ubWriteCapture) WithAttrs([]slog.Attr) slog.Handler { return h } -func (h *ubWriteCapture) WithGroup(string) slog.Handler { return h } -func (h *ubWriteCapture) snapshot() []slog.Record { - h.mu.Lock() - defer h.mu.Unlock() - return append([]slog.Record(nil), h.records...) -} - -type ubWriteResponseWriter struct { - data []byte - n int - err error -} - -func (w *ubWriteResponseWriter) Header() http.Header { return nil } -func (w *ubWriteResponseWriter) WriteHeader(int) {} -func (w *ubWriteResponseWriter) Write(p []byte) (int, error) { - w.data = append(w.data, p...) - if w.n >= 0 { - return w.n, w.err - } - return len(p), w.err -} diff --git a/worker_internal_test.go b/worker_internal_test.go index db7b7e9797..a54ff41b43 100644 --- a/worker_internal_test.go +++ b/worker_internal_test.go @@ -2,12 +2,15 @@ package frankenphp import ( "context" + "log/slog" "net/http/httptest" "os" "path/filepath" + "reflect" "runtime" "testing" "time" + "unsafe" "github.com/dunglas/frankenphp/internal/state" "github.com/stretchr/testify/assert" @@ -165,3 +168,44 @@ func TestInitReturnsWhenARebootWaitsOnAWorkerThatGivesUp(t *testing.T) { t.Fatal("Init() hung behind a reboot waiting for a worker thread that ended on its own") } } + +func TestWorkerStartupLogOwnsMessage(t *testing.T) { + const message = "worker startup\x00\xff" + data := []byte(message) + capture := &workerStartupLogCapture{} + fc := &frankenPHPContext{ctx: context.Background(), logger: slog.New(capture)} + oldThreads := phpThreads + phpThreads = []*phpThread{{handler: ®ularThread{fc: fc}}} + t.Cleanup(func() { phpThreads = oldThreads }) + + // Reflection bridges C argument types, which cannot be imported in a Go test file. + fn := reflect.ValueOf(go_ub_write) + signature := fn.Type() + result := fn.Call([]reflect.Value{ + reflect.Zero(signature.In(0)), + reflect.NewAt(signature.In(1).Elem(), unsafe.Pointer(unsafe.SliceData(data))), + reflect.ValueOf(len(data)).Convert(signature.In(2)), + }) + require.Equal(t, uint64(len(data)), result[0].Uint()) + require.False(t, result[1].Bool()) + + // A log handler may retain the record after PHP reuses its output buffer. + for i := range data { + data[i] = 'Z' + } + require.Len(t, capture.records, 1) + assert.Equal(t, message, capture.records[0].Message) + assert.Equal(t, slog.LevelInfo, capture.records[0].Level) +} + +type workerStartupLogCapture struct { + records []slog.Record +} + +func (h *workerStartupLogCapture) Enabled(context.Context, slog.Level) bool { return true } +func (h *workerStartupLogCapture) Handle(_ context.Context, record slog.Record) error { + h.records = append(h.records, record.Clone()) + return nil +} +func (h *workerStartupLogCapture) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *workerStartupLogCapture) WithGroup(string) slog.Handler { return h }