actually allow multiple workers per script filename

Author: henderkes

caddy/caddy.go

@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/davecgh/go-spew/spew"
 	"net/http"
 	"path/filepath"
 	"strconv"
@@ -121,11 +122,13 @@ func (f *FrankenPHPApp) Start() error {
 	// Add workers from FrankenPHPApp configuration
 	for _, w := range f.Workers {
 		opts = append(opts, frankenphp.WithWorkers(w.Name, repl.ReplaceKnown(w.FileName, ""), w.Num, w.Env, w.Watch, w.ModuleID))
+		caddy.Log().Warn(fmt.Sprintf("Starting FrankenPHP with worker: %s", spew.Sdump(w)))
 	}
 
 	// Add workers from FrankenPHPModule configurations
 	for _, w := range sharedState.Workers {
 		opts = append(opts, frankenphp.WithWorkers(w.Name, repl.ReplaceKnown(w.FileName, ""), w.Num, w.Env, w.Watch, w.ModuleID))
+		caddy.Log().Warn(fmt.Sprintf("Starting FrankenPHP with worker: %s", spew.Sdump(w)))
 	}
 
 	frankenphp.Shutdown()
@@ -444,8 +447,8 @@ func (f *FrankenPHPModule) Provision(ctx caddy.Context) error {
 		hash := sha256.Sum256(data)
 		f.ModuleID = binary.LittleEndian.Uint64(hash[:8])
 
-		for _, w := range f.Workers {
-			w.ModuleID = f.ModuleID
+		for i := range f.Workers {
+			f.Workers[i].ModuleID = f.ModuleID
 		}
 		sharedState.Workers = append(sharedState.Workers, f.Workers...)
 		f.logger.Warn(fmt.Sprintf("Initialized FrankenPHP module %d with %d workers", f.ModuleID, len(f.Workers)))

frankenphp.go

@@ -32,6 +32,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"github.com/davecgh/go-spew/spew"
 	"io"
 	"net/http"
 	"os"
@@ -405,11 +406,23 @@ func ServeHTTP(responseWriter http.ResponseWriter, request *http.Request) error
 	}
 
 	// Detect if a worker is available to handle this request
-	if worker, ok := workers[fc.scriptFilename]; ok {
-		// can handle with a global worker, or a module worker from the matching module
-		if worker.moduleID == 0 || worker.moduleID == fc.moduleID {
-			worker.handleRequest(fc)
-			return nil
+	if workersList, ok := workers[fc.scriptFilename]; ok {
+		workersString := ""
+		for _, worker := range workersList {
+			envString := ""
+			for k, v := range worker.env {
+				envString += k + "=" + v + ","
+			}
+			workersString += strconv.FormatUint(worker.moduleID, 10) + ":" + envString + "\n"
+		}
+		fc.logger.Info(fmt.Sprintf("Available workers: %s", workersString))
+		// Look for a worker with matching moduleID or a global worker (moduleID == 0)
+		for _, worker := range workersList {
+			if worker.moduleID == 0 || worker.moduleID == fc.moduleID {
+				fc.logger.Warn(fmt.Sprintf("Handled %d by worker %d with env: %s", fc.moduleID, worker.moduleID, spew.Sdump(worker.env)))
+				worker.handleRequest(fc)
+				return nil
+			}
 		}
 	}
 

phpmainthread_test.go

@@ -179,7 +179,7 @@ func TestFinishBootingAWorkerScript(t *testing.T) {
 
 func getDummyWorker(fileName string) *worker {
 	if workers == nil {
-		workers = make(map[string]*worker)
+		workers = make(map[string][]*worker)
 	}
 	worker, _ := newWorker(workerOpt{
 		fileName: testDataPath + "/" + fileName,
@@ -211,9 +211,9 @@ func allPossibleTransitions(worker1Path string, worker2Path string) []func(*phpT
 				thread.boot()
 			}
 		},
-		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker1Path]) },
+		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker1Path][0]) },
 		convertToInactiveThread,
-		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker2Path]) },
+		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker2Path][0]) },
 		convertToInactiveThread,
 	}
 }

scaling.go

