feat: frankenphp_send_task(), frankenphp_receive_task() and friends

The task half of #2319, on top of the background workers and their shared
vars: a request, an HTTP worker or another background worker hands work to
a named background worker with frankenphp_send_task(), which returns a
stream carrying the updates the worker sends back with
frankenphp_update_task(). The worker dequeues tasks with
frankenphp_receive_task() after reading a "task\n" line on its handle: the
one handle of #2617 carries both the drain EOF and the wake-ups, so a
script keeps a single stream_select() loop. The line is a wake-up, not a
count: every thread of a pool gets one per task, the first one back in its
loop takes the task and the others get null.

send_task() blocks until a thread of the worker picks the task up and
throws on timeout, so a busy worker pushes back on its senders instead of
queueing without bounds; tasks queued while a thread restarts are signaled
again on its next run. The wait also ends when the sender's own thread is
drained for a restart or the shutdown, since the target's threads are
drained too. Names resolve like frankenphp_get_vars() does.

Each task gets a socket pair. The sender's stream is a socket stream over
one end, one byte per update and EOF at completion, so stream_select()
bounds the wait or multiplexes tasks, and a blocking read parks as well;
closing it abandons the task. The receiver's stream is a socket stream
over the other end: updates go through update_task(), the stream itself
reports the sender's close as EOF to stream_select() and feof(), so a long
task learns that nobody waits for its result, and update_task() throws.
Closing it completes the task, unless the close is the resource cleanup of
request shutdown, which means the script ended with the task open: the
sender's next read throws instead of returning null. Sixteen updates are
buffered per task, past that update_task() waits for the sender to read.

Waking threads is what a task costs, so wake-ups are kept to a minimum. A
send wakes one parked thread of the worker, round-robin, with the line on
its handle; a thread that reads its handle while tasks are queued gets the
line from the read op itself, so no wake-up is lost whichever loop shape
the script uses, and after 10ms without pickup every thread is woken as a
fallback. The sender waits for the pickup in the kernel, on its end of the
task's pair, rather than in a Go select: waking a PHP thread parked inside
a cgo callback costs Go a P hand-off, a byte on a socket does not. The
thread taking the task writes that byte, a watcher goroutine does when the
wait must end without a pickup. The queue mutex is never held across a
syscall and taken once per wake-up, as a thread inside a cgo callback that
loses it parks the same expensive way. In the Docker builder image this
takes a task from 549 to 285us with one thread, a pool of 8 from 1745 to
320us, and 8 senders on 8 threads from 2k to 32k tasks/s.

Payloads and updates follow the set_vars() whitelist and travel as
persistent tables through the Go side, which owns them until they are
copied into request memory. The streams reference their task through a
cgo handle; the task is freed once both sides closed, or by the sender
when no thread picked it up. The stop sockets of a worker's threads are now
guarded by its task queue mutex, since senders write to them.

Compared to #2319: no queue ahead of pickup and no cancellation before it,
no dedicated signaling stream, no global task table.

Author: nicolas-grekas

docs/worker.md

@@ -253,6 +253,34 @@ $vars = frankenphp_get_vars('config');
 
 `frankenphp_get_vars()` blocks until the worker reached its ready point, which can only happen between background workers reading each other while booting; a cycle between them throws instead of hanging. It also throws when the name is unknown, or when the worker is ready but has not published anything.
 
+### Sending tasks to background workers
+
+A request, an HTTP worker or another background worker hands work to a background worker with `frankenphp_send_task()`, by worker name, resolved like `frankenphp_get_vars()` does. The payload follows the same rules as `frankenphp_set_vars()`: null, scalars, arrays or enums. The call 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. It returns a stream: `frankenphp_read_task()` blocks for the next update and returns `null` once the worker completed the task, and `stream_select()` works on the stream to wait on several tasks or to bound the wait. Closing the stream abandons the task.
+
+On the worker side, each task sent wakes one parked thread of the worker with a `task\n` line on its handle, so the loop reads the handle: `fgets()` returns `"task\n"` when there is work and `false` once the worker is drained. The line is a wake-up, not a count: `frankenphp_receive_task()` dequeues a task without blocking, `[$stream, $payload]`, 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 reads its handle while tasks are queued gets a line at once, whichever loop shape it uses. `frankenphp_update_task()` sends progress or a result back and closing the stream completes the task; a script that ends with the stream still open, a close from a destructor or a shutdown function at that point included, makes the sender's next `frankenphp_read_task()` throw. When the sender closes its stream instead, the worker's stream reaches EOF, so `stream_select()` or `feof()` on it tell a long task that nobody waits for its result, and `frankenphp_update_task()` throws.
+
+```php
+// background worker
+$handle = frankenphp_get_worker_handle();
+
+while (false !== fgets($handle)) {
+    while ($task = frankenphp_receive_task()) {
+        [$stream, $payload] = $task;
+        frankenphp_update_task($stream, ['progress' => 50]);
+        frankenphp_update_task($stream, ['result' => process($payload)]);
+        fclose($stream);
+    }
+}
+
+// request, HTTP worker or another background worker
+$task = frankenphp_send_task('jobs', ['file' => 'photo.jpg']);
+while (null !== $update = frankenphp_read_task($task)) {
+    // ['progress' => 50], then ['result' => ...]
+}
+```
+
+Sixteen updates are buffered per task; past that, `frankenphp_update_task()` waits for the sender to read, and it throws once the sender closed its stream.
+
 ## Superglobals behavior
 
 [PHP superglobals](https://www.php.net/manual/language.variables.superglobals.php) (`$_SERVER`, `$_ENV`, `$_GET`...)