feat: SentTaskHandle and ReceivedTaskHandle

Author: nicolas-grekas

caddy/caddy_test.go

@@ -989,7 +989,7 @@ func TestWorkerMetrics(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads 2
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 	# TYPE frankenphp_busy_workers gauge
 	frankenphp_busy_workers{worker="` + workerName + `"} 0
 
@@ -1146,7 +1146,7 @@ func TestNamedWorkerMetrics(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads 2
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
         # TYPE frankenphp_busy_workers gauge
         frankenphp_busy_workers{worker="my_app"} 0
 
@@ -1242,7 +1242,7 @@ func TestAutoWorkerConfig(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads ` + workers + `
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 	# TYPE frankenphp_busy_workers gauge
 	frankenphp_busy_workers{worker="` + workerName + `"} 0
 
@@ -1516,6 +1516,7 @@ func TestMaxWaitTimeWorker(t *testing.T) {
 	require.NoError(t, err)
 
 	expectedMetrics := `
+	# HELP frankenphp_worker_queue_depth Number of queued requests for this worker, or of tasks waiting for a thread of a background worker
 	# TYPE frankenphp_worker_queue_depth gauge
 	frankenphp_worker_queue_depth{worker="service"} 0
 	`
@@ -1616,7 +1617,7 @@ func TestMultiWorkersMetrics(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads 5
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 	# TYPE frankenphp_busy_workers gauge
 	frankenphp_busy_workers{worker="service1"} 0
 

caddy/config_test.go

@@ -390,6 +390,39 @@ func TestWorkerBackgroundVars(t *testing.T) {
 	require.Equal(t, []string{"redis", "feature_flags"}, module.Workers[0].Vars)
 }
 
+func TestWorkerBackgroundTasks(t *testing.T) {
+	d := caddyfile.NewTestDispenser(`
+	{
+		php_server {
+			worker {
+				file ../testdata/worker-with-env.php
+				background
+				num 4
+				tasks image.resize image.thumbnail
+			}
+		}
+	}`)
+	module := &FrankenPHPModule{}
+
+	require.NoError(t, module.UnmarshalCaddyfile(d))
+	require.Equal(t, []string{"image.resize", "image.thumbnail"}, module.Workers[0].Tasks)
+}
+
+func TestWorkerTasksRequireBackground(t *testing.T) {
+	d := caddyfile.NewTestDispenser(`
+	{
+		php_server {
+			worker {
+				file ../testdata/worker-with-env.php
+				tasks jobs
+			}
+		}
+	}`)
+	module := &FrankenPHPModule{}
+
+	require.ErrorContains(t, module.UnmarshalCaddyfile(d), `"tasks" is only supported for background workers`)
+}
+
 func TestWorkerVarsRequireBackground(t *testing.T) {
 	d := caddyfile.NewTestDispenser(`
 	{
@@ -482,3 +515,11 @@ func TestWorkerVarsReachTheCoreWithoutBackground(t *testing.T) {
 
 	require.ErrorContains(t, frankenphp.Validate(frankenphp.WithWorkers("http", wc.FileName, 1, opts...)), "only background workers publish them")
 }
+
+func TestWorkerTasksReachTheCoreWithoutBackground(t *testing.T) {
+	wc := workerConfig{FileName: "../testdata/worker-with-env.php", Tasks: []string{"jobs"}}
+	opts, err := wc.toWorkerOptions()
+	require.NoError(t, err)
+
+	require.ErrorContains(t, frankenphp.Validate(frankenphp.WithWorkers("http", wc.FileName, 1, opts...)), "only background workers receive them")
+}

caddy/workerconfig.go

@@ -45,6 +45,8 @@ type workerConfig struct {
 	Background bool `json:"background,omitempty"`
 	// Vars lists the namespaces a background worker publishes with WorkerHandle::setVars(), read by frankenphp_get_vars().
 	Vars []string `json:"vars,omitempty"`
+	// Tasks lists the namespaces a background worker receives, the ones a FrankenPHP\SentTaskHandle is sent to.
+	Tasks []string `json:"tasks,omitempty"`
 
 	options []frankenphp.WorkerOption
 }
@@ -169,8 +171,14 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) {
 				return wc, d.ArgErr()
 			}
 			wc.Vars = append(wc.Vars, namespaces...)
+		case "tasks":
+			namespaces := d.RemainingArgs()
+			if len(namespaces) == 0 {
+				return wc, d.ArgErr()
+			}
+			wc.Tasks = append(wc.Tasks, namespaces...)
 		default:
-			return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, max_consecutive_failures, max_threads, background, boot_timeout, vars", v)
+			return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, max_consecutive_failures, max_threads, background, boot_timeout, vars, tasks", v)
 		}
 	}
 
@@ -190,6 +198,10 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) {
 		return wc, d.Err(`"vars" is only supported for background workers`)
 	}
 
+	if !wc.Background && len(wc.Tasks) != 0 {
+		return wc, d.Err(`"tasks" is only supported for background workers`)
+	}
+
 	if frankenphp.EmbeddedAppPath != "" && filepath.IsLocal(wc.FileName) {
 		wc.FileName = filepath.Join(frankenphp.EmbeddedAppPath, wc.FileName)
 	}
@@ -219,12 +231,16 @@ func (wc *workerConfig) toWorkerOptions() ([]frankenphp.WorkerOption, error) {
 		}
 	}
 
-	// whatever the kind of worker: the core refuses vars on an HTTP one, which
-	// a JSON config reaches without the Caddyfile check
+	// whatever the kind of worker: the core refuses vars and tasks on an HTTP
+	// one, which a JSON config reaches without the Caddyfile check
 	if len(wc.Vars) > 0 {
 		opts = append(opts, frankenphp.WithWorkerVars(wc.Vars...))
 	}
 
+	if len(wc.Tasks) > 0 {
+		opts = append(opts, frankenphp.WithWorkerTasks(wc.Tasks...))
+	}
+
 	// copy the caddy match logic and create a unique matcher function for this worker
 	// inject the matcher into frankenphp
 	if len(wc.MatchPath) > 0 {

docs/config.md

@@ -114,6 +114,7 @@ You can also explicitly configure FrankenPHP using the [global option](https://c
 			boot_timeout <duration> # EXPERIMENTAL: how long a background worker may take to reach its first tick() before the start fails, 0 waits forever. Default: 30s on PHP builds without Zend max execution timers, none elsewhere.
 			background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker, whose script ticks its FrankenPHP\WorkerHandle at least once
 			vars <namespace...> # EXPERIMENTAL: the vars namespaces a background worker publishes, read with frankenphp_get_vars()
+			tasks <namespace...> # EXPERIMENTAL: the task namespaces a background worker receives, sent with FrankenPHP\SentTaskHandle
 		}
 	}
 }
@@ -205,6 +206,7 @@ php_server [<matcher>] {
 		boot_timeout <duration> # EXPERIMENTAL: how long a background worker may take to reach its first tick() before the start fails, 0 waits forever. Default: 30s on PHP builds without Zend max execution timers, none elsewhere.
 		background # EXPERIMENTAL: marks this worker as a background (non-HTTP) worker, whose script ticks its FrankenPHP\WorkerHandle at least once
 		vars <namespace...> # EXPERIMENTAL: the vars namespaces a background worker publishes, read with frankenphp_get_vars()
+		tasks <namespace...> # EXPERIMENTAL: the task namespaces a background worker receives, sent with FrankenPHP\SentTaskHandle
 	}
 	worker <other_file> <num> # Can also use the short form like in the global frankenphp block.
 }

docs/metrics.md

@@ -16,13 +16,15 @@ When [Caddy metrics](https://caddyserver.com/docs/metrics) are enabled, FrankenP
 - `frankenphp_busy_threads`: The number of PHP threads currently processing a request (running workers always consume a thread).
 - `frankenphp_queue_depth`: The number of regular queued requests.
 - `frankenphp_total_workers{worker="[worker_name]"}`: The total number of workers.
-- `frankenphp_busy_workers{worker="[worker_name]"}`: The number of workers currently processing a request.
+- `frankenphp_busy_workers{worker="[worker_name]"}`: The number of workers currently processing a request, or a task for a background worker.
 - `frankenphp_worker_request_time{worker="[worker_name]"}`: The time spent processing requests by all workers.
 - `frankenphp_worker_request_count{worker="[worker_name]"}`: The number of requests processed by all workers.
 - `frankenphp_ready_workers{worker="[worker_name]"}`: The number of workers that have reached their ready point at least once: `frankenphp_handle_request()` for HTTP workers, `WorkerHandle::tick()` for background workers.
 - `frankenphp_worker_crashes{worker="[worker_name]"}`: The number of times a worker has unexpectedly terminated.
 - `frankenphp_worker_restarts{worker="[worker_name]"}`: The number of times a worker has been deliberately restarted.
-- `frankenphp_worker_queue_depth{worker="[worker_name]"}`: The number of queued requests.
+- `frankenphp_worker_queue_depth{worker="[worker_name]"}`: The number of queued requests, or of tasks waiting for a thread of a background worker.
+- `frankenphp_worker_task_count{worker="[worker_name]",outcome="[outcome]"}`: The number of tasks sent to a background worker, by outcome: `completed`, `aborted` (the script ended with the task open), `abandoned` (the sender closed its stream first), `timeout` (no thread picked the task up in time) or `cancelled` (the wait for a thread ended on a restart or the shutdown).
+- `frankenphp_worker_task_time{worker="[worker_name]"}`: The time spent on tasks by all threads of a background worker, from pickup to the close of the task's stream.
 - `frankenphp_opcache_restarts{reason="[reason]"}`: (experimental) The number of restarts of opcache's shared memory scheduled, by reason (`oom`, `hash`, `manual`). PHP 8.4 and up.
 
 For worker metrics, the `[worker_name]` placeholder is replaced by the worker name in the Caddyfile, otherwise the absolute path of the worker file will be used.
@@ -76,8 +78,8 @@ Each entry in `ThreadDebugStates` contains:
 | `Index` | integer | The index of the thread. |
 | `Name` | string | The name of the thread (e.g., the worker file path). |
 | `State` | string | The internal state of the thread (e.g., `ready`, `shutting down`). |
-| `IsWaiting` | boolean | Whether the thread is waiting for a request. |
-| `IsBusy` | boolean | Whether the thread is currently processing a request. |
+| `IsWaiting` | boolean | Whether the thread is waiting for a request, or for a task in a background worker. |
+| `IsBusy` | boolean | Whether the thread is currently processing a request, or a task in a background worker. |
 | `WaitingSinceMilliseconds` | integer | How long the thread has been idle, in milliseconds. `0` if the thread is busy. |
 | `CurrentURI` | string | The URI currently being processed. Empty if the thread is idle. |
 | `CurrentMethod` | string | The HTTP method of the current request (e.g., `GET`, `POST`). Empty if the thread is idle. |

docs/worker.md

@@ -310,6 +310,62 @@ Values must be null, scalars, arrays or enums. Each call replaces the whole snap
 
 `frankenphp_get_vars()` can only block while workers boot, a background worker or the bootstrap of an HTTP worker waiting for the publisher of a namespace. A cycle between background workers throws instead of hanging, and so does a wait once the start failed. It also throws when no worker publishes the namespace.
 
+### Sending tasks to background workers
+
+A background worker receives the task namespaces declared with `tasks` on it, and a request, an HTTP worker or another background worker hands work to it by constructing a `FrankenPHP\SentTaskHandle` with one of those namespaces, resolved the way `frankenphp_get_vars()` resolves vars: within the `php_server` first, then among global workers.
+
+```caddyfile
+php_server {
+	worker {
+		file images.php
+		background
+		num 4
+		tasks image.resize image.thumbnail
+	}
+}
+```
+
+A namespace has one receiving worker per scope, and the threads of that worker share one queue for all its namespaces, so a pool is sized for the work it takes on, and `getNamespace()` on a received task tells which namespace it was sent to. The payload follows the same rules as `setVars()`: null, scalars, arrays or enums. Constructing blocks until a thread of the worker picks the task up and throws if none did before the timeout, so a busy worker pushes back on its senders instead of queueing without bounds. `read()` then blocks for the next update and returns `null` once the worker completed the task, and `abandon()`, like dropping the handle, gives up on the task. Waiting on a task, alone or with others, goes through a context as it does for the worker's handle, or through `getStream()` for a library that takes a stream: reading that stream steals the signal `read()` needs, writing to it goes nowhere, and closing it ends the task on that side.
+
+On the worker side, each task sent wakes one parked thread of the worker through its handle, in the loop of the previous section: `tick()` consumes the wake-up, which is not a count and in a pool may belong to a task a sibling thread took. `receive()` dequeues a task without blocking, a `FrankenPHP\ReceivedTaskHandle`, or `null` when another thread of the pool got there first, so the example below drains the queue on each wake-up and treats `null` as the normal outcome. A thread that ticks while tasks are queued finds its handle readable at once, whichever loop shape it uses. `update()` sends progress back, `complete()` ends the task with a last update when one is given, and a script that ends with a task still open, a `complete()` from a destructor or a shutdown function at that point included, makes the sender's next `read()` throw. When the sender gives up instead, the task's side of the worker becomes readable and its stream reaches EOF, so a context or a `feof()` tells a long task that nobody waits for its result, and `update()` throws.
+
+```php
+// background worker, in the loop of the previous section
+while ($handle->tick()) {
+    foreach ($poll->wait() as $watcher) {
+    }
+
+    while ($task = $handle->receive()) {
+        $task->update(['progress' => 50]);
+        $task->complete(['result' => process($task->getNamespace(), $task->getPayload())]);
+    }
+}
+
+// request, HTTP worker or another background worker
+$task = new FrankenPHP\SentTaskHandle('image.resize', ['file' => 'photo.jpg']);
+while (null !== $update = $task->read()) {
+    // ['progress' => 50], then ['result' => ...]
+}
+```
+
+Both task handles are `Io\Poll\Handle` as well, so a context follows several tasks at once, and a script that never asks for a stream has none built for it:
+
+```php
+use Io\Poll\{Context, Event};
+
+$poll = new Context();
+$task = new FrankenPHP\SentTaskHandle('image.resize', ['file' => 'photo.jpg']);
+$poll->add($task, [Event::Read], 'photo');
+
+foreach ($poll->wait() as $watcher) {
+    $update = $task->read();
+}
+```
+
+`getStream()` is there for the libraries that take a stream, and `stream_select()` or a blocking read work on it as they do on the worker's handle.
+
+Sixteen updates are buffered per task; past that, `update()` waits for the sender to read, and it throws once the sender gave up. The channel of a task is a pair of eventfd descriptors on Linux and of kqueue descriptors on macOS, pooled between tasks, and a socket pair elsewhere.
+
 ## Superglobals behavior
 
 [PHP superglobals](https://www.php.net/manual/language.variables.superglobals.php) (`$_SERVER`, `$_ENV`, `$_GET`...)