suggestion: external worker api

frankenphp.go

@@ -216,7 +216,7 @@ func Init(options ...Option) error {
 
 	// add registered external workers
 	for _, ew := range extensionWorkers {
-		options = append(options, WithWorkers(ew.Name(), ew.FileName(), ew.MinThreads(), WithWorkerEnv(ew.Env())))
+		options = append(options, WithWorkers(ew.name, ew.fileName, ew.num, ew.options...))
 	}
 
 	opt := &opt{}

options.go

@@ -35,6 +35,9 @@ type workerOpt struct {
 	env                    PreparedEnv
 	watch                  []string
 	maxConsecutiveFailures int
+	onReady                func(int)
+	onShutdown             func(int)
+	onServerShutdown       func(int)
 }
 
 // WithNumThreads configures the number of PHP threads to start.
@@ -116,6 +119,30 @@ func WithWorkerMaxFailures(maxFailures int) WorkerOption {
 	}
 }
 
+func WithWorkerOnReady(f func(int)) WorkerOption {
+	return func(w *workerOpt) error {
+		w.onReady = f
+
+		return nil
+	}
+}
+
+func WithWorkerOnShutdown(f func(int)) WorkerOption {
+	return func(w *workerOpt) error {
+		w.onShutdown = f
+
+		return nil
+	}
+}
+
+func WithWorkerOnServerShutdown(f func(int)) WorkerOption {
+	return func(w *workerOpt) error {
+		w.onServerShutdown = f
+
+		return nil
+	}
+}
+
 // WithLogger configures the global logger to use.
 func WithLogger(l *slog.Logger) Option {
 	return func(o *opt) error {

testdata/message-worker.php

@@ -0,0 +1,8 @@
+<?php
+
+while (frankenphp_handle_request(function ($message) {
+    echo $message;
+    return "received message: $message";
+})) {
+    // continue handling requests
+}

threadworker.go

@@ -20,13 +20,10 @@ type workerThread struct {
 	dummyContext    *frankenPHPContext
 	workerContext   *frankenPHPContext
 	backoff         *exponentialBackoff
-	externalWorker  Worker
 	isBootingScript bool // true if the worker has not reached frankenphp_handle_request yet
 }
 
 func convertToWorkerThread(thread *phpThread, worker *worker) {
-	externalWorker := extensionWorkers[worker.name]
-
 	thread.setHandler(&workerThread{
 		state:  thread.state,
 		thread: thread,
@@ -36,7 +33,6 @@ func convertToWorkerThread(thread *phpThread, worker *worker) {
 			minBackoff:             100 * time.Millisecond,
 			maxConsecutiveFailures: worker.maxConsecutiveFailures,
 		},
-		externalWorker: externalWorker,
 	})
 	worker.attachThread(thread)
 }
@@ -45,27 +41,27 @@ func convertToWorkerThread(thread *phpThread, worker *worker) {
 func (handler *workerThread) beforeScriptExecution() string {
 	switch handler.state.get() {
 	case stateTransitionRequested:
-		if handler.externalWorker != nil {
-			handler.externalWorker.OnServerShutdown(handler.thread.threadIndex)
+		if handler.worker.onShutdown != nil {
+			handler.worker.onShutdown(handler.thread.threadIndex)
 		}
 		handler.worker.detachThread(handler.thread)
 		return handler.thread.transitionToNewHandler()
 	case stateRestarting:
-		if handler.externalWorker != nil {
-			handler.externalWorker.OnShutdown(handler.thread.threadIndex)
+		if handler.worker.onShutdown != nil {
+			handler.worker.onShutdown(handler.thread.threadIndex)
 		}
 		handler.state.set(stateYielding)
 		handler.state.waitFor(stateReady, stateShuttingDown)
 		return handler.beforeScriptExecution()
 	case stateReady, stateTransitionComplete:
-		if handler.externalWorker != nil {
-			handler.externalWorker.OnReady(handler.thread.threadIndex)
+		if handler.worker.onReady != nil {
+			handler.worker.onReady(handler.thread.threadIndex)
 		}
 		setupWorkerScript(handler, handler.worker)
 		return handler.worker.fileName
 	case stateShuttingDown:
-		if handler.externalWorker != nil {
-			handler.externalWorker.OnServerShutdown(handler.thread.threadIndex)
+		if handler.worker.onServerShutdown != nil {
+			handler.worker.onServerShutdown(handler.thread.threadIndex)
 		}
 		handler.worker.detachThread(handler.thread)
 		// signal to stop

worker.go

@@ -23,6 +23,9 @@ type worker struct {
 	threadMutex            sync.RWMutex
 	allowPathMatching      bool
 	maxConsecutiveFailures int
+	onReady                func(int)
+	onShutdown             func(int)
+	onServerShutdown       func(int)
 }
 
 var (
@@ -51,12 +54,6 @@ func initWorkers(opt []workerOpt) error {
 			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 startWorker(w, workerThread.externalWorker, thread)
-				}
 				workersReady.Done()
 			}()
 		}
@@ -131,6 +128,9 @@ func newWorker(o workerOpt) (*worker, error) {
 		threads:                make([]*phpThread, 0, o.num),
 		allowPathMatching:      allowPathMatching,
 		maxConsecutiveFailures: o.maxConsecutiveFailures,
+		onReady:                o.onReady,
+		onShutdown:             o.onShutdown,
+		onServerShutdown:       o.onServerShutdown,
 	}
 
 	return w, nil

workerextension.go

@@ -1,11 +1,8 @@
 package frankenphp
 
 import (
-	"context"
-	"log/slog"
+	"errors"
 	"net/http"
-	"sync"
-	"sync/atomic"
 )
 
 // EXPERIMENTAL: Worker allows you to register a worker where, instead of calling FrankenPHP handlers on
@@ -27,154 +24,73 @@ import (
 // Extension workers receive the lowest priority when determining thread allocations. If MinThreads 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 Worker interface {
-	// Name returns the worker name
-	Name() string
-	// FileName returns the PHP script filename
-	FileName() string
-	// Env returns the environment variables available in the worker script.
-	Env() PreparedEnv
-	// MinThreads returns the minimum number of threads to reserve from the FrankenPHP thread pool.
-	// This number must be positive.
-	MinThreads() int
-	// OnReady is called when the worker is assigned to a thread and receives an opaque thread ID as parameter.
-	// This is a time for setting up any per-thread resources.
-	OnReady(threadId int)
-	// OnShutdown is called when the worker is shutting down and receives an opaque thread ID as parameter.
-	// This is a time for cleaning up any per-thread resources.
-	OnShutdown(threadId int)
-	// OnServerShutdown is called when FrankenPHP is shutting down.
-	OnServerShutdown(threadId int)
-	// GetRequest is called once at least one thread is ready.
-	// The returned request will be passed to the worker script.
-	GetRequest() *WorkerRequest
-	// SendRequest sends a request to the worker script. The callback function of frankenphp_handle_request() will be called.
-	SendRequest(r *WorkerRequest)
-}
-
-// EXPERIMENTAL: WorkerRequest represents a request to pass to a worker script.
-type WorkerRequest struct {
-	// Request is an optional HTTP request for your worker script to handle
-	Request *http.Request
-	// Response is an optional 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 or left as-is if it's an unsafe.Pointer and passed as parameter to the PHP callback
-	CallbackParameters any
-	// 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 any)
+type Worker struct {
+	name     string
+	fileName string
+	num      int
+	options  []WorkerOption
 }
 
 var extensionWorkers = make(map[string]Worker)
-var extensionWorkersMutex sync.Mutex
 
 // EXPERIMENTAL: RegisterWorker registers a custom worker script.
 func RegisterWorker(worker Worker) {
-	extensionWorkersMutex.Lock()
-	defer extensionWorkersMutex.Unlock()
-
-	extensionWorkers[worker.Name()] = worker
+	extensionWorkers[worker.name] = worker
 }
 
-// startWorker creates a pipe from a worker to the main worker.
-func startWorker(w *worker, extensionWorker Worker, thread *phpThread) {
-	for {
-		rq := extensionWorker.GetRequest()
-
-		var fc *frankenPHPContext
-		if rq.Request == nil {
-			fc = newFrankenPHPContext()
-			fc.logger = logger
-		} else {
-			fr, err := NewRequestWithContext(rq.Request, WithOriginalRequest(rq.Request))
-			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
-			}
-
-			var ok bool
-			if fc, ok = fromContext(fr.Context()); !ok {
-				continue
-			}
-		}
-
-		fc.worker = w
-
-		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)
-				}
-			}()
-		}
-	}
-}
+// EXPERIMENTAL: SendRequest sends an HTTP request to the worker and writes the response to the provided ResponseWriter.
+func (w Worker) SendRequest(rw http.ResponseWriter, r *http.Request) error {
+	worker := getWorkerByName(w.name)
 
-// EXPERIMENTAL: NewWorker creates a Worker instance to embed in a custom struct implementing the Worker interface.
-// The returned instance may be sufficient on its own for simple use cases.
-func NewWorker(name, fileName string, minThreads int, env PreparedEnv) Worker {
-	return &defaultWorker{
-		name:           name,
-		fileName:       fileName,
-		env:            env,
-		minThreads:     minThreads,
-		requestChan:    make(chan *WorkerRequest),
-		activatedCount: atomic.Int32{},
-		drainCount:     atomic.Int32{},
+	if worker == nil {
+		return errors.New("worker not found: " + w.name)
 	}
-}
 
-type defaultWorker struct {
-	name           string
-	fileName       string
-	env            PreparedEnv
-	minThreads     int
-	requestChan    chan *WorkerRequest
-	activatedCount atomic.Int32
-	drainCount     atomic.Int32
-}
+	fr, err := NewRequestWithContext(
+		r,
+		WithOriginalRequest(r),
+		WithWorkerName(w.name),
+	)
 
-func (w *defaultWorker) Name() string {
-	return w.name
-}
+	if err != nil {
+		return err
+	}
 
-func (w *defaultWorker) FileName() string {
-	return w.fileName
-}
+	err = ServeHTTP(rw, fr)
 
-func (w *defaultWorker) Env() PreparedEnv {
-	return w.env
-}
+	if err != nil {
+		return err
+	}
 
-func (w *defaultWorker) MinThreads() int {
-	return w.minThreads
+	return nil
 }
 
-func (w *defaultWorker) OnReady(_ int) {
-	w.activatedCount.Add(1)
-}
+// EXPERIMENTAL: SendMessage sends a message to the worker and waits for a response.
+func (w Worker) SendMessage(message any, rw http.ResponseWriter) (any, error) {
+	internalWorker := getWorkerByName(w.name)
 
-func (w *defaultWorker) OnShutdown(_ int) {
-	w.drainCount.Add(1)
-}
+	if internalWorker == nil {
+		return nil, errors.New("worker not found: " + w.name)
+	}
 
-func (w *defaultWorker) OnServerShutdown(_ int) {
-	w.drainCount.Add(-1)
-	w.activatedCount.Add(-1)
-}
+	fc := newFrankenPHPContext()
+	fc.logger = logger
+	fc.worker = internalWorker
+	fc.responseWriter = rw
+	fc.handlerParameters = message
+
+	internalWorker.handleRequest(fc)
 
-func (w *defaultWorker) GetRequest() *WorkerRequest {
-	return <-w.requestChan
+	return fc.handlerReturn, nil
 }
 
-func (w *defaultWorker) SendRequest(r *WorkerRequest) {
-	w.requestChan <- r
+// EXPERIMENTAL: NewWorker creates a Worker instance to embed in a custom struct implementing the Worker interface.
+// The returned instance may be sufficient on its own for simple use cases.
+func NewWorker(name string, fileName string, num int, options ...WorkerOption) Worker {
+	return Worker{
+		name:     name,
+		fileName: fileName,
+		num:      num,
+		options:  options,
+	}
 }