feat: custom workers initial support (#1795)

* create a simple thread framework Signed-off-by: Robert Landers <landers.robert@gmail.com> * add tests Signed-off-by: Robert Landers <landers.robert@gmail.com> * fix comment Signed-off-by: Robert Landers <landers.robert@gmail.com> * remove mention of an old function that no longer exists Signed-off-by: Robert Landers <landers.robert@gmail.com> * simplify providing a request Signed-off-by: Robert Landers <landers.robert@gmail.com> * satisfy linter Signed-off-by: Robert Landers <landers.robert@gmail.com> * add error handling and handle shutdowns Signed-off-by: Robert Landers <landers.robert@gmail.com> * add tests Signed-off-by: Robert Landers <landers.robert@gmail.com> * pipes are tied to workers, not threads Signed-off-by: Robert Landers <landers.robert@gmail.com> * fix test Signed-off-by: Robert Landers <landers.robert@gmail.com> * add a way to detect when a request is completed Signed-off-by: Robert Landers <landers.robert@gmail.com> * we never shutdown workers or remove them, so we do not need this Signed-off-by: Robert Landers <landers.robert@gmail.com> * add more comments Signed-off-by: Robert Landers <landers.robert@gmail.com> * Simplify modular threads (#1874) * Simplify * remove unused variable * log thread index * feat: allow passing parameters to the PHP callback and accessing its return value (#1881) * fix formatting Signed-off-by: Robert Landers <landers.robert@gmail.com> * fix test compilation Signed-off-by: Robert Landers <landers.robert@gmail.com> * fix segfaults Signed-off-by: Robert Landers <landers.robert@gmail.com> * Update frankenphp.c Co-authored-by: Kévin Dunglas <kevin@dunglas.fr> --------- Signed-off-by: Robert Landers <landers.robert@gmail.com> Co-authored-by: Kévin Dunglas <kevin@dunglas.fr>
2025-09-26 19:41:13 +08:00 · 2025-09-18 09:21:49 +02:00
parent fe7aa2cae4
commit 52df300f86
7 changed files with 307 additions and 16 deletions
--- a/context.go
+++ b/context.go
@@ -29,6 +29,8 @@ type frankenPHPContext struct {
 	isDone bool

 	responseWriter    http.ResponseWriter
+	handlerParameters any
+	handlerReturn     any

 	done      chan any
 	startedAt time.Time
--- a/frankenphp.c
+++ b/frankenphp.c
@@ -432,10 +432,11 @@ PHP_FUNCTION(frankenphp_handle_request) {
  zend_unset_timeout();
 #endif

-  bool has_request = go_frankenphp_worker_handle_request_start(thread_index);
+  struct go_frankenphp_worker_handle_request_start_return result =
+      go_frankenphp_worker_handle_request_start(thread_index);
  if (frankenphp_worker_request_startup() == FAILURE
      /* Shutting down */
-      || !has_request) {
+      || !result.r0) {
    RETURN_FALSE;
  }

@@ -450,10 +451,15 @@ PHP_FUNCTION(frankenphp_handle_request) {

  /* Call the PHP func passed to frankenphp_handle_request() */
  zval retval = {0};
+  zval *callback_ret = NULL;
+
  fci.size = sizeof fci;
  fci.retval = &retval;
-  if (zend_call_function(&fci, &fcc) == SUCCESS) {
-    zval_ptr_dtor(&retval);
+  fci.params = result.r1;
+  fci.param_count = result.r1 == NULL ? 0 : 1;
+
+  if (zend_call_function(&fci, &fcc) == SUCCESS && Z_TYPE(retval) != IS_UNDEF) {
+    callback_ret = &retval;
  }

  /*
@@ -467,7 +473,13 @@ PHP_FUNCTION(frankenphp_handle_request) {
  }

  frankenphp_worker_request_shutdown();
-  go_frankenphp_finish_worker_request(thread_index);
+  go_frankenphp_finish_worker_request(thread_index, callback_ret);
+  if (result.r1 != NULL) {
+    zval_ptr_dtor(result.r1);
+  }
+  if (callback_ret != NULL) {
+    zval_ptr_dtor(&retval);
+  }

  RETURN_TRUE;
 }
--- a/frankenphp.go
+++ b/frankenphp.go
@@ -214,6 +214,11 @@ func Init(options ...Option) error {

 	registerExtensions()

+	// add registered external workers
+	for _, ew := range externalWorkers {
+		options = append(options, WithWorkers(ew.Name(), ew.FileName(), ew.GetMinThreads(), WithWorkerEnv(ew.Env())))
+	}
+
 	opt := &opt{}
 	for _, o := range options {
 		if err := o(opt); err != nil {
--- a/threadFramework.go
+++ b/threadFramework.go
@@ -0,0 +1,101 @@
+package frankenphp
+
+import (
+	"context"
+	"log/slog"
+	"net/http"
+	"sync"
+)
+
+// EXPERIMENTAL: WorkerExtension allows you to register an external worker where instead of calling frankenphp handlers on
+// frankenphp_handle_request(), the ProvideRequest method is called. You are responsible for providing a standard
+// http.Request that will be conferred to the underlying worker script.
+//
+// A worker script with the provided Name and FileName will be registered, along with the provided
+// configuration. You can also provide any environment variables that you want through Env. GetMinThreads allows you to
+// reserve a minimum number of threads from the frankenphp thread pool. This number must be positive.
+// These methods are only called once at startup, so register them in an init() function.
+//
+// When a thread is activated and nearly ready, ThreadActivatedNotification will be called with an opaque threadId;
+// this is a time for setting up any per-thread resources. When a thread is about to be returned to the thread pool,
+// you will receive a call to ThreadDrainNotification that will inform you of the threadId.
+// After the thread is returned to the thread pool, ThreadDeactivatedNotification will be called.
+//
+// Once you have at least one thread activated, you will receive calls to ProvideRequest where you should respond with
+// a request. FrankenPHP will automatically pipe these requests to the worker script and handle the response.
+// The piping process is designed to run indefinitely and will be gracefully shut down when FrankenPHP shuts down.
+//
+// Note: External workers receive the lowest priority when determining thread allocations. If GetMinThreads cannot be
+// allocated, then frankenphp will panic and provide this information to the user (who will need to allocate more
+// total threads). Don't be greedy.
+type WorkerExtension interface {
+	Name() string
+	FileName() string
+	Env() PreparedEnv
+	GetMinThreads() int
+	ThreadActivatedNotification(threadId int)
+	ThreadDrainNotification(threadId int)
+	ThreadDeactivatedNotification(threadId int)
+	ProvideRequest() *WorkerRequest[any, any]
+}
+
+// EXPERIMENTAL
+type WorkerRequest[P any, R any] struct {
+	// The request for your worker script to handle
+	Request *http.Request
+	// Response is a response writer that provides the output of the provided request, it must not be nil to access the request body
+	Response http.ResponseWriter
+	// CallbackParameters is an optional field that will be converted in PHP types and passed as parameter to the PHP callback
+	CallbackParameters P
+	// AfterFunc is an optional function that will be called after the request is processed with the original value, the return of the PHP callback, converted in Go types, is passed as parameter
+	AfterFunc func(callbackReturn R)
+}
+
+var externalWorkers = make(map[string]WorkerExtension)
+var externalWorkerMutex sync.Mutex
+
+// EXPERIMENTAL
+func RegisterExternalWorker(worker WorkerExtension) {
+	externalWorkerMutex.Lock()
+	defer externalWorkerMutex.Unlock()
+
+	externalWorkers[worker.Name()] = worker
+}
+
+// startExternalWorkerPipe creates a pipe from an external worker to the main worker.
+func startExternalWorkerPipe(w *worker, externalWorker WorkerExtension, thread *phpThread) {
+	for {
+		rq := externalWorker.ProvideRequest()
+
+		if rq == nil || rq.Request == nil {
+			logger.LogAttrs(context.Background(), slog.LevelWarn, "external worker provided nil request", slog.String("worker", w.name), slog.Int("thread", thread.threadIndex))
+			continue
+		}
+
+		r := rq.Request
+		fr, err := NewRequestWithContext(r, WithOriginalRequest(r), WithWorkerName(w.name))
+		if err != nil {
+			logger.LogAttrs(context.Background(), slog.LevelError, "error creating request for external worker", slog.String("worker", w.name), slog.Int("thread", thread.threadIndex), slog.Any("error", err))
+			continue
+		}
+
+		if fc, ok := fromContext(fr.Context()); ok {
+			fc.responseWriter = rq.Response
+			fc.handlerParameters = rq.CallbackParameters
+
+			// Queue the request and wait for completion if Done channel was provided
+			logger.LogAttrs(context.Background(), slog.LevelInfo, "queue the external worker request", slog.String("worker", w.name), slog.Int("thread", thread.threadIndex))
+
+			w.requestChan <- fc
+			if rq.AfterFunc != nil {
+				go func() {
+					<-fc.done
+
+					if rq.AfterFunc != nil {
+						rq.AfterFunc(fc.handlerReturn)
+					}
+				}()
+			}
+		}
+	}
+}
--- a/threadFramework_test.go
+++ b/threadFramework_test.go
@@ -0,0 +1,136 @@
+package frankenphp
+
+import (
+	"io"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// mockWorkerExtension implements the WorkerExtension interface
+type mockWorkerExtension struct {
+	name             string
+	fileName         string
+	env              PreparedEnv
+	minThreads       int
+	requestChan      chan *WorkerRequest[any, any]
+	activatedCount   int
+	drainCount       int
+	deactivatedCount int
+	mu               sync.Mutex
+}
+
+func newMockWorkerExtension(name, fileName string, minThreads int) *mockWorkerExtension {
+	return &mockWorkerExtension{
+		name:        name,
+		fileName:    fileName,
+		env:         make(PreparedEnv),
+		minThreads:  minThreads,
+		requestChan: make(chan *WorkerRequest[any, any], 10), // Buffer to avoid blocking
+	}
+}
+
+func (m *mockWorkerExtension) Name() string {
+	return m.name
+}
+
+func (m *mockWorkerExtension) FileName() string {
+	return m.fileName
+}
+
+func (m *mockWorkerExtension) Env() PreparedEnv {
+	return m.env
+}
+
+func (m *mockWorkerExtension) GetMinThreads() int {
+	return m.minThreads
+}
+
+func (m *mockWorkerExtension) ThreadActivatedNotification(threadId int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.activatedCount++
+}
+
+func (m *mockWorkerExtension) ThreadDrainNotification(threadId int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.drainCount++
+}
+
+func (m *mockWorkerExtension) ThreadDeactivatedNotification(threadId int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.deactivatedCount++
+}
+
+func (m *mockWorkerExtension) ProvideRequest() *WorkerRequest[any, any] {
+	return <-m.requestChan
+}
+
+func (m *mockWorkerExtension) InjectRequest(r *WorkerRequest[any, any]) {
+	m.requestChan <- r
+}
+
+func (m *mockWorkerExtension) GetActivatedCount() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.activatedCount
+}
+
+func TestWorkerExtension(t *testing.T) {
+	// Create a mock extension
+	mockExt := newMockWorkerExtension("mockWorker", "testdata/worker.php", 1)
+
+	// Register the mock extension
+	RegisterExternalWorker(mockExt)
+
+	// Clean up external workers after test to avoid interfering with other tests
+	defer func() {
+		delete(externalWorkers, mockExt.Name())
+	}()
+
+	// Initialize FrankenPHP with a worker that has a different name than our extension
+	err := Init()
+	require.NoError(t, err)
+	defer Shutdown()
+
+	// Wait a bit for the worker to be ready
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify that the extension's thread was activated
+	assert.GreaterOrEqual(t, mockExt.GetActivatedCount(), 1, "Thread should have been activated")
+
+	// Create a test request
+	req := httptest.NewRequest("GET", "http://example.com/test/?foo=bar", nil)
+	req.Header.Set("X-Test-Header", "test-value")
+
+	w := httptest.NewRecorder()
+
+	// Create a channel to signal when the request is done
+	done := make(chan struct{})
+
+	// Inject the request into the worker through the extension
+	mockExt.InjectRequest(&WorkerRequest[any, any]{
+		Request:  req,
+		Response: w,
+		AfterFunc: func(callbackReturn any) {
+			close(done)
+		},
+	})
+
+	// Wait for the request to be fully processed
+	<-done
+
+	// Check the response - now safe from race conditions
+	resp := w.Result()
+	body, _ := io.ReadAll(resp.Body)
+
+	// The worker.php script should output information about the request
+	// We're just checking that we got a response, not the specific content
+	assert.NotEmpty(t, body, "Response body should not be empty")
+}
--- a/threadworker.go
+++ b/threadworker.go
@@ -7,6 +7,7 @@ import (
 	"log/slog"
 	"path/filepath"
 	"time"
+	"unsafe"
 )

 // representation of a thread assigned to a worker script
@@ -19,10 +20,13 @@ type workerThread struct {
 	dummyContext    *frankenPHPContext
 	workerContext   *frankenPHPContext
 	backoff         *exponentialBackoff
+	externalWorker  WorkerExtension
 	isBootingScript bool // true if the worker has not reached frankenphp_handle_request yet
 }

 func convertToWorkerThread(thread *phpThread, worker *worker) {
+	externalWorker := externalWorkers[worker.name]
+
 	thread.setHandler(&workerThread{
 		state:  thread.state,
 		thread: thread,
@@ -32,6 +36,7 @@ func convertToWorkerThread(thread *phpThread, worker *worker) {
 			minBackoff:             100 * time.Millisecond,
 			maxConsecutiveFailures: worker.maxConsecutiveFailures,
 		},
+		externalWorker: externalWorker,
 	})
 	worker.attachThread(thread)
 }
@@ -40,16 +45,28 @@ func convertToWorkerThread(thread *phpThread, worker *worker) {
 func (handler *workerThread) beforeScriptExecution() string {
 	switch handler.state.get() {
 	case stateTransitionRequested:
+		if handler.externalWorker != nil {
+			handler.externalWorker.ThreadDeactivatedNotification(handler.thread.threadIndex)
+		}
 		handler.worker.detachThread(handler.thread)
 		return handler.thread.transitionToNewHandler()
 	case stateRestarting:
+		if handler.externalWorker != nil {
+			handler.externalWorker.ThreadDrainNotification(handler.thread.threadIndex)
+		}
 		handler.state.set(stateYielding)
 		handler.state.waitFor(stateReady, stateShuttingDown)
 		return handler.beforeScriptExecution()
 	case stateReady, stateTransitionComplete:
+		if handler.externalWorker != nil {
+			handler.externalWorker.ThreadActivatedNotification(handler.thread.threadIndex)
+		}
 		setupWorkerScript(handler, handler.worker)
 		return handler.worker.fileName
 	case stateShuttingDown:
+		if handler.externalWorker != nil {
+			handler.externalWorker.ThreadDeactivatedNotification(handler.thread.threadIndex)
+		}
 		handler.worker.detachThread(handler.thread)
 		// signal to stop
 		return ""
@@ -143,7 +160,7 @@ func tearDownWorkerScript(handler *workerThread, exitStatus int) {
 }

 // waitForWorkerRequest is called during frankenphp_handle_request in the php worker script.
-func (handler *workerThread) waitForWorkerRequest() bool {
+func (handler *workerThread) waitForWorkerRequest() (bool, any) {
 	// unpin any memory left over from previous requests
 	handler.thread.Unpin()

@@ -179,7 +196,7 @@ func (handler *workerThread) waitForWorkerRequest() bool {
 			C.frankenphp_reset_opcache()
 		}

-		return false
+		return false, nil
 	case fc = <-handler.thread.requestChan:
 	case fc = <-handler.worker.requestChan:
 	}
@@ -189,23 +206,35 @@ func (handler *workerThread) waitForWorkerRequest() bool {

 	logger.LogAttrs(ctx, slog.LevelDebug, "request handling started", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex), slog.String("url", fc.request.RequestURI))

-	return true
+	return true, fc.handlerParameters
 }

 // go_frankenphp_worker_handle_request_start is called at the start of every php request served.
 //
 //export go_frankenphp_worker_handle_request_start
-func go_frankenphp_worker_handle_request_start(threadIndex C.uintptr_t) C.bool {
+func go_frankenphp_worker_handle_request_start(threadIndex C.uintptr_t) (C.bool, unsafe.Pointer) {
 	handler := phpThreads[threadIndex].handler.(*workerThread)
-	return C.bool(handler.waitForWorkerRequest())
+	hasRequest, parameters := handler.waitForWorkerRequest()
+
+	if parameters != nil {
+		p := PHPValue(parameters)
+		handler.thread.Pin(p)
+
+		return C.bool(hasRequest), p
+	}
+
+	return C.bool(hasRequest), nil
 }

 // go_frankenphp_finish_worker_request is called at the end of every php request served.
 //
 //export go_frankenphp_finish_worker_request
-func go_frankenphp_finish_worker_request(threadIndex C.uintptr_t) {
+func go_frankenphp_finish_worker_request(threadIndex C.uintptr_t, retval *C.zval) {
 	thread := phpThreads[threadIndex]
 	fc := thread.getRequestContext()
+	if retval != nil {
+		fc.handlerReturn = GoValue(unsafe.Pointer(retval))
+	}

 	fc.closeContext()
 	thread.handler.(*workerThread).workerContext = nil
--- a/worker.go
+++ b/worker.go
@@ -44,13 +44,19 @@ func initWorkers(opt []workerOpt) error {
 		workers = append(workers, w)
 	}

-	for _, worker := range workers {
-		workersReady.Add(worker.num)
-		for i := 0; i < worker.num; i++ {
+	for _, w := range workers {
+		workersReady.Add(w.num)
+		for i := 0; i < w.num; i++ {
 			thread := getInactivePHPThread()
-			convertToWorkerThread(thread, worker)
+			convertToWorkerThread(thread, w)
 			go func() {
 				thread.state.waitFor(stateReady)
+
+				// create a pipe from the external worker to the main worker
+				// note: this is locked to the initial thread size the external worker requested
+				if workerThread, ok := thread.handler.(*workerThread); ok && workerThread.externalWorker != nil {
+					go startExternalWorkerPipe(w, workerThread.externalWorker, thread)
+				}
 				workersReady.Done()
 			}()
 		}