Adds state tests.

state_test.go

@@ -0,0 +1,48 @@
+package frankenphp
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test2GoroutinesYieldToEachOtherViaStates(t *testing.T) {
+	threadState := &threadState{currentState: stateBooting}
+
+	go func() {
+		threadState.waitFor(stateInactive)
+		assert.True(t, threadState.is(stateInactive))
+		threadState.set(stateReady)
+	}()
+
+	threadState.set(stateInactive)
+	threadState.waitFor(stateReady)
+	assert.True(t, threadState.is(stateReady))
+}
+
+func TestStateShouldHaveCorrectAmountOfSubscribers(t *testing.T) {
+	threadState := &threadState{currentState: stateBooting}
+
+	// 3 subscribers waiting for different states
+	go threadState.waitFor(stateInactive)
+	go threadState.waitFor(stateInactive, stateShuttingDown)
+	go threadState.waitFor(stateShuttingDown)
+
+	time.Sleep(1 * time.Millisecond)
+	assertNumberOfSubscribers(t, threadState, 3)
+
+	threadState.set(stateInactive)
+	time.Sleep(1 * time.Millisecond)
+	assertNumberOfSubscribers(t, threadState, 1)
+
+	threadState.set(stateShuttingDown)
+	time.Sleep(1 * time.Millisecond)
+	assertNumberOfSubscribers(t, threadState, 0)
+}
+
+func assertNumberOfSubscribers(t *testing.T, threadState *threadState, expected int) {
+	threadState.mu.RLock()
+	assert.Len(t, threadState.subscribers, expected)
+	threadState.mu.RUnlock()
+}

thread-worker.go

@@ -35,7 +35,7 @@ func convertToWorkerThread(thread *phpThread, worker *worker) {
 			maxConsecutiveFailures: 6,
 		},
 	})
-	worker.addThread(thread)
+	worker.attachThread(thread)
 	if worker.fileName == "" {
 		panic("worker script is empty")
 	}
@@ -54,7 +54,7 @@ func (handler *workerThread) beforeScriptExecution() string {
 	switch handler.state.get() {
 	case stateTransitionRequested:
 		thread := handler.thread
-		handler.worker.removeThread(handler.thread)
+		handler.worker.detachThread(handler.thread)
 		thread.state.set(stateTransitionInProgress)
 		thread.state.waitFor(stateTransitionComplete, stateShuttingDown)
 

worker.go

@@ -39,7 +39,8 @@ func initWorkers(opt []workerOpt) error {
 			return err
 		}
 		for i := 0; i < worker.num; i++ {
-			worker.startNewThread()
+			thread := getInactivePHPThread()
+			convertToWorkerThread(thread, worker)
 		}
 	}
 
@@ -112,9 +113,21 @@ func getDirectoriesToWatch(workerOpts []workerOpt) []string {
 	return directoriesToWatch
 }
 
-func (worker *worker) startNewThread() {
-	thread := getInactivePHPThread()
-	convertToWorkerThread(thread, worker)
+func (worker *worker) attachThread(thread *phpThread) {
+	worker.threadMutex.Lock()
+	worker.threads = append(worker.threads, thread)
+	worker.threadMutex.Unlock()
+}
+
+func (worker *worker) detachThread(thread *phpThread) {
+	worker.threadMutex.Lock()
+	for i, t := range worker.threads {
+		if t == thread {
+			worker.threads = append(worker.threads[:i], worker.threads[i+1:]...)
+			break
+		}
+	}
+	worker.threadMutex.Unlock()
 }
 
 func (worker *worker) handleRequest(r *http.Request, fc *FrankenPHPContext) {
@@ -140,20 +153,3 @@ func (worker *worker) handleRequest(r *http.Request, fc *FrankenPHPContext) {
 	<-fc.done
 	metrics.StopWorkerRequest(worker.fileName, time.Since(fc.startedAt))
 }
-
-func (worker *worker) addThread(thread *phpThread) {
-	worker.threadMutex.Lock()
-	worker.threads = append(worker.threads, thread)
-	worker.threadMutex.Unlock()
-}
-
-func (worker *worker) removeThread(thread *phpThread) {
-	worker.threadMutex.Lock()
-	for i, t := range worker.threads {
-		if t == thread {
-			worker.threads = append(worker.threads[:i], worker.threads[i+1:]...)
-			break
-		}
-	}
-	worker.threadMutex.Unlock()
-}