From 04aa3adc9a0e007e70eb25d41b743865deb177b6 Mon Sep 17 00:00:00 2001 From: Edmond <1571649+edmonddantes@users.noreply.github.com> Date: Thu, 9 Jul 2026 07:29:03 +0000 Subject: [PATCH] =?UTF-8?q?fix(grpc/http):=20review=20follow-ups=20?= =?UTF-8?q?=E2=80=94=20mask=20drift,=20body=20caps,=20flow-control,=20HEAD?= =?UTF-8?q?,=20compression=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the merged gRPC branch (#104) surfaced 7 confirmed bugs plus layering/cleanup debt. Fixes, no behaviour change to passing paths: Correctness: - protocol→mask: single http_protocol_registration_mask() used by both addXHandler registration and the worker-side transit rebuild, so a gRPC-only server keeps the HTTP2 bit on worker threads (h2c no longer silently refused). - H2/H3 streaming request bodies: restore the cumulative max_body_size ceiling that a live-only cap had dropped; waived per-stream only for gRPC (unbounded by design). In-flight cap still bounds memory. - H3 buffered→streaming upgrade: track body_precredited so spliced prefix bytes aren't flow-control-credited twice. - HEAD: response->is_head stamped at dispatch; send() drops chunks so no DATA follows the header block (RFC 9110 §9.3.2) on every transport. - grpc-timeout: clamp to UINT64_MAX instead of uint64 overflow wrap. - worker sink: check STREAM_HEADERS post result before building the chunk wire; replace the 100ms blocking nanosleep retry with a hidden one-shot timer + per-thread FIFO so the worker thread never blocks. - reactor_pool_exec: escape the done-wait loop when the reactor reaches DONE (latent shutdown-hang guard). Layering / performance: - Classify gRPC once: http_request_classify_protocols() stamps grpc_mode + body policy flags at headers-complete; transports read neutral flags and no longer include grpc.h or re-parse content-type per readMessage. - Compression API: setGrpcEncoding('gzip') declared before the first message (per-call, matching grpc-go/java/c++); drop the per-message writeMessage($msg, $compress) arg — a compressed frame with no declared encoding can no longer be expressed. - Event-driven credit backpressure: reactor wakes the parked producer via stream_credit_wake instead of 2–32ms timer polling; same for the H3 static throttle hysteresis wake. Cleanup: - response_wire_discard() replaces three copy-pasted discard blocks. - async_coroutine_sleep_ms() shared helper (was 4 near-identical copies). - http_protocol_pick_handler() in the H3 coroutine entry. - delete the dead http3_build_listener_local wrapper. --- CHANGELOG.md | 8 +- README.md | 5 +- include/core/async_plain_event.h | 5 + include/core/stream_credit.h | 42 ++- include/core/worker_dispatch.h | 6 + include/http1/http_parser.h | 11 + include/php_http_server.h | 4 + src/core/async_plain_event.c | 16 ++ src/core/http_connection.c | 2 + src/core/http_protocol_handlers.c | 19 ++ src/core/http_protocol_handlers.h | 12 + src/core/http_protocol_strategy.h | 19 ++ src/core/reactor_pool.c | 7 + src/core/worker_dispatch.c | 109 ++++---- src/grpc/grpc.c | 14 +- src/grpc/grpc.h | 10 +- src/http1/http_parser.c | 6 +- src/http2/http2_session.c | 22 +- src/http2/http2_strategy.c | 2 + src/http3/http3_callbacks.c | 23 +- src/http3/http3_connection.c | 26 +- src/http3/http3_dispatch.c | 32 +-- src/http3/http3_internal.h | 9 - src/http3/http3_io.c | 4 +- src/http3/http3_static_response.c | 59 ++++- src/http_body_stream.c | 23 +- src/http_request.c | 2 +- src/http_response.c | 86 ++++-- src/http_response_internal.h | 3 + src/http_response_server_api.c | 5 + src/http_server_class.c | 248 +++++++++++++----- src/log/http_log.c | 72 +++-- stubs/HttpResponse.php | 35 ++- stubs/HttpResponse.php_arginfo.h | 7 +- .../server/grpc/008-grpc-compression.phpt | 14 +- 35 files changed, 679 insertions(+), 288 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 27b2ae87..e42b1295 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,9 +25,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 handler that writes no messages gets the canonical Trailers-Only reply. - **grpc-web (binary)**: `application/grpc-web` calls carry their trailers in-body as a `0x80`-flagged frame, on H2 and H3. - - **Per-message gzip**: inbound `grpc-encoding: gzip` messages inflate - transparently in `readMessage()`; `writeMessage(..., compress: true)` - emits compressed frames. + - **gzip message encoding**: inbound `grpc-encoding: gzip` messages inflate + transparently in `readMessage()`; `setGrpcEncoding('gzip')` declared + before the first `writeMessage()` compresses every reply frame (per-call + declaration, mirroring grpc-go/java/C++ — a compressed frame without a + declared encoding cannot be expressed). - **`grpc-timeout`** request header parsed and exposed via `HttpRequest::getGrpcTimeout()`. - **grpc-web-text**: `application/grpc-web-text` calls carry base64 both diff --git a/README.md b/README.md index 5656a0b2..d19dfdcd 100644 --- a/README.md +++ b/README.md @@ -308,8 +308,9 @@ Unary and all three streaming shapes use this one API; `grpc-status` / `grpc-message` ride real HTTP trailers (or the in-body `0x80` frame for grpc-web, base64-encoded per frame for grpc-web-text), uncaught exceptions map to `13 INTERNAL`, inbound `grpc-encoding: gzip` messages inflate -transparently, and `writeMessage($msg, compress: true)` gzips replies. The -client's deadline is exposed via `$request->getGrpcTimeout()`. +transparently, and `setGrpcEncoding('gzip')` before the first message gzips +every reply frame. The client's deadline is exposed via +`$request->getGrpcTimeout()`. Working examples live under [`examples/`](examples/): [`minimal-server.php`](examples/minimal-server.php), diff --git a/include/core/async_plain_event.h b/include/core/async_plain_event.h index edb72d38..083ab1c6 100644 --- a/include/core/async_plain_event.h +++ b/include/core/async_plain_event.h @@ -21,4 +21,9 @@ static zend_always_inline void async_plain_event_fire(zend_async_event_t *event) } } +/* Suspend `co` for `ms` on a one-shot timer; the worker loop keeps draining + * (mailbox, scheduler) meanwhile. Leaves a resume exception in EG for the + * caller to inspect or clear (a create failure is cleared internally). */ +void async_coroutine_sleep_ms(zend_coroutine_t *co, zend_ulong ms); + #endif /* ASYNC_PLAIN_EVENT_H */ diff --git a/include/core/stream_credit.h b/include/core/stream_credit.h index 48d8e0ec..44cd83c0 100644 --- a/include/core/stream_credit.h +++ b/include/core/stream_credit.h @@ -14,6 +14,7 @@ #include #include "Zend/zend_atomic.h" +#include "Zend/zend_async_API.h" /* zend_async_trigger_event_t */ /* Per-stream flow-control credit shared worker↔reactor. Malloc + atomics * only — the one object both threads touch. Worker creates it with two refs @@ -24,6 +25,8 @@ typedef struct { zend_atomic_int64 acked; /* bytes the reactor has retired (peer ACK) */ zend_atomic_int dead; /* stream died — producer must stop waiting */ zend_atomic_int refs; + zend_atomic_ptr waker; /* worker-owned trigger the reactor signals */ + zend_atomic_int waker_busy; /* a signaller is inside trigger() */ } stream_credit_t; static inline stream_credit_t *stream_credit_create(void) @@ -37,6 +40,8 @@ static inline stream_credit_t *stream_credit_create(void) ZEND_ATOMIC_INT64_INIT(&sc->acked, 0); ZEND_ATOMIC_INT_INIT(&sc->dead, 0); ZEND_ATOMIC_INT_INIT(&sc->refs, 2); /* worker ctx + reactor stream */ + ZEND_ATOMIC_PTR_INIT(&sc->waker, NULL); + ZEND_ATOMIC_INT_INIT(&sc->waker_busy, 0); return sc; } @@ -66,11 +71,46 @@ static inline void stream_credit_mark_dead(stream_credit_t *sc) zend_atomic_int_store_ex(&sc->dead, 1); } -/* NULL-safe. Dead must be set before release so a parked producer sees it. */ +/* Signal the parked producer from any thread. busy brackets the load+trigger + * so stream_credit_clear_waker can't dispose the event mid-signal. */ +static inline void stream_credit_wake(stream_credit_t *sc) +{ + zend_atomic_int_store_ex(&sc->waker_busy, 1); + + zend_async_trigger_event_t *const t = + (zend_async_trigger_event_t *)zend_atomic_ptr_load_ex(&sc->waker); + + if (t != NULL) { + t->trigger(t); + } + + zend_atomic_int_store_ex(&sc->waker_busy, 0); +} + +/* Worker thread: publish/retract the park trigger. clear must complete + * before the worker disposes the event. */ +static inline void stream_credit_set_waker(stream_credit_t *sc, + zend_async_trigger_event_t *t) +{ + zend_atomic_ptr_store_ex(&sc->waker, t); +} + +static inline void stream_credit_clear_waker(stream_credit_t *sc) +{ + zend_atomic_ptr_store_ex(&sc->waker, NULL); + + /* wait out a signal that loaded the pointer just before the NULL store */ + while (zend_atomic_int_load_ex(&sc->waker_busy) != 0) { + } +} + +/* NULL-safe. Dead must be set before release so a parked producer sees it; + * wake so it sees it NOW instead of at its next timeout. */ static inline void stream_credit_abandon(stream_credit_t *sc) { if (sc != NULL) { stream_credit_mark_dead(sc); + stream_credit_wake(sc); stream_credit_release(sc); } } diff --git a/include/core/worker_dispatch.h b/include/core/worker_dispatch.h index 13429a3c..3a13ccd3 100644 --- a/include/core/worker_dispatch.h +++ b/include/core/worker_dispatch.h @@ -38,6 +38,12 @@ typedef struct http_server_object http_server_object; * the stream (STREAM_* fragments must never be dropped silently). */ typedef bool (*worker_response_sink_fn)(response_wire_t *rw, void *sink_arg); +/* Discard an undeliverable wire: abandon its credit (marks the producer's + * stream dead), release the owned chunk, free the wire. Zend-side companion + * to response_wire_free — every drop site must go through this so a new + * owned field can't leak from a forgotten copy. */ +void response_wire_discard(response_wire_t *rw); + /* Take ownership of `req` (a persistent reactor-built or ZMM request, refcount * 1), wrap it in an HttpRequest on THIS (worker) thread, spawn the user handler * coroutine in `scope`, and when it finishes render the HttpResponse into a diff --git a/include/http1/http_parser.h b/include/http1/http_parser.h index 897bd980..02e2b08f 100644 --- a/include/http1/http_parser.h +++ b/include/http1/http_parser.h @@ -215,6 +215,13 @@ struct http_request_t { void *body_h3_conn; int64_t body_h3_stream_id; size_t body_h3_uncredited; /* drained, not yet flushed */ + + /* Bytes spliced into the queue by a buffered→streaming upgrade whose + * flow-control credit was already returned while they sat in the + * transport's buffer (H2 buffered consume / H3 immediate extend). + * http_body_stream_pop drains this before granting new credit so the + * same bytes are never credited twice. */ + size_t body_precredited; int32_t body_h2_consume_pending; void *body_h3_stream; @@ -235,6 +242,10 @@ struct http_request_t { bool body_streaming; bool body_eof; bool body_error; + + /* grpc_mode_t stamped once at headers-complete; body policy derives + * from it (http_request_body_must_buffer / _size_uncapped). */ + uint8_t grpc_mode; /* TODO(issue #26 backpressure): set true when on_body trips * llhttp_pause; readBody clears it via llhttp_resume below the * low-water mark. Unused by the MVP — see TODO in on_body. */ diff --git a/include/php_http_server.h b/include/php_http_server.h index e0ddc6e4..1e1c1cbb 100644 --- a/include/php_http_server.h +++ b/include/php_http_server.h @@ -1126,6 +1126,10 @@ zend_string *http_response_format_static_head(zend_object *obj, bool include_inline_body); void http_response_set_socket(zend_object *obj, php_socket_t fd); 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 + * chunks silently when set. Stamped at dispatch wherever the request is + * known. */ +void http_response_set_head(zend_object *obj, bool is_head); bool http_response_is_closed(zend_object *obj); /* HTTP/2 strategy uses these to build frames without going through diff --git a/src/core/async_plain_event.c b/src/core/async_plain_event.c index 2d15e0cd..be0131fe 100644 --- a/src/core/async_plain_event.c +++ b/src/core/async_plain_event.c @@ -11,6 +11,7 @@ #endif #include "php.h" +#include "Zend/zend_exceptions.h" /* zend_clear_exception */ #include "Zend/zend_async_API.h" #include "core/async_plain_event.h" @@ -52,6 +53,21 @@ static bool plain_dispose(zend_async_event_t *event) return true; } +void async_coroutine_sleep_ms(zend_coroutine_t *co, const zend_ulong ms) +{ + zend_async_timer_event_t *const t = ZEND_ASYNC_NEW_TIMER_EVENT(ms, false); + + if (UNEXPECTED(t == NULL)) { + zend_clear_exception(); + return; + } + + t->base.start(&t->base); + zend_async_resume_when(co, &t->base, true, zend_async_waker_callback_resolve, NULL); + ZEND_ASYNC_SUSPEND(); + ZEND_ASYNC_WAKER_DESTROY(co); +} + zend_async_event_t *async_plain_event_new(void) { zend_async_event_t *event = pecalloc(1, sizeof(*event), 0); diff --git a/src/core/http_connection.c b/src/core/http_connection.c index e7089046..e0d659cf 100644 --- a/src/core/http_connection.c +++ b/src/core/http_connection.c @@ -2262,6 +2262,8 @@ static void http_connection_dispatch_request(http_connection_t *conn, http_reque /* Create HttpResponse PHP object */ object_init_ex(&ctx->response_zv, http_response_ce); http_response_set_protocol_version(Z_OBJ(ctx->response_zv), conn->http_version); + http_response_set_head(Z_OBJ(ctx->response_zv), + http_request_method_is_head(req)); /* Install the HTTP/1 streaming vtable. send() on the response * activates chunked framing on first call; handlers that stick to diff --git a/src/core/http_protocol_handlers.c b/src/core/http_protocol_handlers.c index 9a8f3c5e..90f0f6b2 100644 --- a/src/core/http_protocol_handlers.c +++ b/src/core/http_protocol_handlers.c @@ -12,6 +12,8 @@ #include "php.h" #include "http_protocol_handlers.h" +#include "http1/http_parser.h" /* http_request_t */ +#include "grpc/grpc.h" /* grpc_request_mode */ /* {{{ http_protocol_type_to_string */ const char* http_protocol_type_to_string(http_protocol_type_t type) @@ -127,6 +129,23 @@ bool http_protocol_has_handler(HashTable *handlers, http_protocol_type_t protoco } /* }}} */ +/* {{{ http_request_classify_protocols */ +void http_request_classify_protocols(http_request_t *req) +{ + req->grpc_mode = (uint8_t)grpc_request_mode(req); +} +/* }}} */ + +bool http_request_body_must_buffer(const http_request_t *req) +{ + return req->grpc_mode == GRPC_MODE_WEB_TEXT; +} + +bool http_request_body_size_uncapped(const http_request_t *req) +{ + return req->grpc_mode != GRPC_MODE_NONE; +} + /* {{{ http_protocol_pick_handler */ zend_fcall_t *http_protocol_pick_handler(HashTable *handlers, const bool is_grpc) { diff --git a/src/core/http_protocol_handlers.h b/src/core/http_protocol_handlers.h index a2397b60..a553a486 100644 --- a/src/core/http_protocol_handlers.h +++ b/src/core/http_protocol_handlers.h @@ -35,6 +35,18 @@ bool http_protocol_has_handler(HashTable *handlers, http_protocol_type_t protoco * classified) → HTTP1 → HTTP2. NULL when nothing matches. */ zend_fcall_t *http_protocol_pick_handler(HashTable *handlers, bool is_grpc); +/* Stamp the request's grpc_mode once at headers-complete, before any + * body-streaming decision; transports never classify themselves and read + * the body policy through the predicates below. */ +void http_request_classify_protocols(struct http_request_t *req); + +/* Body policy derived from the stamped grpc_mode, keeping transports + * gRPC-agnostic. must_buffer: never stream (grpc-web-text decodes the whole + * body). size_uncapped: waive the cumulative max_body_size cap (gRPC streams + * are unbounded; the in-flight window still bounds memory). */ +bool http_request_body_must_buffer(const struct http_request_t *req); +bool http_request_body_size_uncapped(const struct http_request_t *req); + /* Remove handler from HashTable */ void http_protocol_remove_handler(HashTable *handlers, http_protocol_type_t protocol); diff --git a/src/core/http_protocol_strategy.h b/src/core/http_protocol_strategy.h index 354e9e3d..223acfe7 100644 --- a/src/core/http_protocol_strategy.h +++ b/src/core/http_protocol_strategy.h @@ -129,6 +129,25 @@ typedef uint32_t http_protocol_mask_t; HTTP_PROTO_MASK_SSE | HTTP_PROTO_MASK_GRPC) #define HTTP_PROTO_MASK_HAS(mask, type) (((mask) & (1u << (type))) != 0) +/* + * Mask bits a registered handler of `type` enables. The single source for + * the protocol→mask rule — used both by the addXHandler methods at + * registration and by the worker-side transit rebuild, so the two can't + * drift (a plain HTTP handler serves h1+h2 on one port; gRPC rides h2). + */ +static zend_always_inline http_protocol_mask_t +http_protocol_registration_mask(http_protocol_type_t type) +{ + switch (type) { + case HTTP_PROTOCOL_HTTP1: + return HTTP_PROTO_MASK_HTTP1 | HTTP_PROTO_MASK_HTTP2; + case HTTP_PROTOCOL_GRPC: + return HTTP_PROTO_MASK_GRPC | HTTP_PROTO_MASK_HTTP2; + default: + return (http_protocol_mask_t)(1u << type); + } +} + /* * Sentinel returned internally when the byte prefix matches no * protocol the server accepts (e.g. HTTP/2 preface on a listener with diff --git a/src/core/reactor_pool.c b/src/core/reactor_pool.c index 9e9d9587..f38ccdc3 100644 --- a/src/core/reactor_pool.c +++ b/src/core/reactor_pool.c @@ -281,6 +281,13 @@ bool reactor_pool_exec(reactor_pool_t *rp, const int idx, const reactor_exec_fn /* Acquire: pair with the reactor's release store once fn has run. */ while (zend_atomic_int_load_ex(&done) == 0) { + if (zend_atomic_int_load_ex(&rc->phase) == REACTOR_PHASE_DONE) { + /* Loop exited: mailbox_free discards still-queued cmds (they can + * never run — no stack UAF), it runs before the DONE store. One + * final check catches a cmd that executed in the last drain. */ + return zend_atomic_int_load_ex(&done) != 0; + } + reactor_pool_msleep(); } diff --git a/src/core/worker_dispatch.c b/src/core/worker_dispatch.c index f066f045..35f674e9 100644 --- a/src/core/worker_dispatch.c +++ b/src/core/worker_dispatch.c @@ -26,7 +26,6 @@ #include "Zend/zend_hrtime.h" /* zend_hrtime — request-service sampling */ #define WORKER_STREAM_INFLIGHT_CAP (1024 * 1024) -#define WORKER_CREDIT_POLL_MS 2 #include @@ -68,6 +67,7 @@ typedef struct { bool stream_failed; /* terminal wire becomes STREAM_ABORT */ stream_credit_t *credit; /* shared with the reactor */ + zend_async_trigger_event_t *credit_wake; /* worker-owned; reactor signals it */ uint64_t posted_bytes; } worker_dispatch_ctx_t; @@ -203,6 +203,21 @@ static void worker_wire_copy_trailers(response_wire_t *rw, zend_object *resp) } ZEND_HASH_FOREACH_END(); } +void response_wire_discard(response_wire_t *rw) +{ + /* nobody adopts the credit ref — release it or the producer hangs */ + stream_credit_abandon((stream_credit_t *)response_wire_credit(rw)); + + zend_string *const orphan_chunk = + (zend_string *)response_wire_take_chunk(rw); + + if (orphan_chunk != NULL) { + zend_string_release(orphan_chunk); + } + + response_wire_free(rw); +} + /* The sink owns the wire in every outcome. */ static bool worker_wire_post(worker_dispatch_ctx_t *ctx, response_wire_t *rw) { @@ -212,17 +227,7 @@ static bool worker_wire_post(worker_dispatch_ctx_t *ctx, response_wire_t *rw) if (ctx->sink != NULL) { delivered = ctx->sink(rw, ctx->sink_arg); } else { - /* nobody adopts the credit ref — release it or the producer hangs */ - stream_credit_abandon((stream_credit_t *)response_wire_credit(rw)); - - zend_string *const orphan_chunk = - (zend_string *)response_wire_take_chunk(rw); - - if (orphan_chunk != NULL) { - zend_string_release(orphan_chunk); - } - - response_wire_free(rw); + response_wire_discard(rw); } if (!delivered && kind != RESPONSE_WIRE_FULL) { @@ -233,23 +238,9 @@ static bool worker_wire_post(worker_dispatch_ctx_t *ctx, response_wire_t *rw) return delivered; } -/* Suspend the producer for `ms` on a one-shot timer; no-op outside a coroutine. */ -static void worker_credit_sleep_ms(zend_coroutine_t *co, const zend_ulong ms) -{ - zend_async_timer_event_t *const t = ZEND_ASYNC_NEW_TIMER_EVENT(ms, false); - - if (UNEXPECTED(t == NULL)) { - zend_clear_exception(); - return; - } - - t->base.start(&t->base); - zend_async_resume_when(co, &t->base, true, zend_async_waker_callback_resolve, NULL); - ZEND_ASYNC_SUSPEND(); - ZEND_ASYNC_WAKER_DESTROY(co); -} - -/* Park until in-flight < cap; false = stream dead. */ +/* Park until in-flight < cap; false = stream dead. Suspends on a trigger the + * reactor signals per ack; a write-timeout timer bounds a peer that stops + * ACKing. */ static bool worker_stream_wait_credit(worker_dispatch_ctx_t *ctx) { if (ctx->credit == NULL) { @@ -258,9 +249,6 @@ static bool worker_stream_wait_credit(worker_dispatch_ctx_t *ctx) const uint32_t timeout_ms = http_server_get_write_timeout_s(ctx->server) * 1000u; - uint64_t waited_ms = 0; - uint64_t poll_ms = WORKER_CREDIT_POLL_MS; - uint64_t last_acked = stream_credit_acked(ctx->credit); while (ctx->posted_bytes - stream_credit_acked(ctx->credit) >= WORKER_STREAM_INFLIGHT_CAP) { @@ -274,25 +262,37 @@ static bool worker_stream_wait_credit(worker_dispatch_ctx_t *ctx) return true; /* can't suspend — degrade to unbounded */ } - worker_credit_sleep_ms(co, poll_ms); + if (ctx->credit_wake == NULL) { + ctx->credit_wake = ZEND_ASYNC_NEW_TRIGGER_EVENT(); - if (EG(exception) != NULL) { - return false; /* cancelled while parked */ + if (UNEXPECTED(ctx->credit_wake == NULL)) { + zend_clear_exception(); + return true; /* no waker — degrade to unbounded */ + } + + stream_credit_set_waker(ctx->credit, ctx->credit_wake); + continue; /* re-check: an ack may have landed pre-publish */ } - waited_ms += poll_ms; + if (UNEXPECTED(ZEND_ASYNC_WAKER_NEW(co) == NULL)) { + return false; + } + + zend_async_resume_when(co, &ctx->credit_wake->base, false, + zend_async_waker_callback_resolve, NULL); - if (timeout_ms > 0 && waited_ms >= timeout_ms) { - return false; /* peer stopped reading — treat as dead */ + if (timeout_ms > 0) { + zend_async_event_t *const timer = + &ZEND_ASYNC_NEW_TIMER_EVENT((zend_ulong)timeout_ms, false)->base; + zend_async_resume_when(co, timer, true, + zend_async_waker_callback_timeout, NULL); } - const uint64_t acked = stream_credit_acked(ctx->credit); + ZEND_ASYNC_SUSPEND(); + zend_async_waker_clean(co); - if (acked == last_acked) { - poll_ms = poll_ms < 32 ? poll_ms * 2 : 32; - } else { - last_acked = acked; - poll_ms = WORKER_CREDIT_POLL_MS; + if (EG(exception) != NULL) { + return false; /* write timeout or cancelled while parked */ } } @@ -324,8 +324,14 @@ static int worker_stream_append_chunk(void *vctx, zend_string *chunk) response_wire_set_kind(hw, RESPONSE_WIRE_STREAM_HEADERS); response_wire_set_credit(hw, ctx->credit); worker_wire_copy_head(hw, Z_OBJ(ctx->response_zv)); - worker_wire_post(ctx, hw); ctx->stream_started = true; + + /* headers undeliverable → the stream never opened; don't copy and + * post a chunk wire the reactor would only throw away */ + if (UNEXPECTED(!worker_wire_post(ctx, hw))) { + zend_string_release(chunk); + return HTTP_STREAM_APPEND_STREAM_DEAD; + } } response_wire_t *const cw = @@ -535,10 +541,22 @@ static void worker_dispatch_dispose(zend_coroutine_t *coroutine) } if (ctx->credit != NULL) { + if (ctx->credit_wake != NULL) { + /* retract the waker (fences out an in-flight reactor signal) + * before disposing its uv_async on this thread */ + stream_credit_clear_waker(ctx->credit); + } + stream_credit_release(ctx->credit); /* worker-side ref */ ctx->credit = NULL; } + if (ctx->credit_wake != NULL) { + ZEND_ASYNC_EVENT_SET_CLOSED(&ctx->credit_wake->base); + ctx->credit_wake->base.dispose(&ctx->credit_wake->base); + ctx->credit_wake = NULL; + } + if (!Z_ISUNDEF(ctx->request_zv)) { zval_ptr_dtor(&ctx->request_zv); ZVAL_UNDEF(&ctx->request_zv); @@ -601,6 +619,7 @@ bool worker_dispatch_request(http_server_object *server, object_init_ex(&ctx->response_zv, http_response_ce); http_response_set_protocol_version(Z_OBJ(ctx->response_zv), "3.0"); + http_response_set_head(Z_OBJ(ctx->response_zv), is_head); http_response_install_stream_ops(Z_OBJ(ctx->response_zv), &worker_stream_ops, ctx); diff --git a/src/grpc/grpc.c b/src/grpc/grpc.c index d0f38fe6..02937bb1 100644 --- a/src/grpc/grpc.c +++ b/src/grpc/grpc.c @@ -64,14 +64,10 @@ grpc_mode_t grpc_request_mode(const http_request_t *req) return GRPC_MODE_NATIVE; } -bool grpc_request_is_grpc_web_text(const http_request_t *req) -{ - return grpc_request_mode(req) == GRPC_MODE_WEB_TEXT; -} - grpc_mode_t grpc_classify(const http_request_t *req, HashTable *handlers) { - const grpc_mode_t mode = grpc_request_mode(req); + /* mode was stamped at headers-complete; here we only gate on a handler */ + const grpc_mode_t mode = (grpc_mode_t)req->grpc_mode; if (mode == GRPC_MODE_NONE || !http_protocol_has_handler(handlers, HTTP_PROTOCOL_GRPC)) { @@ -222,6 +218,12 @@ uint64_t grpc_parse_timeout_ns(const http_request_t *req) default: return 0; } + /* 8 digits × the hour factor exceeds uint64 — clamp instead of wrapping + * to a bogus small deadline. */ + if (value > UINT64_MAX / unit_ns) { + return UINT64_MAX; + } + return value * unit_ns; } diff --git a/src/grpc/grpc.h b/src/grpc/grpc.h index 11550276..6a7220bf 100644 --- a/src/grpc/grpc.h +++ b/src/grpc/grpc.h @@ -50,13 +50,13 @@ typedef enum { GRPC_MODE_WEB_TEXT, /* application/grpc-web-text — WEB + per-frame base64 */ } grpc_mode_t; -/* Delivery mode from the request content-type; NONE for non-gRPC. */ +/* Delivery mode from the request content-type; NONE for non-gRPC. Called + * only by http_request_classify_protocols — everyone else reads the + * req->grpc_mode stamp. */ grpc_mode_t grpc_request_mode(const struct http_request_t *req); -bool grpc_request_is_grpc_web_text(const struct http_request_t *req); - -/* Mode gated on a registered gRPC handler; every dispatch path (H2 / H3 / - * worker) must classify through this. */ +/* Stamped mode gated on a registered gRPC handler; every dispatch path + * (H2 / H3 / worker) must classify through this. */ grpc_mode_t grpc_classify(const struct http_request_t *req, HashTable *handlers); /* New string; decode returns NULL on malformed input. */ diff --git a/src/http1/http_parser.c b/src/http1/http_parser.c index 5dfd5486..8243ddb9 100644 --- a/src/http1/http_parser.c +++ b/src/http1/http_parser.c @@ -20,6 +20,7 @@ #include "core/body_pool.h" #include "log/trace_context.h" #include "http_body_stream.h" +#include "core/http_protocol_handlers.h" /* http_request_classify_protocols */ #include #ifndef PHP_WIN32 @@ -663,13 +664,16 @@ static int on_headers_complete(llhttp_t* llhttp_parser) } } + http_request_classify_protocols(req); + /* Streaming body mode (issue #26). Three-case policy by Content-Length: * CL >= AUTO_THRESHOLD or CL == 0 (chunked) → stream immediately; * CL >= SMALL but < AUTO → buffer, upgrade if readBody is called; * CL < SMALL → buffer, never upgrade. * Multipart bypasses streaming entirely (no current use case for * raw-multipart streaming; see plan §5 edge-case). */ - if (!req->use_multipart && parser->conn != NULL && parser->conn->view != NULL + if (!req->use_multipart && !http_request_body_must_buffer(req) + && parser->conn != NULL && parser->conn->view != NULL && parser->conn->view->body_streaming_enabled) { if (req->content_length == 0 || req->content_length >= HTTP_BODY_STREAM_AUTO_THRESHOLD) { diff --git a/src/http2/http2_session.c b/src/http2/http2_session.c index 677f1268..b6007f7a 100644 --- a/src/http2/http2_session.c +++ b/src/http2/http2_session.c @@ -26,7 +26,6 @@ #include "log/trace_context.h" #include "http_body_stream.h" #include "core/async_plain_event.h" -#include "grpc/grpc.h" /* web-text is buffered-only (policy gate) */ #include @@ -384,6 +383,9 @@ static void http2_request_body_upgrade(http_request_t *req) zend_string *initial = stream->request_body_buf.s; stream->request_body_buf.s = NULL; + /* the buffered branch already consumed these bytes — pop must not + * credit them a second time */ + req->body_precredited += ZSTR_LEN(initial); http_body_stream_push(req, initial); zend_string_release(initial); } @@ -444,8 +446,7 @@ static int cb_on_data_chunk_recv(nghttp2_session *ng, #endif /* Streaming mode (issue #26) — push the chunk into the per-request - * queue instead of accumulating into stream->request_body_buf. The - * cumulative max_body_size cap still applies as a hard ceiling. + * queue instead of accumulating into stream->request_body_buf. * Backpressure: no_auto_window_update is set; peer credit is granted * per chunk in http_body_stream_pop after the handler drains it. */ if (stream->request != NULL && stream->request->body_streaming) { @@ -456,9 +457,12 @@ static int cb_on_data_chunk_recv(nghttp2_session *ng, body_cap = HTTP2_MAX_BODY_SIZE; } - /* cap the LIVE (queued) bytes, not the cumulative total — a - * streaming body is legitimately unbounded in total */ - if (req->body_bytes_queued + len > body_cap) { + /* Live bytes cap memory for everyone; the cumulative cap is the + * operator's max_body_size contract, waived only for gRPC. */ + if (req->body_bytes_queued + len > body_cap + || (!http_request_body_size_uncapped(req) + && req->body_bytes_consumed + req->body_bytes_queued + len + > body_cap)) { return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE; } @@ -617,16 +621,16 @@ static int cb_on_frame_recv(nghttp2_session *ng, http_request_parse_trace_context(stream->request); } + http_request_classify_protocols(stream->request); + /* Streaming body mode (issue #26). Three-case policy by * Content-Length, see H1 parser for the full doc: * CL == 0 (chunked) or CL >= AUTO → stream immediately * SMALL <= CL < AUTO → buffer + install upgrade hook * CL < SMALL → buffer, never stream. */ - /* grpc-web-text stays buffered: readMessage() decodes the - * assembled body */ if (session->conn != NULL && session->conn->view != NULL && session->conn->view->body_streaming_enabled - && !grpc_request_is_grpc_web_text(stream->request)) { + && !http_request_body_must_buffer(stream->request)) { http_request_t *r = stream->request; if (r->content_length == 0 diff --git a/src/http2/http2_strategy.c b/src/http2/http2_strategy.c index 815874fe..f682ddac 100644 --- a/src/http2/http2_strategy.c +++ b/src/http2/http2_strategy.c @@ -267,6 +267,8 @@ static void http2_strategy_dispatch(struct http_request_t *request, object_init_ex(&stream->response_zv, http_response_ce); http_response_set_protocol_version(Z_OBJ(stream->response_zv), self->conn->http_version); + 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 * the vtable. Ops installed once at dispatch; diff --git a/src/http3/http3_callbacks.c b/src/http3/http3_callbacks.c index 70d24b43..ef658b78 100644 --- a/src/http3/http3_callbacks.c +++ b/src/http3/http3_callbacks.c @@ -41,7 +41,6 @@ #include "core/response_wire.h" /* response_wire_* (reverse path) */ #include "core/stream_credit.h" /* reverse-path flow control */ #include "http_body_stream.h" /* streaming request body (issue #26) */ -#include "grpc/grpc.h" /* web-text is buffered-only (policy gate) */ #include "http3/http3_stream.h" /* http3_stream_t */ #include /* ngtcp2_crypto_* callback ptrs */ @@ -387,6 +386,9 @@ static void http3_request_body_upgrade(http_request_t *req) zend_string *const initial = s->body_buf.s; s->body_buf.s = NULL; + /* the buffered branch already extended the window for these bytes — + * pop must not credit them a second time */ + req->body_precredited += ZSTR_LEN(initial); http_body_stream_push(req, initial); zend_string_release(initial); } @@ -406,17 +408,20 @@ static int h3_end_headers_cb(nghttp3_conn *conn, int64_t stream_id, return 0; } + if (s->request != NULL) { + http_request_classify_protocols(s->request); + } + /* Reactor mode: defer dispatch to h3_end_stream_cb — the body must be * fully assembled before hand-off (body queue is same-thread only). */ if (c != NULL && http3_listener_reactor_ctx(c->listener) != NULL) { return 0; } - /* Content-Length policy, mirror of H2 cb_on_frame_recv. grpc-web-text - * stays buffered: readMessage() base64-decodes the assembled body. */ + /* Content-Length policy, mirror of H2 cb_on_frame_recv. */ if (c != NULL && c->view != NULL && c->view->body_streaming_enabled && s->request != NULL - && !grpc_request_is_grpc_web_text(s->request)) { + && !http_request_body_must_buffer(s->request)) { http_request_t *const r = s->request; r->body_h3_conn = c; @@ -476,7 +481,12 @@ static int h3_recv_data_cb(nghttp3_conn *conn, int64_t stream_id, body_cap = HTTP3_MAX_BODY_BYTES; } - if (UNEXPECTED(req->body_bytes_queued + datalen > body_cap)) { + /* Live bytes cap memory for everyone; the cumulative cap is the + * operator's max_body_size contract, waived only for gRPC. */ + if (UNEXPECTED(req->body_bytes_queued + datalen > body_cap + || (!http_request_body_size_uncapped(req) + && req->body_bytes_consumed + req->body_bytes_queued + + datalen > body_cap))) { http3_packet_stats_t *const stats = c != NULL ? http3_listener_packet_stats(c->listener) : NULL; @@ -1527,9 +1537,10 @@ static int h3_acked_stream_data_cb(nghttp3_conn *conn, int64_t stream_id, return 0; } - /* retire acked bytes so a worker producer parked on credit resumes */ + /* retire acked bytes; wake a producer parked on credit */ if (s->wire_credit != NULL) { stream_credit_ack((stream_credit_t *)s->wire_credit, datalen); + stream_credit_wake((stream_credit_t *)s->wire_credit); } s->chunk_ack_credit += datalen; diff --git a/src/http3/http3_connection.c b/src/http3/http3_connection.c index 4b1d07e6..a9532702 100644 --- a/src/http3/http3_connection.c +++ b/src/http3/http3_connection.c @@ -134,24 +134,6 @@ void http3_debug_logger(void *user_data, const char *fmt, ...) http_logf_debug(st, "h3.ngtcp2 %s", buf); } -/* Build a sockaddr_storage from the listener's bound (host, port). - * - * ngtcp2_path matching is strict: the local addr passed to read_pkt / - * writev_stream must be the same value on every call after server_new. - * The proper plumbing for this is `zend_async_udp_sockname`, which is - * not yet available; until it lands we fabricate the sockaddr from the - * bind config so at least it is stable across calls. peer_family lets us produce v4 / v6 to match - * the inbound datagram. The result is precomputed once per family at listener - * spawn; this just copies the cached value. Returns 0 on success. */ -int http3_build_listener_local(const http3_listener_t *l, - int peer_family, - struct sockaddr_storage *out, - socklen_t *out_len) -{ - http3_listener_local_sockaddr(l, peer_family, out, out_len); - return 0; -} - /* ------------------------------------------------------------------------ * TLS attach (per-connection SSL bound to ngtcp2 via crypto_ossl) * ------------------------------------------------------------------------ */ @@ -316,11 +298,11 @@ static http3_connection_t *http3_connection_accept( orig_dcid.datalen = c->original_dcidlen; /* Stable local sockaddr derived from the listener bind config. See - * http3_build_listener_local() — works around the missing + * http3_listener_local_sockaddr() — works around the missing * zend_async_udp_sockname API. */ struct sockaddr_storage local_addr; socklen_t local_addr_len = 0; - http3_build_listener_local(listener, peer->sa_family, &local_addr, &local_addr_len); + http3_listener_local_sockaddr(listener, peer->sa_family, &local_addr, &local_addr_len); ngtcp2_path path = { .local = { .addr = (struct sockaddr *)&local_addr, .addrlen = local_addr_len }, @@ -738,12 +720,12 @@ bool http3_connection_dispatch( * subsequent packets advance the handshake or carry 1-RTT data. * * Same fabricated local addr as accept / drain — see - * http3_build_listener_local() for the rationale. The path remote is the + * http3_listener_local_sockaddr() for the rationale. The path remote is the * datagram's actual source (not conn->peer) so ngtcp2 sees a migrated / * NAT-rebound client (RFC 9000 §9); identical to conn->peer otherwise. */ struct sockaddr_storage local_addr; socklen_t local_addr_len = 0; - http3_build_listener_local(listener, peer->sa_family, + http3_listener_local_sockaddr(listener, peer->sa_family, &local_addr, &local_addr_len); ngtcp2_path_storage ps = {0}; diff --git a/src/http3/http3_dispatch.c b/src/http3/http3_dispatch.c index bf2e9b1b..0798ed60 100644 --- a/src/http3/http3_dispatch.c +++ b/src/http3/http3_dispatch.c @@ -29,6 +29,7 @@ #include "grpc/grpc_call.h" /* gRPC call lifecycle policy */ #include "static/static_handler.h" /* http_static_try_serve / count */ #include "core/response_wire.h" /* response_wire_* (reverse path) */ +#include "core/worker_dispatch.h" /* response_wire_discard */ /* Defined in src/http_request.c. Declared here because the public * php_http_server.h header doesn't expose it (it lives in the C boundary @@ -234,17 +235,8 @@ void http3_reactor_apply_response(void *arg) http3_connection_t *const c = (s != NULL) ? s->conn : NULL; if (c == NULL || c->closed || c->nghttp3_conn == NULL) { - /* stream gone: nobody adopts the credit ref — unblock the producer */ - stream_credit_abandon((stream_credit_t *)response_wire_credit(rw)); - - zend_string *const orphan_chunk = - (zend_string *)response_wire_take_chunk(rw); - - if (orphan_chunk != NULL) { - zend_string_release(orphan_chunk); - } - - response_wire_free(rw); + /* stream gone: abandon credit / release chunk — unblock the producer */ + response_wire_discard(rw); return; } @@ -331,6 +323,8 @@ static bool http3_reactor_try_static(http3_connection_t *c, http3_stream_t *s, object_init_ex(&s->response_zv, http_response_ce); 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)); const http_static_result_t rc = http_static_try_serve_mounts( (const http_static_handler_t *const *)rctx->static_mounts, @@ -439,6 +433,8 @@ void http3_stream_dispatch(http3_connection_t *c, http3_stream_t *s) object_init_ex(&s->response_zv, http_response_ce); 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 * handler enqueues into our chunk_queue. setBody/end (REST) handlers * never touch this; they go through the buffered submit_response in @@ -571,19 +567,7 @@ static void h3_handler_coroutine_entry(void) } HashTable *handlers = http_server_get_protocol_handlers(server); - zend_fcall_t *fcall = http_protocol_get_handler(handlers, HTTP_PROTOCOL_HTTP1); - - if (fcall == NULL) { - fcall = http_protocol_get_handler(handlers, HTTP_PROTOCOL_HTTP2); - } - - /* gRPC streams route to the addGrpcHandler callable. */ - if (s->is_grpc) { - zend_fcall_t *g = http_protocol_get_handler(handlers, HTTP_PROTOCOL_GRPC); - if (g != NULL) { - fcall = g; - } - } + zend_fcall_t *fcall = http_protocol_pick_handler(handlers, s->is_grpc); if (fcall == NULL) return; diff --git a/src/http3/http3_internal.h b/src/http3/http3_internal.h index a0bc2f40..dd6e685a 100644 --- a/src/http3/http3_internal.h +++ b/src/http3/http3_internal.h @@ -82,15 +82,6 @@ void http3_ensure_ossl_crypto_init(void); /* ngtcp2 log_printf-compatible bridge into http_log at DEBUG. */ void http3_debug_logger(void *user_data, const char *fmt, ...); -/* Build a sockaddr_storage from the listener's bound (host, port). - * Required because the reactor doesn't yet expose the actual bound - * sockname; ngtcp2_path matching is strict so we must reproduce the - * same value on every read/write. peer_family selects v4/v6. */ -int http3_build_listener_local(const http3_listener_t *l, - int peer_family, - struct sockaddr_storage *out, - socklen_t *out_len); - /* ===== Packet/timer machinery (defined in http3_io.c) ===== */ diff --git a/src/http3/http3_io.c b/src/http3/http3_io.c index d72e06e8..f2b0e880 100644 --- a/src/http3/http3_io.c +++ b/src/http3/http3_io.c @@ -257,7 +257,7 @@ void http3_connection_drain_out(http3_connection_t *c) * the path. */ struct sockaddr_storage local_addr; socklen_t local_addr_len = 0; - http3_build_listener_local(c->listener, c->peer.ss_family, + http3_listener_local_sockaddr(c->listener, c->peer.ss_family, &local_addr, &local_addr_len); ngtcp2_path_storage ps; ngtcp2_path_storage_init(&ps, @@ -654,7 +654,7 @@ void http3_connection_emit_close(http3_connection_t *c) struct sockaddr_storage local_addr; socklen_t local_addr_len = 0; - http3_build_listener_local(c->listener, c->peer.ss_family, + http3_listener_local_sockaddr(c->listener, c->peer.ss_family, &local_addr, &local_addr_len); ngtcp2_path_storage ps = {0}; ngtcp2_path_storage_init(&ps, diff --git a/src/http3/http3_static_response.c b/src/http3/http3_static_response.c index 9a5b4000..c89457dd 100644 --- a/src/http3/http3_static_response.c +++ b/src/http3/http3_static_response.c @@ -66,6 +66,7 @@ #include "http3_internal.h" #include "http3_listener.h" #include "http3/http3_stream.h" +#include "core/async_plain_event.h" /* throttle wake for parked pumps */ #include #include @@ -77,13 +78,21 @@ #define H3_STATIC_BUDGET_FLOOR (4u * 1024u * 1024u) #define H3_STATIC_BUDGET_FALLBACK (64u * 1024u * 1024u) /* memory_limit = -1 */ #define H3_STATIC_BUDGET_RESERVE_FRAC 8u /* hard top = 87.5% */ -#define H3_STATIC_THROTTLE_POLL_MS 2u /* pump re-check cadence */ +#define H3_STATIC_RESUME_NUM 4u +#define H3_STATIC_RESUME_DEN 5u /* resume below 80% */ +#define H3_STATIC_THROTTLE_TICK_MS 100u /* defensive re-check */ /* 0 = not yet initialised (h3_static_budget_init_once). */ ZEND_TLS size_t h3_static_budget_bytes = 0; ZEND_TLS size_t h3_static_global_bytes_in_flight = 0; ZEND_TLS size_t h3_static_global_high_water = 0; +/* Hysteresis wake for over-budget pumps: parked coroutines suspend on this + * plain event; account_debit fires it once usage drops below the resume + * threshold (mirror of the H2 static throttled-list kick). */ +ZEND_TLS zend_async_event_t *h3_static_throttle_event = NULL; +ZEND_TLS bool h3_static_global_throttled = false; + static void h3_static_budget_init_once(void) { if (h3_static_budget_bytes != 0) { @@ -124,6 +133,15 @@ void h3_static_account_debit(const size_t n) * debited on the ACK path if a release raced, so never underflow. */ h3_static_global_bytes_in_flight -= (n <= h3_static_global_bytes_in_flight) ? n : h3_static_global_bytes_in_flight; + + /* hysteresis: wake every parked pump once usage falls below 80% */ + if (h3_static_global_throttled + && h3_static_global_bytes_in_flight + < h3_static_budget_bytes * H3_STATIC_RESUME_NUM + / H3_STATIC_RESUME_DEN) { + h3_static_global_throttled = false; + async_plain_event_fire(h3_static_throttle_event); + } } static bool h3_static_over_budget(void) @@ -131,20 +149,38 @@ static bool h3_static_over_budget(void) return h3_static_global_bytes_in_flight >= h3_static_budget_bytes; } -/* Suspend the pump on a one-shot timer so the reactor keeps draining/ACKing. */ -static void h3_static_throttle_sleep(zend_coroutine_t *co, const zend_ulong ms) +/* Park the pump until the debit path fires the throttle event (plus a + * defensive tick so peer_closed is still noticed if nothing drains). */ +static void h3_static_throttle_park(zend_coroutine_t *co) { - zend_async_timer_event_t *const t = ZEND_ASYNC_NEW_TIMER_EVENT(ms, false); + if (h3_static_throttle_event == NULL) { + h3_static_throttle_event = async_plain_event_new(); + + if (UNEXPECTED(h3_static_throttle_event == NULL)) { + return; /* no event — the caller's loop degrades to a spin */ + } + } - if (UNEXPECTED(t == NULL)) { - zend_clear_exception(); + h3_static_global_throttled = true; + + if (UNEXPECTED(ZEND_ASYNC_WAKER_NEW(co) == NULL)) { return; } - t->base.start(&t->base); - zend_async_resume_when(co, &t->base, true, zend_async_waker_callback_resolve, NULL); + zend_async_resume_when(co, h3_static_throttle_event, false, + zend_async_waker_callback_resolve, NULL); + + zend_async_timer_event_t *const t = + ZEND_ASYNC_NEW_TIMER_EVENT((zend_ulong)H3_STATIC_THROTTLE_TICK_MS, false); + + if (t != NULL) { + t->base.start(&t->base); + zend_async_resume_when(co, &t->base, true, + zend_async_waker_callback_resolve, NULL); + } + ZEND_ASYNC_SUSPEND(); - ZEND_ASYNC_WAKER_DESTROY(co); + zend_async_waker_clean(co); } typedef struct { @@ -204,9 +240,10 @@ static void h3_static_pump_entry(void) while (state->bytes_sent < state->body_length && !state->stream->peer_closed) { - /* over budget: pace this stream; queued bytes keep draining, so no deadlock */ + /* over budget: park until the debit path wakes us; queued bytes keep + * draining, so no deadlock */ while (h3_static_over_budget() && !state->stream->peer_closed) { - h3_static_throttle_sleep(co, H3_STATIC_THROTTLE_POLL_MS); + h3_static_throttle_park(co); if (EG(exception) != NULL) { zend_clear_exception(); diff --git a/src/http_body_stream.c b/src/http_body_stream.c index 283d6c7d..b16b59c8 100644 --- a/src/http_body_stream.c +++ b/src/http_body_stream.c @@ -100,22 +100,31 @@ zend_string *http_body_stream_pop(http_request_t *req) req->body_bytes_consumed += ZSTR_LEN(data); /* Flow-control credit: grant peer permission to send another len bytes - * now that PHP has actually drained them. h_session_schedule_emit pushes - * the resulting WINDOW_UPDATE on the wire. */ + * now that PHP has actually drained them — minus any prefix that was + * already credited while buffered pre-upgrade (body_precredited). */ + size_t credit = ZSTR_LEN(data); + + if (req->body_precredited > 0) { + const size_t skip = credit < req->body_precredited + ? credit : req->body_precredited; + req->body_precredited -= skip; + credit -= skip; + } + #ifdef HAVE_HTTP2 - if (req->body_h2_session != NULL) { + if (req->body_h2_session != NULL && credit > 0) { http2_session_t *s = req->body_h2_session; - (void)nghttp2_session_consume(s->ng, req->body_h2_stream_id, - ZSTR_LEN(data)); + (void)nghttp2_session_consume(s->ng, req->body_h2_stream_id, credit); h2_session_schedule_emit(s); } #endif #ifdef HAVE_HTTP_SERVER_HTTP3 /* Same discipline over QUIC: return window credit for the drained bytes - * (coalesced inside consume). */ + * (coalesced inside consume; called even at credit==0 so a queue-empty + * pop still flushes previously coalesced credit). */ if (req->body_h3_conn != NULL) { - http3_request_body_consume(req, ZSTR_LEN(data), + http3_request_body_consume(req, credit, req->body_queue_head == NULL); } #endif diff --git a/src/http_request.c b/src/http_request.c index 76892dbe..21b0b32d 100644 --- a/src/http_request.c +++ b/src/http_request.c @@ -286,7 +286,7 @@ ZEND_METHOD(TrueAsync_HttpRequest, readMessage) int rc; /* grpc-web-text: decode the base64 body once, deframe from the result */ - const bool web_text = grpc_request_is_grpc_web_text(req); + const bool web_text = (req->grpc_mode == GRPC_MODE_WEB_TEXT); if (web_text && req->grpc_text_body == NULL) { if (req->body == NULL) { diff --git a/src/http_response.c b/src/http_response.c index 1617e491..ff2ce8da 100644 --- a/src/http_response.c +++ b/src/http_response.c @@ -974,6 +974,11 @@ ZEND_METHOD(TrueAsync_HttpResponse, send) return; } + /* HEAD must carry no body (RFC 9110 §9.3.2); drop the chunk. */ + if (response->is_head) { + 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) { @@ -1013,17 +1018,72 @@ ZEND_METHOD(TrueAsync_HttpResponse, send) } /* }}} */ -/* {{{ proto HttpResponse::writeMessage(string $message, bool $compress = false): static - * Stream one gRPC length-prefixed message; first call commits, like send(). */ +/* {{{ proto HttpResponse::setGrpcEncoding(string $encoding): static + * Declare the response message encoding (grpc-encoding header) before the + * first writeMessage(). Mirrors grpc-java setCompression / C++ + * set_compression_algorithm: enabling compression is a per-call decision; + * per-message the wire only allows *skipping* it. */ +ZEND_METHOD(TrueAsync_HttpResponse, setGrpcEncoding) +{ + zend_string *encoding; + + ZEND_PARSE_PARAMETERS_START(1, 1) + Z_PARAM_STR(encoding) + ZEND_PARSE_PARAMETERS_END(); + + http_response_object *response = Z_HTTP_RESPONSE_P(ZEND_THIS); + + if (response->grpc_mode == GRPC_MODE_NONE) { + zend_throw_exception(http_server_runtime_exception_ce, + "setGrpcEncoding() is only available on gRPC responses", 0); + return; + } + + if (response->closed || response->streaming || response->headers_sent) { + zend_throw_exception(http_server_runtime_exception_ce, + "setGrpcEncoding() must be called before the first writeMessage()", 0); + return; + } + + if (zend_string_equals_literal_ci(encoding, "identity")) { + response->grpc_compress = false; + zend_hash_str_del(response->headers, "grpc-encoding", + sizeof("grpc-encoding") - 1); + RETURN_OBJ_COPY(Z_OBJ_P(ZEND_THIS)); + } + + if (!zend_string_equals_literal_ci(encoding, "gzip")) { + zend_throw_exception_ex(http_server_runtime_exception_ce, 0, + "Unsupported grpc-encoding \"%s\" (supported: gzip, identity)", + ZSTR_VAL(encoding)); + return; + } + +#ifdef HAVE_HTTP_COMPRESSION + response->grpc_compress = true; + + zval enc; + ZVAL_STRING(&enc, "gzip"); + zend_hash_str_update(response->headers, "grpc-encoding", + sizeof("grpc-encoding") - 1, &enc); + + RETURN_OBJ_COPY(Z_OBJ_P(ZEND_THIS)); +#else + zend_throw_exception(http_server_runtime_exception_ce, + "gzip grpc-encoding requires the compression module", 0); +#endif +} +/* }}} */ + +/* {{{ proto HttpResponse::writeMessage(string $message): static + * Stream one gRPC length-prefixed message; first call commits, like send(). + * Compressed automatically when setGrpcEncoding('gzip') was declared. */ ZEND_METHOD(TrueAsync_HttpResponse, writeMessage) { zend_string *message; - bool compress = false; - ZEND_PARSE_PARAMETERS_START(1, 2) + ZEND_PARSE_PARAMETERS_START(1, 1) Z_PARAM_STR(message) - Z_PARAM_OPTIONAL - Z_PARAM_BOOL(compress) ZEND_PARSE_PARAMETERS_END(); http_response_object *response = Z_HTTP_RESPONSE_P(ZEND_THIS); @@ -1046,29 +1106,21 @@ ZEND_METHOD(TrueAsync_HttpResponse, writeMessage) return; } - /* grpc-encoding must ride the initial HEADERS — declare before commit */ zend_string *payload = message; bool gzipped = false; #ifdef HAVE_HTTP_COMPRESSION - if (compress) { + if (response->grpc_compress) { zend_string *gz = grpc_message_deflate_gzip( ZSTR_VAL(message), ZSTR_LEN(message)); + /* deflate failure → identity for this message (compressed-flag 0), + * which the spec permits under a declared encoding */ if (gz != NULL) { payload = gz; gzipped = true; - - if (!response->streaming) { - zval enc; - ZVAL_STRING(&enc, "gzip"); - zend_hash_str_update(response->headers, "grpc-encoding", - sizeof("grpc-encoding") - 1, &enc); - } } } -#else - (void)compress; #endif if (!response->streaming) { diff --git a/src/http_response_internal.h b/src/http_response_internal.h index 6a61803d..43bd3fa5 100644 --- a/src/http_response_internal.h +++ b/src/http_response_internal.h @@ -58,11 +58,14 @@ typedef struct { 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) */ /* grpc_mode_t stamped at dispatch; picks the per-frame transform. * 0 = not a gRPC call. */ uint8_t grpc_mode; + bool grpc_compress; /* setGrpcEncoding('gzip') declared */ + /* Compression module state (issue #8). Opaque ptr — owned by the * compression TU; allocated by http_compression_attach at dispatch * and freed by http_compression_state_free at object dtor. */ diff --git a/src/http_response_server_api.c b/src/http_response_server_api.c index 6a01512f..a2f07a43 100644 --- a/src/http_response_server_api.c +++ b/src/http_response_server_api.c @@ -34,6 +34,11 @@ void http_response_set_socket(zend_object *obj, php_socket_t fd) http_response_from_obj(obj)->socket_fd = fd; } +void http_response_set_head(zend_object *obj, bool is_head) +{ + http_response_from_obj(obj)->is_head = is_head; +} + void http_response_set_protocol_version(zend_object *obj, const char *version) { http_response_object *response = http_response_from_obj(obj); diff --git a/src/http_server_class.c b/src/http_server_class.c index 7ee9abad..7d4b2613 100644 --- a/src/http_server_class.c +++ b/src/http_server_class.c @@ -33,6 +33,7 @@ #include "core/worker_registry.h" #include "core/response_wire.h" #include "core/stream_credit.h" +#include "core/async_plain_event.h" /* async_coroutine_sleep_ms */ #include "log/http_log.h" #ifndef PHP_WIN32 #include @@ -1412,9 +1413,8 @@ ZEND_METHOD(TrueAsync_HttpServer, addHttpHandler) /* addHttpHandler registers a handler for plain HTTP requests. * Historically this serves both HTTP/1 and HTTP/2 on the same port * (curl --http1.1 and curl --http2 both end up in the same PHP - * callback). Enable both bits so dual-protocol listeners keep - * working; h2-only deployments use addHttp2Handler exclusively. */ - server->view.protocol_mask |= HTTP_PROTO_MASK_HTTP1 | HTTP_PROTO_MASK_HTTP2; + * callback); h2-only deployments use addHttp2Handler exclusively. */ + server->view.protocol_mask |= http_protocol_registration_mask(HTTP_PROTOCOL_HTTP1); } /* }}} */ @@ -1509,7 +1509,7 @@ ZEND_METHOD(TrueAsync_HttpServer, addWebSocketHandler) HTTP_PROTOCOL_WEBSOCKET, Z_OBJ_P(ZEND_THIS) ); - server->view.protocol_mask |= HTTP_PROTO_MASK_WS; + server->view.protocol_mask |= http_protocol_registration_mask(HTTP_PROTOCOL_WEBSOCKET); } /* }}} */ @@ -1531,7 +1531,7 @@ ZEND_METHOD(TrueAsync_HttpServer, addHttp2Handler) HTTP_PROTOCOL_HTTP2, Z_OBJ_P(ZEND_THIS) ); - server->view.protocol_mask |= HTTP_PROTO_MASK_HTTP2; + server->view.protocol_mask |= http_protocol_registration_mask(HTTP_PROTOCOL_HTTP2); } /* }}} */ @@ -1553,8 +1553,7 @@ ZEND_METHOD(TrueAsync_HttpServer, addGrpcHandler) Z_OBJ_P(ZEND_THIS) ); - /* gRPC rides h2 — a gRPC-only server must still accept h2c/h2 ALPN */ - server->view.protocol_mask |= HTTP_PROTO_MASK_GRPC | HTTP_PROTO_MASK_HTTP2; + server->view.protocol_mask |= http_protocol_registration_mask(HTTP_PROTOCOL_GRPC); } /* }}} */ @@ -2552,73 +2551,197 @@ static void http_server_reactor_pool_down(http_server_object *server) #endif } -/* Worker response sink: post the rendered response back to the originating - * reactor for nghttp3 encode + send. Runs on the worker thread (from - * the handler coroutine's dispose). reactor_id (echoed on the wire) selects the - * reverse channel; ownership of `rw` transfers to the reactor apply on success. - * On failure (no pool / full mailbox) drop it — the client times out and the - * slab is still reclaimed by the consumed that follows. */ -static bool http_server_worker_response_sink(response_wire_t *rw, void *arg) +#ifdef HAVE_HTTP_SERVER_HTTP3 +/* Deferred-wire retry (per worker thread). A full reactor mailbox never + * blocks the worker: the wire is parked in a thread-local FIFO and a hidden + * one-shot timer retries the post on the next tick, so other coroutines + * keep running while the reactor drains. Order is preserved (head blocks + * the queue) and each wire gets ~100 ms before it is discarded — + * response_wire_discard marks the producer's credit dead, the same failure + * signal the old blocking retry produced. */ +typedef struct pending_wire_s { + response_wire_t *rw; + uint64_t deadline_ns; + int reactor; + struct pending_wire_s *next; +} pending_wire_t; + +ZEND_TLS pending_wire_t *pending_wire_head = NULL; +ZEND_TLS pending_wire_t *pending_wire_tail = NULL; +ZEND_TLS bool pending_wire_armed = false; + +#define PENDING_WIRE_RETRY_MS 1u +#define PENDING_WIRE_TTL_NS (100ull * 1000000ull) + +static bool pending_wire_arm_timer(void); + +static void pending_wire_flush(void) { - (void)arg; + const uint64_t now = zend_hrtime(); -#ifdef HAVE_HTTP_SERVER_HTTP3 - if (g_reactor_pool != NULL) { - const int reactor = (int)response_wire_reactor_id(rw); + while (pending_wire_head != NULL) { + pending_wire_t *const n = pending_wire_head; - if (reactor_pool_post_exec(g_reactor_pool, reactor, - http3_reactor_apply_response, rw)) { - return true; /* the reactor owns rw now */ + const bool posted = g_reactor_pool != NULL + && reactor_pool_post_exec(g_reactor_pool, n->reactor, + http3_reactor_apply_response, n->rw); + + if (!posted) { + if (g_reactor_pool != NULL && now < n->deadline_ns) { + break; /* mailbox still full — keep order, retry next tick */ + } + response_wire_discard(n->rw); /* expired / pool gone → ABORT */ } - /* STREAM_* wires are ordered fragments — dropping one corrupts the - * stream. Retry ~100 ms on a full mailbox, then fail fast (→ ABORT). */ - if (response_wire_kind(rw) != RESPONSE_WIRE_FULL) { - for (int attempt = 0; attempt < 100; attempt++) { -#ifdef PHP_WIN32 - Sleep(1); -#else - const struct timespec ts = { 0, 1000000 }; /* 1 ms */ - nanosleep(&ts, NULL); -#endif + pending_wire_head = n->next; - if (reactor_pool_post_exec(g_reactor_pool, reactor, - http3_reactor_apply_response, rw)) { - return true; - } - } + if (pending_wire_head == NULL) { + pending_wire_tail = NULL; } + + pefree(n, 1); } -#endif - /* undeliverable: nobody adopts the credit ref — unblock the producer */ - stream_credit_abandon((stream_credit_t *)response_wire_credit(rw)); + if (pending_wire_head != NULL && !pending_wire_arm_timer()) { + /* can't schedule another tick — fail the whole queue now */ + while (pending_wire_head != NULL) { + pending_wire_t *const n = pending_wire_head; + response_wire_discard(n->rw); + pending_wire_head = n->next; + pefree(n, 1); + } + pending_wire_tail = NULL; + } +} - zend_string *const orphan_chunk = (zend_string *)response_wire_take_chunk(rw); +static void pending_wire_timer_fn(zend_async_event_t *event, + zend_async_event_callback_t *callback, + void *result, zend_object *exception) +{ + (void)event; (void)callback; (void)result; - if (orphan_chunk != NULL) { - zend_string_release(orphan_chunk); + pending_wire_armed = false; /* one-shot: fired (auto-closes) */ + + if (exception != NULL) { + return; /* loop teardown — flush would re-arm on a dying reactor */ } - response_wire_free(rw); - return false; + pending_wire_flush(); } -/* Suspend the current coroutine for `ms` on a one-shot timer; lets the worker - * loop keep draining (mailbox, scheduler) while we wait. */ -static void hot_reload_sleep_ms(zend_coroutine_t *co, const zend_ulong ms) +static bool pending_wire_arm_timer(void) { - zend_async_timer_event_t *t = ZEND_ASYNC_NEW_TIMER_EVENT(ms, false); + if (pending_wire_armed) { + return true; + } + + zend_async_timer_event_t *const t = + ZEND_ASYNC_NEW_TIMER_EVENT(PENDING_WIRE_RETRY_MS, /*periodic*/ false); if (UNEXPECTED(t == NULL)) { zend_clear_exception(); - return; + return false; + } + + zend_async_event_callback_t *const cb = + ZEND_ASYNC_EVENT_CALLBACK(pending_wire_timer_fn); + + if (UNEXPECTED(cb == NULL || !t->base.add_callback(&t->base, cb))) { + if (cb != NULL) { + ZEND_ASYNC_EVENT_CALLBACK_RELEASE(cb); + } + t->base.dispose(&t->base); + return false; + } + + if (UNEXPECTED(!t->base.start(&t->base))) { + zend_async_callbacks_remove(&t->base, cb); + t->base.dispose(&t->base); + return false; } - t->base.start(&t->base); - zend_async_resume_when(co, &t->base, true, zend_async_waker_callback_resolve, NULL); - ZEND_ASYNC_SUSPEND(); - ZEND_ASYNC_WAKER_DESTROY(co); + pending_wire_armed = true; + return true; +} + +static bool pending_wire_defer(response_wire_t *rw, int reactor) +{ + pending_wire_t *const n = pemalloc(sizeof(*n), 1); + + n->rw = rw; + n->reactor = reactor; + n->deadline_ns = zend_hrtime() + PENDING_WIRE_TTL_NS; + n->next = NULL; + + if (pending_wire_tail != NULL) { + pending_wire_tail->next = n; + } else { + pending_wire_head = n; + } + pending_wire_tail = n; + + if (!pending_wire_arm_timer()) { + /* unlink what we just queued; caller discards the wire */ + if (pending_wire_head == n) { + pending_wire_head = NULL; + pending_wire_tail = NULL; + } else { + pending_wire_t *p = pending_wire_head; + while (p->next != n) { + p = p->next; + } + p->next = NULL; + pending_wire_tail = p; + } + pefree(n, 1); + return false; + } + + return true; +} +#endif /* HAVE_HTTP_SERVER_HTTP3 */ + +/* Worker response sink: post the rendered response back to the originating + * reactor for nghttp3 encode + send. Runs on the worker thread (handler + * coroutine or its dispose). reactor_id (echoed on the wire) selects the + * reverse channel; ownership of `rw` transfers to the reactor apply on + * success. A FULL wire that doesn't fit is dropped (the client times out); + * STREAM_* wires are ordered fragments, so they defer to the hidden retry + * timer instead — never blocking the worker thread. */ +static bool http_server_worker_response_sink(response_wire_t *rw, void *arg) +{ + (void)arg; + +#ifdef HAVE_HTTP_SERVER_HTTP3 + if (g_reactor_pool != NULL) { + const int reactor = (int)response_wire_reactor_id(rw); + const bool is_stream = response_wire_kind(rw) != RESPONSE_WIRE_FULL; + + /* Once anything is deferred, stream wires queue behind it so + * fragments never overtake each other. FULL wires are whole + * responses on their own streams — always try directly. */ + if ((!is_stream || pending_wire_head == NULL) + && reactor_pool_post_exec(g_reactor_pool, reactor, + http3_reactor_apply_response, rw)) { + return true; /* the reactor owns rw now */ + } + + if (is_stream && pending_wire_defer(rw, reactor)) { + return true; /* parked; retried by the timer, ~100 ms TTL */ + } + } +#endif + + /* undeliverable: abandoning the credit unblocks the producer */ + response_wire_discard(rw); + return false; +} + +/* Suspend the current coroutine for `ms` on a one-shot timer; lets the worker + * loop keep draining (mailbox, scheduler) while we wait. */ +static void hot_reload_sleep_ms(zend_coroutine_t *co, const zend_ulong ms) +{ + async_coroutine_sleep_ms(co, ms); if (EG(exception)) { zend_clear_exception(); @@ -5369,23 +5492,8 @@ static zend_object *http_server_transfer_obj( * dispatch a parsed request to user code; without this, the * loaded handler sits in the HashTable but plain HTTP/1 * requests are silently dropped on each worker thread. */ - switch (transit->entries[i].protocol) { - case HTTP_PROTOCOL_HTTP1: - dst_obj->view.protocol_mask |= - HTTP_PROTO_MASK_HTTP1 | HTTP_PROTO_MASK_HTTP2; - break; - case HTTP_PROTOCOL_HTTP2: - dst_obj->view.protocol_mask |= HTTP_PROTO_MASK_HTTP2; - break; - case HTTP_PROTOCOL_WEBSOCKET: - dst_obj->view.protocol_mask |= HTTP_PROTO_MASK_WS; - break; - case HTTP_PROTOCOL_GRPC: - dst_obj->view.protocol_mask |= HTTP_PROTO_MASK_GRPC; - break; - default: - break; - } + dst_obj->view.protocol_mask |= + http_protocol_registration_mask(transit->entries[i].protocol); } } diff --git a/src/log/http_log.c b/src/log/http_log.c index 7c7e0d44..3cb12a0e 100644 --- a/src/log/http_log.c +++ b/src/log/http_log.c @@ -17,6 +17,7 @@ #include "Zend/zend_async_API.h" #include "log/http_log.h" +#include "core/async_plain_event.h" /* drain-flush wakeup */ #include "../../stubs/LogSeverity.php_arginfo.h" #include @@ -64,10 +65,9 @@ static void *g_formatter_ud = NULL; #define HTTP_LOG_PENDING_MAX (64u * 1024u) -/* Stop-time drain of in-flight log writes: yield to the reactor in short beats - * until the writer is idle, bounded so a wedged sink can't pin teardown. */ -#define HTTP_LOG_STOP_DRAIN_SLEEP_MS 1u -#define HTTP_LOG_STOP_DRAIN_MAX_SPINS 3000u /* ~3s ceiling → leak-fallback */ +/* Ceiling on the stop-time flush wait; a wedged sink past this leaks rather + * than pinning teardown. */ +#define HTTP_LOG_STOP_DRAIN_BUDGET_MS 3000u struct http_log_writer_cb { zend_async_event_callback_t base; @@ -77,6 +77,10 @@ struct http_log_writer_cb { char *pending_buf; size_t pending_len; size_t pending_cap; + /* Fired by writer_complete_cb once the write chain fully drains, so + * http_log_server_stop can wait for flush without polling. Same thread + * as the completion, so a plain (non-uv) event suffices. */ + zend_async_event_t *drain_event; }; /* ------------------------------------------------------------------------ */ @@ -260,6 +264,11 @@ static void writer_complete_cb( } writer_kick_next(cb); + + /* Chain fully drained — wake a stop() waiting to flush before teardown. */ + if (cb->drain_event != NULL && cb->active_req == NULL) { + async_plain_event_fire(cb->drain_event); + } } static void writer_callback_dispose( @@ -525,6 +534,7 @@ void http_log_server_start(http_log_state_t *state, cb->pending_buf = NULL; cb->pending_len = 0; cb->pending_cap = 0; + cb->drain_event = NULL; if (UNEXPECTED(!io->event.add_callback(&io->event, &cb->base))) { efree(cb); @@ -545,23 +555,6 @@ void http_log_server_start(http_log_state_t *state, state->severity = severity; } -/* Yield to the reactor so a pending write completion can land. Best-effort. */ -static void http_log_stop_yield(zend_coroutine_t *co) -{ - zend_async_timer_event_t *const t = - ZEND_ASYNC_NEW_TIMER_EVENT((zend_ulong)HTTP_LOG_STOP_DRAIN_SLEEP_MS, false); - - if (UNEXPECTED(t == NULL)) { - zend_clear_exception(); - return; - } - - t->base.start(&t->base); - zend_async_resume_when(co, &t->base, true, zend_async_waker_callback_resolve, NULL); - ZEND_ASYNC_SUSPEND(); - ZEND_ASYNC_WAKER_DESTROY(co); -} - void http_log_server_stop(http_log_state_t *state) { if (state == NULL) { @@ -570,17 +563,36 @@ void http_log_server_stop(http_log_state_t *state) state->severity = HTTP_LOG_OFF; - /* Drain in-flight writes before teardown. Must NOT async_io_req_await: - * writer_complete_cb frees the req before the awaiter re-reads it (UAF - * under ASAN) — yield to the reactor and re-poll instead. */ + /* Flush the in-flight write chain before teardown. We wait on a plain + * event fired by writer_complete_cb (same thread) rather than awaiting + * the io req directly — the completion frees the req, so awaiting it is + * a UAF. A timer caps the wait so a wedged write falls to the leak path. */ if (state->async_io != NULL && state->writer_cb != NULL + && state->writer_cb->active_req != NULL && ZEND_ASYNC_CURRENT_COROUTINE != NULL) { zend_coroutine_t *const co = ZEND_ASYNC_CURRENT_COROUTINE; - unsigned spins = 0; + http_log_writer_cb_t *const cb = state->writer_cb; + + if (cb->drain_event == NULL) { + cb->drain_event = async_plain_event_new(); + } - while (state->writer_cb->active_req != NULL - && spins++ < HTTP_LOG_STOP_DRAIN_MAX_SPINS) { - http_log_stop_yield(co); + /* Same thread: no completion can fire between this check and SUSPEND, + * so no lost wakeup. drain_event only fires once the chain is empty. */ + if (cb->drain_event != NULL && ZEND_ASYNC_WAKER_NEW(co) != NULL) { + zend_async_resume_when(co, cb->drain_event, false, + zend_async_waker_callback_resolve, NULL); + zend_async_event_t *const timer = + &ZEND_ASYNC_NEW_TIMER_EVENT( + (zend_ulong)HTTP_LOG_STOP_DRAIN_BUDGET_MS, false)->base; + zend_async_resume_when(co, timer, true, + zend_async_waker_callback_timeout, NULL); + ZEND_ASYNC_SUSPEND(); + zend_async_waker_clean(co); + + if (EG(exception)) { + zend_clear_exception(); /* budget elapsed → leak path below */ + } } } @@ -610,6 +622,10 @@ void http_log_server_stop(http_log_state_t *state) efree(cb->pending_buf); } + if (cb->drain_event != NULL) { + cb->drain_event->dispose(cb->drain_event); + } + efree(cb); } diff --git a/stubs/HttpResponse.php b/stubs/HttpResponse.php index 5a193110..cf7d355b 100644 --- a/stubs/HttpResponse.php +++ b/stubs/HttpResponse.php @@ -179,24 +179,35 @@ public function write(string $data): static {} public function send(string $chunk): static {} /** - * Frame and stream one gRPC message. + * Declare the gRPC response message encoding. + * + * Must be called before the first writeMessage() — the encoding rides + * the initial HEADERS as the grpc-encoding header, and every subsequent + * writeMessage() compresses automatically. Supported: "gzip" and + * "identity" (the default; clears a previous declaration). Enabling + * compression is per-call by design — a compressed message without a + * declared encoding is a gRPC protocol error, so it cannot be expressed. * - * Prepends the 5-byte gRPC length prefix (identity encoding) 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 server-streaming. Pass the already - * protobuf-encoded bytes; the grpc-status is carried separately via - * setTrailer() (defaults to 0 when unset). + * @param string $encoding "gzip" or "identity". + * @return static + */ + public function setGrpcEncoding(string $encoding): static {} + + /** + * Frame and stream one gRPC message. * - * Pass $compress = true to gzip this message (sets the per-message - * compressed flag and, on the first message, the grpc-encoding: gzip - * response header). Falls back to identity when gzip is not built in. + * 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 + * 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') + * was declared. * * @param string $message Protobuf-encoded message bytes. - * @param bool $compress Gzip this message (grpc-encoding: gzip). * @return static */ - public function writeMessage(string $message, bool $compress = false): static {} + public function writeMessage(string $message): static {} /** * Advisory, non-blocking backpressure check for streaming responses. diff --git a/stubs/HttpResponse.php_arginfo.h b/stubs/HttpResponse.php_arginfo.h index 20294d58..97a37d03 100644 --- a/stubs/HttpResponse.php_arginfo.h +++ b/stubs/HttpResponse.php_arginfo.h @@ -68,9 +68,12 @@ ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_sen 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_setGrpcEncoding, 0, 1, IS_STATIC, 0) + ZEND_ARG_TYPE_INFO(0, encoding, IS_STRING, 0) +ZEND_END_ARG_INFO() + ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_writeMessage, 0, 1, IS_STATIC, 0) ZEND_ARG_TYPE_INFO(0, message, IS_STRING, 0) - ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, compress, _IS_BOOL, 0, "false") ZEND_END_ARG_INFO() ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_HttpResponse_sendable, 0, 0, _IS_BOOL, 0) @@ -156,6 +159,7 @@ ZEND_METHOD(TrueAsync_HttpResponse, getProtocolName); ZEND_METHOD(TrueAsync_HttpResponse, getProtocolVersion); ZEND_METHOD(TrueAsync_HttpResponse, write); ZEND_METHOD(TrueAsync_HttpResponse, send); +ZEND_METHOD(TrueAsync_HttpResponse, setGrpcEncoding); ZEND_METHOD(TrueAsync_HttpResponse, writeMessage); ZEND_METHOD(TrueAsync_HttpResponse, sendable); ZEND_METHOD(TrueAsync_HttpResponse, setNoCompression); @@ -196,6 +200,7 @@ static const zend_function_entry class_TrueAsync_HttpResponse_methods[] = { 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, 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) diff --git a/tests/phpt/server/grpc/008-grpc-compression.phpt b/tests/phpt/server/grpc/008-grpc-compression.phpt index 1dae862b..7d84402a 100644 --- a/tests/phpt/server/grpc/008-grpc-compression.phpt +++ b/tests/phpt/server/grpc/008-grpc-compression.phpt @@ -11,11 +11,12 @@ if (!function_exists('gzencode')) die('skip zlib ext not available'); ?> --FILE-- addGrpcHandler(function($req, $resp) { $msg = $req->readMessage(); // auto-inflated $empty = $req->readMessage(); // compressed empty message -> "" $tail = ($empty === '') ? ':empty-ok' : ':empty-bad'; - $resp->writeMessage('echo:' . $msg . $tail, true); // gzip the reply + $resp->setGrpcEncoding('gzip'); + $resp->writeMessage('echo:' . $msg . $tail); // compressed automatically }); $client = spawn(function() use ($port, $server) {