Starts separate opcache_reset request flow once all threads are stopped.

Author: AlliBalliBaba

frankenphp.c

@@ -343,11 +343,9 @@ PHP_FUNCTION(frankenphp_getenv) {
 
 /* {{{ thread-safe opcache reset */
 PHP_FUNCTION(frankenphp_opcache_reset) {
-  if (go_schedule_opcache_reset(thread_index)) {
-    orig_opcache_reset(INTERNAL_FUNCTION_PARAM_PASSTHRU);
-  }
+  go_schedule_opcache_reset(thread_index);
 
-  RETVAL_FALSE;
+  RETVAL_TRUE;
 } /* }}} */
 
 /* {{{ Fetch all HTTP request headers */
@@ -1270,11 +1268,15 @@ int frankenphp_execute_script_cli(char *script, int argc, char **argv,
 }
 
 int frankenphp_reset_opcache(void) {
+  php_request_startup();
   zend_function *opcache_reset =
       zend_hash_str_find_ptr(CG(function_table), ZEND_STRL("opcache_reset"));
   if (opcache_reset) {
+    ((zend_internal_function *)opcache_reset)->handler = orig_opcache_reset;
     zend_call_known_function(opcache_reset, NULL, NULL, NULL, 0, NULL, NULL);
+    ((zend_internal_function *)opcache_reset)->handler = ZEND_FN(frankenphp_opcache_reset);
   }
+  php_request_shutdown((void *)0);
 
   return 0;
 }

frankenphp.go

@@ -59,10 +59,9 @@ var (
 	contextKey   = contextKeyStruct{}
 	serverHeader = []string{"FrankenPHP"}
 
-	isRunning            bool
-	isOpcacheResetting   atomic.Bool
-	threadsAreRestarting atomic.Bool
-	onServerShutdown     []func()
+	isRunning        bool
+	restartCounter   atomic.Int32
+	onServerShutdown []func()
 
 	// Set default values to make Shutdown() idempotent
 	globalMu     sync.Mutex
@@ -759,68 +758,49 @@ func go_is_context_done(threadIndex C.uintptr_t) C.bool {
 }
 
 //export go_schedule_opcache_reset
-func go_schedule_opcache_reset(threadIndex C.uintptr_t) C.bool {
-	if isOpcacheResetting.CompareAndSwap(false, true) {
-		restartThreadsForOpcacheReset(phpThreads[threadIndex])
-		return C.bool(true)
+func go_schedule_opcache_reset(threadIndex C.uintptr_t) {
+	if restartCounter.CompareAndSwap(0, 1) {
+		go restartThreadsForOpcacheReset()
 	}
-
-	// always call the original opcache_reset if already restarting
-	return C.bool(phpThreads[threadIndex].state.Is(state.Restarting))
 }
 
 // restart all threads for an opcache_reset
-func restartThreadsForOpcacheReset(callingThread *phpThread) {
-	if threadsAreRestarting.Load() {
-		// ignore reloads while a restart is already ongoing
-		return
-	}
-
+func restartThreadsForOpcacheReset() {
 	// disallow scaling threads while restarting workers
 	scalingMu.Lock()
 	defer scalingMu.Unlock()
 
-	// drain workers
-	globalLogger.Info("Restarting all PHP threads for opcache_reset")
-	threadsToRestart := drainWorkerThreads()
-
-	// drain regular threads
-	globalLogger.Info("Draining regular PHP threads for opcache_reset")
-	wg := sync.WaitGroup{}
-	for _, thread := range regularThreads {
-		if thread.state.Is(state.Ready) {
-			threadsToRestart = append(threadsToRestart, thread)
-			thread.state.Set(state.Restarting)
-			close(thread.drainChan)
-
-			wg.Go(func() {
-				thread.state.WaitFor(state.Yielding)
-			})
-		}
+	threadsToRestart := drainWorkerThreads(true)
+
+	for _, thread := range threadsToRestart {
+		thread.drainChan = make(chan struct{})
+		thread.state.Set(state.Ready)
 	}
+}
 
-	wg.Done() // ignore the calling thread
-	done := make(chan struct{})
-	go func() {
-		wg.Wait()
-		close(done)
-	}()
+func scheduleOpcacheReset(thread *phpThread) {
+	restartCounter.Add(-1)
+	if restartCounter.Load() != 1 {
+		return // only the last restarting thread will trigger an actual opcache_reset
+	}
+	workerThread, ok := thread.handler.(*workerThread)
+	fc, _ := newDummyContext("/opcache_reset")
+	if ok && workerThread.worker != nil {
+		workerThread.dummyFrankenPHPContext = fc
+		defer func() { workerThread.dummyFrankenPHPContext = nil }()
+	}
 
-	select {
-	case <-done:
-		// all other threads are drained
-	case <-time.After(time.Second):
-		// probably a deadlock, continue anyway and hope for the best
+	regularThread, ok := thread.handler.(*regularThread)
+	if ok {
+		regularThread.contextHolder.frankenPHPContext = fc
+		defer func() { regularThread.contextHolder.frankenPHPContext = nil }()
 	}
 
-	go func() {
-		callingThread.state.WaitFor(state.Yielding)
-		for _, thread := range threadsToRestart {
-			thread.drainChan = make(chan struct{})
-			thread.state.Set(state.Ready)
-			isOpcacheResetting.Store(false)
-		}
-	}()
+	globalLogger.Info("resetting opcache in all threads")
+	C.frankenphp_reset_opcache()
+
+	// all threads should have restarted now
+	restartCounter.Store(0)
 }
 
 // ExecuteScriptCLI executes the PHP script passed as parameter.

threadregular.go

@@ -50,6 +50,7 @@ func (handler *regularThread) beforeScriptExecution() string {
 		return handler.waitForRequest()
 	case state.Restarting:
 		handler.state.Set(state.Yielding)
+		scheduleOpcacheReset(handler.thread)
 		handler.state.WaitFor(state.Ready, state.ShuttingDown)
 		return handler.beforeScriptExecution()
 

threadworker.go

@@ -51,6 +51,7 @@ func (handler *workerThread) beforeScriptExecution() string {
 			handler.worker.onThreadShutdown(handler.thread.threadIndex)
 		}
 		handler.state.Set(state.Yielding)
+		scheduleOpcacheReset(handler.thread)
 		handler.state.WaitFor(state.Ready, state.ShuttingDown)
 		return handler.beforeScriptExecution()
 	case state.Ready, state.TransitionComplete:
@@ -226,12 +227,6 @@ func (handler *workerThread) waitForWorkerRequest() (bool, any) {
 			globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "shutting down", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex))
 		}
 
-		// flush the opcache when restarting due to watcher or admin api
-		// note: this is done right before frankenphp_handle_request() returns 'false'
-		if handler.state.Is(state.Restarting) {
-			C.frankenphp_reset_opcache()
-		}
-
 		return false, nil
 	case requestCH = <-handler.thread.requestChan:
 	case requestCH = <-handler.worker.requestChan:

worker.go

@@ -171,10 +171,10 @@ func newWorker(o workerOpt) (*worker, error) {
 
 // EXPERIMENTAL: DrainWorkers finishes all worker scripts before a graceful shutdown
 func DrainWorkers() {
-	_ = drainWorkerThreads()
+	_ = drainWorkerThreads(false)
 }
 
-func drainWorkerThreads() []*phpThread {
+func drainWorkerThreads(withRegularThreads bool) []*phpThread {
 	var (
 		ready          sync.WaitGroup
 		drainedThreads []*phpThread
@@ -192,7 +192,7 @@ func drainWorkerThreads() []*phpThread {
 				// we'll proceed to restart all other threads anyway
 				continue
 			}
-
+			restartCounter.Add(1)
 			close(thread.drainChan)
 			drainedThreads = append(drainedThreads, thread)
 
@@ -205,6 +205,31 @@ func drainWorkerThreads() []*phpThread {
 		worker.threadMutex.RUnlock()
 	}
 
+	if withRegularThreads {
+		regularThreadMu.RLock()
+		ready.Add(len(regularThreads))
+
+		for _, thread := range regularThreads {
+			if !thread.state.RequestSafeStateChange(state.Restarting) {
+				ready.Done()
+
+				// no state change allowed == thread is shutting down
+				// we'll proceed to restart all other threads anyway
+				continue
+			}
+			restartCounter.Add(1)
+			close(thread.drainChan)
+			drainedThreads = append(drainedThreads, thread)
+
+			go func(thread *phpThread) {
+				thread.state.WaitFor(state.Yielding)
+				ready.Done()
+			}(thread)
+		}
+
+		regularThreadMu.RUnlock()
+	}
+
 	ready.Wait()
 
 	return drainedThreads
@@ -213,13 +238,14 @@ func drainWorkerThreads() []*phpThread {
 // RestartWorkers attempts to restart all workers gracefully
 // All workers must be restarted at the same time to prevent issues with opcache resetting.
 func RestartWorkers() {
-	threadsAreRestarting.Store(true)
-	defer threadsAreRestarting.Store(false)
+	if !restartCounter.CompareAndSwap(0, 1) {
+		return // another restart is already in progress
+	}
 	// disallow scaling threads while restarting workers
 	scalingMu.Lock()
 	defer scalingMu.Unlock()
 
-	threadsToRestart := drainWorkerThreads()
+	threadsToRestart := drainWorkerThreads(false)
 
 	for _, thread := range threadsToRestart {
 		thread.drainChan = make(chan struct{})