fix: reboot threads instead of opcache_reset (#2364)

Instead of resetting the opcache (manually or on overflow), reboot all threads including the main thread while keeping requests queued. This avoids multiple races that exist in opcache and opcache ZTS. Does also affect resets when watching.
---------

Co-authored-by: henderkes <m@pyc.ac>

Author: AlliBalliBaba

caddy/admin_test.go

@@ -40,6 +40,9 @@ func TestRestartWorkerViaAdminApi(t *testing.T) {
 		}
 		`, "caddyfile")
 
+	// make sure workers are not still running from any previous tests
+	assertAdminResponse(t, tester, "POST", "workers/restart", http.StatusOK, "workers restarted successfully\n")
+
 	tester.AssertGetResponse("http://localhost:"+testPort+"/", http.StatusOK, "requests:1")
 	tester.AssertGetResponse("http://localhost:"+testPort+"/", http.StatusOK, "requests:2")
 

caddy/caddy_test.go

@@ -1794,6 +1794,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 := 500
+	wg.Add(numRequests)
+	for i := 0; i < numRequests; i++ {
+
+		// introduce a delay every 10 requests
+		if i%10 == 0 {
+			time.Sleep(time.Millisecond * 10)
+		}
+
+		go func(i int) {
+			defer wg.Done()
+			// spam opcache_reset on intervals
+			if i%10 > 7 {
+				tester.AssertGetResponse(
+					"http://localhost:"+testPort+"/opcache_reset.php",
+					http.StatusOK,
+					"opcache reset done",
+				)
+				return
+			}
+
+			// otherwise call sleep.php with different 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),
+			)
+		}(i)
+	}
+
+	wg.Wait()
+}
+
 func TestLog(t *testing.T) {
 	tester := caddytest.NewTester(t)
 	initServer(t, tester, `

caddy/hotreload_test.go

@@ -62,16 +62,20 @@ func TestHotReload(t *testing.T) {
 		req = req.WithContext(cx)
 		resp := tester.AssertResponseCode(req, http.StatusOK)
 
-		connected.Done()
-
 		var receivedBody strings.Builder
 
 		buf := make([]byte, 1024)
+		isConnected := false
 		for {
 			n, err := resp.Body.Read(buf)
 			if n > 0 {
 				receivedBody.Write(buf[:n])
 			}
+			if !isConnected {
+				// wait for the first bytes before marking the client as connected
+				isConnected = true
+				connected.Done()
+			}
 			if strings.Contains(receivedBody.String(), "index.php") {
 				cancel()
 

frankenphp.c

@@ -934,6 +934,31 @@ PHP_FUNCTION(frankenphp_log) {
   }
 }
 
+static void frankenphp_opcache_restart_hook(int reason) {
+  (void)reason;
+  go_schedule_opcache_reset(frankenphp_thread_index());
+}
+
+/* {{{ thread-safe opcache reset */
+PHP_FUNCTION(frankenphp_opcache_reset) {
+  frankenphp_opcache_restart_hook(0);
+
+  RETVAL_TRUE;
+} /* }}} */
+
+/* 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);
+  }
+}
+
 #ifdef FRANKENPHP_TEST
 /* Test-only entry point that exercises zval.h end-to-end:
  * validate -> persist (request -> persistent memory) ->
@@ -1005,6 +1030,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;
 }
 
@@ -1023,7 +1052,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; }
@@ -1392,6 +1430,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);
 
@@ -1571,6 +1615,13 @@ static void *php_main(void *arg) {
 
   frankenphp_sapi_module.startup(&frankenphp_sapi_module);
 
+#if defined(ZTS) && PHP_VERSION_ID >= 80400
+  /* Also restart everything on opcache memory overflow or similar events, the
+   * hook is triggered right before an opcache reset is scheduled
+   */
+  zend_accel_schedule_restart_hook = frankenphp_opcache_restart_hook;
+#endif
+
   /* check if a default filter is set in php.ini and only filter if
    * it is, this is deprecated and will be removed in PHP 9 */
   char *default_filter;
@@ -1784,16 +1835,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

@@ -375,7 +375,6 @@ func Shutdown() {
 	}
 
 	drainWatchers()
-	drainAutoScaling()
 	drainPHPThreads()
 
 	metrics.Shutdown()
@@ -761,6 +760,13 @@ 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) {
+	if mainThread != nil {
+		go mainThread.rebootAllThreads()
+	}
+}
+
 func convertArgs(args []string) (C.int, []*C.char) {
 	argc := C.int(len(args))
 	argv := make([]*C.char, argc)

frankenphp.h

@@ -204,7 +204,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

@@ -12,19 +12,15 @@ type State int
 
 const (
 	// lifecycle States of a thread
-	Reserved State = iota
+	Reserved State = iota // stable
 	Booting
 	BootRequested
 	ShuttingDown
 	Done
 
 	// these States are 'stable' and safe to transition from at any time
-	Inactive
-	Ready
-
-	// States necessary for restarting workers
-	Restarting
-	Yielding
+	Inactive // stable
+	Ready    // stable
 
 	// States necessary for transitioning between different handlers
 	TransitionRequested
@@ -33,8 +29,11 @@ const (
 
 	// thread is exiting the C loop for a full ZTS restart (max_requests)
 	Rebooting
+	ForceRebooting
 	// 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 +52,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:
@@ -65,8 +60,12 @@ func (s State) String() string {
 		return "transition complete"
 	case Rebooting:
 		return "rebooting"
+	case ForceRebooting:
+		return "rebooting (force)"
 	case RebootReady:
 		return "reboot ready"
+	case YieldingForReboot:
+		return "yielding for reboot"
 	default:
 		return "unknown"
 	}
@@ -168,16 +167,46 @@ func (ts *ThreadState) WaitFor(states ...State) {
 	<-sub.ch
 }
 
+func (ts *ThreadState) WaitForStateWithTimeout(timeout time.Duration, states ...State) bool {
+	ts.mu.Lock()
+	if slices.Contains(states, ts.currentState) {
+		ts.mu.Unlock()
+
+		return true
+	}
+
+	sub := stateSubscriber{
+		states: states,
+		ch:     make(chan struct{}),
+	}
+	ts.subscribers = append(ts.subscribers, sub)
+	ts.mu.Unlock()
+
+	select {
+	case <-sub.ch:
+		return true
+	case <-time.After(timeout):
+		ts.mu.Lock()
+		// remove subscriber so there is no leak potential
+		ts.subscribers = slices.DeleteFunc(ts.subscribers, func(s stateSubscriber) bool {
+			return sub.ch == s.ch
+		})
+		ts.mu.Unlock()
+		return false
+	}
+}
+
 // RequestSafeStateChange safely requests a state change from a different goroutine
+// returns false if the thread is already reserved, shutting down, or done
 func (ts *ThreadState) RequestSafeStateChange(nextState State) bool {
 	ts.mu.Lock()
 	switch ts.currentState {
-	// disallow state changes if shutting down or done
-	case ShuttingDown, Done, Reserved:
+	// disallow state changes when already shutting down or done
+	case Reserved, ShuttingDown, Done:
 		ts.mu.Unlock()
 
 		return false
-	// ready and inactive are safe states to transition from
+	// ready and inactive are stable states to transition from
 	case Ready, Inactive:
 		ts.currentState = nextState
 		ts.notifySubscribers(nextState)
@@ -187,8 +216,8 @@ func (ts *ThreadState) RequestSafeStateChange(nextState State) bool {
 	}
 	ts.mu.Unlock()
 
-	// wait for the state to change to a safe state
-	ts.WaitFor(Ready, Inactive, ShuttingDown)
+	// wait for the state to change to a stable state
+	ts.WaitFor(Ready, Inactive, Reserved)
 
 	return ts.RequestSafeStateChange(nextState)
 }