fix: bound the reboot wait so an unkillable thread can't wedge the pool

internal/state/state.go

@@ -34,6 +34,13 @@ const (
 	RebootReady
 	// all threads are yielding for main thread reboot
 	YieldingForReboot
+
+	// thread failed to yield even after a force-kill (e.g. stuck in a
+	// blocking write to a stalled client, unreachable by force-kill, which
+	// only interrupts the Zend VM at the next opcode boundary). Terminal:
+	// the underlying OS thread may still be alive, so its slot is abandoned
+	// rather than reused, permanently excluding it from dispatch.
+	Abandoned
 )
 
 func (s State) String() string {
@@ -66,6 +73,8 @@ func (s State) String() string {
 		return "reboot ready"
 	case YieldingForReboot:
 		return "yielding for reboot"
+	case Abandoned:
+		return "abandoned"
 	default:
 		return "unknown"
 	}
@@ -201,8 +210,9 @@ func (ts *ThreadState) WaitForStateWithTimeout(timeout time.Duration, states ...
 func (ts *ThreadState) RequestSafeStateChange(nextState State) bool {
 	ts.mu.Lock()
 	switch ts.currentState {
-	// disallow state changes when already shutting down or done
-	case Reserved, ShuttingDown, Done:
+	// disallow state changes when already shutting down, done, or abandoned
+	// (abandoned is terminal: the thread never reaches a stable state again)
+	case Reserved, ShuttingDown, Done, Abandoned:
 		ts.mu.Unlock()
 
 		return false

phpmainthread.go

@@ -159,6 +159,12 @@ func (mainThread *phpMainThread) rebootAllThreads() bool {
 		}
 	}
 
+	// force_kill_thread only interrupts the Zend VM at the next opcode
+	// boundary (see frankenphp_force_kill_thread): a thread parked in a
+	// blocking write to a stalled client (e.g. an unresponsive HTTP/2 peer)
+	// has already left PHP execution and can't be reached by it. Give up on
+	// such threads instead of waiting on them forever, so one stuck thread
+	// can't wedge the reboot of every other thread.
 	for _, thread := range rebootingThreads {
 		rebootWg.Go(func() {
 			close(thread.drainChan)
@@ -176,7 +182,24 @@ func (mainThread *phpMainThread) rebootAllThreads() bool {
 				slog.String("timeout", rebootGracePeriod.String()),
 			)
 			thread.sendKillSignal()
-			thread.state.WaitFor(state.YieldingForReboot)
+			if thread.state.WaitForStateWithTimeout(rebootGracePeriod, state.YieldingForReboot) {
+				return
+			}
+
+			// still hasn't yielded: the underlying OS thread may still be alive
+			// (e.g. stuck in a blocking write), so its slot can't be reused
+			// without risking two OS threads sharing the same TSRM slot.
+			// Abandon it permanently rather than block every other thread's
+			// reboot on it; it stays out of Ready/Inactive forever, so
+			// dispatch naturally stops routing requests to it.
+			globalLogger.LogAttrs(
+				globalCtx,
+				slog.LevelError,
+				"thread unresponsive after force-kill, abandoning it permanently",
+				slog.String("name", thread.name()),
+				slog.String("state", thread.state.Name()),
+			)
+			thread.state.Set(state.Abandoned)
 		})
 	}
 
@@ -189,11 +212,19 @@ func (mainThread *phpMainThread) rebootAllThreads() bool {
 	}
 	mainThread.state.WaitFor(state.Ready)
 
+	rebootedThreads := 0
+	abandonedThreads := 0
 	for _, thread := range rebootingThreads {
+		if thread.state.Is(state.Abandoned) {
+			abandonedThreads++
+			continue
+		}
+
 		thread.drainChan = make(chan struct{})
 		if thread.state.CompareAndSwap(state.YieldingForReboot, state.RebootReady) {
 			// wait for any of the stable states
 			thread.state.WaitFor(state.Ready, state.Inactive, state.Reserved)
+			rebootedThreads++
 		} else {
 			panic("unexpected state on reboot: " + thread.state.Name() + " for thread " + thread.name())
 		}
@@ -204,7 +235,8 @@ func (mainThread *phpMainThread) rebootAllThreads() bool {
 		slog.LevelInfo,
 		"thread reboot finished",
 		slog.String("duration", time.Since(rebootStart).String()),
-		slog.Int("num_threads", len(rebootingThreads)),
+		slog.Int("num_threads", rebootedThreads),
+		slog.Int("abandoned_threads", abandonedThreads),
 	)
 
 	return true

reboot_stuck_thread_test.go

@@ -0,0 +1,151 @@
+package frankenphp_test
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/dunglas/frankenphp"
+	"github.com/stretchr/testify/require"
+)
+
+// blockingResponseWriter simulates a client whose connection is stalled: any
+// Write() call after headers are sent blocks forever. This reproduces a PHP
+// thread stuck inside go_ub_write with no way for FrankenPHP's force-kill
+// (which only interrupts the Zend VM at the next opcode boundary, see
+// frankenphp_force_kill_thread) to unstick it, since the thread has already
+// left PHP execution and is parked in a Go-level network write.
+type blockingResponseWriter struct {
+	header  http.Header
+	started chan struct{}
+	once    sync.Once
+}
+
+func newBlockingResponseWriter() *blockingResponseWriter {
+	return &blockingResponseWriter{header: make(http.Header), started: make(chan struct{})}
+}
+
+func (w *blockingResponseWriter) Header() http.Header { return w.header }
+
+func (w *blockingResponseWriter) WriteHeader(int) {}
+
+func (w *blockingResponseWriter) Write(p []byte) (int, error) {
+	w.once.Do(func() { close(w.started) })
+	select {} // block forever, like a write to a dead/stalled TCP connection
+}
+
+// TestRebootDoesNotHangOnUnkillableThread reproduces the hang described on
+// https://github.com/php/frankenphp/pull/2570#issuecomment-5099392587 and
+// https://github.com/php/frankenphp/issues/2553: a thread genuinely stuck in
+// a blocking write (e.g. to a stalled HTTP/2 client) can't be interrupted by
+// force-kill, so rebootAllThreads()'s unbounded post-force-kill wait blocks
+// forever, wedging the whole pool (scalingMu never released, isRebooting
+// never cleared) rather than just losing the one stuck thread.
+func TestRebootDoesNotHangOnUnkillableThread(t *testing.T) {
+	require.NoError(t, frankenphp.Init(frankenphp.WithNumThreads(2), frankenphp.WithMaxThreads(2)))
+	defer frankenphp.Shutdown()
+
+	cwd, _ := os.Getwd()
+
+	w := newBlockingResponseWriter()
+	go func() {
+		req := httptest.NewRequest("GET", "http://example.com/index.php", nil)
+		fr, err := frankenphp.NewRequestWithContext(req, frankenphp.WithRequestDocumentRoot(cwd+"/testdata/", false))
+		if err != nil {
+			t.Errorf("NewRequestWithContext: %v", err)
+			return
+		}
+		// echo.php-style output: any script that writes output triggers go_ub_write
+		_ = frankenphp.ServeHTTP(w, fr)
+	}()
+
+	select {
+	case <-w.started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the stuck request never started writing")
+	}
+
+	// give go_ub_write a moment to actually be blocked in Write()
+	time.Sleep(100 * time.Millisecond)
+
+	done := make(chan struct{})
+	go func() {
+		frankenphp.RestartWorkers()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(20 * time.Second):
+		t.Fatal("RestartWorkers() hung forever on a thread stuck in a blocking write (force-kill can't interrupt it)")
+	}
+
+	// the pool must still be usable after giving up on the stuck thread: a
+	// fresh request on a *new* writer must complete normally.
+	body, _ := testGet(fmt.Sprintf("http://example.com/index.php?i=%d", 1), func(rw http.ResponseWriter, r *http.Request) {
+		fr, err := frankenphp.NewRequestWithContext(r, frankenphp.WithRequestDocumentRoot(cwd+"/testdata/", false))
+		require.NoError(t, err)
+		require.NoError(t, frankenphp.ServeHTTP(rw, fr))
+	}, t)
+	require.Contains(t, body, "I am by birth a Genevese")
+}
+
+// TestShutdownDoesNotHangWithAbandonedThread proves that Shutdown() itself
+// bounds its wait once a thread has been abandoned: before Abandoned was a
+// real terminal state, RequestSafeStateChange's unbounded
+// WaitFor(Ready, Inactive, Reserved) fallback (and thread.shutdown()'s own
+// unbounded WaitFor(Done) after a failed force-kill) would recurse/block
+// forever on the same stuck thread that rebootAllThreads() gave up on.
+func TestShutdownDoesNotHangWithAbandonedThread(t *testing.T) {
+	require.NoError(t, frankenphp.Init(frankenphp.WithNumThreads(2), frankenphp.WithMaxThreads(2)))
+
+	cwd, _ := os.Getwd()
+
+	w := newBlockingResponseWriter()
+	go func() {
+		req := httptest.NewRequest("GET", "http://example.com/index.php", nil)
+		fr, err := frankenphp.NewRequestWithContext(req, frankenphp.WithRequestDocumentRoot(cwd+"/testdata/", false))
+		if err != nil {
+			t.Errorf("NewRequestWithContext: %v", err)
+			return
+		}
+		_ = frankenphp.ServeHTTP(w, fr)
+	}()
+
+	select {
+	case <-w.started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the stuck request never started writing")
+	}
+	time.Sleep(100 * time.Millisecond)
+
+	// Reboot first: this is what actually abandons the stuck thread.
+	rebootDone := make(chan struct{})
+	go func() {
+		frankenphp.RestartWorkers()
+		close(rebootDone)
+	}()
+
+	select {
+	case <-rebootDone:
+	case <-time.After(20 * time.Second):
+		t.Fatal("RestartWorkers() hung")
+	}
+
+	// Shutdown() must not wait forever on the now-abandoned thread.
+	shutdownDone := make(chan struct{})
+	go func() {
+		frankenphp.Shutdown()
+		close(shutdownDone)
+	}()
+
+	select {
+	case <-shutdownDone:
+	case <-time.After(10 * time.Second):
+		t.Fatal("Shutdown() hung forever on the abandoned thread")
+	}
+}

threadinactive.go

@@ -34,7 +34,7 @@ func (handler *inactiveThread) beforeScriptExecution() string {
 		thread.state.MarkAsWaiting(false)
 
 		return handler.beforeScriptExecution()
-	case state.Rebooting, state.ForceRebooting:
+	case state.Rebooting, state.ForceRebooting, state.Abandoned:
 		return ""
 	case state.RebootReady:
 		thread.state.Set(state.Inactive)

threadregular.go

@@ -53,7 +53,10 @@ func (handler *regularThread) beforeScriptExecution() string {
 		return handler.waitForRequest()
 	case state.Ready:
 		return handler.waitForRequest()
-	case state.Rebooting, state.ForceRebooting:
+	case state.Rebooting, state.ForceRebooting, state.Abandoned:
+		// Abandoned: this thread failed to yield during a reboot and was
+		// given up on (see rebootAllThreads); if the blocking call it was
+		// stuck in ever does return, just let the underlying thread exit.
 		return ""
 	case state.RebootReady:
 		handler.requestCount = 0

threadworker.go

@@ -56,7 +56,7 @@ func (handler *workerThread) beforeScriptExecution() string {
 		setupWorkerScript(handler, handler.worker)
 
 		return handler.worker.fileName
-	case state.Rebooting, state.ForceRebooting:
+	case state.Rebooting, state.ForceRebooting, state.Abandoned:
 		return ""
 	case state.RebootReady:
 		handler.requestCount = 0