diff --git a/CHANGELOG.md b/CHANGELOG.md index 551b4200..90f3f72a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,18 +9,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **A streaming handler can offer a chunk without waiting for room (#177).** `HttpResponse::tryWrite()` returns false when the outbound queue is full, having queued nothing, so the same chunk can be offered again; a client that has gone still throws `HttpException` 499, because "wait" and "stop" need opposite reactions and one bool cannot carry both. The pair mirrors `WebSocket::send()`/`trySend()` and shares their high-water mark, `HttpServerConfig::setStreamWriteBufferBytes()`. The transport answers where the chunk is queued rather than through a predicate read beforehand: `append_chunk` gained a `nonblocking` argument, and a transport that would have parked returns `HTTP_STREAM_APPEND_BACKPRESSURE` instead. HTTP/1 is the exception in both halves — it keeps no queue of its own, so it never refuses and an accepted chunk waits for the socket as `write()` does; #179 removes that. +- **A refused chunk can be waited out instead of spun on (#177).** `HttpResponse::awaitWritable()` suspends until the outbound queue has room and reports whether it has. Without it the only shapes after a `false` were a sleep-and-retry loop or a fall back to the blocking `send()`, and the drain event each transport already maintains was reachable from C only. The wait belongs to the transport, which keeps its own deadline and re-pumps its drain on each wake — assembling it at the PHP boundary instead would drop all three. HTTP/1 has no queue and so answers at once; a transport that can be full but cannot be waited on answers false rather than true, because a handler told to go ahead would spin without yielding, and on a pool worker that freezes every other request on the thread. The refusal granularity differs by transport and decides which shape is right — HTTP/2 refuses on 8 live slots or `setStreamWriteBufferBytes()`, HTTP/3 on any chunk not yet handed to nghttp3, so on HTTP/3 a refusal is expected once per chunk under a congested path. - **A handler can ask whether the client is still there (#175).** `HttpResponse::isWritable()` reports whether output is still possible — `end()` was not called, the response is not sealed by `sendFile()`, and the peer has not gone. The only predicate before it was `sendable()`, which also answers false on a full queue, so a streaming loop could not separate "yield and continue" from "stop"; our own SSE example read it as the latter, and so did the loop that truncated a proxied body at ~100 KB in YanGusik/laravel-spawn#60. A false answer from `isWritable()` is final, which is what makes it safe to break on. An optional `is_alive` op on the stream vtable backs it in all four transports; on HTTP/1 a peer's departure only becomes visible when a write fails, so that discovery is recorded on the request and answered afterwards instead of being rediscovered by a second doomed write. ### Changed +- **BC: `write()` streams, and the buffered append moved to `appendBody()` (#180).** `HttpResponse::write()` appended to a buffer and put nothing on the wire until `end()`, while Node, Swoole and Go all stream under that name — both field reports behind this contract work (YanGusik/laravel-spawn#50, #60) came from the API rather than from the adapter's code. A handler that used `write()` for buffered appending keeps parsing and starts streaming: the first call commits status and headers, so every later `setHeader()` or `setStatusCode()` throws where it used to work. Rename those calls to `appendBody()`, which is the old behaviour under a name that says it. +- **BC: `send()` is removed; the call is `write()` (#180).** No alias is kept. An alias would have covered one call in the shipped laravel-spawn adapter (`src/Server/TrueAsyncServer.php:492`) while `isClosed()`, removed in the same release, breaks three others beside it — the adapter needs a release either way, and a deprecated spelling left behind only postpones the same edit. A call to `send()` now fails as an undefined method, at the line that has to change. +- **BC: `isClosed()` is now `isEnded()` (#180).** The method returned `response->closed`, the flag `end()` sets, and reported nothing about the connection — a handler reading it as "the peer is gone" got a wrong answer for the whole life of the response. `isWritable()` is the call that answers liveness. `isClosed()` no longer exists; a call fails as an undefined method. +- **BC: `sendable()` is removed, and its declaration is a tombstone (#180).** One bool answered four questions — closed, sealed by `sendFile()`, detached, full — and `README.md` documented it as a liveness check until #174, which is the loop that truncated a proxied body in YanGusik/laravel-spawn#60. Our own `examples/sse-server.php` broke its loop on it too, and now stops on `!isWritable()`. Calling `sendable()` raises `HttpServerRuntimeException` naming both replacements: `isWritable()` for liveness, `tryWrite()`/`awaitWritable()` for room. The declaration stays one minor release so shipped adapter code is told what to call rather than failing as an undefined method. +- **BC: `getBodyStream()` and `setBodyStream()` are removed (#180).** Neither ever had an implementation: the first returned null, the second threw "Body stream support is not yet implemented". A handler wanting a file on the wire calls `sendFile()`; one wanting incremental output calls `write()`. - **BC: `RoomDeliveryException` extends `HttpServerException`, not `WebSocketException`.** A build configured with `--disable-websocket` serves rooms, and in it `WebSocketException` does not exist. A handler that caught `WebSocketException` around `Room::send()`, `Room::trySend()` or `HttpServer::send()` no longer catches it — catch `RoomDeliveryException` or `HttpServerException` instead. Nothing else about the exception changed: the `delivered` and `pending` counts and the message are what they were. - **Rooms build without WebSocket.** The pub/sub core is `src/room/` and no longer knows what a connection is; `--disable-websocket` compiles it, registers `Room` and `RoomDeliveryException`, and delivers a publish from one thread to a `recv()` in another. `getRuntimeStats()` reports the room counters in every build, and the request-shutdown sweep that detaches a subscribed thread now runs in every build — without WebSocket it did not, so such a thread leaked its mailbox and left a live libuv handle behind. ### Fixed -- **A streamed response was held by the compressor until the stream ended (#170).** `HttpResponse::send()` fed every chunk to the encoder in continue mode (`Z_NO_FLUSH`, `ZSTD_e_continue`, `BROTLI_OPERATION_PROCESS`) and a block was closed only by `finish()` at end of stream, so a progress feed, a log tail or a row-by-row export reached the client in one burst at the end. Text compresses well enough that a whole stream fits inside that holdback, which is why the failure hit exactly the payloads people stream; incompressible bodies crossed the buffer on the first chunk and streamed normally. Reproduction in the issue: 350 KB of CSV emitted in five bursts 300 ms apart arrived as one 10 KB burst after 1.5 s under `Accept-Encoding: gzip`, against arrivals every 300 ms without it. The encoder vtable now carries a `flush` op (`Z_SYNC_FLUSH`, `ZSTD_e_flush`, `BROTLI_OPERATION_FLUSH`) and the streaming wrapper calls it once per non-empty chunk the handler hands over, so the boundary the handler chose is the flush granularity. Measured with `013-h1-streaming-gzip-flush.phpt`: a client reading a stream whose handler is still parked decoded 0 of 4600 bytes before, and the whole first chunk after; Brotli and zstd went from 0 bytes on the wire to a decodable block. The cost is one closed block per chunk — 7.7 bytes for gzip, 9.9 for Brotli, 9.8 for zstd, measured over 80 chunks of 51 bytes (`dev/BENCHMARKS.md`) — so a handler streaming row by row trades ratio for immediacy and still sends 4.8 times less than identity. An empty chunk skips the flush, and a buffered response is untouched: it still compresses in one shot. +- **A buffered body was discarded without error when the handler then streamed (#181).** `setBody()`, `appendBody()`, `json()` or `html()` followed by a streaming call put only the streamed chunks on the wire; the buffer was never read, because the streaming path commits its own headers and the buffered dispose path runs only while `streaming` is false. The reverse direction has always thrown, so the failure looked symmetric and reported on one side only. The guard every streaming entry point shares now refuses a non-empty buffer and names both modes; an empty one does not count, since `setBody('')` commits the handler to nothing. Test `tests/phpt/server/core/062-body-api-names.phpt`, route `/mixed`: without the guard the route answers `Transfer-Encoding: chunked` with body `streamed`, with it `Content-Length: 8` and body `buffered`. +- **A cancelled handler could seal a half-written chunk (#177).** An HTTP/1 chunk is three writes — size line, body, CRLF — and the coroutine suspends between them, so a cancellation lands mid-frame: parse-error cancellation, `ThreadPool::stop()`, a scope teardown. `mark_ended` then wrote the terminal zero-chunk regardless, telling the peer the body had ended cleanly and handing the connection on for reuse — while the peer read that terminator as the first bytes of the chunk the orphaned size line had promised. A frame interrupted this way is now recorded as a dead stream: no terminator, and the connection is not kept alive. +- **A dropped chunk was reported as written (#177).** When the reactor's mailbox refused a wire after its retries, `worker_stream_append_chunk` answered OK, so a pool-dispatched handler was told it had written bytes the peer will never see. It now reports the stream dead, which is what the abort already sent on the next call. +- **`sendable()` answered a constant true under compression, and on HTTP/3 (#177).** A transport that can refuse must publish the predicate, because the compressing wrapper reads it to decide whether it may feed the encoder, and an encoder cannot be un-fed: without it a refusal on HTTP/3 threw away a block deflate had already emitted, and the retry the caller was told to make wrote the same bytes into a window that already held them. The compressing stream wrapper installed on the first `write()` carried neither a `sendable` nor an `is_alive` slot, and a NULL slot means "no queue of its own, report writable" — so on a compressed response both questions were answered by the absence of an implementation rather than by the transport holding the queue. HTTP/3 had no `sendable` at all, for a stream that does queue. Both now report from the transport: the wrapper delegates, and HTTP/3 answers whether the previous chunk has reached nghttp3, which is the granularity its `append_chunk` waits on. +- **A streamed response was held by the compressor until the stream ended (#170).** `HttpResponse::write()` fed every chunk to the encoder in continue mode (`Z_NO_FLUSH`, `ZSTD_e_continue`, `BROTLI_OPERATION_PROCESS`) and a block was closed only by `finish()` at end of stream, so a progress feed, a log tail or a row-by-row export reached the client in one burst at the end. Text compresses well enough that a whole stream fits inside that holdback, which is why the failure hit exactly the payloads people stream; incompressible bodies crossed the buffer on the first chunk and streamed normally. Reproduction in the issue: 350 KB of CSV emitted in five bursts 300 ms apart arrived as one 10 KB burst after 1.5 s under `Accept-Encoding: gzip`, against arrivals every 300 ms without it. The encoder vtable now carries a `flush` op (`Z_SYNC_FLUSH`, `ZSTD_e_flush`, `BROTLI_OPERATION_FLUSH`) and the streaming wrapper calls it once per non-empty chunk the handler hands over, so the boundary the handler chose is the flush granularity. Measured with `013-h1-streaming-gzip-flush.phpt`: a client reading a stream whose handler is still parked decoded 0 of 4600 bytes before, and the whole first chunk after; Brotli and zstd went from 0 bytes on the wire to a decodable block. The cost is one closed block per chunk — 7.7 bytes for gzip, 9.9 for Brotli, 9.8 for zstd, measured over 80 chunks of 51 bytes (`dev/BENCHMARKS.md`) — so a handler streaming row by row trades ratio for immediacy and still sends 4.8 times less than identity. An empty chunk skips the flush, and a buffered response is untouched: it still compresses in one shot. - **A compressing stream wrapper returned a faulted encoder to the pool.** The buffered path destroys an encoder that answered `HTTP_ENC_ERROR`, because its internal state is indeterminate; the streaming path left it attached to the response, and teardown handed it back to the per-thread pool for the next response to reuse. It is now destroyed on the spot, and `mark_ended` writes no trailer when the encoder is gone. - ## [0.12.0] - 2026-08-15 ### Added diff --git a/dev/BENCHMARKS.md b/dev/BENCHMARKS.md index b7e2179b..cfd0e054 100644 --- a/dev/BENCHMARKS.md +++ b/dev/BENCHMARKS.md @@ -3,6 +3,51 @@ One entry per measurement, newest first. An entry names the machine, the build and the scenario, because a number without them cannot be compared with the next one. +## 2026-08-20 — what the three writes per HTTP/1 chunk cost (#179) + +Branch `180-body-rename` at 2e72f41. Machine: WSL2, Linux 6.6.114.1, 16 cores. +PHP 8.6.0-dev ZTS **release** (`--disable-debug`), built for this measurement into +`/home/edmond/php-release-24` from the same php-src the debug build uses, because +the installed release PHP is ABI v0.23 and the extension needs v0.24. Load: +`wrk -t1 -c4 -d6s`, three runs per cell, median reported. Server: +`tests/perf/servers/server_stream.php` in `h1` mode, one worker. + +`h1_stream_append_chunk` sends a chunk as three awaited writes — size line, body, +CRLF (`src/http1/http1_stream.c:154`). Counted with `strace -e trace=write,epoll_pwait` +on one request of four 16 KiB chunks: **3 `write(2)` and 3 zero-timeout +`epoll_pwait` per chunk**, whatever the chunk size, plus one write for the headers +and one for the terminator. The loop turn after each write is the coroutine +suspending: `async_io_req_await` returns early only on `req->completed`, and +libuv's inline-write fast path does not fire for back-to-back writes on one stream. + +The comparison holds the body at 64 KiB and moves only the chunk count. The second +build differs by one hunk: the three pieces are copied into one buffer and sent as +a single awaited write. + +| chunks | chunk | three writes | one write | gain | µs per chunk, before → after | +|---|---|---|---|---|---| +| 1 | 64 KiB | 15746 | 17207 | +9.3% | — | +| 4 | 16 KiB | 8320 | 11876 | +42.7% | 18.9 → 8.7 | +| 16 | 4 KiB | 2938 | 5209 | +77.3% | 18.5 → 8.9 | +| 64 | 1 KiB | 893 | 1650 | +84.8% | 16.8 → 8.7 | + +Taken. Per-chunk cost is flat in the chunk size and halves when the frame goes out +as one write: the two extra syscalls and two extra loop turns are worth about +10 µs per chunk. The 1 KiB row is the noisiest — its three base runs were 1135, +893 and 781 — and the others repeat within 3%. + +The prototype copies the whole chunk to coalesce it, and still wins by that much. +A vectored write would avoid the copy, but the ABI has no awaitable one: +`io_pipe_writev_cb` (`php-src/ext/async/libuv_reactor.c:4947`) sends no NOTIFY and +frees the request itself, so `ZEND_ASYNC_IO_WRITEV` cannot be awaited. Removing the +copy means adding that op to ext/async. + +What this decides for #179: the win is reachable without a queue, without an +ordering hazard between two writers and without a per-response structure — the +three things both rejected designs foundered on. Whatever else #179 wants, a +non-blocking `tryWrite()` on HTTP/1, has to be argued on its own; this measurement +does not support it. + ## 2026-08-19 — cost of the per-chunk flush on a streamed response (#170) Base commit 22a8d37 plus the #170 working tree. Machine: WSL2, Linux 6.6.114.1, diff --git a/dev/PLAN.md b/dev/PLAN.md index c6a0fecc..5872a30d 100644 --- a/dev/PLAN.md +++ b/dev/PLAN.md @@ -37,34 +37,63 @@ wire unverified. The contract is settled; the steps are ordered so the documentation lands first, because the reporter is writing a proxy recipe against it and expects a tag within days. -- [ ] **Document the three body modes first.** `docs/USAGE.md` says nothing about +- [x] **Document the three body modes first.** `docs/USAGE.md` says nothing about the response body at all. It gains a section naming the modes — buffered (`setBody`), streamed (`write`), file (`sendFile`) — the state each commits, and a framing table: a buffered body gets its `Content-Length` computed, an undeclared stream is chunked or DATA frames, a declared stream keeps the header. `README.md:277` and the `write()` docblock (`stubs/HttpResponse.php:160`, which never says that nothing leaves before `end()`) are corrected in the same step. -- [ ] **`isWritable(): bool` — liveness, with the op behind it.** A new optional + Done in #174: the README guard is gone and both docblocks say what they mean. + The `docs/USAGE.md` section landed with the renames as §3.5, written once + against the final names. +- [x] **`isWritable(): bool` — liveness, with the op behind it.** A new optional `is_alive` in `http_response_stream_ops_t` (`include/php_http_server.h:706`); every backend already computes it inside `append_chunk` (`peer_closed` for H2, `stream_credit_is_dead` for the worker). Sound as a predicate because every input is a one-way latch, unlike queue depth. -- [ ] **`write()` becomes the streaming call.** `send()` stays one minor release as - a deprecated alias; buffered incremental appending keeps its behaviour under - `appendBody()`; `isClosed()` becomes `isEnded()`, which is all it ever reported; - `sendable()` is removed with a tombstone naming its two replacements, because - shipped adapter code calls it; `setBodyStream()`/`getBodyStream()` - (`stubs/HttpResponse.php:257,265`) are deleted — one throws "not yet - implemented", the other returns null. -- [ ] **`tryWrite(): bool` and the dialect twins.** The non-blocking half of the +- [x] **`write()` becomes the streaming call.** Done in #180. `send()` is removed + outright rather than kept as a deprecated alias: it would have covered one call + in the shipped laravel-spawn adapter while `isClosed()` breaks three others + beside it (`src/Server/TrueAsyncServer.php:106,395,413,492`), so the adapter + needs a release either way. Buffered incremental appending keeps its behaviour + under `appendBody()`; `isClosed()` became `isEnded()`; + `sendable()` throws a tombstone naming `isWritable()` and `tryWrite()`; + `setBodyStream()`/`getBodyStream()` are deleted. Evidence: + `tests/phpt/server/core/062-body-api-names.phpt` reads the wire for each mode, + and `h2/023-h2-sendable-tombstone.phpt` asserts the throw on a live H2 stream + where the method used to answer. `docs/USAGE.md` §3.5 documents the three modes. + + Two defects surfaced while doing it, both fixed here. **#181**: a buffered body + followed by a streaming call was discarded with no error — the streaming path + never reads `response->body`, while the reverse direction has always thrown, so + only one side of a symmetric-looking mistake reported. **The `stream` perf + profile had never run**: `tests/perf/servers/server_stream.php` called `->send()` + with no argument against an arginfo requiring one, so the profile answered 500 + with `expects exactly 1 argument, 0 given` before measuring anything, and its + chunk loop buffered through the old `write()`. +- [~] **`tryWrite(): bool` and the dialect twins.** In #178, without the twins. The non-blocking half of the pair, matching `WebSocket::trySend()`; `trySseEvent()` and `tryWriteMessage()` follow, so the idiom is not half-applied. Invariant: false means nothing was queued and no header was committed, and a dead peer is still the 499 exception. Blocked by the compressing wrapper — `ws_append_chunk` feeds the encoder and closes a block before it consults the underlying ops, so a refusal there is not retryable, and the capacity check has to move ahead of the encoder. - `compressing_stream_ops` (`src/compression/http_compression_response.c:701`) has - no `sendable` slot either, so under compression the answer is a constant true. + Three review passes reshaped it. The refusal moved into `append_chunk` as a + `nonblocking` argument, because a predicate read beforehand cannot be atomic at + the PHP boundary; the wait moved into a `wait_writable` op, because each + transport's own wait carries a deadline, a wake source and a re-pump of the + drain that a wait assembled outside would drop; `awaitWritable()` answers false + rather than true where a transport can be full but offers no wait, since "go + ahead" spins a handler that trusts it. `compressing_stream_ops` and + `h3_stream_ops` gained the missing `sendable` slots — without them a refusal + under compression threw away a block the encoder had already emitted, and the + retry the caller was told to make corrupted the deflate stream. + + **HTTP/1 is the open exception**: it keeps no queue of its own, so it never + refuses and an accepted chunk waits for the socket. Two ways to close it were + tried and rejected — see below. The twins (`trySseEvent`, `tryWriteMessage`) + wait for that to settle. - [ ] **Framing by declared length.** A `Content-Length` set before the first `write()` reaches the client verbatim on every protocol, and the server becomes the auditor: excess throws at the offending write, a shortfall aborts the stream @@ -73,10 +102,60 @@ it and expects a tag within days. that passes a handler value today; H2, H3 and the worker strip it in `http_response_header_allowed_h2h3`. Needs the abort op from #171 — the vtable carries only the clean `mark_ended` (`include/php_http_server.h:723`). -- [ ] **Migration.** Seven BC entries in the CHANGELOG. laravel-spawn is a two-line - diff: `Sse::connected()` calls `isWritable()`, `send()` becomes `write()`. Its - docblock was corrected ahead of the rename in YanGusik/laravel-spawn#63, so the - wording stops teaching the loop that truncated #60 in the meantime. +- [~] **Migration.** The CHANGELOG entries are written (#180, five bullets covering + the seven renames, plus #181 under Fixed). What is left is laravel-spawn, and it + is five call sites rather than the two this plan assumed: `send()` → `write()` + at `src/Server/TrueAsyncServer.php:492`, `isClosed()` → `isEnded()` at 106, 395 + and 413, and `Sse::connected()` → `isWritable()` at `src/Sse/Sse.php:42`. To land + once a build carrying the renames is tagged. Its docblock was + corrected ahead of the rename in YanGusik/laravel-spawn#63, so the wording stops + teaching the loop that truncated #60 in the meantime. + +## HTTP/1 has no non-blocking write, and the two candidate fixes are both wrong + +`tryWrite()` cannot refuse on HTTP/1: the streaming path writes through +`http_connection_send` → `send_raw`, which submits a `uv_write` and awaits it, so +backpressure is the kernel socket buffer and there is no depth to read. Two +designs were worked out and both fail on something mechanical. + +- **A second writer for the non-blocking case** (`http_connection_send_batched`, + the one WebSocket uses). It is an unordered channel: a chunk body parked in + `out_pending_buf` waits behind an in-flight write while the headers, a blocking + `send()` and `mark_ended`'s terminal chunk go out through the raw path and reach + the peer first. Chunked framing does not survive that. +- **A queue on the response** (`http1_request_ctx_t`). Dead twice over: on TLS the + drain writer would run in scheduler context, where `tls_wait_space` refuses + outright (`src/core/http_connection_tls.c:97`), so it could not push a byte on + HTTPS; and the context is freed in `http_request_finalize`, while a queue drained + by write completions outlives the handler by definition. The precedents cited for + it — the H2 per-stream ring and the wslay FIFO — both live on connection-lifetime + objects, not on a per-request one. + +- [x] **Measure the HTTP/1 chunk path before deciding.** Taken on 2026-08-20, on a + release PHP built for it (`dev/BENCHMARKS.md`). `strace` confirms three `write(2)` + and three loop turns per chunk, flat in the chunk size. A one-hunk prototype that + copies the three pieces into one awaited write gains 42.7% at four chunks, 77.3% + at sixteen and 84.8% at sixty-four, cutting the per-chunk cost from ~18.5 µs to + ~8.8 µs. The copy is only there because the ABI has no awaitable vectored write: + `io_pipe_writev_cb` sends no NOTIFY and frees the request itself. + + What it decides: the win needs no queue, no second writer and no per-response + structure, so it is not an argument for #179. Next step is to land the coalesced + frame as its own change, with the copy or with a new ext/async op. + +- [ ] **Answer from the queues the connection already has.** Plaintext: + `out_pending_buf` carries a byte count, a high-water predicate on the same knob, + low-water hysteresis, a drain hook and a destroy defer gate — all implemented and + all exercised by WebSocket. TLS: `BIO_ctrl_get_write_guarantee` on the plaintext + BIO is the exact predicate `tls_wait_space` loops on, so a refusal built from it + is exact by construction. Neither needs a new structure. What it does need is the + out-of-band writers brought under one order first — `send_strv_owned` ignores the + pending tail, and `emit_parse_error` writes with a direct `send(2)` syscall. + Measure before deciding: the case for it is three submits and up to three + suspensions per chunk, and that number has never been taken. +- [ ] **#179 — one serialized outbound path per HTTP/1 connection.** The larger + version of the same idea. Filed, and to be judged against the measurement rather + than against the argument. ## The cmocka suite rots unnoticed diff --git a/docs/COMPRESSION.md b/docs/COMPRESSION.md index 4ad64339..d2e6c93b 100644 --- a/docs/COMPRESSION.md +++ b/docs/COMPRESSION.md @@ -188,11 +188,11 @@ codec to bypass the bomb cap. ## Streaming -When handlers stream via `$response->send($chunk)`, the encoder is +When handlers stream via `$response->write($chunk)`, the encoder is installed transparently on the first call (subject to negotiation). The wrapper accumulates compressed output across an entire encoder iteration and ships it as a single underlying chunk — one chunked-H1 -size line, one H2 DATA frame per `send()` call, regardless of how many +size line, one H2 DATA frame per `write()` call, regardless of how many internal inflate passes deflate needed. `mark_ended()` (called by `$response->end()`) drains the gzip trailer diff --git a/docs/USAGE.md b/docs/USAGE.md index eb98adb1..cfb333ff 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -134,6 +134,86 @@ returns or when `$res->end()` is called explicitly. --- +## 3.5. The response body + +A body is produced in one of three modes. The first body call fixes the mode +for that response, and a call belonging to another one throws +`HttpServerRuntimeException`. + +| Mode | Calls | Reaches the client | Framing | +|---|---|---|---| +| Buffered | `setBody()`, `appendBody()`, `json()`, `html()` | at `end()` | `Content-Length`, computed from the buffer | +| Streamed | `write()`, `sseEvent()`, `writeMessage()` | at each call | chunked encoding (HTTP/1) or DATA frames closed by END_STREAM (HTTP/2, HTTP/3) | +| File | `sendFile()` | after the handler returns | `Content-Length` from the file; a satisfiable `Range` yields `206` with `Content-Range` | + +### Buffered + +```php +$res->setBody('one ') // replaces the buffer + ->appendBody('two') // appends to it + ->setHeader('Content-Type', 'text/plain') + ->end(); // 7 bytes leave here, Content-Length: 7 +``` + +Nothing is committed until `end()`, so status and headers stay writable for as +long as the handler runs, and `getBody()` returns what has accumulated. This is +the mode to use when the body fits in memory: one write syscall, and the +compressor sees the whole payload at once. + +### Streamed + +```php +$res->setStatusCode(200)->setHeader('Content-Type', 'text/plain'); +foreach ($rows as $row) { + $res->write(format($row)); // the first call commits status + headers +} +$res->end(); +``` + +The first `write()` commits the status line and the headers; from then on +`setStatusCode()`, `setHeader()` and `setBody()` throw, and `isHeadersSent()` +answers true. A `Content-Length` the handler set beforehand is dropped and the +response is framed by chunked encoding — honouring a declared length is issue +[#171](https://github.com/true-async/server/issues/171)'s successor, not +today's behaviour. + +`write()` parks the handler coroutine while the outbound queue is full: HTTP/2 +and HTTP/3 park once every ring slot is live or the queued bytes reach +`HttpServerConfig::setStreamWriteBufferBytes()` (256 KiB by default), HTTP/1 +parks on the socket write itself. Three calls cover the cases where parking is +the wrong answer: + +- `tryWrite($chunk)` queues the chunk or answers false, having queued nothing — + the same chunk can be offered again. HTTP/1 keeps no queue of its own, so it + never refuses and an accepted chunk still waits for the socket. +- `awaitWritable($timeoutMs = null)` waits for the room `tryWrite()` refused. + It answers false where a transport can be full and offers no wait, since + "go ahead" would spin a handler that trusts it. +- `isWritable()` reports whether output is still possible at all: `end()` not + called, no `sendFile()` seal, peer still there. A false answer is final, + which is what makes it the right condition for leaving a streaming loop. + +A peer that departs mid-stream arrives as `HttpException` with code 499 out of +the next call, so a `try`/`catch` around the loop is how a handler winds down. +`isEnded()` reports the response, not the connection: it stays false until the +handler calls `end()`. + +`send()` is the previous spelling of `write()` and still works, one minor +release long. + +### File + +```php +$res->sendFile('/srv/assets/app.js'); +``` + +`sendFile()` seals the response and returns at once; the file is delivered +after the handler returns. Every mutating call afterwards throws, including a +second `sendFile()`. Options — cache headers, download disposition, precomputed +compressed variants — go in a `SendFileOptions` passed as the second argument. + +--- + ## 4. TLS Once any listener has `tls: true` (or any HTTP/3 listener exists at all), diff --git a/examples/sse-server.php b/examples/sse-server.php index 7b921d8b..34cef4c0 100644 --- a/examples/sse-server.php +++ b/examples/sse-server.php @@ -44,9 +44,9 @@ id: (string) $i, ); - // sendable() is an advisory backpressure check — skip the sleep and - // bail early if the peer has gone away. - if (!$res->sendable()) { + // Stop when the peer has gone. sseEvent() waits for room on its own, + // so a full queue is not a reason to leave the loop. + if (!$res->isWritable()) { break; } diff --git a/ide-stubs/true-async-server.php b/ide-stubs/true-async-server.php index 91691bd4..f5d8564d 100644 --- a/ide-stubs/true-async-server.php +++ b/ide-stubs/true-async-server.php @@ -900,9 +900,9 @@ public function getDrainCooldownMs(): int {} // === Streaming responses === /** - * Per-stream chunk-queue cap for HttpResponse::send() backpressure. + * Per-stream chunk-queue cap for HttpResponse::write() backpressure. * - * When handler's send() call grows the stream's chunk queue past + * When the handler's write() call grows the stream's chunk queue past * this many bytes, the coroutine suspends until nghttp2 drains * enough to drop below. HTTP/2 only; HTTP/1 chunked path uses * the kernel send buffer instead. @@ -2273,46 +2273,61 @@ public function getProtocolVersion(): string {} // === Body methods === /** - * Write data to response body buffer. + * Stream a chunk to the client. * - * @param string $data Data to write - * @return static + * The first call commits status and headers; afterwards setStatusCode(), + * setHeader() and setBody() throw. Later calls append chunked-transfer + * segments (HTTP/1) or DATA frames (HTTP/2, HTTP/3). To append to a + * buffered body instead, call appendBody(). + * + * Parks the handler coroutine only under backpressure: HTTP/2 and HTTP/3 + * park while every ring slot is live or the queued bytes stand at + * HttpServerConfig::setStreamWriteBufferBytes (256 KiB by default), + * HTTP/1 parks on the socket write. tryWrite() offers a chunk without + * committing to that wait. A peer that has gone throws HttpException 499. */ - public function write(string $data): static {} + public function write(string $chunk): static {} /** - * Send a chunk to the client (streaming response). + * Removed. One bool answered four questions, and a loop that read it as + * liveness stopped streams that were merely slow. * - * First call commits status + headers (they can no longer be - * changed). Subsequent calls append DATA frames (HTTP/2) or - * chunked-transfer segments (HTTP/1). + * Ask the two questions separately: isWritable() reports whether output is + * still possible, tryWrite() and awaitWritable() report whether the + * outbound queue has room. * - * Blocks the handler coroutine ONLY under backpressure — when the - * per-stream staging buffer is full (HTTP/2: all ring slots live - * OR queued bytes reach HttpServerConfig::setStreamWriteBufferBytes, - * default 256 KiB). Otherwise returns immediately. send() is always - * safe to call; use sendable() to check first if you'd rather do - * other work than block. + * The declaration stays for one minor release so a call names its + * replacements instead of failing as an undefined method. * - * @param string $chunk - * @return static + * @throws HttpServerRuntimeException always */ - public function send(string $chunk): static {} + public function sendable(): bool {} /** - * Advisory, non-blocking backpressure check for streaming responses. - * - * Returns true when send() would accept a chunk without suspending - * the handler coroutine — the per-stream staging buffer has room. - * Returns false when send() would block on backpressure, or when the - * response is closed / sealed by sendFile() / not streaming-capable. - * - * send() is always safe to call regardless; sendable() just lets a - * handler do other work instead of blocking on a slow peer. + * Offer a chunk without waiting for room: false means the outbound queue + * had no room and nothing was queued, so the same chunk can be offered + * again later. A client that has gone throws HttpException 499 instead of + * answering false, because "wait" and "stop" need opposite reactions. * - * @return bool + * HTTP/1 keeps no queue of its own, so it never refuses and an accepted + * chunk waits on the socket for as long as a blocking write() would. */ - public function sendable(): bool {} + public function tryWrite(string $chunk): bool {} + + /** + * Wait until the outbound queue has room again, and report whether it has. + * True at once on a transport with no queue; false without waiting on one + * that cannot be waited on. A timeout or a cancellation arrives as an + * exception. + */ + public function awaitWritable(?int $timeoutMs = null): bool {} + + /** + * True while output is still possible: end() was not called, the response + * is not sealed by sendFile(), and the client has not gone. A false answer + * is final, unlike the queue depth tryWrite() reports. + */ + public function isWritable(): bool {} // === Server-Sent Events === @@ -2408,7 +2423,7 @@ public function setGrpcEncoding(string $encoding): static {} /** * Frame and stream one gRPC message: the 5-byte length prefix is prepended - * for you. The first call activates streaming, exactly as send() does — so + * for you. The first call activates streaming, exactly as write() does — so * call it once for a unary reply and repeatedly for server-streaming. * * Pass already-protobuf-encoded bytes. The grpc-status travels separately, on @@ -2437,26 +2452,17 @@ public function getBody(): string {} /** * Set body content (replaces buffer). - * - * @param string $body Body content - * @return static */ public function setBody(string $body): static {} /** - * Get body stream. + * Append to the buffered response body. * - * @return mixed Stream resource or null + * Nothing reaches the client here: the whole body goes out on end(), with + * Content-Length computed from it. Call write() to stream instead — that + * is the call which commits headers and applies backpressure. */ - public function getBodyStream(): mixed {} - - /** - * Set body stream. - * - * @param mixed $stream Stream resource - * @return static - */ - public function setBodyStream(mixed $stream): static {} + public function appendBody(string $data): static {} // === Helper methods === @@ -2547,9 +2553,13 @@ public function sendFile(string $path, ?SendFileOptions $options = null): void { public function isHeadersSent(): bool {} /** - * Check if response is closed. + * True once end() has been called. + * + * Reports the response, not the connection: a peer that has gone leaves + * this false until the handler ends the response. Use isWritable() for + * liveness. */ - public function isClosed(): bool {} + public function isEnded(): bool {} } // --------------------------------------------------------------------------- diff --git a/include/http2/http2_session.h b/include/http2/http2_session.h index 06e473ec..2df4d602 100644 --- a/include/http2/http2_session.h +++ b/include/http2/http2_session.h @@ -246,7 +246,7 @@ int http2_session_submit_response(http2_session_t *session, /* Submit a streaming response: HEADERS go on the wire immediately, * but the DATA source is the stream's chunk_queue (populated by - * `HttpResponse::send()`). The data + * `HttpResponse::write()`). The data * provider returns NGHTTP2_ERR_DEFERRED whenever the queue is * transiently empty; caller must call * `nghttp2_session_resume_data(stream_id)` after each queue append diff --git a/include/http2/http2_stream.h b/include/http2/http2_stream.h index d05de2b9..56388db1 100644 --- a/include/http2/http2_stream.h +++ b/include/http2/http2_stream.h @@ -83,14 +83,14 @@ struct http2_stream_t { /* Streaming-response chunk queue. * - * Active only when the handler called HttpResponse::send(); a + * Active only when the handler called HttpResponse::write(); a * plain setBody() handler leaves these NULL and uses the legacy * response_body pointer path above. * * Grow-only ring-ish queue: chunks are appended at tail, drained * from head. We never shrink the array — steady-state traffic * reaches a plateau. Each slot holds a refcount'ed zend_string - * handed over from send()'s zval; released once fully drained. */ + * handed over from write()'s zval; released once fully drained. */ zend_string **chunk_queue; size_t chunk_queue_cap; size_t chunk_queue_head; /* next chunk to drain from */ diff --git a/include/http3/http3_stream.h b/include/http3/http3_stream.h index 36f1fbfc..cecf16c7 100644 --- a/include/http3/http3_stream.h +++ b/include/http3/http3_stream.h @@ -75,12 +75,12 @@ struct _http3_stream_s { * on http3_stream_release. Set in http3_stream_submit_response * from http_response_get_body; mutually exclusive with the * streaming chunk queue below — the data_reader picks one or the - * other depending on whether HttpResponse::send() was called. */ + * other depending on whether HttpResponse::write() was called. */ zend_string *response_body; size_t response_body_offset; /* Streaming response chunk queue. - * Active only when the handler called HttpResponse::send(); a plain + * Active only when the handler called HttpResponse::write(); a plain * setBody() handler leaves these NULL and uses response_body above. * * Three positions instead of H2's two — nghttp3 keeps iov pointers diff --git a/include/php_http_server.h b/include/php_http_server.h index 56cfd4b4..5e80419f 100644 --- a/include/php_http_server.h +++ b/include/php_http_server.h @@ -192,7 +192,7 @@ struct _http_server_config_t { uint32_t drain_cooldown_ms; /* HTTP/2 streaming response per-stream queue cap. - * When handler's chunk queue exceeds this, send() suspends + * When handler's chunk queue exceeds this, write() suspends * the coroutine until drain brings it back under. HTTP/1 chunked * path ignores this — the kernel send buffer IS the queue. */ uint32_t stream_write_buffer_bytes; @@ -689,7 +689,7 @@ void http_response_set_alt_svc_if_unset(zend_object *obj, /* * Streaming response — binary interface (vtable) that protocol - * strategies install on an HttpResponse object so HttpResponse::send() + * strategies install on an HttpResponse object so HttpResponse::write() * can route chunks without either side seeing the other's layout. * * HTTP/2 + HTTP/3 plug in stream-aware impls at dispatch time; HTTP/1 @@ -706,9 +706,16 @@ typedef struct http_response_stream_ops_t http_response_stream_ops_t; struct http_response_stream_ops_t { /* Append a chunk (caller already bumped its refcount). Returns * one of http_stream_append_result_t. The op itself knows the - * threshold (it lives in the context), so send() doesn't need - * to see server config. */ - int (*append_chunk)(void *ctx, zend_string *chunk); + * threshold (it lives in the context), so write() doesn't need + * to see server config. + * + * `nonblocking` forbids suspending the calling coroutine: a transport + * that would have parked returns HTTP_STREAM_APPEND_BACKPRESSURE + * INSTEAD, having queued nothing and committed nothing, so the caller + * may offer the same chunk again. Deciding inside the op is what makes + * that atomic — a predicate consulted beforehand answers about a moment + * that has already passed. */ + int (*append_chunk)(void *ctx, zend_string *chunk, bool nonblocking); /* Advisory, non-blocking: true when append_chunk would accept a * chunk without suspending the handler (the per-stream staging @@ -716,6 +723,12 @@ struct http_response_stream_ops_t { * protocols without a userspace staging ring (HTTP/1, paced by the * kernel socket buffer) leave it NULL and sendable() reports true. */ bool (*sendable)(void *ctx); + /* REQUIRED of any op whose append_chunk can answer + * HTTP_STREAM_APPEND_BACKPRESSURE: the compressing wrapper reads it to + * decide whether it may feed the encoder, and an encoder cannot be + * un-fed. A NULL slot therefore promises "this transport never refuses", + * and a transport that refuses anyway loses the bytes of a flushed + * block on every refusal. */ /* True while output is still possible: the peer has not gone and the * transport can still carry bytes. Every input is a one-way latch, so a @@ -730,6 +743,15 @@ struct http_response_stream_ops_t { * Idempotent. */ void (*mark_ended)(void *ctx); + /* Wait until append_chunk would accept a chunk, and report whether it + * would. `timeout_ms` of 0 means the transport's own write deadline. + * Each transport keeps what its internal wait already does — its + * deadline, its re-pump of the drain, its wake source — which a wait + * assembled at the PHP boundary would drop. MAY be NULL: the caller + * then falls back to get_wait_event, and a transport with neither + * cannot be waited on at all. */ + bool (*wait_writable)(void *ctx, uint32_t timeout_ms); + /* Lazily-created trigger event the handler awaits on under * backpressure. Fired by the drain side when the queue drops * below threshold. Returns NULL only on alloc failure — callers @@ -781,7 +803,7 @@ struct http_response_stream_ops_t { }; /* Install the streaming vtable + ctx on a response object. The - * protocol strategy calls this once at dispatch; send() reads it. */ + * protocol strategy calls this once at dispatch; write() reads it. */ void http_response_install_stream_ops(zend_object *response_obj, const http_response_stream_ops_t *ops, void *ctx); @@ -1349,7 +1371,7 @@ size_t http_sockaddr_ip(const struct sockaddr *addr, socklen_t addr_len, char *out, size_t out_len); uint16_t http_sockaddr_port(const struct sockaddr *addr, socklen_t addr_len); void http_response_set_protocol_version(zend_object *obj, const char *version); -/* RFC 9110 §9.3.2 — HEAD responses must not carry a body; send() drops +/* RFC 9110 §9.3.2 — HEAD responses must not carry a body; write() drops * chunks silently when set. Stamped at dispatch wherever the request is * known. */ void http_response_set_head(zend_object *obj, bool is_head); @@ -1454,7 +1476,7 @@ const char *http_response_status_line_http11(int code, size_t *out_len); * http_connection.c, HTTP/2 in src/http2/http2_strategy.c). */ bool http_response_is_committed (zend_object *obj); void http_response_set_committed (zend_object *obj); -bool http_response_is_streaming (zend_object *obj); /* send() activated streaming */ +bool http_response_is_streaming (zend_object *obj); /* write() activated streaming */ void http_response_reset_to_error (zend_object *obj, int status_code, const char *message); diff --git a/src/compression/http_compression_response.c b/src/compression/http_compression_response.c index e553fc30..5e422cd7 100644 --- a/src/compression/http_compression_response.c +++ b/src/compression/http_compression_response.c @@ -14,7 +14,7 @@ * smart_str body, mutates headers in place. The * buffered path knows the body length up-front, so * the size-threshold check is exact. - * - stream wrapper : on first send() we substitute the installed + * - stream wrapper : on first write() we substitute the installed * stream_ops with a compressing one. The wrapper's * append_chunk feeds each chunk through the encoder, * closes the block with flush() so the client can @@ -556,14 +556,24 @@ typedef struct { * wire. Compared with emitting per-loop slices, this trades a small * temporary buffer for fewer protocol-level frames (H2 DATA / chunked * size-line). zs is consumed; the underlying owns the refcount. */ -static int forward_compressed(ws_ctx_t *w, zend_string *zs) +static int forward_compressed(ws_ctx_t *w, zend_string *zs, const bool nonblocking) { if (UNEXPECTED(zs == NULL || ZSTR_LEN(zs) == 0)) { if (zs) zend_string_release(zs); return HTTP_STREAM_APPEND_OK; } - return w->underlying_ops->append_chunk(w->underlying_ctx, zs); + const int rc = w->underlying_ops->append_chunk(w->underlying_ctx, zs, nonblocking); + + /* By now the encoder has eaten the chunk and closed a block, so a refusal + * is not retryable: the same plaintext offered again would be deflated + * against a window the decoder never saw. A truncated body with a 499 is + * recoverable; a corrupted stream is not. */ + if (UNEXPECTED(rc == HTTP_STREAM_APPEND_BACKPRESSURE)) { + return HTTP_STREAM_APPEND_STREAM_DEAD; + } + + return rc; } /* An encoder that answered HTTP_ENC_ERROR is left mid-block and cannot @@ -585,19 +595,29 @@ static void drop_faulted_encoder(ws_ctx_t *w) w->encoder = NULL; } -static int ws_append_chunk(void *ctx_opaque, zend_string *chunk) +static int ws_append_chunk(void *ctx_opaque, zend_string *chunk, + const bool nonblocking) { ws_ctx_t *w = (ws_ctx_t *)ctx_opaque; /* An earlier chunk faulted the encoder and dropped it. The stream * cannot be resumed mid-block, so a handler that caught the 499 and - * called send() again gets the same refusal rather than a NULL + * called write() again gets the same refusal rather than a NULL * encoder handed to encoder_drain_write. */ if (UNEXPECTED(w->encoder == NULL)) { zend_string_release(chunk); return HTTP_STREAM_APPEND_STREAM_DEAD; } + /* Asked before the encoder is fed: the encoder cannot be un-fed, and a + * closed block would leave the stream one boundary ahead of what the + * transport actually took. */ + if (nonblocking && w->underlying_ops->sendable != NULL + && !w->underlying_ops->sendable(w->underlying_ctx)) { + zend_string_release(chunk); + return HTTP_STREAM_APPEND_BACKPRESSURE; + } + if (UNEXPECTED(!w->first_chunk_done)) { /* Header mutation deferred to first chunk: by now the handler * has finalised setHeader/setStatusCode (committed=true was set @@ -622,7 +642,7 @@ static int ws_append_chunk(void *ctx_opaque, zend_string *chunk) smart_str_alloc(&out, in_len + 32, 0); /* Encode, then close the block so the client decodes this chunk now - * rather than at end of stream: handing a chunk to send() is the + * rather than at end of stream: handing a chunk to write() is the * handler stating that this much is ready to go. An empty chunk * skips the flush — a block boundary with no payload behind it * costs bytes and tells the client nothing. */ @@ -645,7 +665,7 @@ static int ws_append_chunk(void *ctx_opaque, zend_string *chunk) } smart_str_0(&out); - return forward_compressed(w, out.s); /* transfers ownership */ + return forward_compressed(w, out.s, nonblocking); /* transfers ownership */ } static void ws_mark_ended(void *ctx_opaque) @@ -676,7 +696,7 @@ static void ws_mark_ended(void *ctx_opaque) if (out.s != NULL && ZSTR_LEN(out.s) > 0) { smart_str_0(&out); - (void)forward_compressed(w, out.s); /* transfers ownership */ + (void)forward_compressed(w, out.s, false); /* transfers ownership */ } else { smart_str_free(&out); } @@ -695,11 +715,48 @@ static void ws_mark_ended(void *ctx_opaque) static zend_async_event_t *ws_get_wait_event(void *ctx_opaque) { ws_ctx_t *w = (ws_ctx_t *)ctx_opaque; + + if (w->underlying_ops->get_wait_event == NULL) { + return NULL; + } + return w->underlying_ops->get_wait_event(w->underlying_ctx); } +static bool ws_wait_writable(void *ctx_opaque, const uint32_t timeout_ms) +{ + ws_ctx_t *w = (ws_ctx_t *)ctx_opaque; + + if (w->underlying_ops->wait_writable == NULL) { + return true; + } + + return w->underlying_ops->wait_writable(w->underlying_ctx, timeout_ms); +} + +/* The wrapper holds no queue of its own, so both answers come from the + * transport underneath rather than from the encoder. */ +static bool ws_sendable(void *ctx_opaque) +{ + const ws_ctx_t *w = (const ws_ctx_t *)ctx_opaque; + + return w->underlying_ops->sendable == NULL + || w->underlying_ops->sendable(w->underlying_ctx); +} + +static bool ws_is_alive(void *ctx_opaque) +{ + const ws_ctx_t *w = (const ws_ctx_t *)ctx_opaque; + + return w->underlying_ops->is_alive == NULL + || w->underlying_ops->is_alive(w->underlying_ctx); +} + static const http_response_stream_ops_t compressing_stream_ops = { .append_chunk = ws_append_chunk, + .sendable = ws_sendable, + .is_alive = ws_is_alive, + .wait_writable = ws_wait_writable, .mark_ended = ws_mark_ended, .get_wait_event = ws_get_wait_event, }; diff --git a/src/core/worker_dispatch.c b/src/core/worker_dispatch.c index 114ca4ad..ac6ae69e 100644 --- a/src/core/worker_dispatch.c +++ b/src/core/worker_dispatch.c @@ -313,7 +313,8 @@ static bool worker_stream_wait_credit(worker_dispatch_ctx_t *ctx) return true; } -static int worker_stream_append_chunk(void *vctx, zend_string *chunk) +static int worker_stream_append_chunk(void *vctx, zend_string *chunk, + const bool nonblocking) { worker_dispatch_ctx_t *const ctx = (worker_dispatch_ctx_t *)vctx; @@ -323,7 +324,18 @@ static int worker_stream_append_chunk(void *vctx, zend_string *chunk) return HTTP_STREAM_APPEND_STREAM_DEAD; } - /* first send(): open the stream; the reactor adopts one credit ref */ + /* Refused on the depth already in flight, letting this chunk overshoot the + * cap — the rule H2 applies too. Counting the candidate's length instead + * would refuse a chunk larger than the cap for ever, whatever the peer + * did, and the caller would spin on it. */ + if (nonblocking && ctx->credit != NULL + && ctx->posted_bytes - stream_credit_acked(ctx->credit) + >= WORKER_STREAM_INFLIGHT_CAP) { + zend_string_release(chunk); + return HTTP_STREAM_APPEND_BACKPRESSURE; + } + + /* first write(): open the stream; the reactor adopts one credit ref */ if (!ctx->stream_started) { response_wire_t *const hw = response_wire_create(ctx->reactor_id, ctx->stream_id, ctx->conn); @@ -366,11 +378,22 @@ static int worker_stream_append_chunk(void *vctx, zend_string *chunk) const size_t chunk_len = ZSTR_LEN(chunk); - worker_wire_post(ctx, cw); + /* A refused wire is a dropped chunk: the sink exhausted its retries and + * worker_wire_post has already marked the stream failed. Reporting OK here + * would tell the handler it wrote bytes the peer will never see. */ + if (UNEXPECTED(!worker_wire_post(ctx, cw))) { + zend_string_release(chunk); + return HTTP_STREAM_APPEND_STREAM_DEAD; + } + zend_string_release(chunk); /* bytes copied into the wire arena */ ctx->posted_bytes += chunk_len; + if (nonblocking) { + return HTTP_STREAM_APPEND_OK; /* room was checked above; never parks */ + } + if (!worker_stream_wait_credit(ctx)) { ctx->stream_failed = true; /* credit timeout / cancelled while parked */ return HTTP_STREAM_APPEND_STREAM_DEAD; @@ -437,12 +460,24 @@ static void worker_stream_mark_ended(void *vctx) worker_wire_post(ctx, ew); } +/* The credit wait the blocking path takes, offered to a non-blocking caller + * that asked to be told when room comes back. */ +static bool worker_stream_wait_writable(void *vctx, const uint32_t timeout_ms) +{ + worker_dispatch_ctx_t *const ctx = (worker_dispatch_ctx_t *)vctx; + + (void)timeout_ms; /* the credit wait uses the configured write deadline */ + + return worker_stream_wait_credit(ctx); +} + static const http_response_stream_ops_t worker_stream_ops = { .append_chunk = worker_stream_append_chunk, .sendable = worker_stream_sendable, .is_alive = worker_stream_is_alive, + .wait_writable = worker_stream_wait_writable, .mark_ended = worker_stream_mark_ended, - .get_wait_event = NULL, /* backpressure parks inside append_chunk */ + .get_wait_event = NULL, /* the wait above is the one to take */ }; /* grpc-web in-body trailer frame; consumes the ref. */ @@ -452,7 +487,7 @@ static void worker_grpc_append_frame_and_end(void *vctx, zend_string *frame) if (http_response_is_streaming(Z_OBJ(ctx->response_zv))) { /* append_chunk consumes the ref (success or failure). */ - if (worker_stream_append_chunk(ctx, frame) == HTTP_STREAM_APPEND_OK) { + if (worker_stream_append_chunk(ctx, frame, false) == HTTP_STREAM_APPEND_OK) { worker_stream_mark_ended(ctx); } @@ -678,7 +713,7 @@ static void worker_dispatch_dispose(zend_coroutine_t *coroutine) http_server_get_log_state(ctx->server)); } - /* ctx dies below; a late send() on a kept $response must throw, not UAF */ + /* ctx dies below; a late write() on a kept $response must throw, not UAF */ http_response_replace_stream_ops(resp, NULL, NULL); } diff --git a/src/http1/http1_format.c b/src/http1/http1_format.c index 910440a3..e85b02ca 100644 --- a/src/http1/http1_format.c +++ b/src/http1/http1_format.c @@ -361,7 +361,7 @@ zend_string *http_response_format(zend_object *obj) * `Transfer-Encoding: chunked` in its place. Headers end with the * empty line; the caller writes the body as a sequence of chunks. * - * Used by h1_stream_ops at first send(). Separate from http_response_format + * Used by h1_stream_ops at first write(). Separate from http_response_format * because the latter builds status + Content-Length + headers + body * as a single atomic payload, which is exactly what chunked avoids. */ zend_string *http_response_format_streaming_headers(zend_object *obj) diff --git a/src/http1/http1_stream.c b/src/http1/http1_stream.c index f89ebcf1..0ed09aea 100644 --- a/src/http1/http1_stream.c +++ b/src/http1/http1_stream.c @@ -80,8 +80,16 @@ static bool h1_emit_headers_once(http1_request_ctx_t *ctx) return ok; } -static int h1_stream_append_chunk(void *opaque, zend_string *chunk) +/* `nonblocking` is accepted and ignored: HTTP/1 keeps no queue of its own, so + * there is no depth to refuse from. Backpressure here belongs to the kernel + * socket buffer, and the only way to learn of it is to write and wait. Issue + * #179 gives the connection one outbound queue; a refusal becomes possible + * then, and this signature is already the one it will use. */ +static int h1_stream_append_chunk(void *opaque, zend_string *chunk, + const bool nonblocking) { + (void)nonblocking; + http1_request_ctx_t *ctx = (http1_request_ctx_t *)opaque; if (ctx == NULL || ctx->conn == NULL) { @@ -101,9 +109,9 @@ static int h1_stream_append_chunk(void *opaque, zend_string *chunk) return HTTP_STREAM_APPEND_STREAM_DEAD; } - /* First send() — commit status + headers with chunked framing. + /* First write() — commit status + headers with chunked framing. * We track wire-commit on ctx->h1_stream_headers_sent rather than - * response->committed because send() sets committed=true before + * response->committed because write() sets committed=true before * calling us (committed means "no more setHeader / setStatusCode * allowed", which happens at the PHP boundary, not on the wire). */ if (!ctx->h1_stream_headers_sent) { @@ -151,6 +159,16 @@ static int h1_stream_append_chunk(void *opaque, zend_string *chunk) return HTTP_STREAM_APPEND_STREAM_DEAD; } + /* Each of those writes suspends. A cancellation that lands between them + * returns success for the writes already done, leaving the frame partly + * written — the same state a failure leaves, and it is recorded the same + * way so mark_ended does not seal it. */ + if (UNEXPECTED(EG(exception) != NULL)) { + ctx->stream_dead = true; + zend_string_release(chunk); + return HTTP_STREAM_APPEND_STREAM_DEAD; + } + zend_string_release(chunk); http_server_on_stream_send(conn->counters, chunk_len); @@ -169,7 +187,7 @@ static void h1_stream_mark_ended(void *opaque) http_connection_t *conn = ctx->conn; - /* If send() was never called but mark_ended fires anyway (rare: + /* If write() was never called but mark_ended fires anyway (rare: * handler flipped streaming mode then immediately closed), we * still need to commit the headers so the peer isn't left * waiting for a response that never starts. */ @@ -181,6 +199,17 @@ static void h1_stream_mark_ended(void *opaque) ctx->h1_stream_headers_sent = true; } + /* A chunk is three writes — size line, body, CRLF — with a suspension + * between them, so a cancellation can leave a frame half on the wire. + * Sealing that with a terminal chunk would tell the peer the body ended + * cleanly and hand the connection on for reuse, and it would read the + * terminator as the first bytes of the chunk the size line promised. + * Refuse both: no terminator, no keep-alive. */ + if (UNEXPECTED(ctx->stream_dead)) { + conn->keep_alive = false; + return; + } + /* Terminal zero-chunk. Trailers not emitted — RFC requires the * client to opt in via TE: trailers, and the chunked-push path * doesn't surface a trailer API yet. */ @@ -189,7 +218,7 @@ static void h1_stream_mark_ended(void *opaque) /* HTTP/1 push streaming has no internal queue — kernel backpressure * suspends directly inside http_connection_send — so there's nothing - * for the handler to await on. Returning NULL signals to the send() + * for the handler to await on. Returning NULL signals to the write() * implementation that the wait-event path doesn't apply. */ static zend_async_event_t *h1_stream_get_wait_event(void *ctx) { diff --git a/src/http2/http2_session.c b/src/http2/http2_session.c index 07780895..aff04e8a 100644 --- a/src/http2/http2_session.c +++ b/src/http2/http2_session.c @@ -1640,7 +1640,7 @@ static ssize_t h2_dp_streaming_copy(http2_stream_t *stream, /* nghttp2 data provider. Two body sources: buffered (response_body * pointer+length, zero-copy) or streaming (chunk_queue of refcounted * zend_strings). Empty streaming queue returns NGHTTP2_ERR_DEFERRED; - * resume fires from the next send()/end() via resume_stream_data. */ + * resume fires from the next write()/end() via resume_stream_data. */ static ssize_t http2_response_data_read(nghttp2_session *ng, const int32_t stream_id, uint8_t *buf, diff --git a/src/http2/http2_strategy.c b/src/http2/http2_strategy.c index ea36bb04..5ed27e5e 100644 --- a/src/http2/http2_strategy.c +++ b/src/http2/http2_strategy.c @@ -102,7 +102,8 @@ extern const http_response_stream_ops_t h2_stream_ops; * handler skipped $res->end(). Defined further down. */ static void h2_stream_mark_ended(void *ctx); -static int h2_stream_append_chunk(void *ctx, zend_string *chunk); +static int h2_stream_append_chunk(void *ctx, zend_string *chunk, bool nonblocking); +static bool h2_stream_sendable(void *ctx); static void h2_grpc_append_frame_and_end(void *ctx, zend_string *frame); static void h2_grpc_commit(void *ctx); @@ -278,9 +279,9 @@ static void http2_strategy_dispatch(struct http_request_t *request, http_response_set_head(Z_OBJ(stream->response_zv), http_request_method_is_head(stream->request)); - /* Let HttpResponse::send() reach this stream's chunk queue via + /* Let HttpResponse::write() reach this stream's chunk queue via * the vtable. Ops installed once at dispatch; - * streaming mode is a handler opt-in (only activated when send() + * streaming mode is a handler opt-in (only activated when write() * is actually called). */ http_response_install_stream_ops(Z_OBJ(stream->response_zv), &h2_stream_ops, stream); @@ -1113,7 +1114,7 @@ static bool http2_commit_stream_response(http_connection_t *conn, } /* ------------------------------------------------------------------------- - * Streaming response — vtable exported for HttpResponse::send(). + * Streaming response — vtable exported for HttpResponse::write(). * All three ops take the http2_stream_t* ctx that dispatch stashed * into the PHP response object. They rely on the stream's * http2_session + owning connection staying alive for as long as the @@ -1667,7 +1668,8 @@ static bool h2_stream_wait_for_room(http2_stream_t *stream, } } -static int h2_stream_append_chunk(void *ctx, zend_string *chunk) +static int h2_stream_append_chunk(void *ctx, zend_string *chunk, + const bool nonblocking) { http2_stream_t *stream = (http2_stream_t *)ctx; http_connection_t *conn = http2_session_get_conn(stream->session); @@ -1687,6 +1689,13 @@ static int h2_stream_append_chunk(void *ctx, zend_string *chunk) ? http_server_get_stream_write_buffer_bytes(conn->server) : 0; + /* A non-blocking caller gets the refusal the wait would have hidden; + * nothing is queued, so the same chunk may be offered again. */ + if (nonblocking && !h2_stream_sendable(stream)) { + zend_string_release(chunk); + return HTTP_STREAM_APPEND_BACKPRESSURE; + } + if (!h2_stream_wait_for_room(stream, conn, max_bytes)) { zend_string_release(chunk); return HTTP_STREAM_APPEND_STREAM_DEAD; @@ -1715,7 +1724,7 @@ static void h2_grpc_append_frame_and_end(void *ctx, zend_string *frame) return; } - (void)h2_stream_append_chunk(stream, frame); /* consumes the ref */ + (void)h2_stream_append_chunk(stream, frame, false); /* consumes the ref */ h2_stream_mark_ended(stream); } @@ -1781,7 +1790,7 @@ static bool h2_stream_sendable(void *ctx) http2_stream_t *stream = (http2_stream_t *)ctx; if (stream->chunk_queue == NULL) { - return true; /* not started — first send() always proceeds */ + return true; /* not started — first write() always proceeds */ } if (stream->chunk_queue_tail - stream->chunk_queue_head @@ -1812,10 +1821,32 @@ static bool h2_stream_is_alive(void *ctx) return conn != NULL && !conn->write_timed_out; } +/* The same loop append_chunk takes when the ring is full: it re-pumps the + * session on each wake, which a bare park on the drain event would not. */ +static bool h2_stream_wait_writable(void *ctx, const uint32_t timeout_ms) +{ + http2_stream_t *stream = (http2_stream_t *)ctx; + + (void)timeout_ms; /* the drain wait uses conn->write_timeout_ms */ + + http_connection_t *conn = http2_session_get_conn(stream->session); + + if (conn == NULL) { + return false; + } + + const uint32_t max_bytes = conn->server != NULL + ? http_server_get_stream_write_buffer_bytes(conn->server) + : 0; + + return h2_stream_wait_for_room(stream, conn, max_bytes); +} + const http_response_stream_ops_t h2_stream_ops = { .append_chunk = h2_stream_append_chunk, .sendable = h2_stream_sendable, .is_alive = h2_stream_is_alive, + .wait_writable = h2_stream_wait_writable, .mark_ended = h2_stream_mark_ended, .get_wait_event = h2_stream_get_wait_event, .send_static_response = h2_stream_send_static_response, @@ -1850,7 +1881,7 @@ static bool ws_h2_send(void *ctx, const uint8_t *data, size_t len) /* append_chunk takes ownership of the zend_string and suspends the * producer coroutine for backpressure when the ring is full. */ zend_string *z = zend_string_init((const char *)data, len, 0); - return h2_stream_append_chunk(stream, z) == HTTP_STREAM_APPEND_OK; + return h2_stream_append_chunk(stream, z, false) == HTTP_STREAM_APPEND_OK; } static bool ws_h2_send_internal(void *ctx, const uint8_t *data, size_t len) diff --git a/src/http3/http3_callbacks.c b/src/http3/http3_callbacks.c index 315f9d6f..7191e87a 100644 --- a/src/http3/http3_callbacks.c +++ b/src/http3/http3_callbacks.c @@ -876,7 +876,7 @@ bool http3_stream_submit_response(http3_connection_t *c, * runs here too — must precede the headers-flatten loop so the * mutated Content-Encoding/Vary land in the HEADERS frame. The * streaming path (`streaming==true`) is handled by the stream - * wrapper installed at first send(); the apply call is a cheap + * wrapper installed at first write(); the apply call is a cheap * no-op there. */ { extern void http_compression_apply_buffered(zend_object *); @@ -1204,7 +1204,7 @@ void h3_chunk_queue_push(http3_stream_t *s, zend_string *chunk) s->chunk_pending_bytes += ZSTR_LEN(chunk); } -int h3_stream_append_chunk(void *ctx, zend_string *chunk) +int h3_stream_append_chunk(void *ctx, zend_string *chunk, const bool nonblocking) { http3_stream_t *const s = (http3_stream_t *)ctx; @@ -1220,6 +1220,14 @@ int h3_stream_append_chunk(void *ctx, zend_string *chunk) return HTTP_STREAM_APPEND_STREAM_DEAD; } + /* Refusal, not a silent accept: the previous chunk has not drained into + * the peer's window, so queueing this one would grow memory that the + * blocking path bounds by waiting. */ + if (nonblocking && s->chunk_pending_bytes > 0) { + zend_string_release(chunk); + return HTTP_STREAM_APPEND_BACKPRESSURE; + } + const bool first_call = s->chunk_queue == NULL; if (first_call) { @@ -1284,7 +1292,7 @@ int h3_stream_append_chunk(void *ctx, zend_string *chunk) * passes and the suspend never returns: it parks a libuv callback frame that * the sender itself would have had to wake. It backpressures in * h3_static_try_read instead. */ - const bool nonblocking_producer = s->static_body_state != NULL; + const bool nonblocking_producer = s->static_body_state != NULL || nonblocking; /* Pull write_timeout_s once — config can't change mid-handler. * 0 = wait forever (used in tests / bring-up). Pre-multiply to ms so @@ -1333,7 +1341,7 @@ int h3_stream_append_chunk(void *ctx, zend_string *chunk) if (EG(exception) != NULL) { /* Timeout exception expected for genuinely stalled peers; - * cancel-from-RST also lands here. send() surfaces this as + * cancel-from-RST also lands here. write() surfaces this as * HttpException to the user handler. */ return HTTP_STREAM_APPEND_STREAM_DEAD; } @@ -1393,6 +1401,16 @@ static zend_async_event_t *h3_stream_get_wait_event(void *ctx) : NULL; } +/* Room means the previous chunk has reached nghttp3, which is what the + * non-blocking refusal below tests. Published because the compressing wrapper + * asks it before feeding the encoder. */ +static bool h3_stream_sendable(void *ctx) +{ + const http3_stream_t *const s = (const http3_stream_t *)ctx; + + return s != NULL && s->chunk_pending_bytes == 0; +} + /* The four terminal conditions h3_stream_append_chunk refuses on. */ static bool h3_stream_is_alive(void *ctx) { @@ -1404,6 +1422,7 @@ static bool h3_stream_is_alive(void *ctx) const http_response_stream_ops_t h3_stream_ops = { .append_chunk = h3_stream_append_chunk, + .sendable = h3_stream_sendable, .is_alive = h3_stream_is_alive, .mark_ended = h3_stream_mark_ended, .get_wait_event = h3_stream_get_wait_event, @@ -1514,7 +1533,7 @@ static int h3_end_stream_cb(nghttp3_conn *conn, int64_t stream_id, /* Mark the stream peer-closed and wake any handler suspended on * write_event. After this point append_chunk short-circuits - * to STREAM_DEAD so HttpResponse::send() unwinds cleanly with an + * to STREAM_DEAD so HttpResponse::write() unwinds cleanly with an * exception; mirrors the H2 peer_closed discipline. */ static void h3_stream_mark_peer_closed(http3_stream_t *s) { diff --git a/src/http3/http3_dispatch.c b/src/http3/http3_dispatch.c index 62c15ecf..eed9cc93 100644 --- a/src/http3/http3_dispatch.c +++ b/src/http3/http3_dispatch.c @@ -615,7 +615,7 @@ void http3_stream_dispatch(http3_connection_t *c, http3_stream_t *s) http_response_set_protocol_version(Z_OBJ(s->response_zv), "3.0"); http_response_set_head(Z_OBJ(s->response_zv), http_request_method_is_head(s->request)); - /* Wire the streaming vtable so HttpResponse::send() in the + /* Wire the streaming vtable so HttpResponse::write() in the * handler enqueues into our chunk_queue. setBody/end (REST) handlers * never touch this; they go through the buffered submit_response in * dispose. */ @@ -956,7 +956,7 @@ static void h3_grpc_append_frame_and_end(void *ctx, zend_string *frame) { http3_stream_t *s = (http3_stream_t *)ctx; - (void)h3_stream_ops.append_chunk(s, frame); /* consumes the ref */ + (void)h3_stream_ops.append_chunk(s, frame, false); /* consumes the ref */ h3_stream_finish_streaming(s); } @@ -1032,7 +1032,7 @@ static void h3_handler_coroutine_dispose(zend_coroutine_t *coroutine) /* Streaming-vs-buffered decision (mirror of H2 dispose). * - * Streaming path: HEADERS were submitted on the first send() via + * Streaming path: HEADERS were submitted on the first write() via * h3_stream_ops.append_chunk; data_reader is already pulling from * chunk_queue. All we have to do here is make sure mark_ended fired * — if the handler forgot to call $res->end(), do it now so the diff --git a/src/http3/http3_internal.h b/src/http3/http3_internal.h index 32d0601b..a4456cd2 100644 --- a/src/http3/http3_internal.h +++ b/src/http3/http3_internal.h @@ -155,7 +155,7 @@ extern const http_response_stream_ops_t h3_stream_ops; /* Buffered REST response commit. The dispose path of the handler * coroutine (in http3_dispatch.c) calls this when nothing was streamed - * via $res->send() — submit_response with the single-slice data_reader. */ + * via $res->write() — submit_response with the single-slice data_reader. */ bool http3_stream_submit_response(http3_connection_t *c, http3_stream_t *s, bool streaming); @@ -188,7 +188,7 @@ void h3_chunk_queue_push(http3_stream_t *s, zend_string *chunk); * (http3_static_response.c). Pumping a file through chunk_queue is * exactly the streaming path: append chunks until EOF, then mark_ended. * The static TU calls these from its coroutine entry. */ -int h3_stream_append_chunk(void *ctx, zend_string *chunk); +int h3_stream_append_chunk(void *ctx, zend_string *chunk, bool nonblocking); void h3_stream_mark_ended(void *ctx); /* Per-worker memory accounting for static delivery: alloc on push, debit on diff --git a/src/http3/http3_static_response.c b/src/http3/http3_static_response.c index cc0e43f3..5aeabd75 100644 --- a/src/http3/http3_static_response.c +++ b/src/http3/http3_static_response.c @@ -513,7 +513,7 @@ static void h3_static_read_dispatch(zend_async_event_t *event, * nghttp3 and drains. Never suspends — s->static_body_state is what tells * append_chunk this producer backpressures itself. */ state->busy = true; - const int rc = h3_stream_append_chunk(state->stream, chunk); + const int rc = h3_stream_append_chunk(state->stream, chunk, false); state->busy = false; if (UNEXPECTED(rc != HTTP_STREAM_APPEND_OK)) { diff --git a/src/http_response.c b/src/http_response.c index 6e513a96..26ff0e8e 100644 --- a/src/http_response.c +++ b/src/http_response.c @@ -40,7 +40,7 @@ static zend_object_handlers http_response_handlers; /* Helper: gate every status/header/body mutation. A response is * no-longer-mutable in two states: * 1. closed — end() has been called; nothing further is possible. - * 2. streaming — send() has been called; status + headers are + * 2. streaming — write() has been called; status + headers are * committed on the wire. Trailers are still allowed * (they're post-DATA) and go through separate * non-guarded setters — see setTrailer/setTrailers. */ @@ -54,7 +54,7 @@ static inline bool response_check_closed(const http_response_object *response) if (response->streaming) { zend_throw_exception(http_server_runtime_exception_ce, - "Cannot modify response — headers already committed by send()", 0); + "Cannot modify response — headers already committed by write()", 0); return true; } @@ -637,8 +637,8 @@ ZEND_METHOD(TrueAsync_HttpResponse, getProtocolVersion) } /* }}} */ -/* {{{ proto HttpResponse::write(string $data): static */ -ZEND_METHOD(TrueAsync_HttpResponse, write) +/* {{{ proto HttpResponse::appendBody(string $data): static */ +ZEND_METHOD(TrueAsync_HttpResponse, appendBody) { zend_string *data; @@ -652,11 +652,10 @@ ZEND_METHOD(TrueAsync_HttpResponse, write) return; } - /* write() is the buffered-mode incremental API: handler calls it - * N times with chunks and the full body goes out on end(). Size is - * unknown up front — scalable-grow flips to doubling above 2 MiB - * so a handler writing a 256 MiB body doesn't take 40 k mremap - * calls. See smart_str_scalable.h. */ + /* Buffered-mode incremental API: the handler calls it N times and the + * full body goes out on end(). Size is unknown up front — scalable-grow + * flips to doubling above 2 MiB so a handler appending a 256 MiB body + * doesn't take 40 k mremap calls. See smart_str_scalable.h. */ response_clear_body_view(response); http_smart_str_append_scalable(&response->body, ZSTR_VAL(data), ZSTR_LEN(data)); @@ -717,32 +716,6 @@ ZEND_METHOD(TrueAsync_HttpResponse, setBody) } /* }}} */ -/* {{{ proto HttpResponse::getBodyStream(): mixed */ -ZEND_METHOD(TrueAsync_HttpResponse, getBodyStream) -{ - ZEND_PARSE_PARAMETERS_NONE(); - - /* TODO: Implement body stream support */ - RETURN_NULL(); -} -/* }}} */ - -/* {{{ proto HttpResponse::setBodyStream(mixed $stream): static */ -ZEND_METHOD(TrueAsync_HttpResponse, setBodyStream) -{ - (void)return_value; - zval *stream; - - ZEND_PARSE_PARAMETERS_START(1, 1) - Z_PARAM_ZVAL(stream) - ZEND_PARSE_PARAMETERS_END(); - - /* TODO: Implement body stream support */ - zend_throw_exception(http_server_runtime_exception_ce, - "Body stream support is not yet implemented", 0); -} -/* }}} */ - /* Wire the per-request JSON-encode default into a freshly-dispatched * response. Called from H1/H2/H3 dispatch alongside compression_attach; * exported (non-static) so the protocol TUs can reach it without @@ -927,7 +900,84 @@ ZEND_METHOD(TrueAsync_HttpResponse, redirect) } /* }}} */ -/* {{{ proto HttpResponse::send(string $chunk): static +/* True when setBody()/appendBody()/json()/html() left bytes waiting for end(). + * An empty buffer does not count: setBody('') commits the handler to nothing. */ +static bool response_has_buffered_body(const http_response_object *response) +{ + if (response->body_view != NULL) { + return ZSTR_LEN(response->body_view) > 0; + } + + return response->body.s != NULL && ZSTR_LEN(response->body.s) > 0; +} + +/* Guards shared by every streaming entry point, so write() and tryWrite() + * cannot drift apart. Returns true after throwing; `method` names the caller + * in the message. */ +static bool response_check_stream_usable(const http_response_object *response, + const char *method) +{ + if (response->closed) { + zend_throw_exception_ex(http_server_runtime_exception_ce, 0, + "Response already closed — cannot %s() after end()", method); + return true; + } + + if (response->sse_mode) { + zend_throw_exception_ex(http_server_runtime_exception_ce, 0, + "Response is in SSE mode — use sseEvent()/sseComment() instead of %s()", method); + return true; + } + + if (response->send_file_req != NULL) { + zend_throw_exception(http_server_runtime_exception_ce, + "Response is sealed by sendFile() — no further mutation allowed", 0); + return true; + } + + if (response->stream_ops == NULL) { + /* No stream ops installed — response is detached from a + * connection (e.g. constructed standalone in user code). */ + zend_throw_exception_ex(http_server_runtime_exception_ce, 0, + "Response streaming (%s()) is not available on this response", method); + return true; + } + + /* A buffered body leaves at end() and the streaming path never reads it, + * so the two modes are exclusive. response_check_closed() refuses the + * other direction; this is the same refusal from this side. */ + if (response_has_buffered_body(response)) { + zend_throw_exception_ex(http_server_runtime_exception_ce, 0, + "Response already has a buffered body — %s() would discard it. " + "Choose one mode: setBody()/appendBody() or %s()", method, method); + return true; + } + + return false; +} + +/* First chunk locks headers and switches to streaming mode. After this, + * setBody / setHeader / setStatusCode throw. */ +static void http_response_stream_commit_once(zend_object *obj, + http_response_object *response) +{ + if (response->streaming) { + return; + } + + response->streaming = true; + response->committed = true; + response->headers_sent = true; +#ifdef HAVE_HTTP_COMPRESSION + /* Wrap stream_ops with a compressing one if Accept-Encoding + + * response state allow gzip. Mutates Content-Encoding/Vary on + * the response so the stream's underlying header-commit picks + * them up on the next line. */ + http_compression_maybe_install_stream_wrapper(obj); +#endif +} + +/* {{{ proto HttpResponse::write(string $chunk): static * * Streaming response — append a chunk to the outbound queue. First * call commits status + headers (they can no longer be changed). @@ -938,7 +988,7 @@ ZEND_METHOD(TrueAsync_HttpResponse, redirect) * * Throws when called on a response that has no stream ops installed * (typically a response detached from a real connection). */ -ZEND_METHOD(TrueAsync_HttpResponse, send) +ZEND_METHOD(TrueAsync_HttpResponse, write) { zend_string *chunk; @@ -948,29 +998,7 @@ ZEND_METHOD(TrueAsync_HttpResponse, send) http_response_object *response = Z_HTTP_RESPONSE_P(ZEND_THIS); - if (response->closed) { - zend_throw_exception(http_server_runtime_exception_ce, - "Response already closed — cannot send() after end()", 0); - return; - } - - if (response->sse_mode) { - zend_throw_exception(http_server_runtime_exception_ce, - "Response is in SSE mode — use sseEvent()/sseComment() instead of send()", 0); - return; - } - - if (response->send_file_req != NULL) { - zend_throw_exception(http_server_runtime_exception_ce, - "Response is sealed by sendFile() — no further mutation allowed", 0); - return; - } - - if (response->stream_ops == NULL) { - /* No stream ops installed — response is detached from a - * connection (e.g. constructed standalone in user code). */ - zend_throw_exception(http_server_runtime_exception_ce, - "Response streaming (send()) is not available on this response", 0); + if (response_check_stream_usable(response, "write")) { return; } @@ -979,27 +1007,14 @@ ZEND_METHOD(TrueAsync_HttpResponse, send) RETURN_OBJ_COPY(Z_OBJ_P(ZEND_THIS)); } - /* First send() — lock headers and switch to streaming mode. - * After this, setBody / setHeader / setStatusCode throw. */ - if (!response->streaming) { - response->streaming = true; - response->committed = true; - response->headers_sent = true; -#ifdef HAVE_HTTP_COMPRESSION - /* Wrap stream_ops with a compressing one if Accept-Encoding + - * response state allow gzip. Mutates Content-Encoding/Vary on - * the response so the stream's underlying header-commit picks - * them up on the next line. */ - http_compression_maybe_install_stream_wrapper(Z_OBJ_P(ZEND_THIS)); -#endif - } + http_response_stream_commit_once(Z_OBJ_P(ZEND_THIS), response); /* Hand ownership of the chunk to the queue — the ops layer * takes a refcount. Empty chunks are still accepted (some * protocols use them as keepalive signals). */ zend_string_addref(chunk); const int rc = response->stream_ops->append_chunk( - response->stream_ctx, chunk); + response->stream_ctx, chunk, false); if (rc == HTTP_STREAM_APPEND_STREAM_DEAD) { /* Peer aborted between dispatch and now. Emulate the @@ -1018,6 +1033,153 @@ ZEND_METHOD(TrueAsync_HttpResponse, send) } /* }}} */ +/* {{{ proto HttpResponse::tryWrite(string $chunk): bool + * + * Non-blocking write(). Returns false when the outbound queue has no room — + * nothing was queued and no header was committed, so the same chunk can be + * offered again later. A peer that is gone is NOT reported as false: it + * throws HttpException 499, because "wait" and "stop" call for opposite + * reactions and one bool cannot carry both. + * + * The refused chunk is a slice of one byte stream, so dropping it corrupts + * the body — retry it or stop. Only the framed dialects (SSE events, gRPC + * messages) carry droppable units. + * + * HTTP/1 neither refuses nor returns promptly: it keeps no queue of its own, + * so the kernel socket buffer is the queue, and an accepted chunk waits on the + * socket for as long as a blocking write() would. Issue #179 gives the + * connection its own outbound queue, after which both halves hold under this + * same signature. */ +ZEND_METHOD(TrueAsync_HttpResponse, tryWrite) +{ + zend_string *chunk; + + ZEND_PARSE_PARAMETERS_START(1, 1) + Z_PARAM_STR(chunk) + ZEND_PARSE_PARAMETERS_END(); + + http_response_object *response = Z_HTTP_RESPONSE_P(ZEND_THIS); + + if (response_check_stream_usable(response, "tryWrite")) { + return; + } + + /* Dead peer first: false must mean "full", and only that. */ + if (response->stream_ops->is_alive != NULL + && !response->stream_ops->is_alive(response->stream_ctx)) { + zend_throw_exception_ex(http_exception_ce, 499, "stream closed by peer"); + return; + } + + /* HEAD carries no body (RFC 9110 §9.3.2); the chunk is accepted and + * dropped, as write() does. */ + if (response->is_head) { + RETURN_TRUE; + } + + /* The commit precedes the append because the wrapper installed here is + * what encodes the chunk, and the transport emits headers from inside. + * No transport can refuse a first offer — each opens its queue on that + * call and answers "room" while the queue is absent — so a refusal never + * arrives with the response still uncommitted. #179 is where that stops + * being true, and where a refusal will have to unwind the commit and the + * wrapper together. */ + http_response_stream_commit_once(Z_OBJ_P(ZEND_THIS), response); + + zend_string_addref(chunk); + const int rc = response->stream_ops->append_chunk( + response->stream_ctx, chunk, true); + + /* append_chunk consumes the ref on every path, refusals included. */ + if (rc == HTTP_STREAM_APPEND_BACKPRESSURE) { + RETURN_FALSE; + } + + /* The transport may already have thrown a more precise reason — an + * over-sized chunk, say. 499 is the fallback diagnosis, not an override. */ + if (rc == HTTP_STREAM_APPEND_STREAM_DEAD) { + if (EXPECTED(EG(exception) == NULL)) { + zend_throw_exception_ex(http_exception_ce, 499, "stream closed by peer"); + } + + return; + } + + RETURN_TRUE; +} +/* }}} */ + +/* {{{ proto HttpResponse::awaitWritable(?int $timeoutMs = null): bool + * + * Wait until the outbound queue has room again, and report whether it has. + * The companion to tryWrite(): that call says "not now", this one waits for + * "now" instead of spinning. + * + * The wait belongs to the transport, which keeps its own deadline and re-pumps + * its drain on each wake. A transport with no queue (HTTP/1) has nothing to + * wait for and answers true at once. A transport that can be full but offers + * no way to wait answers false rather than true — a caller told "go ahead" + * would spin and never yield, which on a pool worker freezes every other + * request on that thread. */ +ZEND_METHOD(TrueAsync_HttpResponse, awaitWritable) +{ + zend_long timeout_ms = 0; + bool timeout_is_null = true; + + ZEND_PARSE_PARAMETERS_START(0, 1) + Z_PARAM_OPTIONAL + Z_PARAM_LONG_OR_NULL(timeout_ms, timeout_is_null) + ZEND_PARSE_PARAMETERS_END(); + + if (UNEXPECTED(!timeout_is_null && timeout_ms < 0)) { + zend_throw_exception(http_server_runtime_exception_ce, + "awaitWritable(): timeout must not be negative", 0); + return; + } + + http_response_object *response = Z_HTTP_RESPONSE_P(ZEND_THIS); + + if (response_check_stream_usable(response, "awaitWritable")) { + return; + } + + const http_response_stream_ops_t *ops = response->stream_ops; + + /* No queue of its own, or room already: nothing to wait for. */ + if (ops->sendable == NULL || ops->sendable(response->stream_ctx)) { + RETURN_TRUE; + } + + if (ops->wait_writable == NULL) { + RETURN_FALSE; + } + + zend_coroutine_t *co = ZEND_ASYNC_CURRENT_COROUTINE; + + if (co == NULL || ZEND_ASYNC_IS_SCHEDULER_CONTEXT) { + zend_throw_exception(http_server_runtime_exception_ce, + "awaitWritable() needs a coroutine to suspend — call it from a handler", 0); + return; + } + + const bool woken = ops->wait_writable(response->stream_ctx, + timeout_is_null ? 0u : (uint32_t)timeout_ms); + + /* A timeout or a cancellation arrives as the transport's exception; it is + * left to propagate rather than flattened into false, which would hide a + * cancelled request behind "still full". */ + if (EG(exception) != NULL) { + return; + } + + if (!woken) { + RETURN_FALSE; + } + + RETURN_BOOL(ops->sendable == NULL || ops->sendable(response->stream_ctx)); +} +/* }}} */ + /* {{{ proto HttpResponse::setGrpcEncoding(string $encoding): static * Declare the response message encoding (grpc-encoding header) before the * first writeMessage(). Mirrors grpc-java setCompression / C++ @@ -1076,7 +1238,7 @@ ZEND_METHOD(TrueAsync_HttpResponse, setGrpcEncoding) /* }}} */ /* {{{ proto HttpResponse::writeMessage(string $message): static - * Stream one gRPC length-prefixed message; first call commits, like send(). + * Stream one gRPC length-prefixed message; first call commits, like write(). * Compressed automatically when setGrpcEncoding('gzip') was declared. */ ZEND_METHOD(TrueAsync_HttpResponse, writeMessage) { @@ -1147,7 +1309,7 @@ ZEND_METHOD(TrueAsync_HttpResponse, writeMessage) } const int rc = response->stream_ops->append_chunk( - response->stream_ctx, framed); + response->stream_ctx, framed, false); if (rc == HTTP_STREAM_APPEND_STREAM_DEAD) { zend_throw_exception_ex(http_exception_ce, 499, @@ -1163,33 +1325,17 @@ ZEND_METHOD(TrueAsync_HttpResponse, writeMessage) /* {{{ proto HttpResponse::sendable(): bool * - * Advisory, non-blocking backpressure check. Returns true when send() - * would accept a chunk without suspending the handler coroutine — the - * per-stream staging buffer has room. Returns false when send() would - * block on backpressure, or when the response is closed / sealed by - * sendFile() / not streaming-capable. - * - * send() is always safe to call regardless; sendable() just lets a - * handler do other work instead of blocking on a slow peer. */ + * Tombstone: the declaration outlives the method for one minor release, + * because shipped adapter code calls it and its two replacements cannot be + * guessed from the name. */ ZEND_METHOD(TrueAsync_HttpResponse, sendable) { + (void)return_value; ZEND_PARSE_PARAMETERS_NONE(); - http_response_object *response = Z_HTTP_RESPONSE_P(ZEND_THIS); - - if (response->closed - || response->send_file_req != NULL - || response->stream_ops == NULL) { - RETURN_FALSE; - } - - /* Protocol without a userspace staging ring (HTTP/1, paced by the - * kernel socket buffer) leaves the op NULL — report writable. */ - if (response->stream_ops->sendable == NULL) { - RETURN_TRUE; - } - - RETURN_BOOL(response->stream_ops->sendable(response->stream_ctx)); + zend_throw_exception(http_server_runtime_exception_ce, + "sendable() is gone: it answered liveness and queue depth with one bool. " + "Use isWritable() for liveness, tryWrite()/awaitWritable() for room", 0); } /* }}} */ @@ -1311,7 +1457,7 @@ ZEND_METHOD(TrueAsync_HttpResponse, end) if (data != NULL && ZSTR_LEN(data) > 0) { zend_string_addref(data); (void)response->stream_ops->append_chunk( - response->stream_ctx, data); + response->stream_ctx, data, false); } response->stream_ops->mark_ended(response->stream_ctx); @@ -1362,9 +1508,9 @@ ZEND_METHOD(TrueAsync_HttpResponse, isHeadersSent) /* {{{ proto HttpResponse::isWritable(): bool * * True while output is still possible: end() was not called, the response is - * not sealed by sendFile(), and the peer has not gone. Unlike sendable(), - * which swings with queue depth, a false answer here is final — so a - * streaming loop stops on !isWritable() and yields on !sendable(). */ + * not sealed by sendFile(), and the peer has not gone. A false answer is + * final, unlike the queue depth tryWrite() reports — so a streaming loop + * stops on !isWritable() and yields on a refused tryWrite(). */ ZEND_METHOD(TrueAsync_HttpResponse, isWritable) { ZEND_PARSE_PARAMETERS_NONE(); @@ -1386,8 +1532,11 @@ ZEND_METHOD(TrueAsync_HttpResponse, isWritable) } /* }}} */ -/* {{{ proto HttpResponse::isClosed(): bool */ -ZEND_METHOD(TrueAsync_HttpResponse, isClosed) +/* {{{ proto HttpResponse::isEnded(): bool + * + * Reports the response, not the connection: a peer that has gone leaves this + * false until the handler ends the response. isWritable() answers liveness. */ +ZEND_METHOD(TrueAsync_HttpResponse, isEnded) { ZEND_PARSE_PARAMETERS_NONE(); http_response_object *response = Z_HTTP_RESPONSE_P(ZEND_THIS); diff --git a/src/http_response_internal.h b/src/http_response_internal.h index 43bd3fa5..0ddc65bc 100644 --- a/src/http_response_internal.h +++ b/src/http_response_internal.h @@ -40,7 +40,7 @@ typedef struct { zend_string *body_view; /* Streaming ops + ctx. Installed by the protocol strategy at - * dispatch; NULL for buffered-mode responses. send() activates + * dispatch; NULL for buffered-mode responses. write() activates * streaming by reading these; the ops interpret ctx (opaque * pointer to the protocol-specific stream state). */ const http_response_stream_ops_t *stream_ops; @@ -56,9 +56,9 @@ typedef struct { bool headers_sent; bool closed; bool committed; - bool streaming; /* send() has been called — setBody/setHeader now throw */ - bool sse_mode; /* SSE helpers committed the stream — send() now throws, sse* re-entry is allowed */ - bool is_head; /* HEAD: send() drops chunks (RFC 9110 §9.3.2) */ + bool streaming; /* write() has been called — setBody/setHeader now throw */ + bool sse_mode; /* SSE helpers committed the stream — write() now throws, sse* re-entry is allowed */ + bool is_head; /* HEAD: write() drops chunks (RFC 9110 §9.3.2) */ /* grpc_mode_t stamped at dispatch; picks the per-frame transform. * 0 = not a gRPC call. */ diff --git a/src/http_response_server_api.c b/src/http_response_server_api.c index a2f07a43..5031e614 100644 --- a/src/http_response_server_api.c +++ b/src/http_response_server_api.c @@ -60,7 +60,7 @@ bool http_response_is_committed(zend_object *obj) return http_response_from_obj(obj)->committed; } -/* True once HttpResponse::send() has been called. Dispose paths use +/* True once HttpResponse::write() has been called. Dispose paths use * this to skip the buffered-mode commit (headers are already on the * wire, the data provider drives the body via chunk_queue). */ bool http_response_is_streaming(zend_object *obj) @@ -123,7 +123,7 @@ zend_string *http_response_get_body_str(zend_object *obj) } /* Install streaming ops + ctx on the response. Protocol strategies - * call this once at dispatch; reading after send() activates + * call this once at dispatch; reading after write() activates * streaming mode. Passing ops=NULL clears (not currently used). */ void http_response_install_stream_ops(zend_object *obj, const http_response_stream_ops_t *ops, diff --git a/src/http_sse.c b/src/http_sse.c index f648cf9b..63595e5f 100644 --- a/src/http_sse.c +++ b/src/http_sse.c @@ -10,7 +10,7 @@ * * SSE is not a separate protocol — it is a Content-Type convention plus * the small line-oriented framing defined by WHATWG §9.2, layered on top - * of the existing HttpResponse::send() streaming pipeline (HTTP/1 chunked, + * of the existing HttpResponse::write() streaming pipeline (HTTP/1 chunked, * HTTP/2 + HTTP/3 DATA frames). These helpers only (1) set the canonical * headers so a handler can't ship a broken stream behind nginx/a CDN and * (2) format event records correctly so handlers don't reinvent framing. @@ -18,7 +18,7 @@ * Wire commit is lazy: the headers are set and the response is locked into * streaming mode here, but the actual HEADERS frame / status line is * emitted by the protocol stream_ops on the first append_chunk — exactly - * the same path the first send() drives. */ + * the same path the first write() drives. */ #ifdef HAVE_CONFIG_H #include @@ -110,13 +110,13 @@ static bool sse_content_type_conflicts(const HashTable *headers) static bool sse_ensure_started(http_response_object *response) { if (response->streaming) { - /* Already streaming via send() (or another non-SSE path) — emitting + /* Already streaming via write() (or another non-SSE path) — emitting * SSE framing now would ship event records without the event-stream - * headers, and possibly through send()'s gzip wrapper. Reject the + * headers, and possibly through write()'s gzip wrapper. Reject the * misuse instead of silently corrupting the stream. */ if (!response->sse_mode) { zend_throw_exception(http_server_runtime_exception_ce, - "Response is already streaming via send() — cannot switch to SSE", 0); + "Response is already streaming via write() — cannot switch to SSE", 0); return false; } @@ -163,7 +163,7 @@ static bool sse_ensure_started(http_response_object *response) #ifdef HAVE_HTTP_COMPRESSION /* A buffering gzip stream defeats real-time delivery — never compress * an event stream. SSE dispatches through the raw stream_ops (not the - * send() wrapper), but mark it explicitly so intent is unambiguous. */ + * write() wrapper), but mark it explicitly so intent is unambiguous. */ http_compression_mark_no_compression(&response->std); #endif @@ -221,11 +221,11 @@ static void sse_append_field(smart_str *out, const char *field, size_t field_len /* Push a finalised event payload through the installed stream ops. * append_chunk takes ownership of the payload ref (so we never release it) - * and suspends the handler under backpressure on H2/H3. Mirrors send(): + * and suspends the handler under backpressure on H2/H3. Mirrors write(): * a dead stream surfaces as a 499 the handler may catch. */ static void sse_dispatch(http_response_object *response, zend_string *payload) { - const int rc = response->stream_ops->append_chunk(response->stream_ctx, payload); + const int rc = response->stream_ops->append_chunk(response->stream_ctx, payload, false); if (rc == HTTP_STREAM_APPEND_STREAM_DEAD) { zend_throw_exception_ex(http_exception_ce, 499, "stream closed by peer"); diff --git a/stubs/HttpResponse.php b/stubs/HttpResponse.php index 0d3d70ca..3de1988f 100644 --- a/stubs/HttpResponse.php +++ b/stubs/HttpResponse.php @@ -152,35 +152,60 @@ public function getProtocolVersion(): string {} // === Body methods === /** - * Append data to the buffered response body. + * Stream a chunk to the client. * - * Nothing reaches the client here: the whole body goes out on end(), - * with Content-Length computed from it. Use send() to stream instead — - * that is the call which commits headers and applies backpressure. + * The first call commits status and headers; afterwards setStatusCode(), + * setHeader() and setBody() throw. Later calls append chunked-transfer + * segments (HTTP/1) or DATA frames (HTTP/2, HTTP/3). To append to a + * buffered body instead, call appendBody(). * - * @param string $data Data to write - * @return static + * Parks the handler coroutine only under backpressure: HTTP/2 and HTTP/3 + * park while every ring slot is live or the queued bytes stand at + * HttpServerConfig::setStreamWriteBufferBytes (256 KiB by default), + * HTTP/1 parks on the socket write. tryWrite() offers a chunk without + * committing to that wait. A peer that has gone throws HttpException 499. + */ + public function write(string $chunk): static {} + + /** + * Offer a chunk without waiting for room: false means the outbound queue + * had no room and nothing was queued, so the same chunk can be offered + * again later. The transport answers at the moment of queueing, not from + * a predicate read beforehand, so nothing slips in between. + * + * A client that has gone is not reported as false — it throws + * HttpException 499, because "wait" and "stop" need opposite reactions. + * The refused chunk is a slice of one byte stream, so dropping it corrupts + * the body: retry it, or stop. + * + * HTTP/1 is the exception, and it is not a small one: that transport keeps + * no queue of its own, so it never refuses AND an accepted chunk waits for + * the socket for as long as a blocking write() would — up to the write + * timeout. A handler + * that must not be parked has to check getProtocolVersion(). Over HTTP/2, + * HTTP/3 and the worker pool neither happens. Issue #179 removes the + * exception. */ - public function write(string $data): static {} + public function tryWrite(string $chunk): bool {} /** - * Send a chunk to the client (streaming response). + * Wait until the outbound queue has room again, and report whether it has. * - * First call commits status + headers (they can no longer be - * changed). Subsequent calls append DATA frames (HTTP/2) or - * chunked-transfer segments (HTTP/1). + * The companion to tryWrite(): that call says "not now", this one waits for + * "now" instead of spinning. The wait belongs to the transport, which keeps + * its own deadline and re-pumps its drain on each wake. * - * Blocks the handler coroutine ONLY under backpressure — when the - * per-stream staging buffer is full (HTTP/2: all ring slots live - * OR queued bytes reach HttpServerConfig::setStreamWriteBufferBytes, - * default 256 KiB). Otherwise returns immediately. send() is always - * safe to call; use sendable() to check first if you'd rather do - * other work than block. + * True at once on HTTP/1, which keeps no queue and so has nothing to wait + * for. False without waiting on a transport that can be full but offers no + * wait — better than "go ahead", which would spin a handler that trusts it. + * A timeout or a cancellation arrives as an exception; false after a wait + * means the queue is still full. * - * @param string $chunk - * @return static + * @param int|null $timeoutMs Milliseconds to wait; null leaves the deadline + * to the transport, which uses the connection's + * write timeout. */ - public function send(string $chunk): static {} + public function awaitWritable(?int $timeoutMs = null): bool {} /** * Declare the gRPC response message encoding. @@ -202,7 +227,7 @@ public function setGrpcEncoding(string $encoding): static {} * * Prepends the 5-byte gRPC length prefix to $message and streams it as * a single gRPC message. Activates streaming mode on the first call, - * exactly like send(). Call once for a unary reply, repeatedly for + * exactly like write(). Call once for a unary reply, repeatedly for * server-streaming. Pass the already protobuf-encoded bytes; the * grpc-status is carried separately via setTrailer() (defaults to 0 * when unset). Compressed automatically when setGrpcEncoding('gzip') @@ -214,21 +239,17 @@ public function setGrpcEncoding(string $encoding): static {} public function writeMessage(string $message): static {} /** - * Advisory, non-blocking backpressure check for streaming responses. - * - * Returns true when send() would accept a chunk without suspending - * the handler coroutine — the per-stream staging buffer has room. - * Returns false when send() would block on backpressure, or when the - * response is closed / sealed by sendFile() / not streaming-capable. + * Removed. One bool answered four questions, and a loop that read it as + * liveness stopped streams that were merely slow. * - * send() is always safe to call regardless; sendable() just lets a - * handler do other work instead of blocking on a slow peer. + * Ask the two questions separately: isWritable() reports whether output is + * still possible, tryWrite() and awaitWritable() report whether the + * outbound queue has room. * - * False does not report a departed client: a peer that is gone surfaces - * as HttpException 499 out of send(). A loop that breaks on false stops a - * stream that is merely slow. + * The declaration stays for one minor release so a call names its + * replacements instead of failing as an undefined method. * - * @return bool + * @throws HttpServerRuntimeException always */ public function sendable(): bool {} @@ -251,26 +272,17 @@ public function getBody(): string {} /** * Set body content (replaces buffer) - * - * @param string $body Body content - * @return static */ public function setBody(string $body): static {} /** - * Get body stream (TODO) - * - * @return mixed Stream resource or null - */ - public function getBodyStream(): mixed {} - - /** - * Set body stream (TODO) + * Append to the buffered response body. * - * @param mixed $stream Stream resource - * @return static + * Nothing reaches the client here: the whole body goes out on end(), with + * Content-Length computed from it. Call write() to stream instead — that + * is the call which commits headers and applies backpressure. */ - public function setBodyStream(mixed $stream): static {} + public function appendBody(string $data): static {} // === Helper methods === @@ -364,7 +376,7 @@ public function sendFile(string $path, ?SendFileOptions $options = null): void { * response; without it events stall behind the proxy buffer until it * fills) — and marks the response as not-compressible (a buffering * gzip stream would defeat real-time delivery). The response then - * enters streaming mode exactly as the first {@see self::send()} would: + * enters streaming mode exactly as the first {@see self::write()} would: * status + headers are committed and may no longer change, but no event * data is emitted until the first sseEvent()/sseComment(). * @@ -456,13 +468,18 @@ public function isHeadersSent(): bool {} * True while output is still possible: end() was not called, the response * is not sealed by sendFile(), and the client has not gone. * - * A false answer is final, which is what separates this from sendable(): - * stop a streaming loop on !isWritable(), yield on !sendable(). + * A false answer is final: stop a streaming loop on !isWritable(). For the + * separate question of room in the outbound queue, use tryWrite() or + * awaitWritable(). */ public function isWritable(): bool {} /** - * Check if response is closed + * True once end() has been called. + * + * Reports the response, not the connection: a peer that has gone leaves + * this false until the handler ends the response. Use isWritable() for + * liveness. */ - public function isClosed(): bool {} + public function isEnded(): bool {} } diff --git a/stubs/HttpResponse.php_arginfo.h b/stubs/HttpResponse.php_arginfo.h index 8872bd15..840e62bd 100644 --- a/stubs/HttpResponse.php_arginfo.h +++ b/stubs/HttpResponse.php_arginfo.h @@ -1,5 +1,5 @@ /* This is a generated file, edit HttpResponse.php.stub.php instead. - * Stub hash: 8e3381806654b44692470e369c6cf3c01b2d13b7 */ + * Stub hash: a589c1d891e3c758c66b880fa5f28accb295a2dc */ ZEND_BEGIN_ARG_INFO_EX(arginfo_class_TrueAsync_HttpResponse___construct, 0, 0, 0) ZEND_END_ARG_INFO() @@ -61,13 +61,17 @@ ZEND_END_ARG_INFO() #define arginfo_class_TrueAsync_HttpResponse_getProtocolVersion arginfo_class_TrueAsync_HttpResponse_getReasonPhrase ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_write, 0, 1, IS_STATIC, 0) - ZEND_ARG_TYPE_INFO(0, data, IS_STRING, 0) + ZEND_ARG_TYPE_INFO(0, chunk, IS_STRING, 0) ZEND_END_ARG_INFO() -ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_send, 0, 1, IS_STATIC, 0) +ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_tryWrite, 0, 1, _IS_BOOL, 0) ZEND_ARG_TYPE_INFO(0, chunk, IS_STRING, 0) ZEND_END_ARG_INFO() +ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_awaitWritable, 0, 0, _IS_BOOL, 0) + ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, timeoutMs, IS_LONG, 1, "null") +ZEND_END_ARG_INFO() + ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_setGrpcEncoding, 0, 1, IS_STATIC, 0) ZEND_ARG_TYPE_INFO(0, encoding, IS_STRING, 0) ZEND_END_ARG_INFO() @@ -87,11 +91,8 @@ ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_set ZEND_ARG_TYPE_INFO(0, body, IS_STRING, 0) ZEND_END_ARG_INFO() -ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_getBodyStream, 0, 0, IS_MIXED, 0) -ZEND_END_ARG_INFO() - -ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_setBodyStream, 0, 1, IS_STATIC, 0) - ZEND_ARG_TYPE_INFO(0, stream, IS_MIXED, 0) +ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_appendBody, 0, 1, IS_STATIC, 0) + ZEND_ARG_TYPE_INFO(0, data, IS_STRING, 0) ZEND_END_ARG_INFO() ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_json, 0, 1, IS_STATIC, 0) @@ -139,7 +140,7 @@ ZEND_END_ARG_INFO() #define arginfo_class_TrueAsync_HttpResponse_isWritable arginfo_class_TrueAsync_HttpResponse_sendable -#define arginfo_class_TrueAsync_HttpResponse_isClosed arginfo_class_TrueAsync_HttpResponse_sendable +#define arginfo_class_TrueAsync_HttpResponse_isEnded arginfo_class_TrueAsync_HttpResponse_sendable ZEND_METHOD(TrueAsync_HttpResponse, __construct); ZEND_METHOD(TrueAsync_HttpResponse, setStatusCode); @@ -160,15 +161,15 @@ ZEND_METHOD(TrueAsync_HttpResponse, getTrailers); ZEND_METHOD(TrueAsync_HttpResponse, getProtocolName); ZEND_METHOD(TrueAsync_HttpResponse, getProtocolVersion); ZEND_METHOD(TrueAsync_HttpResponse, write); -ZEND_METHOD(TrueAsync_HttpResponse, send); +ZEND_METHOD(TrueAsync_HttpResponse, tryWrite); +ZEND_METHOD(TrueAsync_HttpResponse, awaitWritable); ZEND_METHOD(TrueAsync_HttpResponse, setGrpcEncoding); ZEND_METHOD(TrueAsync_HttpResponse, writeMessage); ZEND_METHOD(TrueAsync_HttpResponse, sendable); ZEND_METHOD(TrueAsync_HttpResponse, setNoCompression); ZEND_METHOD(TrueAsync_HttpResponse, getBody); ZEND_METHOD(TrueAsync_HttpResponse, setBody); -ZEND_METHOD(TrueAsync_HttpResponse, getBodyStream); -ZEND_METHOD(TrueAsync_HttpResponse, setBodyStream); +ZEND_METHOD(TrueAsync_HttpResponse, appendBody); ZEND_METHOD(TrueAsync_HttpResponse, json); ZEND_METHOD(TrueAsync_HttpResponse, html); ZEND_METHOD(TrueAsync_HttpResponse, redirect); @@ -180,7 +181,7 @@ ZEND_METHOD(TrueAsync_HttpResponse, sseComment); ZEND_METHOD(TrueAsync_HttpResponse, sseRetry); ZEND_METHOD(TrueAsync_HttpResponse, isHeadersSent); ZEND_METHOD(TrueAsync_HttpResponse, isWritable); -ZEND_METHOD(TrueAsync_HttpResponse, isClosed); +ZEND_METHOD(TrueAsync_HttpResponse, isEnded); static const zend_function_entry class_TrueAsync_HttpResponse_methods[] = { ZEND_ME(TrueAsync_HttpResponse, __construct, arginfo_class_TrueAsync_HttpResponse___construct, ZEND_ACC_PRIVATE) @@ -202,15 +203,15 @@ static const zend_function_entry class_TrueAsync_HttpResponse_methods[] = { ZEND_ME(TrueAsync_HttpResponse, getProtocolName, arginfo_class_TrueAsync_HttpResponse_getProtocolName, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, getProtocolVersion, arginfo_class_TrueAsync_HttpResponse_getProtocolVersion, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, write, arginfo_class_TrueAsync_HttpResponse_write, ZEND_ACC_PUBLIC) - ZEND_ME(TrueAsync_HttpResponse, send, arginfo_class_TrueAsync_HttpResponse_send, ZEND_ACC_PUBLIC) + ZEND_ME(TrueAsync_HttpResponse, tryWrite, arginfo_class_TrueAsync_HttpResponse_tryWrite, ZEND_ACC_PUBLIC) + ZEND_ME(TrueAsync_HttpResponse, awaitWritable, arginfo_class_TrueAsync_HttpResponse_awaitWritable, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, setGrpcEncoding, arginfo_class_TrueAsync_HttpResponse_setGrpcEncoding, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, writeMessage, arginfo_class_TrueAsync_HttpResponse_writeMessage, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, sendable, arginfo_class_TrueAsync_HttpResponse_sendable, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, setNoCompression, arginfo_class_TrueAsync_HttpResponse_setNoCompression, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, getBody, arginfo_class_TrueAsync_HttpResponse_getBody, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, setBody, arginfo_class_TrueAsync_HttpResponse_setBody, ZEND_ACC_PUBLIC) - ZEND_ME(TrueAsync_HttpResponse, getBodyStream, arginfo_class_TrueAsync_HttpResponse_getBodyStream, ZEND_ACC_PUBLIC) - ZEND_ME(TrueAsync_HttpResponse, setBodyStream, arginfo_class_TrueAsync_HttpResponse_setBodyStream, ZEND_ACC_PUBLIC) + ZEND_ME(TrueAsync_HttpResponse, appendBody, arginfo_class_TrueAsync_HttpResponse_appendBody, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, json, arginfo_class_TrueAsync_HttpResponse_json, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, html, arginfo_class_TrueAsync_HttpResponse_html, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, redirect, arginfo_class_TrueAsync_HttpResponse_redirect, ZEND_ACC_PUBLIC) @@ -222,7 +223,7 @@ static const zend_function_entry class_TrueAsync_HttpResponse_methods[] = { ZEND_ME(TrueAsync_HttpResponse, sseRetry, arginfo_class_TrueAsync_HttpResponse_sseRetry, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, isHeadersSent, arginfo_class_TrueAsync_HttpResponse_isHeadersSent, ZEND_ACC_PUBLIC) ZEND_ME(TrueAsync_HttpResponse, isWritable, arginfo_class_TrueAsync_HttpResponse_isWritable, ZEND_ACC_PUBLIC) - ZEND_ME(TrueAsync_HttpResponse, isClosed, arginfo_class_TrueAsync_HttpResponse_isClosed, ZEND_ACC_PUBLIC) + ZEND_ME(TrueAsync_HttpResponse, isEnded, arginfo_class_TrueAsync_HttpResponse_isEnded, ZEND_ACC_PUBLIC) ZEND_FE_END }; diff --git a/stubs/HttpServerConfig.php b/stubs/HttpServerConfig.php index f640fe7b..c4c371d6 100644 --- a/stubs/HttpServerConfig.php +++ b/stubs/HttpServerConfig.php @@ -371,9 +371,9 @@ public function getDrainCooldownMs(): int {} // === Streaming responses (HTTP/2 Step 5b) === /** - * Per-stream chunk-queue cap for HttpResponse::send() backpressure. + * Per-stream chunk-queue cap for HttpResponse::write() backpressure. * - * When handler's send() call grows the stream's chunk queue past + * When the handler's write() call grows the stream's chunk queue past * this many bytes, the coroutine suspends until nghttp2 drains * enough to drop below. HTTP/2 only; HTTP/1 chunked path uses * the kernel send buffer instead. diff --git a/stubs/HttpServerConfig.php_arginfo.h b/stubs/HttpServerConfig.php_arginfo.h index 0b917937..65340db0 100644 --- a/stubs/HttpServerConfig.php_arginfo.h +++ b/stubs/HttpServerConfig.php_arginfo.h @@ -1,5 +1,5 @@ /* This is a generated file, edit HttpServerConfig.php.stub.php instead. - * Stub hash: 342a57752f851751e705775fa87ef39fa97b365e */ + * Stub hash: 538a3f73aaa3d0daa900ad7e54e17bd5a7b816b0 */ ZEND_BEGIN_ARG_INFO_EX(arginfo_class_TrueAsync_HttpServerConfig___construct, 0, 0, 0) ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, host, IS_STRING, 1, "null") diff --git a/tests/bench/bench_bidi_server.php b/tests/bench/bench_bidi_server.php index ad250491..81d5a32d 100644 --- a/tests/bench/bench_bidi_server.php +++ b/tests/bench/bench_bidi_server.php @@ -5,7 +5,7 @@ * gRPC bidi bench). * * Handler pattern: await the request body, then echo it back in - * 32 KiB chunks via HttpResponse::send(). This exercises: + * 32 KiB chunks via HttpResponse::write(). This exercises: * * - H2 DATA-frame ingestion path (cb_on_data_chunk_recv + the * OOM-guarded smart_str preallocation), @@ -44,7 +44,7 @@ $req->awaitBody(); $body = $req->getBody(); - /* Commit status + headers on first send(); everything afterwards + /* Commit status + headers on first write(); everything afterwards * is DATA frames (Step 4 streaming-OUT). Chunk at 32 KiB so we * exercise WINDOW_UPDATE round-trips — smaller than the default * SETTINGS_INITIAL_WINDOW but large enough that we're not wasting @@ -55,7 +55,7 @@ $len = strlen($body); $chunk = 32 * 1024; for ($off = 0; $off < $len; $off += $chunk) { - $res->send(substr($body, $off, $chunk)); + $res->write(substr($body, $off, $chunk)); } $res->end(); }); diff --git a/tests/perf/servers/server_stream.php b/tests/perf/servers/server_stream.php index 76e0782f..b76bd2cb 100644 --- a/tests/perf/servers/server_stream.php +++ b/tests/perf/servers/server_stream.php @@ -41,10 +41,12 @@ function perf_parse_size(string $s): int return; } $resp->setStatusCode(200) - ->setHeader('Content-Type', 'application/octet-stream') - ->send(); + ->setHeader('Content-Type', 'application/octet-stream'); $payload = str_repeat('x', $chunk); $left = $total; + /* The first write() commits status and headers; a commit call taking no + * chunk never existed, and the one that stood here raised + * ArgumentCountError before the profile measured anything. */ while ($left > 0) { $n = $left < $chunk ? $left : $chunk; $resp->write($n === $chunk ? $payload : substr($payload, 0, $n)); diff --git a/tests/phpt/server/compression/012-h1-streaming-gzip.phpt b/tests/phpt/server/compression/012-h1-streaming-gzip.phpt index 2fe8dbaa..fd89ba9c 100644 --- a/tests/phpt/server/compression/012-h1-streaming-gzip.phpt +++ b/tests/phpt/server/compression/012-h1-streaming-gzip.phpt @@ -26,7 +26,7 @@ $config = (new HttpServerConfig()) $server = new HttpServer($config); -/* Streaming handler emits the same payload over four send() chunks + +/* Streaming handler emits the same payload over four write() chunks + * an end() finaliser. Compression wrapper must produce a single valid * gzip stream regardless of chunk boundaries. */ $payload = str_repeat("Hello, streaming gzip!\n", 100); @@ -34,9 +34,9 @@ $payload = str_repeat("Hello, streaming gzip!\n", 100); $server->addHttpHandler(function ($req, $resp) use ($payload) { $resp->setHeader('Content-Type', 'text/html'); $q = strlen($payload) / 4; - $resp->send(substr($payload, 0, $q)); - $resp->send(substr($payload, $q, $q)); - $resp->send(substr($payload, 2*$q, $q)); + $resp->write(substr($payload, 0, $q)); + $resp->write(substr($payload, $q, $q)); + $resp->write(substr($payload, 2*$q, $q)); $resp->end(substr($payload, 3*$q)); }); diff --git a/tests/phpt/server/compression/013-h1-streaming-gzip-flush.phpt b/tests/phpt/server/compression/013-h1-streaming-gzip-flush.phpt index bd96b158..7b63f424 100644 --- a/tests/phpt/server/compression/013-h1-streaming-gzip-flush.phpt +++ b/tests/phpt/server/compression/013-h1-streaming-gzip-flush.phpt @@ -43,15 +43,15 @@ $server->addHttpHandler(function ($req, $resp) use (&$gate, $head, $tail) { * the gated route above. */ if ($req->getPath() === '/empties') { $resp->setHeader('Content-Type', 'text/html'); - $resp->send(''); - $resp->send(''); - $resp->send($head); + $resp->write(''); + $resp->write(''); + $resp->write($head); $resp->end($tail); return; } $resp->setHeader('Content-Type', 'text/html'); - $resp->send($head); + $resp->write($head); while (!$gate) { delay(10); diff --git a/tests/phpt/server/compression/041-h1-streaming-brotli.phpt b/tests/phpt/server/compression/041-h1-streaming-brotli.phpt index 7ba3d83d..fd8cc80a 100644 --- a/tests/phpt/server/compression/041-h1-streaming-brotli.phpt +++ b/tests/phpt/server/compression/041-h1-streaming-brotli.phpt @@ -45,7 +45,7 @@ $server->addHttpHandler(function ($req, $resp) use ($chunk, $rounds) { } $resp->setHeader('Content-Type', 'text/html'); for ($i = 0; $i < $rounds; $i++) { - $resp->send($chunk); + $resp->write($chunk); } $resp->end(); }); diff --git a/tests/phpt/server/compression/042-h1-streaming-brotli-flush.phpt b/tests/phpt/server/compression/042-h1-streaming-brotli-flush.phpt index 030deedb..bae6cb43 100644 --- a/tests/phpt/server/compression/042-h1-streaming-brotli-flush.phpt +++ b/tests/phpt/server/compression/042-h1-streaming-brotli-flush.phpt @@ -55,7 +55,7 @@ $server->addHttpHandler(function ($req, $resp) use (&$gate, $head, $tail) { } $resp->setHeader('Content-Type', 'text/html'); - $resp->send($head); + $resp->write($head); while (!$gate) { delay(10); diff --git a/tests/phpt/server/compression/051-h1-streaming-zstd-flush.phpt b/tests/phpt/server/compression/051-h1-streaming-zstd-flush.phpt index 56161863..5dd858cd 100644 --- a/tests/phpt/server/compression/051-h1-streaming-zstd-flush.phpt +++ b/tests/phpt/server/compression/051-h1-streaming-zstd-flush.phpt @@ -55,7 +55,7 @@ $server->addHttpHandler(function ($req, $resp) use (&$gate, $head, $tail) { } $resp->setHeader('Content-Type', 'text/html'); - $resp->send($head); + $resp->write($head); while (!$gate) { delay(10); diff --git a/tests/phpt/server/core/023-response-body-api.phpt b/tests/phpt/server/core/023-response-body-api.phpt index 56b9002e..cf7f7e32 100644 --- a/tests/phpt/server/core/023-response-body-api.phpt +++ b/tests/phpt/server/core/023-response-body-api.phpt @@ -27,18 +27,18 @@ $snap = function (string $tag, $val) use (&$lines) { $server->addHttpHandler(function ($req, $res) use ($snap, $server) { // Snapshot value test: getBody() must return a deep copy that - // does NOT change when the body buffer is later mutated by write() + // does NOT change when the body buffer is later mutated by appendBody() // or setBody(). Each $b below is checked AFTER all subsequent // mutations have happened, so any aliasing surfaces as a wrong // value here. $b0 = $res->getBody(); - $res->write('hello '); + $res->appendBody('hello '); $b1 = $res->getBody(); - $res->write('world'); + $res->appendBody('world'); $b2 = $res->getBody(); $res->setBody('replaced'); $b3 = $res->getBody(); - $res->write('+more'); + $res->appendBody('+more'); $b4 = $res->getBody(); $res->setBody(''); $b5 = $res->getBody(); diff --git a/tests/phpt/server/core/025-response-state-api.phpt b/tests/phpt/server/core/025-response-state-api.phpt index b58c3f18..5c568b79 100644 --- a/tests/phpt/server/core/025-response-state-api.phpt +++ b/tests/phpt/server/core/025-response-state-api.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse: state observers — isHeadersSent / isClosed across send + end +HttpResponse: state observers — isHeadersSent / isEnded across setBody + end --EXTENSIONS-- true_async_server true_async @@ -19,23 +19,23 @@ $server = new HttpServer((new HttpServerConfig()) $snap = []; $server->addHttpHandler(function ($req, $res) use (&$snap, $server) { - // Initial state: nothing sent, not closed. + // Initial state: nothing sent, not ended. $snap['init_headers_sent'] = $res->isHeadersSent(); - $snap['init_closed'] = $res->isClosed(); + $snap['init_ended'] = $res->isEnded(); $res->setStatusCode(200) ->setHeader('Content-Type', 'text/plain') ->setBody('state-check'); - // Setting buffer doesn't commit on the wire. + // Setting the buffer doesn't commit on the wire. $snap['post_set_headers_sent'] = $res->isHeadersSent(); - $snap['post_set_closed'] = $res->isClosed(); + $snap['post_set_ended'] = $res->isEnded(); $res->end(); - // After end() the response is closed; isHeadersSent depends on + // After end() the response is ended; isHeadersSent depends on // protocol path (may already be true) — check both consistently. - $snap['post_end_closed'] = $res->isClosed(); + $snap['post_end_ended'] = $res->isEnded(); $server->stop(); }); @@ -65,7 +65,7 @@ connection: close state-check === state === init_headers_sent = false -init_closed = false +init_ended = false post_set_headers_sent = false -post_set_closed = false -post_end_closed = %s +post_set_ended = false +post_end_ended = %s diff --git a/tests/phpt/server/core/062-body-api-names.phpt b/tests/phpt/server/core/062-body-api-names.phpt new file mode 100644 index 00000000..6922e68b --- /dev/null +++ b/tests/phpt/server/core/062-body-api-names.phpt @@ -0,0 +1,160 @@ +--TEST-- +HttpResponse body API — write() streams, appendBody() buffers, and the two modes refuse to mix +--EXTENSIONS-- +true_async_server +true_async +--FILE-- +addListener('127.0.0.1', $port) + ->setReadTimeout(5)->setWriteTimeout(5)); + +$probe = []; +$server->addHttpHandler(function ($req, $res) use (&$probe, $server) { + $path = $req->getPath(); + + if ($path === '/buffered') { + $res->appendBody('one '); + /* Buffered appending commits nothing, so headers stay open. */ + $res->setHeader('X-After-Append', 'yes'); + $res->appendBody('two'); + $probe['buffered_headers_sent'] = $res->isHeadersSent(); + $probe['buffered_body'] = $res->getBody(); + $res->end(); + return; + } + + if ($path === '/mixed') { + $res->appendBody('buffered'); + try { + $res->write('streamed'); + $probe['write_after_append'] = 'NO-THROW'; + } catch (\Throwable $e) { + $probe['write_after_append'] = get_class($e); + } + $res->end(); + return; + } + + if ($path === '/two-chunks') { + $res->write('one-'); + $probe['stream_headers_sent_early'] = $res->isHeadersSent(); + $res->write('two'); + $res->end(); + $probe['stream_ended'] = $res->isEnded(); + return; + } + + $res->write('streamed'); + $probe['stream_headers_sent'] = $res->isHeadersSent(); + try { + $res->setHeader('X-Too-Late', 'yes'); + $probe['header_after_write'] = 'NO-THROW'; + } catch (\Throwable $e) { + $probe['header_after_write'] = get_class($e); + } + $res->end(); + $server->stop(); +}); + +$get = function (int $port, string $path): string { + $fp = stream_socket_client("tcp://127.0.0.1:$port", $errno, $errstr, 2); + stream_set_timeout($fp, 2); + fwrite($fp, "GET $path HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + $buf = ''; + while (!feof($fp)) { + $c = fread($fp, 8192); + if ($c === '' || $c === false) break; + $buf .= $c; + } + fclose($fp); + return preg_replace("/^Date: [^\r\n]*\r?\n/mi", "", $buf); +}; + +$cli = spawn(function () use ($port, $get) { + usleep(30000); + foreach (['/buffered', '/mixed', '/two-chunks', '/streamed'] as $path) { + $wire = $get($port, $path); + [$head, $body] = explode("\r\n\r\n", $wire, 2); + echo "== $path\n"; + echo "content_length=", (preg_match('/^content-length:\s*(\d+)/mi', $head, $m) ? $m[1] : 'none'), "\n"; + echo "chunked=", (int)(bool)preg_match('/^transfer-encoding:\s*chunked/mi', $head), "\n"; + echo "after_append_header=", (int)(bool)preg_match('/^x-after-append:/mi', $head), "\n"; + echo "body=", trim(preg_replace('/^[0-9a-f]+\r\n|\r\n0\r\n\r\n$|\r\n/mi', '', $body)), "\n"; + } +}); + +$server->start(); +await($cli); + +echo "== removed\n"; +echo "send=", (int)method_exists('TrueAsync\\HttpResponse', 'send'), "\n"; +echo "getBodyStream=", (int)method_exists('TrueAsync\\HttpResponse', 'getBodyStream'), "\n"; +echo "setBodyStream=", (int)method_exists('TrueAsync\\HttpResponse', 'setBodyStream'), "\n"; +echo "isClosed=", (int)method_exists('TrueAsync\\HttpResponse', 'isClosed'), "\n"; +echo "sendable=", (int)method_exists('TrueAsync\\HttpResponse', 'sendable'), "\n"; + +echo "== probe\n"; +foreach ($probe as $k => $v) echo "$k = " . var_export($v, true) . "\n"; +?> +--EXPECT-- +== /buffered +content_length=7 +chunked=0 +after_append_header=1 +body=one two +== /mixed +content_length=8 +chunked=0 +after_append_header=0 +body=buffered +== /two-chunks +content_length=none +chunked=1 +after_append_header=0 +body=one-two +== /streamed +content_length=none +chunked=1 +after_append_header=0 +body=streamed +== removed +send=0 +getBodyStream=0 +setBodyStream=0 +isClosed=0 +sendable=1 +== probe +buffered_headers_sent = false +buffered_body = 'one two' +write_after_append = 'TrueAsync\\HttpServerRuntimeException' +stream_headers_sent_early = true +stream_ended = true +stream_headers_sent = true +header_after_write = 'TrueAsync\\HttpServerRuntimeException' diff --git a/tests/phpt/server/h1/013-h1-chunked-basic.phpt b/tests/phpt/server/h1/013-h1-chunked-basic.phpt index c261d836..0e557099 100644 --- a/tests/phpt/server/h1/013-h1-chunked-basic.phpt +++ b/tests/phpt/server/h1/013-h1-chunked-basic.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse::send() — HTTP/1.1 chunked streaming (PLAN_STREAMING Phase 2) +HttpResponse::write() — HTTP/1.1 chunked streaming (PLAN_STREAMING Phase 2) --EXTENSIONS-- true_async_server true_async @@ -27,7 +27,7 @@ $server = new HttpServer( $server->addHttpHandler(function ($req, $res) { $res->setStatusCode(200)->setHeader('Content-Type', 'text/plain'); for ($i = 1; $i <= 5; $i++) { - $res->send("chunk-$i\n"); + $res->write("chunk-$i\n"); } $res->end(); }); diff --git a/tests/phpt/server/h1/014-h1-sse-pattern.phpt b/tests/phpt/server/h1/014-h1-sse-pattern.phpt index 60e28ace..1c2a1423 100644 --- a/tests/phpt/server/h1/014-h1-sse-pattern.phpt +++ b/tests/phpt/server/h1/014-h1-sse-pattern.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse::send() — HTTP/1.1 chunked delivers Server-Sent Events +HttpResponse::write() — HTTP/1.1 chunked delivers Server-Sent Events --EXTENSIONS-- true_async_server true_async @@ -28,9 +28,9 @@ $server->addHttpHandler(function ($req, $res) { $res->setStatusCode(200) ->setHeader('Content-Type', 'text/event-stream') ->setHeader('Cache-Control', 'no-cache'); - $res->send("data: alpha\n\n"); - $res->send("data: bravo\n\n"); - $res->send("data: charlie\n\n"); + $res->write("data: alpha\n\n"); + $res->write("data: bravo\n\n"); + $res->write("data: charlie\n\n"); $res->end(); }); diff --git a/tests/phpt/server/h1/015-h1-stream-edge-cases.phpt b/tests/phpt/server/h1/015-h1-stream-edge-cases.phpt index cbc7a43b..621d9479 100644 --- a/tests/phpt/server/h1/015-h1-stream-edge-cases.phpt +++ b/tests/phpt/server/h1/015-h1-stream-edge-cases.phpt @@ -28,16 +28,16 @@ $server->addHttpHandler(function ($req, $res) use (&$count, $server) { if ($path === '/empty-chunks') { // Empty chunk should be silently dropped (not emit zero-chunk EOF). - $res->send("real1\n"); - $res->send(""); // dropped - $res->send(""); // dropped - $res->send("real2\n"); + $res->write("real1\n"); + $res->write(""); // dropped + $res->write(""); // dropped + $res->write("real2\n"); } elseif ($path === '/no-send') { - // No send() call. end() must still commit headers + zero chunk + // No write() call. end() must still commit headers + zero chunk // (covers h1_stream_mark_ended's "headers-not-sent" branch). } elseif ($path === '/large') { // 8 KB chunk — exercises the hex header path beyond a few digits. - $res->send(str_repeat('A', 8192)); + $res->write(str_repeat('A', 8192)); } else { $res->setStatusCode(404)->setBody('nf'); } diff --git a/tests/phpt/server/h1/016-h1-stream-after-send-many.phpt b/tests/phpt/server/h1/016-h1-stream-after-send-many.phpt index 465b07fb..8ebb9f98 100644 --- a/tests/phpt/server/h1/016-h1-stream-after-send-many.phpt +++ b/tests/phpt/server/h1/016-h1-stream-after-send-many.phpt @@ -25,7 +25,7 @@ $server->addHttpHandler(function ($req, $res) use ($server) { $res->setStatusCode(200)->setHeader('Content-Type', 'text/plain'); // 100 small chunks — exercises the chunk-header sprintf path repeatedly for ($i = 1; $i <= 100; $i++) { - $res->send(sprintf("%03d\n", $i)); + $res->write(sprintf("%03d\n", $i)); } $res->end(); $server->stop(); diff --git a/tests/phpt/server/h1/024-h1-sse-misuse.phpt b/tests/phpt/server/h1/024-h1-sse-misuse.phpt index 9f57e128..d6247cae 100644 --- a/tests/phpt/server/h1/024-h1-sse-misuse.phpt +++ b/tests/phpt/server/h1/024-h1-sse-misuse.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse SSE API — mixing send() and SSE throws (symmetric sse_mode guard) +HttpResponse SSE API — mixing write() and SSE throws (symmetric sse_mode guard) --EXTENSIONS-- true_async_server true_async @@ -9,12 +9,12 @@ if (!shell_exec('which curl')) die('skip curl not installed'); ?> --FILE-- sseEvent()/sseComment()/sseRetry() throw (the + * - write() first -> sseEvent()/sseComment()/sseRetry() throw (the * stream is plain, not text/event-stream), - * - sseStart()/sseEvent() first -> send() throws (the stream is SSE). + * - sseStart()/sseEvent() first -> write() throws (the stream is SSE). * Both raise HttpServerRuntimeException; the handler can catch it and keep * streaming through the channel it already committed to. */ @@ -39,7 +39,7 @@ $mark = function (callable $fn): string { $cls = $e::class; $short = substr($cls, strrpos($cls, '\\') + 1); $m = $e->getMessage(); - $kind = str_contains($m, 'already streaming via send()') ? 'sse-after-send' + $kind = str_contains($m, 'already streaming via write()') ? 'sse-after-send' : (str_contains($m, 'in SSE mode') ? 'send-in-sse' : 'other'); return "$short:$kind"; @@ -49,16 +49,16 @@ $mark = function (callable $fn): string { $server->addHttpHandler(function ($req, $res) use ($mark) { if ($req->getPath() === '/sse-then-send') { $res->sseStart(); - $k = $mark(fn () => $res->send("x")); // SSE committed -> send() throws + $k = $mark(fn () => $res->write("x")); // SSE committed -> write() throws $res->sseEvent($k); // report back over SSE $res->end(); return; } // /send-then-sse - $res->send("a="); // plain stream committed + $res->write("a="); // plain stream committed $k = $mark(fn () => $res->sseEvent("x")); // -> sseEvent() throws - $res->send($k); // report back over the plain stream + $res->write($k); // report back over the plain stream $res->end(); }); diff --git a/tests/phpt/server/h2/013-h2-streaming-basic.phpt b/tests/phpt/server/h2/013-h2-streaming-basic.phpt index b7fc636f..988a617f 100644 --- a/tests/phpt/server/h2/013-h2-streaming-basic.phpt +++ b/tests/phpt/server/h2/013-h2-streaming-basic.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse::send() — HTTP/2 streaming basic round-trip +HttpResponse::write() — HTTP/2 streaming basic round-trip --EXTENSIONS-- true_async_server true_async @@ -10,9 +10,9 @@ h2_skipif(['curl_h2' => true]); ?> --FILE-- addHttpHandler(function ($req, $res) { $res->setStatusCode(200) ->setHeader('Content-Type', 'text/plain'); for ($i = 1; $i <= 5; $i++) { - $res->send("chunk-$i\n"); + $res->write("chunk-$i\n"); } $res->end(); }); diff --git a/tests/phpt/server/h2/014-h2-streaming-large.phpt b/tests/phpt/server/h2/014-h2-streaming-large.phpt index 278c4af2..be44c830 100644 --- a/tests/phpt/server/h2/014-h2-streaming-large.phpt +++ b/tests/phpt/server/h2/014-h2-streaming-large.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse::send() — HTTP/2 streaming multi-chunk body within initial window +HttpResponse::write() — HTTP/2 streaming multi-chunk body within initial window --EXTENSIONS-- true_async_server true_async @@ -39,11 +39,11 @@ $server->addHttpHandler(function ($req, $res) { * multi-chunk queue + data-provider walker work end-to-end. * * Bodies LARGER than the initial window are a Phase 1.1 item - * (needs a DP-triggered wake event so send() can suspend + * (needs a DP-triggered wake event so write() can suspend * properly when flow-control stalls the drain). */ $chunk = str_repeat('A', 4096); for ($i = 0; $i < 12; $i++) { - $res->send($chunk); + $res->write($chunk); } $res->end(); }); diff --git a/tests/phpt/server/h2/015-h2-streaming-backpressure.phpt b/tests/phpt/server/h2/015-h2-streaming-backpressure.phpt index 7bbaf605..005b76c3 100644 --- a/tests/phpt/server/h2/015-h2-streaming-backpressure.phpt +++ b/tests/phpt/server/h2/015-h2-streaming-backpressure.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse::send() — streaming exercises backpressure wake via WINDOW_UPDATE +HttpResponse::write() — streaming exercises backpressure wake via WINDOW_UPDATE --EXTENSIONS-- true_async_server true_async @@ -8,7 +8,7 @@ true_async /* Step 5b Phase 1.1 — the critical test that proves the suspend+ * wake-on-WINDOW_UPDATE path actually works. Server sends 256 KiB * in 32 KiB chunks; that's 4× the default 64 KiB stream initial - * window. Handler must suspend in send() when drain stalls, then + * window. Handler must suspend in write() when drain stalls, then * wake each time our client sends WINDOW_UPDATE. Byte-exact hash * verifies nothing was lost. * @@ -40,7 +40,7 @@ $server->addHttpHandler(function ($req, $res) { * — forces the suspend loop in h2_stream_append_chunk. */ $chunk = str_repeat('A', 32768); for ($i = 0; $i < 8; $i++) { - $res->send($chunk); + $res->write($chunk); } $res->end(); }); diff --git a/tests/phpt/server/h2/016-h2-streaming-telemetry.phpt b/tests/phpt/server/h2/016-h2-streaming-telemetry.phpt index 17f763a6..a18631d6 100644 --- a/tests/phpt/server/h2/016-h2-streaming-telemetry.phpt +++ b/tests/phpt/server/h2/016-h2-streaming-telemetry.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpServer: streaming telemetry counters advance on send() / reset() clears +HttpServer: streaming telemetry counters advance on write() / reset() clears --EXTENSIONS-- true_async_server true_async @@ -10,7 +10,7 @@ h2_skipif(['curl_h2' => true]); ?> --FILE-- addHttpHandler(function ($req, $res) { if ($path === '/stream') { $res->setStatusCode(200) ->setHeader('Content-Type', 'text/plain'); - $res->send("aaa"); // 3 bytes - $res->send("bbbbb"); // 5 bytes + $res->write("aaa"); // 3 bytes + $res->write("bbbbb"); // 5 bytes $res->end(); } else { $res->setStatusCode(200)->setBody("buffered\n")->end(); @@ -53,7 +53,7 @@ $client = spawn(function () use ($port, $server) { echo "after-buffered stream_send_calls=", $t0['stream_send_calls_total'], "\n"; echo "after-buffered stream_bytes_sent=", $t0['stream_bytes_sent_total'], "\n"; - /* Two streaming requests, each with 2 send() calls, 8 bytes total. */ + /* Two streaming requests, each with 2 write() calls, 8 bytes total. */ exec(sprintf('curl --http2-prior-knowledge -s --max-time 3 http://127.0.0.1:%d/stream -o /dev/null', $port)); exec(sprintf('curl --http2-prior-knowledge -s --max-time 3 http://127.0.0.1:%d/stream -o /dev/null', $port)); diff --git a/tests/phpt/server/h2/017-h2-streaming-cancel.phpt b/tests/phpt/server/h2/017-h2-streaming-cancel.phpt index 46a413a9..f2180099 100644 --- a/tests/phpt/server/h2/017-h2-streaming-cancel.phpt +++ b/tests/phpt/server/h2/017-h2-streaming-cancel.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpResponse::send() — HTTP/2 peer RST mid-stream surfaces as HttpException(499) +HttpResponse::write() — HTTP/2 peer RST mid-stream surfaces as HttpException(499) --EXTENSIONS-- true_async_server true_async @@ -11,15 +11,15 @@ h2_skipif(['curl_h2' => true]); --FILE-- addHttpHandler(function ($req, $res) try { $res->setStatusCode(200)->setHeader('Content-Type', 'text/plain'); /* Stream slowly enough that curl's --max-time kills us between - * chunks. One send() per iteration, delay in between. */ + * chunks. One write() per iteration, delay in between. */ for ($i = 0; $i < 20; $i++) { - $res->send("chunk-$i\n"); + $res->write("chunk-$i\n"); $chunks_sent++; delay(100); } @@ -80,7 +80,7 @@ $client = spawn(function () use ($port, $server) { * RST_STREAM emission differs from POSIX). The pure-PHP H2 client * gives the test exact frame-level control: open a stream, wait * for the first DATA frame from the server (handler is past the - * first send()), then send RST_STREAM. The server's + * first write()), then send RST_STREAM. The server's * cb_on_stream_close fires the same code path the curl variant * was exercising. */ $cli = new H2TestClient('127.0.0.1', $port); @@ -95,7 +95,7 @@ $client = spawn(function () use ($port, $server) { continue; } /* First DATA frame on our stream → handler made it past the - * first send(); RST_STREAM now reaches it suspended between + * first write(); RST_STREAM now reaches it suspended between * chunks, mirroring the curl --max-time-during-stream window. */ if ($type === H2_FRAME_DATA && $sid_in === $sid && !$rst_sent) { $cli->sendRstStream($sid, /* CANCEL */ 0x08); diff --git a/tests/phpt/server/h2/022-h2-streaming-ring-full.phpt b/tests/phpt/server/h2/022-h2-streaming-ring-full.phpt index 66469606..375b533d 100644 --- a/tests/phpt/server/h2/022-h2-streaming-ring-full.phpt +++ b/tests/phpt/server/h2/022-h2-streaming-ring-full.phpt @@ -1,11 +1,11 @@ --TEST-- -HttpResponse::send() — streaming fills the 16-slot chunk ring, producer suspends on full +HttpResponse::write() — streaming fills the 16-slot chunk ring, producer suspends on full --EXTENSIONS-- true_async_server true_async --FILE-- > 16 * forces h2_stream_append_chunk's suspend-on-full branch and the @@ -13,7 +13,7 @@ true_async * cycles. * * The single-threaded scheduler makes the ring-fill deterministic: the - * handler runs its send() loop uninterrupted until the ring is full and + * handler runs its write() loop uninterrupted until the ring is full and * it suspends, so the client physically cannot credit the flow-control * window before the suspend has happened at least once. * @@ -50,7 +50,7 @@ $server = new HttpServer($config); $server->addHttpHandler(function ($req, $res) use ($CHUNK_SZ, $N_CHUNKS) { $res->setStatusCode(200)->setHeader('Content-Type', 'application/octet-stream'); for ($i = 0; $i < $N_CHUNKS; $i++) { - $res->send(str_repeat(chr(33 + ($i % 90)), $CHUNK_SZ)); + $res->write(str_repeat(chr(33 + ($i % 90)), $CHUNK_SZ)); } $res->end(); }); diff --git a/tests/phpt/server/h2/023-h2-streaming-sendable.phpt b/tests/phpt/server/h2/023-h2-sendable-tombstone.phpt similarity index 56% rename from tests/phpt/server/h2/023-h2-streaming-sendable.phpt rename to tests/phpt/server/h2/023-h2-sendable-tombstone.phpt index d7f7478c..24029b7f 100644 --- a/tests/phpt/server/h2/023-h2-streaming-sendable.phpt +++ b/tests/phpt/server/h2/023-h2-sendable-tombstone.phpt @@ -1,20 +1,19 @@ --TEST-- -HttpResponse::sendable() — advisory backpressure check flips under a full ring +HttpResponse::sendable() — the tombstone throws on a live stream --EXTENSIONS-- true_async_server true_async --FILE-- false, 'false' => false]; +/* Shared with the handler — what the tombstone raised. */ +$obs = ['class' => '', 'message' => '']; $config = (new HttpServerConfig()) ->addListener('127.0.0.1', $port) @@ -48,12 +47,14 @@ $server = new HttpServer($config); $server->addHttpHandler(function ($req, $res) use ($CHUNK_SZ, $N_CHUNKS, &$obs) { $res->setStatusCode(200)->setHeader('Content-Type', 'application/octet-stream'); for ($i = 0; $i < $N_CHUNKS; $i++) { - if ($res->sendable()) { - $obs['true'] = true; - } else { - $obs['false'] = true; + try { + $res->sendable(); + $obs['class'] = 'NO-THROW'; + } catch (\Throwable $e) { + $obs['class'] = get_class($e); + $obs['message'] = $e->getMessage(); } - $res->send(str_repeat(chr(33 + ($i % 90)), $CHUNK_SZ)); + $res->write(str_repeat(chr(33 + ($i % 90)), $CHUNK_SZ)); } $res->end(); }); @@ -79,8 +80,8 @@ $client = spawn(function () use ($port, $server, $expected) { $server->start(); await($client); -echo "saw_sendable_true=", (int)$obs['true'], "\n"; -echo "saw_sendable_false=", (int)$obs['false'], "\n"; +echo "class=", $obs['class'], "\n"; +echo "message=", $obs['message'], "\n"; echo "done\n"; ?> --EXPECT-- @@ -88,6 +89,6 @@ status=200 len=393216 ended=1 hash_match=1 -saw_sendable_true=1 -saw_sendable_false=1 +class=TrueAsync\HttpServerRuntimeException +message=sendable() is gone: it answered liveness and queue depth with one bool. Use isWritable() for liveness, tryWrite()/awaitWritable() for room done diff --git a/tests/phpt/server/h2/025-h2-try-write.phpt b/tests/phpt/server/h2/025-h2-try-write.phpt new file mode 100644 index 00000000..e80c5453 --- /dev/null +++ b/tests/phpt/server/h2/025-h2-try-write.phpt @@ -0,0 +1,107 @@ +--TEST-- +HttpResponse::tryWrite() — false when the ring is full, and a refusal queues nothing +--EXTENSIONS-- +true_async_server +true_async +--FILE-- +addListener('127.0.0.1', $port) + ->setReadTimeout(15) + ->setWriteTimeout(15); + +$server = new HttpServer($config); +$server->addHttpHandler(function ($req, $res) use ($CHUNK_SZ, $N_CHUNKS, &$refused, &$waited, &$fellBack) { + $res->setStatusCode(200)->setHeader('Content-Type', 'application/octet-stream'); + + for ($i = 0; $i < $N_CHUNKS; $i++) { + $chunk = str_repeat(chr(33 + ($i % 90)), $CHUNK_SZ); + + if (!$res->tryWrite($chunk)) { + $refused++; + + /* Wait for room instead of spinning, then offer the same bytes + * again. If awaitWritable() returned without waiting, the retry + * below would be refused too and $fellBack would rise — which is + * what the expected output rules out. */ + if ($res->awaitWritable(5000)) { + $waited++; + } + + if (!$res->tryWrite($chunk)) { + $fellBack++; + $res->write($chunk); + } + } + } + + $res->end(); +}); + +$client = spawn(function () use ($port, $server, $expected) { + usleep(50000); + try { + $cli = new H2TestClient('127.0.0.1', $port, 15); + $sid = $cli->sendRequest('GET', '/stream', "127.0.0.1:$port"); + [$status, $body, $trailers, $ended] = $cli->collectResponse($sid, true); + $cli->close(); + + echo "status=$status\n"; + echo "len=", strlen($body), "\n"; + echo "hash_match=", (sha1($body) === sha1($expected) ? 1 : 0), "\n"; + } catch (\Throwable $e) { + echo "ERR: ", $e->getMessage(), "\n"; + } + $server->stop(); +}); + +$server->start(); +await($client); + +echo "refused=", $refused > 0 ? 1 : 0, "\n"; +echo "waited=", $waited > 0 ? 1 : 0, "\n"; +echo "fell_back=", $fellBack, "\n"; +echo "done\n"; +?> +--EXPECT-- +status=200 +len=393216 +hash_match=1 +refused=1 +waited=1 +fell_back=0 +done diff --git a/tests/phpt/server/h2/027-h2-streaming-trailers.phpt b/tests/phpt/server/h2/027-h2-streaming-trailers.phpt index 076ab491..a69a308d 100644 --- a/tests/phpt/server/h2/027-h2-streaming-trailers.phpt +++ b/tests/phpt/server/h2/027-h2-streaming-trailers.phpt @@ -10,7 +10,7 @@ h2_skipif(['curl_h2' => true]); ?> --FILE-- addHttpHandler(function($req, $resp) { $resp->setStatusCode(200) ->setHeader('Content-Type', 'application/grpc'); - /* Streaming send() path: commits HEADERS, then two DATA frames. */ - $resp->send('msg-one'); - $resp->send('msg-two'); + /* Streaming write() path: commits HEADERS, then two DATA frames. */ + $resp->write('msg-one'); + $resp->write('msg-two'); /* Trailers set before the stream ends — carried by the terminal * HEADERS(trailers) frame that mark_ended now emits. */ $resp->setTrailer('grpc-status', '0') diff --git a/tests/phpt/server/h3/011-h3-e2e-streaming-send.phpt b/tests/phpt/server/h3/011-h3-e2e-streaming-send.phpt index 1beecd19..1a052235 100644 --- a/tests/phpt/server/h3/011-h3-e2e-streaming-send.phpt +++ b/tests/phpt/server/h3/011-h3-e2e-streaming-send.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpServer: HTTP/3 streaming response — HttpResponse::send() loop, multi-chunk DATA +HttpServer: HTTP/3 streaming response — HttpResponse::write() loop, multi-chunk DATA --EXTENSIONS-- true_async_server true_async @@ -10,7 +10,7 @@ h3_skipif(['openssl_cli' => true, 'h3client' => true]); ?> --FILE-- send() +/* Step 5b regression — handler streams a response via $res->write() * loop, exercising: * - h3_stream_ops.append_chunk first-call HEADERS commit + queue alloc * - h3_read_data_cb chunk_queue branch + chunk_read_idx walking @@ -57,7 +57,7 @@ $config = (new HttpServerConfig()) $server = new HttpServer($config); $server->addHttpHandler(function ($req, $res) use ($chunks) { $res->setStatusCode(200)->setHeader('content-type', 'application/octet-stream'); - foreach ($chunks as $c) { $res->send($c); } + foreach ($chunks as $c) { $res->write($c); } $res->end(); }); diff --git a/tests/phpt/server/h3/047-h3-reactor-pool-streaming.phpt b/tests/phpt/server/h3/047-h3-reactor-pool-streaming.phpt index f98e9cb3..1c2e9b75 100644 --- a/tests/phpt/server/h3/047-h3-reactor-pool-streaming.phpt +++ b/tests/phpt/server/h3/047-h3-reactor-pool-streaming.phpt @@ -1,5 +1,5 @@ --TEST-- -HttpServer: streaming send() crosses the reactor/worker split (#80, gated pool) +HttpServer: streaming write() crosses the reactor/worker split (#80, gated pool) --EXTENSIONS-- true_async_server true_async @@ -14,10 +14,10 @@ TRUE_ASYNC_SERVER_REACTOR_POOL=1 PHP_HTTP3_DISABLE_RETRY=1 --FILE-- addHttpHandler(function ($req, $res) { ->setHeader('content-type', 'text/plain; charset=utf-8'); for ($i = 1; $i <= 5; $i++) { - $res->send("chunk{$i};"); + $res->write("chunk{$i};"); } $res->end(); diff --git a/tests/phpt/server/h3/048-h3-reactor-pool-backpressure.phpt b/tests/phpt/server/h3/048-h3-reactor-pool-backpressure.phpt index e750a142..1a962d1a 100644 --- a/tests/phpt/server/h3/048-h3-reactor-pool-backpressure.phpt +++ b/tests/phpt/server/h3/048-h3-reactor-pool-backpressure.phpt @@ -54,7 +54,7 @@ $server->addHttpHandler(function ($req, $res) use ($chunks, $chunk_len) { for ($i = 0; $i < $chunks; $i++) { /* Deterministic per-chunk fill so truncation/reorder breaks the hash. */ - $res->send(str_repeat(chr(65 + ($i % 26)), $chunk_len)); + $res->write(str_repeat(chr(65 + ($i % 26)), $chunk_len)); } $res->end(); diff --git a/tests/phpt/server/h3/050-h3-reactor-pool-mailbox-overflow.phpt b/tests/phpt/server/h3/050-h3-reactor-pool-mailbox-overflow.phpt index f35b2991..95a4093e 100644 --- a/tests/phpt/server/h3/050-h3-reactor-pool-mailbox-overflow.phpt +++ b/tests/phpt/server/h3/050-h3-reactor-pool-mailbox-overflow.phpt @@ -56,7 +56,7 @@ $server->addHttpHandler(function ($req, $res) use ($chunks) { ->setHeader('content-type', 'text/plain; charset=utf-8'); for ($i = 1; $i <= $chunks; $i++) { - $res->send("chunk{$i};"); + $res->write("chunk{$i};"); } $res->end(); diff --git a/tests/phpt/server/sendfile/003-sendfile-sealed.phpt b/tests/phpt/server/sendfile/003-sendfile-sealed.phpt index d580d5fd..0e1445de 100644 --- a/tests/phpt/server/sendfile/003-sendfile-sealed.phpt +++ b/tests/phpt/server/sendfile/003-sendfile-sealed.phpt @@ -32,6 +32,7 @@ $server->addHttpHandler(function ($req, $res) use ($tmp) { ['resetHeaders', fn() => $res->resetHeaders()], ['setBody', fn() => $res->setBody('x')], ['write', fn() => $res->write('x')], + ['appendBody', fn() => $res->appendBody('x')], ['json', fn() => $res->json(['a'=>1])], ['html', fn() => $res->html('

')], ['redirect', fn() => $res->redirect('/ok')], @@ -80,6 +81,7 @@ addHeader: throw resetHeaders: throw setBody: throw write: throw +appendBody: throw json: throw html: throw redirect: throw diff --git a/tests/phpt/server/telemetry/009-getstats-contract.phpt b/tests/phpt/server/telemetry/009-getstats-contract.phpt index 093c827f..9766e0d9 100644 --- a/tests/phpt/server/telemetry/009-getstats-contract.phpt +++ b/tests/phpt/server/telemetry/009-getstats-contract.phpt @@ -47,8 +47,8 @@ $server = new HttpServer($config); $server->addHttpHandler(function ($req, $res) { if ($req->getPath() === '/stream') { $res->setStatusCode(200); - $res->send('abc'); - $res->send('de'); + $res->write('abc'); + $res->write('de'); $res->end(); return; } diff --git a/tests/phpt/websocket/025-frame-order-under-control-frames.phpt b/tests/phpt/websocket/025-frame-order-under-control-frames.phpt new file mode 100644 index 00000000..553378c1 --- /dev/null +++ b/tests/phpt/websocket/025-frame-order-under-control-frames.phpt @@ -0,0 +1,210 @@ +--TEST-- +WebSocket H1: data frames keep their order while auto-PONGs are emitted through the other connection writer +--EXTENSIONS-- +true_async_server +true_async +--FILE-- +out_pending_buf + * when a write is already in flight and flushes them from the completion + * callback. + * + * Both drain the same wslay byte stream, so if a parked tail were overtaken + * by a later direct submit, the stream would desynchronise exactly as chunked + * framing does. This test floods the server with PINGs while it pushes a + * numbered burst the client is not reading, then checks that every data frame + * arrived, in order, and that the stream parsed at all. */ + +use TrueAsync\HttpServer; +use TrueAsync\HttpServerConfig; +use TrueAsync\WebSocket; +use TrueAsync\HttpRequest; +use function Async\spawn; +use function Async\await; + +require_once __DIR__ . '/../server/_free_port.inc'; + +const N_MSG = 120; +const PAYLOAD = 4096; +const N_PINGS = 200; + +$port = tas_free_port(); +$config = (new HttpServerConfig()) + ->addListener('127.0.0.1', $port) + ->setReadTimeout(10) + ->setWriteTimeout(10) + ->setWsPingIntervalMs(0); // only the client's PINGs drive the internal path + +$server = new HttpServer($config); + +$server->addWebSocketHandler(function (WebSocket $ws, HttpRequest $req) { + $ws->recv(); // wait for "go" + + $pad = str_repeat('.', PAYLOAD); + + for ($i = 0; $i < N_MSG; $i++) { + $ws->send($i . '|' . $pad); + } + + $ws->recv(); // hold the connection open for the reader +}); + +$server->addHttpHandler(function ($req, $resp) { $resp->setStatusCode(404)->end(); }); + +function ws_client_frame(int $opcode, string $payload): string { + $mask = random_bytes(4); + $masked = ''; + + for ($i = 0, $n = strlen($payload); $i < $n; $i++) { + $masked .= chr(ord($payload[$i]) ^ ord($mask[$i & 3])); + } + + $len = strlen($payload); + + if ($len < 126) { + $head = chr(0x80 | $opcode) . chr(0x80 | $len); + } else { + $head = chr(0x80 | $opcode) . chr(0x80 | 126) . pack('n', $len); + } + + return $head . $mask . $masked; +} + +/** Read exactly $n bytes or return null. */ +function read_n($fp, int $n): ?string { + $buf = ''; + + while (strlen($buf) < $n) { + $c = fread($fp, $n - strlen($buf)); + + if ($c === '' || $c === false) { + return null; + } + + $buf .= $c; + } + + return $buf; +} + +/** Read one server frame (unmasked); [opcode, payload] or null at EOF. */ +function read_frame($fp): ?array { + $hdr = read_n($fp, 2); + + if ($hdr === null) { + return null; + } + + $opcode = ord($hdr[0]) & 0x0f; + $len = ord($hdr[1]) & 0x7f; + + if ($len === 126) { + $ext = read_n($fp, 2); + if ($ext === null) return null; + $len = unpack('n', $ext)[1]; + } elseif ($len === 127) { + $ext = read_n($fp, 8); + if ($ext === null) return null; + $len = unpack('J', $ext)[1]; + } + + $data = $len > 0 ? read_n($fp, $len) : ''; + + if ($data === null) { + return null; + } + + return [$opcode, $data]; +} + +$client = spawn(function () use ($port, $server) { + usleep(20000); + $fp = stream_socket_client("tcp://127.0.0.1:$port", $errno, $errstr, 3); + stream_set_timeout($fp, 5); + fwrite($fp, + "GET / HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + . "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n"); + + $hs = ''; + while (!str_contains($hs, "\r\n\r\n")) { + $c = fread($fp, 4096); + if ($c === '' || $c === false) break; + $hs .= $c; + } + + /* Start the burst, then flood control frames without reading: the socket + * buffer fills, writes stop completing inline, and the two writers are + * live at the same time. */ + fwrite($fp, ws_client_frame(0x1, 'go')); + + /* Spaced, not batched: each PING arrives in its own read callback, so + * each auto-PONG is its own flush through the internal writer. A flush + * that lands while the previous one is still in flight is the one that + * parks bytes in the pending tail — the state a later direct submit + * could overtake. */ + for ($i = 0; $i < N_PINGS; $i++) { + fwrite($fp, ws_client_frame(0x9, 'p' . $i)); + usleep(1500); + } + + usleep(200000); + + $seen = []; + $pongs = 0; + $garbled = 0; + + while (count($seen) < N_MSG) { + $frame = read_frame($fp); + + if ($frame === null) { + break; + } + + [$opcode, $payload] = $frame; + + if ($opcode === 0xa) { + $pongs++; + continue; + } + + if ($opcode !== 0x1) { + continue; + } + + $sep = strpos($payload, '|'); + + if ($sep === false || strlen($payload) !== $sep + 1 + PAYLOAD) { + $garbled++; + continue; + } + + $seen[] = (int) substr($payload, 0, $sep); + } + + fclose($fp); + usleep(20000); + $server->stop(); + + $expected = range(0, N_MSG - 1); + + return [count($seen), $seen === $expected ? 1 : 0, $garbled, $pongs > 0 ? 1 : 0]; +}); + +$server->start(); +[$count, $inOrder, $garbled, $sawPong] = await($client); + +echo "messages: $count of ", N_MSG, "\n"; +echo "in order: $inOrder\n"; +echo "garbled: $garbled\n"; +echo "saw pong: $sawPong\n"; +echo "Done\n"; +?> +--EXPECT-- +messages: 120 of 120 +in order: 1 +garbled: 0 +saw pong: 1 +Done