feat: task workers

caddy/app.go

@@ -33,6 +33,8 @@ type FrankenPHPApp struct {
 	MaxThreads int `json:"max_threads,omitempty"`
 	// Workers configures the worker scripts to start.
 	Workers []workerConfig `json:"workers,omitempty"`
+	// TaskWorkers configures the task worker scripts to start.
+	TaskWorkers []workerConfig `json:"task_workers,omitempty"`
 	// Overwrites the default php ini configuration
 	PhpIni map[string]string `json:"php_ini,omitempty"`
 	// The maximum amount of time a request may be stalled waiting for a thread
@@ -127,6 +129,15 @@ func (f *FrankenPHPApp) Start() error {
 
 		opts = append(opts, frankenphp.WithWorkers(w.Name, repl.ReplaceKnown(w.FileName, ""), w.Num, workerOpts...))
 	}
+	for _, tw := range f.TaskWorkers {
+		workerOpts := []frankenphp.WorkerOption{
+			frankenphp.WithWorkerEnv(tw.Env),
+			frankenphp.WithWorkerWatchMode(tw.Watch),
+			frankenphp.AsTaskWorker(true, 0), // TODO: maxQueueLen configurable here?
+		}
+
+		opts = append(opts, frankenphp.WithWorkers(tw.Name, repl.ReplaceKnown(tw.FileName, ""), tw.Num, workerOpts...))
+	}
 
 	frankenphp.Shutdown()
 	if err := frankenphp.Init(opts...); err != nil {
@@ -234,6 +245,13 @@ func (f *FrankenPHPApp) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 					}
 				}
 
