fix(metrics): handle the case where the worker is already assigned to a thread

caddy/watcher_test.go

@@ -15,7 +15,7 @@ func TestWorkerWithInactiveWatcher(t *testing.T) {
 		{
 			skip_install_trust
 			admin localhost:2999
-			http_port 9080
+			http_port `+testPort+`
 
 			frankenphp {
 				worker {
@@ -26,13 +26,13 @@ func TestWorkerWithInactiveWatcher(t *testing.T) {
 			}
 		}
 
-		localhost:9080 {
+		localhost:`+testPort+` {
 			root ../testdata
 			rewrite worker-with-watcher.php
 			php
 		}
 		`, "caddyfile")
 
-	tester.AssertGetResponse("http://localhost:9080", http.StatusOK, "requests:1")
-	tester.AssertGetResponse("http://localhost:9080", http.StatusOK, "requests:2")
+	tester.AssertGetResponse("http://localhost:"+testPort, http.StatusOK, "requests:1")
+	tester.AssertGetResponse("http://localhost:"+testPort, http.StatusOK, "requests:2")
 }

php_thread.go

@@ -6,6 +6,7 @@ import "C"
 import (
 	"net/http"
 	"runtime"
+	"sync"
 	"unsafe"
 )
 
@@ -19,6 +20,7 @@ type phpThread struct {
 	worker            *worker
 	requestChan       chan *http.Request
 	knownVariableKeys map[string]*C.zend_string
+	readiedOnce       sync.Once
 }
 
 func initPHPThreads(numThreads int) {
@@ -28,7 +30,7 @@ func initPHPThreads(numThreads int) {
 	}
 }
 
-func (thread phpThread) getActiveRequest() *http.Request {
+func (thread *phpThread) getActiveRequest() *http.Request {
 	if thread.workerRequest != nil {
 		return thread.workerRequest
 	}
@@ -46,5 +48,5 @@ func (thread *phpThread) pinString(s string) *C.char {
 
 // C strings must be null-terminated
 func (thread *phpThread) pinCString(s string) *C.char {
-	return thread.pinString(s+"\x00")
+	return thread.pinString(s + "\x00")
 }

worker.go

@@ -9,7 +9,6 @@ import (
 	"net/http"
 	"path/filepath"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/dunglas/frankenphp/internal/watcher"
@@ -24,6 +23,7 @@ type worker struct {
 	requestChan chan *http.Request
 	threads     []*phpThread
 	threadMutex sync.RWMutex
+	ready       chan struct{}
 }
 
 const maxWorkerErrorBackoff = 1 * time.Second
@@ -32,34 +32,28 @@ const maxWorkerConsecutiveFailures = 6
 
 var (
 	watcherIsEnabled bool
-	workersReadyWG   sync.WaitGroup
 	workerShutdownWG sync.WaitGroup
-	workersAreReady  atomic.Bool
-	workersAreDone   atomic.Bool
 	workersDone      chan interface{}
 	workers          = make(map[string]*worker)
 )
 
 func initWorkers(opt []workerOpt) error {
 	workersDone = make(chan interface{})
-	workersAreReady.Store(false)
-	workersAreDone.Store(false)
 
 	for _, o := range opt {
 		worker, err := newWorker(o)
 		worker.threads = make([]*phpThread, 0, o.num)
 		if err != nil {
 			return err
 		}
-		workersReadyWG.Add(worker.num)
 		for i := 0; i < worker.num; i++ {
 			go worker.startNewWorkerThread()
 		}
+		for i := 0; i < worker.num; i++ {
+			<-worker.ready
+		}
 	}
 
-	workersReadyWG.Wait()
-	workersAreReady.Store(true)
-
 	return nil
 }
 
@@ -80,7 +74,13 @@ func newWorker(o workerOpt) (*worker, error) {
 	}
 
 	o.env["FRANKENPHP_WORKER\x00"] = "1"
-	w := &worker{fileName: absFileName, num: o.num, env: o.env, requestChan: make(chan *http.Request)}
+	w := &worker{
+		fileName:    absFileName,
+		num:         o.num,
+		env:         o.env,
+		requestChan: make(chan *http.Request),
+		ready:       make(chan struct{}),
+	}
 	workers[absFileName] = w
 
 	return w, nil
@@ -95,14 +95,14 @@ func (worker *worker) startNewWorkerThread() {
 	backingOffLock := sync.RWMutex{}
 
 	for {
-
 		// if the worker can stay up longer than backoff*2, it is probably an application error
 		upFunc := sync.Once{}
 		go func() {
 			backingOffLock.RLock()
 			wait := backoff * 2
 			backingOffLock.RUnlock()
 			time.Sleep(wait)
+			worker.ready <- struct{}{}
 			upFunc.Do(func() {
 				backingOffLock.Lock()
 				defer backingOffLock.Unlock()
@@ -146,8 +146,14 @@ func (worker *worker) startNewWorkerThread() {
 		fc := r.Context().Value(contextKey).(*FrankenPHPContext)
 
 		// if we are done, exit the loop that restarts the worker script
-		if workersAreDone.Load() {
-			break
+		select {
+		case _, ok := <-workersDone:
+			if !ok {
+				return
+			}
+			// continue on since the channel is still open
+		default:
+			// continue on since the channel is still open
 		}
 
 		// on exit status 0 we just run the worker script again
@@ -185,12 +191,7 @@ func (worker *worker) startNewWorkerThread() {
 		metrics.StopWorker(worker.fileName, StopReasonCrash)
 	}
 
-	metrics.StopWorker(worker.fileName, StopReasonShutdown)
-
-	// TODO: check if the termination is expected
-	if c := logger.Check(zapcore.DebugLevel, "terminated"); c != nil {
-		c.Write(zap.String("worker", worker.fileName))
-	}
+	// unreachable
 }
 
 func (worker *worker) handleRequest(r *http.Request) {
@@ -211,7 +212,6 @@ func (worker *worker) handleRequest(r *http.Request) {
 }
 
 func stopWorkers() {
-	workersAreDone.Store(true)
 	close(workersDone)
 }
 
@@ -254,15 +254,11 @@ func restartWorkers(workerOpts []workerOpt) {
 
 func assignThreadToWorker(thread *phpThread) {
 	fc := thread.mainRequest.Context().Value(contextKey).(*FrankenPHPContext)
-	metrics.ReadyWorker(fc.scriptFilename)
 	worker, ok := workers[fc.scriptFilename]
 	if !ok {
 		panic("worker not found for script: " + fc.scriptFilename)
 	}
 	thread.worker = worker
-	if !workersAreReady.Load() {
-		workersReadyWG.Done()
-	}
 	thread.requestChan = make(chan *http.Request)
 	worker.threadMutex.Lock()
 	worker.threads = append(worker.threads, thread)
@@ -273,10 +269,23 @@ func assignThreadToWorker(thread *phpThread) {
 func go_frankenphp_worker_handle_request_start(threadIndex C.uintptr_t) C.bool {
 	thread := phpThreads[threadIndex]
 
+	select {
+	case _, ok := <-workersDone:
+		if !ok {
+			// attempted to restart during shutdown
+			return C.bool(false)
+		}
+	default:
+	}
+
 	// we assign a worker to the thread if it doesn't have one already
 	if thread.worker == nil {
 		assignThreadToWorker(thread)
 	}
+	thread.readiedOnce.Do(func() {
+		// inform metrics that the worker is ready
+		metrics.ReadyWorker(thread.worker.fileName)
+	})
 
 	if c := logger.Check(zapcore.DebugLevel, "waiting for request"); c != nil {
 		c.Write(zap.String("worker", thread.worker.fileName))
@@ -343,5 +352,14 @@ func go_frankenphp_finish_request(threadIndex C.uintptr_t, isWorkerRequest bool)
 
 	if isWorkerRequest {
 		thread.Unpin()
+		workers[fc.scriptFilename].threadMutex.Lock()
+		defer workers[fc.scriptFilename].threadMutex.Unlock()
+		for i, t := range workers[fc.scriptFilename].threads {
+			if t == thread {
+				// remove thread from worker threads
+				workers[fc.scriptFilename].threads = append(workers[fc.scriptFilename].threads[:i], workers[fc.scriptFilename].threads[i+1:]...)
+				break
+			}
+		}
 	}
 }