diff --git a/src/core/http_connection.c b/src/core/http_connection.c index d46602cc..f0a26fac 100644 --- a/src/core/http_connection.c +++ b/src/core/http_connection.c @@ -1413,6 +1413,78 @@ bool http_connection_send_raw(http_connection_t *conn, } /* }}} */ +#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900 +/* {{{ http_connection_send_strv_awaited + * + * Vectored plaintext send the caller waits for. Slots go out in array order; + * every reference is consumed whatever happens, including a cancellation while + * parked — the reactor holds them until libuv is done, which is exactly what + * the awaited single-buffer write cannot promise. */ +bool http_connection_send_strv_awaited(http_connection_t *conn, + zend_string *const *bufs, + const unsigned nbufs) +{ + /* The one refusal that neither throws nor consumes: unrecognisable after. */ + ZEND_ASSERT(nbufs > 0); + + if (UNEXPECTED(conn->write_timed_out)) { + for (unsigned i = 0; i < nbufs; i++) { + zend_string_release(bufs[i]); + } + + return false; + } + + const uint32_t write_timeout_ms = conn->write_timeout_ms; + + if (write_timeout_ms > 0 && !http_write_timer_arm(conn, write_timeout_ms)) { + for (unsigned i = 0; i < nbufs; i++) { + zend_string_release(bufs[i]); + } + + return false; + } + + size_t total = 0; + + for (unsigned i = 0; i < nbufs; i++) { + total += ZSTR_LEN(bufs[i]); + } + + bool ok_total = false; + zend_async_io_req_t *req = ZEND_ASYNC_IO_WRITEV_AWAITED(conn->io, bufs, nbufs); + + if (req != NULL) { + const bool ok = async_io_req_await(req, conn->io, write_timeout_ms, + HTTP_IO_REQ_WRITE, conn->log_state); + const bool had_exc = (req->exception != NULL); + + if (had_exc) { + OBJ_RELEASE(req->exception); + req->exception = NULL; + } + + const ssize_t transferred = req->transferred; + req->dispose(req); + ok_total = ok && !had_exc && transferred == (ssize_t)total; + } else { + /* Past the guard every refusal has released the slots already. */ + http_absorb_io_submission_exception(conn, __func__); + } + + if (write_timeout_ms > 0) { + http_write_timer_stop(conn); + } + + if (UNEXPECTED(conn->write_timed_out)) { + return false; + } + + return ok_total; +} +/* }}} */ +#endif /* async API >= 0.25 */ + /* {{{ http_connection_send_str_owned * * Fire-and-forget plaintext send: transfer ownership of @p body to the @@ -2781,7 +2853,15 @@ void http_handler_coroutine_dispose(zend_coroutine_t *coroutine) conn->state = CONN_STATE_SENDING; if (http_response_is_streaming(Z_OBJ(ctx->response_zv))) { - if (!http_response_is_closed(Z_OBJ(ctx->response_zv))) { + if (UNEXPECTED(ctx->stream_dead)) { + /* A cancellation can cut a frame in half, and the peer has been + * promised the bytes its size line named. Sealing that with the + * terminator would say the body ended cleanly and hand the + * connection on, and the peer would read the terminator as the + * first bytes of what it is still waiting for. mark_ended refuses + * the same way; this is the path that skips mark_ended. */ + conn->keep_alive = 0; + } else if (!http_response_is_closed(Z_OBJ(ctx->response_zv))) { /* Handler fell through without end() — emit the terminator. */ (void)http_connection_send(conn, "0\r\n\r\n", 5); } diff --git a/src/core/http_connection.h b/src/core/http_connection.h index ac52643e..3a13d3d3 100644 --- a/src/core/http_connection.h +++ b/src/core/http_connection.h @@ -461,6 +461,15 @@ bool http_connection_send_batched_writev(http_connection_t *conn, bool http_connection_send_strv_owned(http_connection_t *conn, zend_string * const *bufs, unsigned nbufs); +#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900 +/* Vectored variant the caller waits for. Each slot is an OWNED reference and is + * consumed in every outcome, a cancellation while parked included — the reactor + * keeps them until libuv is done, so a queued write never points at freed bytes. + * Plaintext only — same TLS caveat as send_str_owned. Requires nbufs > 0. */ +bool http_connection_send_strv_awaited(http_connection_t *conn, + zend_string * const *bufs, unsigned nbufs); +#endif + /* Outbound backpressure (transport-level, plaintext batched path). * pending_bytes = coalesced tail waiting behind the single in-flight * batched write — the part that grows under a slow consumer. The diff --git a/src/http1/http1_stream.c b/src/http1/http1_stream.c index 84d78bbc..2baaa373 100644 --- a/src/http1/http1_stream.c +++ b/src/http1/http1_stream.c @@ -59,6 +59,28 @@ ZEND_STATIC_ASSERT(H1_CHUNK_COALESCE_MAX <= HTTP_TLS_PLAINTEXT_RING_BYTES, "a coalesced frame must fit one TLS plaintext ring cycle"); +/* Sends a header block the caller owns, consuming its reference. Where the + * reactor can take it over it does, so a cancellation cannot leave a queued + * write pointing at a released string; elsewhere the copy is unavoidable. */ +static bool h1_send_headers_owned(http_connection_t *conn, zend_string *headers) +{ +#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900 +#ifdef HAVE_OPENSSL + const bool plaintext = conn->tls == NULL; +#else + const bool plaintext = true; +#endif + + if (plaintext) { + return http_connection_send_strv_awaited(conn, &headers, 1); + } +#endif + + const bool ok = http_connection_send(conn, ZSTR_VAL(headers), ZSTR_LEN(headers)); + zend_string_release(headers); + return ok; +} + /* The status line and headers of a streaming response, as bytes. Returns NULL * when the response is gone or formats to nothing; the caller owns the string. * Separate from the send so the first frame can carry the block with it. */ @@ -113,10 +135,7 @@ static bool h1_emit_headers_once(http1_request_ctx_t *ctx) return false; } - const bool ok = http_connection_send(ctx->conn, ZSTR_VAL(headers), - ZSTR_LEN(headers)); - zend_string_release(headers); - return ok; + return h1_send_headers_owned(ctx->conn, headers); } /* `nonblocking` is accepted and ignored: HTTP/1 keeps no queue of its own, so @@ -179,9 +198,7 @@ static int h1_stream_append_chunk(void *opaque, zend_string *chunk, zend_string_release(chunk); if (headers != NULL) { - const bool sent = http_connection_send(conn, ZSTR_VAL(headers), - ZSTR_LEN(headers)); - zend_string_release(headers); + const bool sent = h1_send_headers_owned(conn, headers); if (!sent) { ctx->stream_dead = true; @@ -221,7 +238,51 @@ static int h1_stream_append_chunk(void *opaque, zend_string *chunk, * over — today's ABI offers one or the other, never both. */ const size_t head_len = headers != NULL ? ZSTR_LEN(headers) : 0; const size_t frame_len = head_len + (size_t)header_len + chunk_len + 2; - const bool coalesce = frame_len <= H1_CHUNK_COALESCE_MAX; + +#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900 + /* Plaintext: the pieces go over as slots the reactor owns — one submit, no + * copy of the body, and nothing a cancelled frame could leave dangling. + * TLS keeps the copy below: tls_push copies into the BIO ring anyway, and a + * vectored write at the socket would put plaintext on a TLS connection. */ +#ifdef HAVE_OPENSSL + const bool plaintext = conn->tls == NULL; +#else + const bool plaintext = true; +#endif + + if (plaintext) { + zend_string *slots[4]; + unsigned n = 0; + + if (headers != NULL) { + slots[n++] = headers; /* ownership moves to the reactor */ + headers = NULL; + } + + slots[n++] = zend_string_init(header, (size_t)header_len, 0); + slots[n++] = chunk; /* the caller's ref, handed over */ + /* Two bytes per frame rather than one shared literal: the send releases + * every slot, and interning at runtime would write the process-wide + * permanent table from a worker thread. */ + slots[n++] = zend_string_init("\r\n", 2, 0); + + const bool sent = http_connection_send_strv_awaited(conn, slots, n); + + if (UNEXPECTED(!sent || EG(exception) != NULL)) { + ctx->stream_dead = true; + return HTTP_STREAM_APPEND_STREAM_DEAD; + } + + if (head_len != 0) { + h1_stream_headers_committed(ctx); + } + + http_server_on_stream_send(conn->counters, chunk_len); + return HTTP_STREAM_APPEND_OK; + } +#endif + + const bool coalesce = frame_len <= H1_CHUNK_COALESCE_MAX; bool frame_ok = true; /* Too large to carry the block along: the headers go out on their own, and diff --git a/tests/phpt/server/h1/030-h1-cancel-while-parked-in-write.phpt b/tests/phpt/server/h1/030-h1-cancel-while-parked-in-write.phpt new file mode 100644 index 00000000..41f79f39 --- /dev/null +++ b/tests/phpt/server/h1/030-h1-cancel-while-parked-in-write.phpt @@ -0,0 +1,131 @@ +--TEST-- +HttpResponse::write() — a handler cancelled while parked inside a write unwinds cleanly +--EXTENSIONS-- +true_async_server +true_async +sockets +--FILE-- +addListener('127.0.0.1', $port) + ->setReadTimeout(30)->setWriteTimeout(1) + ->setShutdownTimeout(0)); /* no grace window: cancel at once */ + +$seen = []; + +$server->addHttpHandler(function ($req, $res) use (&$seen) { + $res->setStatusCode(200)->setHeader('Content-Type', 'application/octet-stream'); + + try { + /* Far more than any socket buffer holds; parks after the first chunks. */ + for ($i = 0; $i < 20000; $i++) { + $res->write(str_repeat('x', 2000)); + } + $res->end(); + $seen['outcome'] = 'finished'; + } catch (\Throwable $e) { + $seen['outcome'] = get_class($e) . ':' . $e->getCode(); + $seen['writable_after'] = $res->isWritable(); + } +}); + +$cli = spawn(function () use ($port, $server) { + usleep(50000); + + $fp = stream_socket_client("tcp://127.0.0.1:$port", $errno, $errstr, 5); + $sock = socket_import_stream($fp); + socket_set_option($sock, SOL_SOCKET, SO_RCVBUF, 4096); + + fwrite($fp, "GET /stream HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + + /* Take the opening frames, then stop reading so the handler parks. */ + delay(200); + stream_set_blocking($fp, false); + $head = ''; + + for ($i = 0; $i < 40 && strlen($head) < 16384; $i++) { + $c = fread($fp, 8192); + if ($c !== false) $head .= $c; + usleep(5000); + } + + delay(500); + $server->stop(); + delay(500); + + /* Whatever the cancelled response still had to say. */ + $tail = ''; + + for ($i = 0; $i < 60; $i++) { + $c = fread($fp, 65536); + if ($c !== false) $tail .= $c; + usleep(5000); + } + + fclose($fp); + + [, $rest] = explode("\r\n\r\n", $head, 2); + $sizes = []; + $off = 0; + + while (count($sizes) < 3) { + $eol = strpos($rest, "\r\n", $off); + if ($eol === false) break; + $len = hexdec(substr($rest, $off, $eol - $off)); + $off = $eol + 2; + if ($len === 0 || $off + $len + 2 > strlen($rest)) break; + /* The frame must be exactly what the handler wrote, and end where the + * size line said it would — a released or misplaced slot shows here. */ + if (substr($rest, $off, $len) !== str_repeat('x', $len)) break; + if (substr($rest, $off + $len, 2) !== "\r\n") break; + $sizes[] = $len; + $off += $len + 2; + } + + echo "frames=", implode(',', $sizes), "\n"; + echo "terminator_after_cancel=", (int)(strpos($tail, "0\r\n\r\n") !== false), "\n"; +}); + +$server->start(); +await($cli); + +foreach ($seen as $k => $v) echo "$k = ", var_export($v, true), "\n"; +echo "done\n"; +?> +--EXPECT-- +frames=2000,2000,2000 +terminator_after_cancel=0 +outcome = 'TrueAsync\\HttpException:499' +writable_after = false +done