+			case "task_worker":
+				twc, err := parseWorkerConfig(d)
+				if err != nil {
+					return err
+				}
+				f.TaskWorkers = append(f.TaskWorkers, twc)
+
 			case "worker":
 				wc, err := parseWorkerConfig(d)
 				if err != nil {

caddy/caddy_test.go

@@ -1423,3 +1423,31 @@ func TestWorkerMatchDirectiveWithoutFileServer(t *testing.T) {
 	// the request should completely fall through the php_server module
 	tester.AssertGetResponse("http://localhost:"+testPort+"/static.txt", http.StatusNotFound, "Request falls through")
 }
+
+func TestServerWithTaskWorker(t *testing.T) {
+	tester := caddytest.NewTester(t)
+	taskWorker, err := fastabs.FastAbs("../testdata/tasks/task-worker.php")
+	require.NoError(t, err)
+	tester.InitServer(`
+		{
+			skip_install_trust
+			admin localhost:2999
+
+			frankenphp {
+				num_threads 2
+				task_worker `+taskWorker+` {
+					num 1
+				}
+			}
+		}
+		`, "caddyfile")
+
+	debugState := getDebugState(t, tester)
+	require.Len(t, debugState.ThreadDebugStates, 2, "there should be 3 threads")
+	require.Equal(
+		t,
+		debugState.ThreadDebugStates[1].Name,
+		"Task Worker PHP Thread - "+taskWorker,
+		"the second spawned thread should be the task worker",
+	)
+}

cgi.go

@@ -277,13 +277,13 @@ func splitPos(path string, splitPath []string) int {
 // See: https://github.com/php/php-src/blob/345e04b619c3bc11ea17ee02cdecad6ae8ce5891/main/SAPI.h#L72
 //
 //export go_update_request_info
-func go_update_request_info(threadIndex C.uintptr_t, info *C.sapi_request_info) C.bool {
+func go_update_request_info(threadIndex C.uintptr_t, info *C.sapi_request_info) {
 	thread := phpThreads[threadIndex]
 	fc := thread.getRequestContext()
 	request := fc.request
 
 	if request == nil {
-		return C.bool(fc.worker != nil)
+		return
 	}
 
 	authUser, authPassword, ok := request.BasicAuth()
@@ -311,8 +311,6 @@ func go_update_request_info(threadIndex C.uintptr_t, info *C.sapi_request_info)
 	info.request_uri = thread.pinCString(request.URL.RequestURI())
 
 	info.proto_num = C.int(request.ProtoMajor*1000 + request.ProtoMinor)
-
-	return C.bool(fc.worker != nil)
 }
 
 // SanitizedPathJoin performs filepath.Join(root, reqPath) that

frankenphp.c

@@ -73,6 +73,7 @@ frankenphp_config frankenphp_get_config() {
 bool should_filter_var = 0;
 __thread uintptr_t thread_index;
 __thread bool is_worker_thread = false;
+__thread bool is_task_worker_thread = false;
 __thread zval *os_environment = NULL;
 
 static void frankenphp_update_request_context() {
@@ -82,7 +83,12 @@ static void frankenphp_update_request_context() {
   /* status It is not reset by zend engine, set it to 200. */
   SG(sapi_headers).http_response_code = 200;
 
-  is_worker_thread = go_update_request_info(thread_index, &SG(request_info));
+  go_update_request_info(thread_index, &SG(request_info));
+}
+
+void frankenphp_update_thread_context(bool is_worker, bool is_task_worker) {
+  is_worker_thread = is_worker;
+  is_task_worker_thread = is_task_worker;
 }
 
 static void frankenphp_free_request_context() {
@@ -411,6 +417,49 @@ PHP_FUNCTION(frankenphp_response_headers) /* {{{ */
 }
 /* }}} */
 
+/* Handle a message in task worker mode */
+static bool frankenphp_handle_message(zend_fcall_info fci,
+                                      zend_fcall_info_cache fcc) {
+  zval *arg = go_frankenphp_worker_handle_task(thread_index);
+  if (arg == NULL) {
+    return false;
+  }
+
+  /* Call the PHP func passed to frankenphp_handle_request() */
+  zval retval = {0};
+  fci.size = sizeof fci;
+  fci.retval = &retval;
+  fci.params = arg;
+  fci.param_count = 1;
+  zend_bool status = zend_call_function(&fci, &fcc) == SUCCESS;
+
+  if (!status || Z_TYPE(retval) == IS_UNDEF) {
+    go_frankenphp_finish_task(thread_index, NULL);
+    zval_ptr_dtor(arg);
+  } else {
+    go_frankenphp_finish_task(thread_index, &retval);
+  }
+
+  zval_ptr_dtor(&retval);
+
+  /*
+   * If an exception occurred, print the message to the client before
+   * exiting
+   */
+  if (EG(exception) && !zend_is_unwind_exit(EG(exception)) &&
+      !zend_is_graceful_exit(EG(exception))) {
+    zend_exception_error(EG(exception), E_ERROR);
+    zend_bailout();
+  }
+
+  zend_try { php_output_end_all(); }
+  zend_end_try();
+
+  zval_ptr_dtor(arg);
+
+  return true;
+}
+
 PHP_FUNCTION(frankenphp_handle_request) {
   zend_fcall_info fci;
   zend_fcall_info_cache fcc;
@@ -420,6 +469,13 @@ PHP_FUNCTION(frankenphp_handle_request) {
   ZEND_PARSE_PARAMETERS_END();
 
   if (!is_worker_thread) {
+
+    /* thread is a task worker
+     * handle the message and do not reset globals */
+    if (is_task_worker_thread) {
+      bool keep_running = frankenphp_handle_message(fci, fcc);
+      RETURN_BOOL(keep_running);
+    }
     /* not a worker, throw an error */
     zend_throw_exception(
         spl_ce_RuntimeException,
@@ -484,6 +540,25 @@ PHP_FUNCTION(frankenphp_handle_request) {
   RETURN_TRUE;
 }
 
+PHP_FUNCTION(frankenphp_send_request) {
+  zval *zv;
+  char *worker_name = NULL;
+  size_t worker_name_len = 0;
+
+  ZEND_PARSE_PARAMETERS_START(1, 2);
+  Z_PARAM_ZVAL(zv);
+  Z_PARAM_OPTIONAL
+  Z_PARAM_STRING(worker_name, worker_name_len);
+  ZEND_PARSE_PARAMETERS_END();
+
+  char *error = go_frankenphp_send_request(thread_index, zv, worker_name,
+                                           worker_name_len);
+  if (error) {
+    zend_throw_exception(spl_ce_RuntimeException, error, 0);
+    RETURN_THROWS();
+  }
+}
+
 PHP_FUNCTION(headers_send) {
   zend_long response_code = 200;
 

frankenphp.go

@@ -37,6 +37,8 @@ import (
 	"unsafe"
 	// debug on Linux
 	//_ "github.com/ianlancetaylor/cgosymbolizer"
+
+	"github.com/dunglas/frankenphp/internal/watcher"
 )
 
 type contextKeyStruct struct{}
@@ -52,7 +54,8 @@ var (
 	ErrScriptExecution        = errors.New("error during PHP script execution")
 	ErrNotRunning             = errors.New("FrankenPHP is not running. For proper configuration visit: https://frankenphp.dev/docs/config/#caddyfile-config")
 
-	isRunning bool
+	isRunning        bool
+	watcherIsEnabled bool
 
 	loggerMu sync.RWMutex
 	logger   *slog.Logger
@@ -151,6 +154,10 @@ func calculateMaxThreads(opt *opt) (int, int, int, error) {
 		numWorkers += opt.workers[i].num
 	}
 
+	for _, tw := range opt.taskWorkers {
+		numWorkers += tw.num
+	}
+
 	numThreadsIsSet := opt.numThreads > 0
 	maxThreadsIsSet := opt.maxThreads != 0
 	maxThreadsIsAuto := opt.maxThreads < 0 // maxthreads < 0 signifies auto mode (see phpmaintread.go)
@@ -279,10 +286,23 @@ func Init(options ...Option) error {
 		convertToRegularThread(getInactivePHPThread())
 	}
 
+	directoriesToWatch := getDirectoriesToWatch(append(opt.workers, opt.taskWorkers...))
+	watcherIsEnabled = len(directoriesToWatch) > 0 // watcherIsEnabled needs to be set before initWorkers()
+
 	if err := initWorkers(opt.workers); err != nil {
 		return err
 	}
 
+	if err := initTaskWorkers(opt.taskWorkers); err != nil {
+		return err
+	}
+
+	if watcherIsEnabled {
+		if err := watcher.InitWatcher(directoriesToWatch, RestartWorkers, logger); err != nil {
+			return err
+		}
+	}
+
 	initAutoScaling(mainThread)
 
 	ctx := context.Background()
@@ -300,8 +320,11 @@ func Shutdown() {
 		return
 	}
 
-	drainWatcher()
+	if watcherIsEnabled {
+		watcher.DrainWatcher()
+	}
 	drainAutoScaling()
+	drainTaskWorkers()
 	drainPHPThreads()
 
 	metrics.Shutdown()
@@ -629,3 +652,12 @@ func timeoutChan(timeout time.Duration) <-chan time.Time {
 
 	return time.After(timeout)
 }
+
+func getDirectoriesToWatch(workerOpts []workerOpt) []string {
+	directoriesToWatch := []string{}
+	for _, w := range workerOpts {
+		directoriesToWatch = append(directoriesToWatch, w.watch...)
+	}
+
+	return directoriesToWatch
+}

frankenphp.h

@@ -86,4 +86,6 @@ void frankenphp_register_bulk(
 
 void register_extensions(zend_module_entry *m, int len);
 
+void frankenphp_update_thread_context(bool is_worker, bool is_task_worker);
+
 #endif

frankenphp.stub.php

@@ -4,6 +4,8 @@
 
 function frankenphp_handle_request(callable $callback): bool {}
 
+function frankenphp_send_request(mixed $task, string $workerName = ''): void {}
+
 function headers_send(int $status = 200): int {}
 
 function frankenphp_finish_request(): bool {}

frankenphp_arginfo.h

@@ -6,6 +6,12 @@ ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_frankenphp_handle_request, 0, 1,
 ZEND_ARG_TYPE_INFO(0, callback, IS_CALLABLE, 0)
 ZEND_END_ARG_INFO()
 
+ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_frankenphp_send_request, 0, 1,
+                                        IS_VOID, 0)
+ZEND_ARG_TYPE_INFO(0, task, IS_MIXED, 0)
+ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, worker_name, IS_STRING, 0, "\"\"")
+ZEND_END_ARG_INFO()
+
 ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_headers_send, 0, 0, IS_LONG, 0)
 ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, status, IS_LONG, 0, "200")
 ZEND_END_ARG_INFO()
@@ -31,6 +37,7 @@ ZEND_END_ARG_INFO()
 #define arginfo_apache_response_headers arginfo_frankenphp_response_headers
 
 ZEND_FUNCTION(frankenphp_handle_request);
+ZEND_FUNCTION(frankenphp_send_request);
 ZEND_FUNCTION(headers_send);
 ZEND_FUNCTION(frankenphp_finish_request);
 ZEND_FUNCTION(frankenphp_request_headers);
@@ -39,6 +46,7 @@ ZEND_FUNCTION(frankenphp_response_headers);
 // clang-format off
 static const zend_function_entry ext_functions[] = {
   ZEND_FE(frankenphp_handle_request, arginfo_frankenphp_handle_request)
+  ZEND_FE(frankenphp_send_request, arginfo_frankenphp_send_request)
   ZEND_FE(headers_send, arginfo_headers_send)
   ZEND_FE(frankenphp_finish_request, arginfo_frankenphp_finish_request)
   ZEND_FALIAS(fastcgi_finish_request, frankenphp_finish_request, arginfo_fastcgi_finish_request)

options.go

@@ -22,6 +22,7 @@ type opt struct {
 	numThreads  int
 	maxThreads  int
 	workers     []workerOpt
+	taskWorkers []workerOpt
 	logger      *slog.Logger
 	metrics     Metrics
 	phpIni      map[string]string
@@ -35,6 +36,8 @@ type workerOpt struct {
 	env                    PreparedEnv
 	watch                  []string
 	maxConsecutiveFailures int
+	isTaskWorker           bool
+	maxQueueLen            int
 }
 
 // WithNumThreads configures the number of PHP threads to start.
@@ -80,7 +83,11 @@ func WithWorkers(name string, fileName string, num int, options ...WorkerOption)
 			}
 		}
 
-		o.workers = append(o.workers, worker)
+		if worker.isTaskWorker {
+			o.taskWorkers = append(o.taskWorkers, worker)
+		} else {
+			o.workers = append(o.workers, worker)
+		}
 
 		return nil
 	}
@@ -141,3 +148,14 @@ func WithMaxWaitTime(maxWaitTime time.Duration) Option {
 		return nil
 	}
 }
+
+// EXPERIMENTAL: AsTaskWorker configures the worker as a task worker.
+// no http requests will be handled.
+// no globals resetting will be performed between tasks.
+func AsTaskWorker(isTaskWorker bool, maxQueueLen int) WorkerOption {
+	return func(w *workerOpt) error {
+		w.isTaskWorker = isTaskWorker
+		w.maxQueueLen = maxQueueLen
+		return nil
+	}
+}

phpthread.go

@@ -132,6 +132,10 @@ func (thread *phpThread) pinCString(s string) *C.char {
 	return thread.pinString(s + "\x00")
 }
 
+func (thread *phpThread) updateContext(isWorker bool, isTaskWorker bool) {
+	C.frankenphp_update_thread_context(C.bool(isWorker), C.bool(isTaskWorker))
+}
+
 //export go_frankenphp_before_script_execution
 func go_frankenphp_before_script_execution(threadIndex C.uintptr_t) *C.char {
 	thread := phpThreads[threadIndex]