fix: reboot threads on opcache reset and watcher

caddy/caddy_test.go

@@ -3,6 +3,7 @@ package caddy_test
 import (
 	"bytes"
 	"fmt"
+	"math/rand/v2"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -1730,6 +1731,75 @@ func TestDd(t *testing.T) {
 	)
 }
 
+// test to force the opcache segfault race condition under concurrency (~1.7s)
+func TestOpcacheReset(t *testing.T) {
+	tester := caddytest.NewTester(t)
+	tester.Client.Timeout = 60 * time.Second
+	tester.InitServer(`
+		{
+			skip_install_trust
+			admin localhost:2999
+			http_port `+testPort+`
+			metrics
+
+			frankenphp {
+				num_threads 40
+				php_ini {
+					opcache.enable 1
+					opcache.log_verbosity_level 4
+					max_execution_time 30s
+				}
+			}
+		}
+
+		localhost:`+testPort+` {
+			php {
+				root ../testdata
+				worker {
+                    file sleep.php
+                    match /sleep*
+                    num 20
+                }
+			}
+		}
+		`, "caddyfile")
+
+	wg := sync.WaitGroup{}
+	numRequests := 1000
+	wg.Add(numRequests)
+	for i := 0; i < numRequests; i++ {
+
+		// introduce some random delay
+		if rand.IntN(10) > 8 {
+			time.Sleep(time.Millisecond * 10)
+		}
+
+		go func() {
+			defer wg.Done()
+			// randomly call opcache_reset
+			if rand.IntN(10) > 7 {
+				tester.AssertGetResponse(
+					"http://localhost:"+testPort+"/opcache_reset.php",
+					http.StatusOK,
+					"opcache reset done",
+				)
+				return
+			}
+
+			// otherwise call sleep.php with random sleep and work values
+			sleep := i % 100
+			work := i % 100
+			tester.AssertGetResponse(
+				fmt.Sprintf("http://localhost:%s/sleep.php?sleep=%d&work=%d", testPort, sleep, work),
+				http.StatusOK,
+				fmt.Sprintf("slept for %d ms and worked for %d iterations", sleep, work),
+			)
+		}()
+	}
+
+	wg.Wait()
+}
+
 func TestLog(t *testing.T) {
 	tester := caddytest.NewTester(t)
 	initServer(t, tester, `

frankenphp.c

@@ -92,6 +92,22 @@ static bool is_forked_child = false;
 static void frankenphp_fork_child(void) { is_forked_child = true; }
 #endif
 
+/* Forward declaration */
+PHP_FUNCTION(frankenphp_opcache_reset);
+
+/* Try to override opcache_reset if opcache is loaded.
+ * instead of resetting opcache, reboot all threads */
+static void frankenphp_override_opcache_reset(void) {
+  zend_function *func = zend_hash_str_find_ptr(
+      CG(function_table), "opcache_reset", sizeof("opcache_reset") - 1);
+  if (func != NULL && func->type == ZEND_INTERNAL_FUNCTION &&
+      ((zend_internal_function *)func)->handler !=
+          ZEND_FN(frankenphp_opcache_reset)) {
+    ((zend_internal_function *)func)->handler =
+        ZEND_FN(frankenphp_opcache_reset);
+  }
+}
+
 void frankenphp_update_local_thread_context(bool is_worker) {
   is_worker_thread = is_worker;
 
@@ -467,6 +483,13 @@ PHP_FUNCTION(frankenphp_getenv) {
   }
 } /* }}} */
 
+/* {{{ thread-safe opcache reset */
+PHP_FUNCTION(frankenphp_opcache_reset) {
+  go_schedule_opcache_reset(thread_index);
+
+  RETVAL_TRUE;
+} /* }}} */
+
 /* {{{ Fetch all HTTP request headers */
 PHP_FUNCTION(frankenphp_request_headers) {
   ZEND_PARSE_PARAMETERS_NONE();
@@ -734,6 +757,10 @@ PHP_MINIT_FUNCTION(frankenphp) {
     php_error(E_WARNING, "Failed to find built-in getenv function");
   }
 
+  // Override opcache_reset (may not be available yet if opcache loads as a
+  // shared extension in PHP 8.4 and below)
+  frankenphp_override_opcache_reset();
+
   return SUCCESS;
 }
 
@@ -752,7 +779,16 @@ static zend_module_entry frankenphp_module = {
 static int frankenphp_startup(sapi_module_struct *sapi_module) {
   php_import_environment_variables = get_full_env;
 
-  return php_module_startup(sapi_module, &frankenphp_module);
+  int result = php_module_startup(sapi_module, &frankenphp_module);
+#if PHP_VERSION_ID < 80500
+  if (result == SUCCESS) {
+    /* Override opcache here again if loaded as a shared extension
+     * (php 8.4 and under) */
+    frankenphp_override_opcache_reset();
+  }
+#endif
+
+  return result;
 }
 
 static int frankenphp_deactivate(void) { return SUCCESS; }
@@ -1092,6 +1128,12 @@ static void *php_thread(void *arg) {
         zend_bailout();
       }
 
+#if PHP_VERSION_ID < 80500
+      /* Override opcache here again if loaded as a shared extension
+       * (php 8.4 and under) */
+      frankenphp_override_opcache_reset();
+#endif
+
       zend_file_handle file_handle;
       zend_stream_init_filename(&file_handle, scriptName);
 
@@ -1457,16 +1499,6 @@ int frankenphp_execute_script_cli(char *script, int argc, char **argv,
   return (intptr_t)exit_status;
 }
 
-int frankenphp_reset_opcache(void) {
-  zend_function *opcache_reset =
-      zend_hash_str_find_ptr(CG(function_table), ZEND_STRL("opcache_reset"));
-  if (opcache_reset) {
-    zend_call_known_function(opcache_reset, NULL, NULL, NULL, 0, NULL, NULL);
-  }
-
-  return 0;
-}
-
 int frankenphp_get_current_memory_limit() { return PG(memory_limit); }
 
 void frankenphp_init_thread_metrics(int max_threads) {

frankenphp.go

@@ -756,6 +756,11 @@ func go_is_context_done(threadIndex C.uintptr_t) C.bool {
 	return C.bool(phpThreads[threadIndex].frankenPHPContext().isDone)
 }
 
+//export go_schedule_opcache_reset
+func go_schedule_opcache_reset(threadIndex C.uintptr_t) {
+	go mainThread.rebootAllThreads()
+}
+
 func convertArgs(args []string) (C.int, []*C.char) {
 	argc := C.int(len(args))
 	argv := make([]*C.char, argc)

frankenphp.h

@@ -182,7 +182,6 @@ void frankenphp_register_server_vars(zval *track_vars_array,
                                      frankenphp_server_vars vars);
 
 zend_string *frankenphp_init_persistent_string(const char *string, size_t len);
-int frankenphp_reset_opcache(void);
 int frankenphp_get_current_memory_limit();
 
 typedef struct {

internal/state/state.go

@@ -22,10 +22,6 @@ const (
 	Inactive
 	Ready
 
-	// States necessary for restarting workers
-	Restarting
-	Yielding
-
 	// States necessary for transitioning between different handlers
 	TransitionRequested
 	TransitionInProgress
@@ -35,6 +31,8 @@ const (
 	Rebooting
 	// C thread has exited and ZTS state is cleaned up, ready to spawn a new C thread
 	RebootReady
+	// all threads are yielding for main thread reboot
+	YieldingForReboot
 )
 
 func (s State) String() string {
@@ -53,10 +51,6 @@ func (s State) String() string {
 		return "inactive"
 	case Ready:
 		return "ready"
-	case Restarting:
-		return "restarting"
-	case Yielding:
-		return "yielding"
 	case TransitionRequested:
 		return "transition requested"
 	case TransitionInProgress:
@@ -67,6 +61,8 @@ func (s State) String() string {
 		return "rebooting"
 	case RebootReady:
 		return "reboot ready"
+	case YieldingForReboot:
+		return "yielding for reboot"
 	default:
 		return "unknown"
 	}

phpmainthread.go

@@ -9,6 +9,7 @@ import (
 	"log/slog"
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/dunglas/frankenphp/internal/memory"
 	"github.com/dunglas/frankenphp/internal/phpheaders"
@@ -18,11 +19,12 @@ import (
 // represents the main PHP thread
 // the thread needs to keep running as long as all other threads are running
 type phpMainThread struct {
-	state      *state.ThreadState
-	done       chan struct{}
-	numThreads int
-	maxThreads int
-	phpIni     map[string]string
+	state       *state.ThreadState
+	done        chan struct{}
+	numThreads  int
+	maxThreads  int
+	phpIni      map[string]string
+	isRebooting atomic.Bool
 }
 
 var (
@@ -121,6 +123,47 @@ func (mainThread *phpMainThread) start() error {
 	return nil
 }
 
+// rebootAllThreads reboots all underlying C threads, but keeps the go side alive
+func (mainThread *phpMainThread) rebootAllThreads() bool {
+	if !mainThread.isRebooting.CompareAndSwap(false, true) {
+		return false
+	}
+
+	scalingMu.Lock()
+	defer scalingMu.Unlock()
+	globalLogger.Info("rebooting all threads")
+
+	wg := sync.WaitGroup{}
+	rebootingThreads := []*phpThread{}
+	for _, thread := range phpThreads {
+		if thread.reboot() {
+			rebootingThreads = append(rebootingThreads, thread)
+			wg.Go(func() {
+				close(thread.drainChan)
+				thread.state.WaitFor(state.YieldingForReboot)
+			})
+		}
+	}
+	wg.Wait()
+
+	mainThread.state.Set(state.Rebooting)
+	mainThread.state.WaitFor(state.YieldingForReboot)
+	if C.frankenphp_new_main_thread(C.int(mainThread.numThreads)) != 0 {
+		panic("unable to recreate main thread after reboot")
+	}
+	mainThread.state.WaitFor(state.Ready)
+
+	for _, thread := range rebootingThreads {
+		thread.drainChan = make(chan struct{})
+		thread.state.Set(state.RebootReady)
+		thread.state.WaitFor(state.Ready, state.ShuttingDown)
+	}
+
+	mainThread.isRebooting.Store(false)
+
+	return true
+}
+
 func getInactivePHPThread() *phpThread {
 	for _, thread := range phpThreads {
 		if thread.state.Is(state.Inactive) {
@@ -146,7 +189,7 @@ func go_frankenphp_main_thread_is_ready() {
 	}
 
 	mainThread.state.Set(state.Ready)
-	mainThread.state.WaitFor(state.Done)
+	mainThread.state.WaitFor(state.Done, state.Rebooting)
 }
 
 // max_threads = auto
@@ -172,6 +215,9 @@ func (mainThread *phpMainThread) setAutomaticMaxThreads() {
 
 //export go_frankenphp_shutdown_main_thread
 func go_frankenphp_shutdown_main_thread() {
+	if mainThread.state.CompareAndSwap(state.Rebooting, state.YieldingForReboot) {
+		return
+	}
 	mainThread.state.Set(state.Reserved)
 }
 

phpthread.go

@@ -208,7 +208,13 @@ func go_frankenphp_on_thread_shutdown(threadIndex C.uintptr_t) {
 	thread := phpThreads[threadIndex]
 	thread.Unpin()
 	if thread.state.Is(state.Rebooting) {
-		thread.state.Set(state.RebootReady)
+		if mainThread.isRebooting.Load() {
+			// if all threads are rebooting, yield
+			thread.state.Set(state.YieldingForReboot)
+		} else {
+			// if only this thread is rebooting, set to ready
+			thread.state.Set(state.RebootReady)
+		}
 	} else {
 		thread.state.Set(state.Done)
 	}

testdata/opcache_reset.php

@@ -0,0 +1,9 @@
+<?php
+
+require_once __DIR__.'/_executor.php';
+
+return function () {
+    require __DIR__ .'/require.php';
+    opcache_reset();
+    echo "opcache reset done";
+};

testdata/require.php

@@ -0,0 +1,6 @@
+<?php
+
+// dummy require file for opcache_reset test
+return function (){
+    echo "";
+};

threadregular.go

@@ -1,5 +1,7 @@
 package frankenphp
 
+// #include "frankenphp.h"
+import "C"
 import (
 	"context"
 	"log/slog"
@@ -48,10 +50,10 @@ func (handler *regularThread) beforeScriptExecution() string {
 		handler.state.Set(state.Ready)
 
 		return handler.waitForRequest()
-
 	case state.Ready:
 		return handler.waitForRequest()
-
+	case state.Rebooting:
+		return ""
 	case state.RebootReady:
 		handler.requestCount = 0
 		handler.state.Set(state.Ready)