@@ -160,8 +160,19 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext,
 			}
 
 			// if the request has been stalled long enough, scale
-			if worker, ok := workers[fc.scriptFilename]; ok {
-				scaleWorkerThread(worker)
+			if workersList, ok := workers[fc.scriptFilename]; ok {
+				// Look for a worker with matching moduleID or a global worker (moduleID == 0)
+				workerFound := false
+				for _, worker := range workersList {
+					if worker.moduleID == 0 || worker.moduleID == fc.moduleID {
+						scaleWorkerThread(worker)
+						workerFound = true
+						break
+					}
+				}
+				if !workerFound {
+					scaleRegularThread()
+				}
 			} else {
 				scaleRegularThread()
 			}

scaling_test.go

@@ -43,7 +43,7 @@ func TestScaleAWorkerThreadUpAndDown(t *testing.T) {
 	autoScaledThread := phpThreads[2]
 
 	// scale up
-	scaleWorkerThread(workers[workerPath])
+	scaleWorkerThread(workers[workerPath][0])
 	assert.Equal(t, stateReady, autoScaledThread.state.get())
 
 	// on down-scale, the thread will be marked as inactive

worker.go

@@ -4,6 +4,7 @@ package frankenphp
 import "C"
 import (
 	"fmt"
+	"sort"
 	"sync"
 	"time"
 
@@ -24,12 +25,12 @@ type worker struct {
 }
 
 var (
-	workers          map[string]*worker
+	workers          map[string][]*worker
 	watcherIsEnabled bool
 )
 
 func initWorkers(opt []workerOpt) error {
-	workers = make(map[string]*worker, len(opt))
+	workers = make(map[string][]*worker, len(opt))
 	workersReady := sync.WaitGroup{}
 	directoriesToWatch := getDirectoriesToWatch(opt)
 	watcherIsEnabled = len(directoriesToWatch) > 0
@@ -84,7 +85,17 @@ func newWorker(o workerOpt) (*worker, error) {
 		requestChan: make(chan *frankenPHPContext),
 		moduleID:    o.moduleID,
 	}
-	workers[absFileName] = w
+
+	// Check if we already have workers for this filename
+	if _, ok := workers[absFileName]; !ok {
+		workers[absFileName] = make([]*worker, 0)
+	}
+	workers[absFileName] = append(workers[absFileName], w)
+
+	// Sort workers by descending moduleID, this way FrankenPHPApp::ServeHTTP will prefer a module-specific worker over a global one
+	sort.Slice(workers[absFileName], func(i, j int) bool {
+		return workers[absFileName][i].moduleID > workers[absFileName][j].moduleID
+	})
 
 	return w, nil
 }
@@ -97,23 +108,25 @@ func DrainWorkers() {
 func drainWorkerThreads() []*phpThread {
 	ready := sync.WaitGroup{}
 	drainedThreads := make([]*phpThread, 0)
-	for _, worker := range workers {
-		worker.threadMutex.RLock()
-		ready.Add(len(worker.threads))
-		for _, thread := range worker.threads {
-			if !thread.state.requestSafeStateChange(stateRestarting) {
-				// no state change allowed == thread is shutting down
-				// we'll proceed to restart all other threads anyways
-				continue
+	for _, workersList := range workers {
+		for _, worker := range workersList {
+			worker.threadMutex.RLock()
+			ready.Add(len(worker.threads))
+			for _, thread := range worker.threads {
+				if !thread.state.requestSafeStateChange(stateRestarting) {
+					// no state change allowed == thread is shutting down
+					// we'll proceed to restart all other threads anyways
+					continue
+				}
+				close(thread.drainChan)
+				drainedThreads = append(drainedThreads, thread)
+				go func(thread *phpThread) {
+					thread.state.waitFor(stateYielding)
+					ready.Done()
+				}(thread)
 			}
-			close(thread.drainChan)
-			drainedThreads = append(drainedThreads, thread)
-			go func(thread *phpThread) {
-				thread.state.waitFor(stateYielding)
-				ready.Done()
-			}(thread)
+			worker.threadMutex.RUnlock()
 		}
-		worker.threadMutex.RUnlock()
 	}
 	ready.Wait()