diff --git a/frameworks/proxygen-coro/README.md b/frameworks/proxygen-coro/README.md new file mode 100644 index 000000000..9d39ea47e --- /dev/null +++ b/frameworks/proxygen-coro/README.md @@ -0,0 +1,61 @@ +# proxygen-coro + +The same workload as the [`proxygen`](../proxygen/README.md) entry, served +through Proxygen's native coroutine stack instead of the callback +`RequestHandler` / `HTTPTransactionHandler` APIs. + +`src/CoroServer.cpp` implements `proxygen::coro::HTTPHandler`, consumes requests +through `HTTPSourceHolder` and returns `HTTPFixedSource` responses. Uploads are +counted while asynchronously draining BODY events. WebSockets use a long-lived +custom `HTTPSource`: its response calls `setEgressWebsocketUpgrade()`, then it +parses and emits RFC 6455 frames over the raw upgraded BODY event stream. + +Response compression is proxygen's coro `CompressionFilter`, attached by the +handler to the JSON responses rather than registered as a +`ServerCompressionFilterFactory` on `HTTPServer::Config`. The reason is that +`HTTPFilterFactoryHandler` calls `makeFilters()` *before* the request headers +have been read, so a server-level factory cannot be conditional: it allocates a +`SharedCtx`, a `VisitorFilter` holding a capturing lambda, a `CompressionFilter` +and a coroutine frame on **every** request, including `baseline`, where nothing +is compressible. That measured at ~10% of baseline throughput on this workload. +The classic `HTTPServer` path has no equivalent cost, because +`CompressionFilterFactory::onRequest` can see the request and returns the +handler unwrapped when there is nothing to compress. + +## Listener layout + +One process owns all five listeners: + +- `8080/tcp` HTTP/1.1 and WebSockets +- `8081/tcp` HTTP/1.1 over TLS, ALPN `http/1.1` +- `8082/tcp` prior-knowledge h2c +- `8443/tcp` HTTP/2 over TLS, ALPN `h2` +- `8443/udp` HTTP/3 over QUIC, ALPN `h3` + +The four TCP listeners are acceptors on one coro `HTTPServer`, so they share a +single affinity-aware I/O pool. Proxygen's coro API selects either TCP or QUIC +per server, so HTTP/3 uses a second in-process pool; whichever transport is not +being benchmarked stays idle. Size the two pools with `PROXYGEN_THREADS` and +`PROXYGEN_H3_THREADS` (`0` = available CPUs). + +HTTP/2 advertises 1024 concurrent streams with a 1 MiB stream window and a +10 MiB connection window. HTTP/3 mirrors Proxygen's coroutine benchmark +settings: GSO batches of 48 packets, continuous-memory writes, a large +congestion window and a 48-packet connection write limit. + +## Build + +This entry has no Dockerfile of its own. `build.sh` builds the +`frameworks/proxygen` context with `--build-arg TARGET=coro`, the same pattern +`sark-h3` uses to reuse `sark`. See +[the proxygen README](../proxygen/README.md#shared-source-tree) for the layout +and image details. + +```bash +./scripts/validate.sh proxygen-coro +./scripts/run.sh proxygen-coro +``` + +The implementation follows the upstream coroutine echo server and the +`H12DownstreamSessionTest.WebSocketUpgrade` test, which documents upgraded raw +bytes flowing through coroutine BODY events. diff --git a/frameworks/proxygen-coro/build.sh b/frameworks/proxygen-coro/build.sh new file mode 100755 index 000000000..9a14d9771 --- /dev/null +++ b/frameworks/proxygen-coro/build.sh @@ -0,0 +1,10 @@ +#!/usr/bin/env bash +# proxygen-coro reuses the proxygen build context and Dockerfile, selecting the +# coroutine server with --build-arg TARGET=coro (the sark-h3 -> sark pattern). +set -euo pipefail +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd -P)" +CONTEXT_DIR="$(cd "$SCRIPT_DIR/../proxygen" && pwd -P)" +docker build -t httparena-proxygen-coro \ + --build-arg TARGET=coro \ + -f "$CONTEXT_DIR/Dockerfile" \ + "$CONTEXT_DIR" diff --git a/frameworks/proxygen-coro/meta.json b/frameworks/proxygen-coro/meta.json new file mode 100644 index 000000000..fc75921e9 --- /dev/null +++ b/frameworks/proxygen-coro/meta.json @@ -0,0 +1,27 @@ +{ + "display_name": "proxygen-coro", + "language": "C++", + "type": "engine", + "engine": "proxygen", + "description": "Meta's Proxygen native coroutine HTTPServer and HTTPSource APIs across HTTP/1.1, HTTP/1.1 TLS, h2c, HTTP/2 TLS, HTTP/3 QUIC, and RFC 6455 WebSockets.", + "repo": "https://github.com/facebook/proxygen", + "enabled": true, + "tests": [ + "baseline", + "json-comp", + "json-tls", + "static-tls", + "pipelined", + "limited-conn", + "baseline-h2", + "baseline-h2c", + "json-h2c", + "static-h2", + "baseline-h3", + "static-h3", + "echo-ws", + "echo-ws-pipeline", + "echo-ws-limited" + ], + "maintainers": [] +} diff --git a/frameworks/proxygen/CMakeLists.txt b/frameworks/proxygen/CMakeLists.txt new file mode 100644 index 000000000..176ca4658 --- /dev/null +++ b/frameworks/proxygen/CMakeLists.txt @@ -0,0 +1,56 @@ +cmake_minimum_required(VERSION 3.20) + +project(httparena-proxygen LANGUAGES CXX) + +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_CXX_EXTENSIONS OFF) + +# Which server to build. Both HttpArena entries share this source tree: +# classic -> proxygen::HTTPServer + the mvfst-backed HQServer (`proxygen`) +# coro -> proxygen::coro::HTTPServer (`proxygen-coro`) +set(SERVER_TARGET "classic" CACHE STRING "classic or coro") + +# Proxygen's exported Fizz package calls find_dependency(Sodium). The official +# builder image keeps that upstream find module with the Proxygen source tree. +list(APPEND CMAKE_MODULE_PATH "/proxygen/build/fbcode_builder/CMake") + +# Proxygen's export refers to the historical un-namespaced c-ares target. +# Resolve the installed package and provide the alias expected by that export. +find_package(c-ares CONFIG REQUIRED) +add_library(cares ALIAS c-ares::cares) + +find_package(proxygen CONFIG REQUIRED) + +if(SERVER_TARGET STREQUAL "coro") + add_executable(arena-server src/CoroServer.cpp) + target_link_libraries( + arena-server + PRIVATE + proxygen::proxygen + proxygen::proxygen_coro + proxygen::proxygen_coro_server + proxygen::proxygen_http_coro_filters_compression_filter_factory + Folly::folly_init_init + Folly::folly_portability_gflags + ) +elseif(SERVER_TARGET STREQUAL "classic") + add_executable(arena-server src/ClassicServer.cpp src/ArenaHQServer.cpp) + target_link_libraries( + arena-server + PRIVATE + proxygen::proxygen + proxygen::proxygenhttpserver + proxygen::proxygen_hq_samples + proxygen::proxygen_hq_server + proxygen::proxygen_transport_persistent_quic_psk_cache + proxygen::proxygen_httpserver + Folly::folly_init_init + Folly::folly_portability_gflags + ) +else() + message(FATAL_ERROR "SERVER_TARGET must be 'classic' or 'coro', got '${SERVER_TARGET}'") +endif() + +target_include_directories(arena-server PRIVATE src) +target_compile_options(arena-server PRIVATE -Wall -Wextra -Wpedantic) diff --git a/frameworks/proxygen/Dockerfile b/frameworks/proxygen/Dockerfile new file mode 100644 index 000000000..2c67721b2 --- /dev/null +++ b/frameworks/proxygen/Dockerfile @@ -0,0 +1,66 @@ +# Shared image for both HttpArena Proxygen entries. +# +# TARGET=classic -> proxygen::HTTPServer + mvfst HQServer (`proxygen`) +# TARGET=coro -> proxygen::coro::HTTPServer (`proxygen-coro`) +# +# `frameworks/proxygen-coro/build.sh` builds this same context with +# `--build-arg TARGET=coro`, so every fix to src/ lands in both entries at once. +# +# PROXYGEN_BASE is pinned by digest so a benchmark round is reproducible; bump +# it deliberately rather than tracking a moving `:latest`. +ARG PROXYGEN_BASE=ghcr.io/facebook/proxygen/base@sha256:3f9c745ea7a3e065bdf6887f7255f2547c17c67fc1210ff120e9e52daa8eb3dc +ARG MIMALLOC_TAG=v2.1.7 + +# glibc malloc is the bottleneck on the allocation-heavy profiles: the JSON +# routes build a live folly::dynamic per request (up to 50 item copies plus the +# serialized string), and proxygen's coroutine server allocates a frame per +# event. Measured on this workload, mimalloc is worth +35% on `json` and +# +39%/+88% (classic/coro) on `json-comp`, against roughly -5% on `static`, +# where the large per-file IOBuf hits mimalloc's large-object path. +FROM ${PROXYGEN_BASE} AS mimalloc +ARG MIMALLOC_TAG +RUN git clone --depth 1 -b "${MIMALLOC_TAG}" https://github.com/microsoft/mimalloc.git /src/mimalloc \ + && cmake -S /src/mimalloc -B /src/build -DCMAKE_BUILD_TYPE=Release -DMI_BUILD_TESTS=OFF \ + && cmake --build /src/build --parallel "$(nproc)" \ + && cp "$(readlink -f /src/build/libmimalloc.so)" /src/libmimalloc.so + +FROM ${PROXYGEN_BASE} AS build +ARG TARGET=classic + +WORKDIR /arena +COPY CMakeLists.txt ./ +COPY src ./src +RUN cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -DSERVER_TARGET="${TARGET}" \ + && cmake --build build --parallel "$(nproc)" \ + && strip build/arena-server + +# Follow Proxygen's quic-interop image pattern: preserve the resolved shared +# library paths, then copy only those libraries and the server binary to +# a same-distro runtime image. +RUN set -eux; \ + ldd build/arena-server \ + | awk '/=> \// { print $3 } /^\// { print $1 }' | sort -u > /tmp/runtime-libs.txt; \ + tar -chf /tmp/runtime-libs.tar --files-from=/tmp/runtime-libs.txt + +FROM ubuntu:24.04@sha256:019e8eb29a85e74d64925745884f2ec79aa27e3feab36353d24656f4d6b89467 + +ENV LD_LIBRARY_PATH=/opt/proxygen/lib \ + LD_PRELOAD=/usr/local/lib/libmimalloc.so \ + MIMALLOC_PURGE_DELAY=1 + +COPY --from=build /tmp/runtime-libs.tar /tmp/runtime-libs.tar +RUN tar -xf /tmp/runtime-libs.tar -C / \ + && rm /tmp/runtime-libs.tar + +COPY --from=mimalloc /src/libmimalloc.so /usr/local/lib/libmimalloc.so +COPY --from=build /arena/build/arena-server /usr/local/bin/arena-server +COPY entrypoint.sh /usr/local/bin/arena-entrypoint +RUN chmod +x /usr/local/bin/arena-entrypoint \ + && groupadd --system --gid 10001 httparena \ + && useradd --system --uid 10001 --gid httparena --no-create-home \ + --home-dir /nonexistent --shell /usr/sbin/nologin httparena + +EXPOSE 8080/tcp 8081/tcp 8082/tcp 8443/tcp 8443/udp + +USER httparena +ENTRYPOINT ["/usr/local/bin/arena-entrypoint"] diff --git a/frameworks/proxygen/README.md b/frameworks/proxygen/README.md new file mode 100644 index 000000000..7754f7896 --- /dev/null +++ b/frameworks/proxygen/README.md @@ -0,0 +1,108 @@ +# Proxygen + +[Meta's Proxygen](https://github.com/facebook/proxygen) serving every protocol +HttpArena exercises, using Proxygen's callback server APIs: + +- `proxygen::HTTPServer` on TCP 8080 for HTTP/1.1 and RFC 6455 WebSocket + upgrades, TCP 8081 for HTTP/1.1 over TLS (ALPN `http/1.1` only), TCP 8082 for + prior-knowledge h2c, and TCP 8443 with TLS/ALPN `h2`. +- Proxygen's mvfst-backed `HQServer` with ALPN `h3` on UDP 8443. + +Both server APIs run in one process, each with its own affinity-aware I/O pool, +so whichever transport is being benchmarked can use the full CPU set while the +other pool sleeps. Size them independently with `PROXYGEN_THREADS` and +`PROXYGEN_H3_THREADS` (`0` = available CPUs). + +| Listener | Endpoints | Subscribed profiles | +| --- | --- | --- | +| HTTP/1.1 `:8080` | `/baseline11`, `/pipeline`, `/json/{count}`, `/upload`, `/static/*`, `/ws` | `baseline`, `pipelined`, `limited-conn`, `json-comp`, `echo-ws`, `echo-ws-pipeline`, `echo-ws-limited` | +| HTTP/1.1 TLS `:8081` | `/json/{count}`, `/static/*` | `json-tls`, `static-tls` | +| h2c `:8082` | `/baseline2`, `/json/{count}` | `baseline-h2c`, `json-h2c` | +| HTTP/2 TLS `:8443` | `/baseline2`, `/static/*` | `baseline-h2`, `static-h2` | +| HTTP/3 QUIC `:8443` | `/baseline2`, `/static/*` | `baseline-h3`, `static-h3` | + +The JSON routes load the immutable dataset once, then build each requested slice +and its derived `total` fields per request and serialize the live object with +Folly. Uploads count bytes delivered through the body callbacks rather than +trusting `Content-Length`. + +## Static assets + +Files are read from the mounted directory on every request — one `open`, one +`fstat`, one `readFull` straight into the response buffer, no intermediate +`std::string`. There is no cache: the arena's static rules require the served +bytes to follow the disk, and explicitly exclude a cache assembled in the entry +("no reading the directory into a map at startup"). `validate.sh` enforces this +with a staleness probe that swaps the file underneath a running server. + +The precompressed `.br`/`.gz` siblings are served when the client accepts them, +which the rules allow "by selecting the variant in the entry off +`Accept-Encoding`" — those bytes already exist on disk, so choosing one is a +file read rather than compression. Selection is a delimited token match that +honours `q=0`, preferring brotli, then gzip, then the byte-exact original, +which is what a client sending no `Accept-Encoding` always gets. + +That removes the on-the-fly gzip of every CSS/JS/HTML response, which was the +dominant cost in the static profiles. + +Response compression via Proxygen's `CompressionFilter` is therefore scoped to +`application/json`, the only content type still compressed at request time and +the one `json-comp` is scored on. The gzip level is Proxygen's default +(`Z_DEFAULT_COMPRESSION`); level 9 measured as a net loss, trading 2.3% of +throughput for 0.35% smaller bodies under the `(minBpr/myBpr)²` scoring. + +The WebSocket handler uses Proxygen's upgrade handshake (including its +per-connection `Sec-WebSocket-Accept` calculation) and implements incremental +RFC 6455 frame parsing: client frames are unmasked before text or binary data is +echoed, and fragmented messages, multiple frames per read, ping/pong and close +frames are all handled. This entry claims WebSocket support over HTTP/1.1 only, +which is the protocol HttpArena's WebSocket profiles exercise. + +## Shared source tree + +`frameworks/proxygen` is the build context for **both** Proxygen entries. The +`TARGET` build arg picks the server, the same way `sark-h3` reuses `sark`: + +| Entry | Build | Server API | +| --- | --- | --- | +| `proxygen` | `docker build frameworks/proxygen` | `src/ClassicServer.cpp` + `src/ArenaHQServer.cpp` | +| `proxygen-coro` | `frameworks/proxygen-coro/build.sh` (`--build-arg TARGET=coro`) | `src/CoroServer.cpp` | + +`src/ArenaCommon.h` holds the routing, parsing, static-asset and validation +helpers both servers share, so a fix lands in both entries at once. + +## Allocator + +Both images `LD_PRELOAD` mimalloc (pinned by tag in the Dockerfile). glibc malloc +is the binding constraint on the allocation-heavy profiles — the JSON routes build +a live `folly::dynamic` per request (up to 50 item copies plus the serialized +string), and the coroutine server allocates a frame per event. Measured here it is +worth roughly +41%/+82% (classic/coro) on `json-comp`. +`MIMALLOC_PURGE_DELAY=1` is set because the default (10 ms) retains ~17% more +memory on `json-comp` at 16384 connections for no throughput gain; going to 0 +would cut memory 4x further but costs 44% of that profile, so it is not used. + +## Image + +The builder tracks `ghcr.io/facebook/proxygen/base`, pinned by digest in +`ARG PROXYGEN_BASE` so a benchmark round is reproducible — bump it deliberately +rather than tracking `:latest`. The final stage contains only the Arena binary +and the shared libraries `ldd` resolves for it, on a digest-pinned Ubuntu 24.04, +running as the unprivileged `httparena` user (UID/GID 10001). + +Note that the prebuilt Proxygen, folly, fizz, wangle and mvfst libraries in the +base image are built `RelWithDebInfo` (`-O2 -g -DNDEBUG`): `getdeps.py` defaults +`--build-type` to `RelWithDebInfo` and Proxygen's `docker/base.Dockerfile` does +not override it. Only this repository's translation unit is `-O3`. + +HttpArena mounts `/certs/server.crt`, `/certs/server.key`, `/data/dataset.json` +and `/data/static` at runtime; no benchmark data is baked into the image. + +## Local validation + +From the repository root, on a host where the standard ports are free: + +```bash +./scripts/validate.sh proxygen +./scripts/benchmark.sh proxygen baseline +``` diff --git a/frameworks/proxygen/entrypoint.sh b/frameworks/proxygen/entrypoint.sh new file mode 100644 index 000000000..e63b25e1d --- /dev/null +++ b/frameworks/proxygen/entrypoint.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Shared by the `proxygen` and `proxygen-coro` images; the binary is the same +# name in both, only the server API compiled into it differs. +# +# PROXYGEN_THREADS TCP (H1/H2) I/O threads, 0 = available CPUs +# PROXYGEN_H3_THREADS QUIC I/O threads, 0 = available CPUs +exec /usr/local/bin/arena-server \ + --ip=:: \ + --http_port=8080 \ + --tls_port=8081 \ + --h2c_port=8082 \ + --h2_port=8443 \ + --h3_port=8443 \ + --cert=/certs/server.crt \ + --key=/certs/server.key \ + --threads="${PROXYGEN_THREADS:-0}" \ + --h3_threads="${PROXYGEN_H3_THREADS:-0}" diff --git a/frameworks/proxygen/meta.json b/frameworks/proxygen/meta.json new file mode 100644 index 000000000..709e20633 --- /dev/null +++ b/frameworks/proxygen/meta.json @@ -0,0 +1,27 @@ +{ + "display_name": "proxygen", + "language": "C++", + "type": "engine", + "engine": "proxygen", + "description": "Meta's Proxygen HTTP engine: HTTPServer for HTTP/1.1, HTTP/1.1 TLS, h2c, HTTP/2 TLS, and RFC 6455 WebSockets, plus the mvfst-backed HQ server for HTTP/3 over QUIC.", + "repo": "https://github.com/facebook/proxygen", + "enabled": true, + "tests": [ + "baseline", + "json-comp", + "json-tls", + "static-tls", + "pipelined", + "limited-conn", + "baseline-h2", + "baseline-h2c", + "json-h2c", + "static-h2", + "baseline-h3", + "static-h3", + "echo-ws", + "echo-ws-pipeline", + "echo-ws-limited" + ], + "maintainers": [] +} diff --git a/frameworks/proxygen/src/ArenaCommon.h b/frameworks/proxygen/src/ArenaCommon.h new file mode 100644 index 000000000..a9cfa60ed --- /dev/null +++ b/frameworks/proxygen/src/ArenaCommon.h @@ -0,0 +1,365 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace httparena { + +inline constexpr uint64_t kMaxWebSocketMessage = 16ULL * 1024 * 1024; +inline constexpr size_t kMaxRequestBody = 1024; +inline constexpr std::string_view kJsonPrefix = "/json/"; +inline constexpr std::string_view kStaticPrefix = "/static/"; +inline constexpr std::string_view kStaticRoot = "/data/static/"; + +inline bool parseInteger(std::string_view input, int64_t &value) { + while (!input.empty() && + std::isspace(static_cast(input.front()))) { + input.remove_prefix(1); + } + while (!input.empty() && + std::isspace(static_cast(input.back()))) { + input.remove_suffix(1); + } + if (input.empty()) { + return false; + } + const auto result = + std::from_chars(input.data(), input.data() + input.size(), value); + return result.ec == std::errc() && result.ptr == input.data() + input.size(); +} + +inline bool checkedAdd(int64_t lhs, int64_t rhs, int64_t &result) { +#if defined(__GNUC__) || defined(__clang__) + return !__builtin_add_overflow(lhs, rhs, &result); +#else + if ((rhs > 0 && lhs > std::numeric_limits::max() - rhs) || + (rhs < 0 && lhs < std::numeric_limits::min() - rhs)) { + return false; + } + result = lhs + rhs; + return true; +#endif +} + +inline bool checkedMultiply(int64_t lhs, int64_t rhs, int64_t &result) { +#if defined(__GNUC__) || defined(__clang__) + return !__builtin_mul_overflow(lhs, rhs, &result); +#else + if (lhs > 0) { + if ((rhs > 0 && lhs > std::numeric_limits::max() / rhs) || + (rhs < 0 && rhs < std::numeric_limits::min() / lhs)) { + return false; + } + } else if (lhs < 0) { + if ((rhs > 0 && lhs < std::numeric_limits::min() / rhs) || + (rhs < 0 && rhs < std::numeric_limits::max() / lhs)) { + return false; + } + } + result = lhs * rhs; + return true; +#endif +} + +// Returns the value for `name` in a raw `a=1&b=2` query string, or nullopt if +// it is absent. Scanning beats HTTPMessage::getQueryParam() on the hot path: +// that accessor materialises a std::map of every +// parameter on first use, i.e. two string allocations per parameter per +// request, and `baseline` is the most-requested endpoint in the suite. +inline std::optional queryValue(std::string_view query, + std::string_view name) { + while (!query.empty()) { + const auto amp = query.find('&'); + const std::string_view field = query.substr(0, amp); + const auto eq = field.find('='); + if (eq != std::string_view::npos && field.substr(0, eq) == name) { + return field.substr(eq + 1); + } + if (amp == std::string_view::npos) { + break; + } + query.remove_prefix(amp + 1); + } + return std::nullopt; +} + +inline std::string_view contentType(std::string_view name) { + const auto endsWith = [name](std::string_view suffix) { + return name.size() >= suffix.size() && + name.substr(name.size() - suffix.size()) == suffix; + }; + if (endsWith(".css")) { + return "text/css"; + } + if (endsWith(".js")) { + return "application/javascript"; + } + if (endsWith(".html")) { + return "text/html"; + } + if (endsWith(".json")) { + return "application/json"; + } + if (endsWith(".svg")) { + return "image/svg+xml"; + } + if (endsWith(".webp")) { + return "image/webp"; + } + if (endsWith(".woff2")) { + return "font/woff2"; + } + return "application/octet-stream"; +} + +// ── Static assets ─────────────────────────────────────────────────────────── +// +// Read from the mounted directory on every request. The arena's static rules +// require the served bytes to follow the disk — replace a file and the next +// response must carry the new bytes — and explicitly exclude a cache assembled +// in the entry ("no reading the directory into a map at startup"). validate.sh +// enforces this with a staleness probe that swaps the file underneath a running +// server, so a preloaded table fails outright. +// +// Serving the precompressed .br/.gz siblings IS allowed, "by selecting the +// variant in the entry off Accept-Encoding" — those bytes already exist on +// disk, so choosing one is a file read rather than compression. That is where +// the remaining win lives: no gzip on the event-base thread per request. + +enum class ContentEncoding : uint8_t { Identity, Gzip, Brotli }; + +inline std::string_view encodingToken(ContentEncoding encoding) { + switch (encoding) { + case ContentEncoding::Brotli: + return "br"; + case ContentEncoding::Gzip: + return "gzip"; + case ContentEncoding::Identity: + break; + } + return {}; +} + +// Token search rather than a full RFC 7231 qvalue parse: the token must be +// delimited so "br" does not match inside another coding, and an explicit +// `q=0` disqualifies it. +inline bool acceptsToken(std::string_view header, std::string_view token) { + size_t pos = 0; + while ((pos = header.find(token, pos)) != std::string_view::npos) { + const bool leftOk = + pos == 0 || header[pos - 1] == ' ' || header[pos - 1] == ','; + const size_t after = pos + token.size(); + const bool rightOk = after == header.size() || header[after] == ',' || + header[after] == ';' || header[after] == ' '; + if (leftOk && rightOk) { + const auto end = header.find(',', after); + const auto params = + header.substr(after, end == std::string_view::npos ? end : end - after); + const auto q = params.find("q="); + if (q == std::string_view::npos) { + return true; + } + const auto qv = params.substr(q); + return !(qv == "q=0" || qv == "q=0.0" || qv.rfind("q=0,", 0) == 0); + } + pos = after; + } + return false; +} + +// Reads `root/name` into a fresh IOBuf: one open, one fstat, one readFull, no +// intermediate std::string. Returns nullptr when the file is absent. +inline std::unique_ptr readFileToIOBuf(const std::string &path) { + int fd = folly::openNoInt(path.c_str(), O_RDONLY | O_CLOEXEC); + if (fd < 0) { + return nullptr; + } + folly::File file(fd, /*ownsFd=*/true); + struct stat info {}; + if (::fstat(fd, &info) != 0 || !S_ISREG(info.st_mode)) { + return nullptr; + } + const auto size = static_cast(info.st_size); + auto buffer = folly::IOBuf::create(size); + if (size > 0) { + const ssize_t got = folly::readFull(fd, buffer->writableTail(), size); + if (got < 0 || static_cast(got) != size) { + return nullptr; + } + } + buffer->append(size); + return buffer; +} + +struct StaticResponse { + std::unique_ptr body; + std::string_view contentType; + ContentEncoding encoding{ContentEncoding::Identity}; +}; + +// `name` must already be rejected if it contains a separator or "..". +// Prefers the precompressed sibling this client accepts, falling back to the +// original — which is what a client sending no Accept-Encoding always gets. +inline StaticResponse serveStatic(std::string_view name, + std::string_view acceptEncoding) { + StaticResponse out; + std::string path; + path.reserve(kStaticRoot.size() + name.size() + 3); + path.append(kStaticRoot).append(name); + const size_t baseLen = path.size(); + + if (acceptsToken(acceptEncoding, "br")) { + path.append(".br"); + out.body = readFileToIOBuf(path); + if (out.body) { + out.encoding = ContentEncoding::Brotli; + } + path.resize(baseLen); + } + if (!out.body && acceptsToken(acceptEncoding, "gzip")) { + path.append(".gz"); + out.body = readFileToIOBuf(path); + if (out.body) { + out.encoding = ContentEncoding::Gzip; + } + path.resize(baseLen); + } + if (!out.body) { + out.body = readFileToIOBuf(path); + out.encoding = ContentEncoding::Identity; + } + if (out.body) { + out.contentType = contentType(name); + } + return out; +} + +inline std::shared_ptr loadDataset() { + std::string contents; + if (!folly::readFile("/data/dataset.json", contents)) { + throw std::runtime_error("cannot open /data/dataset.json"); + } + auto dataset = folly::parseJson(contents); + if (!dataset.isArray() || dataset.size() < 50) { + throw std::runtime_error("/data/dataset.json must contain 50 items"); + } + return std::make_shared(std::move(dataset)); +} + +inline bool validWebSocketKey(std::string_view key) noexcept { + if (key.size() != 24) { + return false; + } + std::array decoded{}; + const auto result = folly::base64Decode(key, decoded.data()); + return result.is_success && result.o == decoded.data() + 16; +} + +inline bool validUtf8(const uint8_t *data, size_t size) noexcept { + const auto continuation = [](uint8_t byte) { + return byte >= 0x80 && byte <= 0xbf; + }; + + size_t index = 0; + while (index < size) { + const uint8_t first = data[index]; + if (first <= 0x7f) { + ++index; + continue; + } + if (first >= 0xc2 && first <= 0xdf) { + if (index + 1 >= size || !continuation(data[index + 1])) { + return false; + } + index += 2; + continue; + } + if (first == 0xe0) { + if (index + 2 >= size || data[index + 1] < 0xa0 || + data[index + 1] > 0xbf || !continuation(data[index + 2])) { + return false; + } + index += 3; + continue; + } + if ((first >= 0xe1 && first <= 0xec) || (first >= 0xee && first <= 0xef)) { + if (index + 2 >= size || !continuation(data[index + 1]) || + !continuation(data[index + 2])) { + return false; + } + index += 3; + continue; + } + if (first == 0xed) { + if (index + 2 >= size || data[index + 1] < 0x80 || + data[index + 1] > 0x9f || !continuation(data[index + 2])) { + return false; + } + index += 3; + continue; + } + if (first == 0xf0) { + if (index + 3 >= size || data[index + 1] < 0x90 || + data[index + 1] > 0xbf || !continuation(data[index + 2]) || + !continuation(data[index + 3])) { + return false; + } + index += 4; + continue; + } + if (first >= 0xf1 && first <= 0xf3) { + if (index + 3 >= size || !continuation(data[index + 1]) || + !continuation(data[index + 2]) || !continuation(data[index + 3])) { + return false; + } + index += 4; + continue; + } + if (first == 0xf4) { + if (index + 3 >= size || data[index + 1] < 0x80 || + data[index + 1] > 0x8f || !continuation(data[index + 2]) || + !continuation(data[index + 3])) { + return false; + } + index += 4; + continue; + } + return false; + } + return true; +} + +inline bool validUtf8(const std::vector &data) noexcept { + return validUtf8(data.data(), data.size()); +} + +inline bool validWebSocketCloseCode(uint16_t code) noexcept { + const bool definedProtocolCode = code >= 1000 && code <= 1014 && + code != 1004 && code != 1005 && code != 1006; + const bool applicationCode = code >= 3000 && code <= 4999; + return definedProtocolCode || applicationCode; +} + +} // namespace httparena diff --git a/frameworks/proxygen/src/ArenaHQServer.cpp b/frameworks/proxygen/src/ArenaHQServer.cpp new file mode 100644 index 000000000..5dc8d040d --- /dev/null +++ b/frameworks/proxygen/src/ArenaHQServer.cpp @@ -0,0 +1,77 @@ +#include "ArenaHQServer.h" + +#include + +#include +#include +#include + +namespace { + +quic::samples::HQServerParams makeHQParams(size_t ioThreads) { + quic::samples::HQServerParams params; + params.serverThreads = ioThreads; + // HQServerParams leaves these at 0, i.e. the system default (typically + // 208 KB), which is small for a loopback benchmark writing 48-packet GSO + // batches per connection. + params.udpSendBufferSize = 4 * 1024 * 1024; + params.udpRecvBufferSize = 4 * 1024 * 1024; + params.transportSettings.maxNumPTOs = 1000; + params.transportSettings.maxCwndInMss = quic::kLargeMaxCwndInMss; + params.transportSettings.batchingMode = + quic::QuicBatchingMode::BATCHING_MODE_GSO; + params.transportSettings.maxBatchSize = 48; + params.transportSettings.dataPathType = quic::DataPathType::ContinuousMemory; + params.transportSettings.writeConnectionDataPacketsLimit = 48; + return params; +} + +} // namespace + +namespace httparena { + +class ArenaHQServer::Impl final { +public: + Impl(const std::string &certificatePath, const std::string &privateKeyPath, + size_t ioThreads, HandlerProvider handlerProvider) + : server_(makeHQParams(ioThreads), std::move(handlerProvider), nullptr, + quic::samples::createFizzServerContext( + quic::samples::kDefaultSupportedAlpns, + fizz::server::ClientAuthMode::None, certificatePath, + privateKeyPath)) {} + + ~Impl() { stop(); } + + void start(const folly::SocketAddress &address) { + started_ = true; + server_.start(address); + server_.getAddress(); + } + + void stop() { + if (started_) { + server_.stop(); + started_ = false; + } + } + +private: + quic::samples::HQServer server_; + bool started_{false}; +}; + +ArenaHQServer::ArenaHQServer(const std::string &certificatePath, + const std::string &privateKeyPath, + size_t ioThreads, HandlerProvider handlerProvider) + : impl_(std::make_unique(certificatePath, privateKeyPath, ioThreads, + std::move(handlerProvider))) {} + +ArenaHQServer::~ArenaHQServer() = default; + +void ArenaHQServer::start(const folly::SocketAddress &address) { + impl_->start(address); +} + +void ArenaHQServer::stop() { impl_->stop(); } + +} // namespace httparena diff --git a/frameworks/proxygen/src/ArenaHQServer.h b/frameworks/proxygen/src/ArenaHQServer.h new file mode 100644 index 000000000..34c985b36 --- /dev/null +++ b/frameworks/proxygen/src/ArenaHQServer.h @@ -0,0 +1,38 @@ +#pragma once + +#include +#include +#include +#include + +#include + +namespace proxygen { +class HTTPMessage; +class HTTPTransactionHandler; +} // namespace proxygen + +namespace httparena { + +class ArenaHQServer final { +public: + using HandlerProvider = std::function; + + ArenaHQServer(const std::string &certificatePath, + const std::string &privateKeyPath, size_t ioThreads, + HandlerProvider handlerProvider); + ~ArenaHQServer(); + + ArenaHQServer(const ArenaHQServer &) = delete; + ArenaHQServer &operator=(const ArenaHQServer &) = delete; + + void start(const folly::SocketAddress &address); + void stop(); + +private: + class Impl; + std::unique_ptr impl_; +}; + +} // namespace httparena diff --git a/frameworks/proxygen/src/ArenaWebSocket.h b/frameworks/proxygen/src/ArenaWebSocket.h new file mode 100644 index 000000000..a0446befb --- /dev/null +++ b/frameworks/proxygen/src/ArenaWebSocket.h @@ -0,0 +1,381 @@ +#pragma once + +// Incremental RFC 6455 frame codec shared by ClassicServer and CoroServer. +// +// Reads normally carry whole frames, so the common path parses straight out of +// the transport's buffer with no intermediate queue: each frame is unmasked in +// place, and only a trailing partial frame is copied into `carry_`. +// +// Echoes are accumulated into one egress buffer per read, so a read carrying N +// pipelined frames produces one write rather than N. Small payloads (the echo +// profiles use ~128 bytes) are memcpy'd into that buffer, which is cheaper than +// the per-frame IOBuf bookkeeping a zero-copy view would cost. Large payloads +// cross over, and are emitted as a clone view of the read buffer instead: the +// answer to a frame reuses the same length encoding as the question, so the +// server header lands exactly on the four mask bytes preceding the payload. + +#include +#include +#include +#include +#include +#include + +#include +#include + +#include "ArenaCommon.h" + +namespace httparena { + +class WebSocketEcho { +public: + enum class Opcode : uint8_t { + Continuation = 0x0, + Text = 0x1, + Binary = 0x2, + Close = 0x8, + Ping = 0x9, + Pong = 0xa, + }; + + // Payloads at or above this are echoed as a view of the read buffer; below + // it, copying into the shared egress buffer costs less than a second IOBuf. + static constexpr size_t kZeroCopyThreshold = 4096; + + void onIngress(std::unique_ptr data) { + if (finished_ || !data) { + return; + } + if (data->isChained()) { + data->coalesce(); + } + if (carry_.empty()) { + const size_t consumed = + parse(data->writableData(), data->length(), data.get()); + if (consumed < data->length()) { + carry_.assign(data->data() + consumed, data->data() + data->length()); + } + return; + } + append(carry_, data->data(), data->length()); + // No owning IOBuf for the carried bytes, so this path always copies. It + // only runs when a previous read ended mid-frame. + const size_t consumed = parse(carry_.data(), carry_.size(), nullptr); + if (consumed == carry_.size()) { + carry_.clear(); + } else if (consumed > 0) { + carry_.erase(carry_.begin(), + carry_.begin() + static_cast(consumed)); + } + } + + // Everything queued since the last call, or nullptr if there is nothing. + std::unique_ptr takeEgress() { + return egress_.empty() ? nullptr : egress_.move(); + } + + // At most `limit` bytes, for the coro HTTPSource, which is handed a per-read + // size cap. + std::unique_ptr takeEgress(size_t limit) { + if (egress_.empty()) { + return nullptr; + } + if (egress_.chainLength() <= limit) { + return egress_.move(); + } + return egress_.split(limit); + } + + bool hasEgress() const { return !egress_.empty(); } + + // True once a close frame has been answered or the stream is unusable: the + // caller should send EOM after flushing whatever is still queued. + bool finished() const { return finished_; } + +private: + // Writes an unmasked server frame header for `payloadLength` into `out`, + // which must have room for 10 bytes. Returns the header length. + static size_t writeHeader(uint8_t *out, uint8_t opcode, + uint64_t payloadLength) { + out[0] = static_cast(0x80U | opcode); + if (payloadLength <= 125) { + out[1] = static_cast(payloadLength); + return 2; + } + if (payloadLength <= std::numeric_limits::max()) { + out[1] = 126; + out[2] = static_cast((payloadLength >> 8) & 0xff); + out[3] = static_cast(payloadLength & 0xff); + return 4; + } + out[1] = 127; + for (int shift = 56, index = 2; shift >= 0; shift -= 8, ++index) { + out[index] = static_cast((payloadLength >> shift) & 0xff); + } + return 10; + } + + void queueFrame(uint8_t opcode, const uint8_t *payload, size_t length) { + std::array header; + const size_t headerLength = writeHeader(header.data(), opcode, length); + egress_.append(header.data(), headerLength); + if (length > 0) { + egress_.append(payload, length); + } + } + + void closeWith(uint16_t status) { + if (closeSent_ || finished_) { + return; + } + const std::array payload = { + static_cast((status >> 8) & 0xff), + static_cast(status & 0xff)}; + queueFrame(static_cast(Opcode::Close), payload.data(), + payload.size()); + closeSent_ = true; + finished_ = true; + } + + void protocolError() { closeWith(1002); } + void invalidPayload() { closeWith(1007); } + void messageTooBig() { closeWith(1009); } + + // resize + memcpy rather than vector::insert from a raw pointer range: + // same work, and GCC 13 cannot bound the iterator form, which trips a + // false-positive -Wstringop-overflow. + static void append(std::vector &out, const uint8_t *data, + size_t length) { + const size_t offset = out.size(); + out.resize(offset + length); + if (length > 0) { + std::memcpy(out.data() + offset, data, length); + } + } + + // Unmasks `length` bytes in place. The mask repeats every four bytes, so it + // is applied as a rotating 32-bit word rather than byte-at-a-time modulo. + static void unmask(uint8_t *payload, size_t length, + const uint8_t *mask) noexcept { + uint32_t maskWord; + std::memcpy(&maskWord, mask, sizeof(maskWord)); + size_t index = 0; + for (; index + sizeof(uint32_t) <= length; index += sizeof(uint32_t)) { + uint32_t chunk; + std::memcpy(&chunk, payload + index, sizeof(chunk)); + chunk ^= maskWord; + std::memcpy(payload + index, &chunk, sizeof(chunk)); + } + for (; index < length; ++index) { + payload[index] ^= mask[index & 3U]; + } + } + + // Echoes a large frame without copying its payload. `frame` points at the + // start of the received frame inside `owner`: + // [header(headerLength)][mask(4)][payload] + // The answer is [header(headerLength)][payload] with the same length + // encoding, so the server header lands on offset 4 and the view starts + // there. The mask has already been consumed by unmask(). + void echoAsView(folly::IOBuf *owner, uint8_t *frame, uint8_t opcode, + size_t headerLength, size_t payloadLength) { + writeHeader(frame + 4, opcode, payloadLength); + auto view = owner->cloneOne(); + const size_t offset = static_cast(frame - owner->writableData()); + view->trimStart(offset + 4); + view->trimEnd(view->length() - (headerLength + payloadLength)); + egress_.append(std::move(view), /*pack=*/false); + } + + void handleControlFrame(uint8_t opcode, uint8_t *payload, size_t length) { + switch (static_cast(opcode)) { + case Opcode::Close: + if (length == 1) { + protocolError(); + return; + } + if (length >= 2) { + const auto status = static_cast( + (static_cast(payload[0]) << 8) | payload[1]); + if (!validWebSocketCloseCode(status)) { + protocolError(); + return; + } + if (!validUtf8(payload + 2, length - 2)) { + invalidPayload(); + return; + } + } + if (!closeSent_) { + queueFrame(static_cast(Opcode::Close), payload, length); + closeSent_ = true; + } + finished_ = true; + return; + case Opcode::Ping: + queueFrame(static_cast(Opcode::Pong), payload, length); + return; + case Opcode::Pong: + return; + default: + protocolError(); + return; + } + } + + // Consumes as many whole frames as `length` holds, unmasking in place. + // `owner` is the IOBuf backing `base`, or nullptr when parsing carried bytes + // (in which case large frames are copied rather than viewed). Returns the + // number of bytes consumed. + size_t parse(uint8_t *base, size_t length, folly::IOBuf *owner) { + size_t cursor = 0; + while (!finished_) { + const size_t available = length - cursor; + if (available < 2) { + break; + } + uint8_t *frame = base + cursor; + const uint8_t first = frame[0]; + const uint8_t second = frame[1]; + const bool fin = (first & 0x80U) != 0; + const uint8_t opcode = first & 0x0fU; + const bool control = (opcode & 0x08U) != 0; + const uint8_t encodedLength = second & 0x7fU; + + // Reserved bits must be clear, clients must mask, and control frames may + // not use an extended length encoding even for a short payload. + if ((first & 0x70U) != 0 || (second & 0x80U) == 0 || + (control && encodedLength > 125)) { + protocolError(); + return length; + } + + size_t headerLength = 2; + uint64_t payloadLength = encodedLength; + if (encodedLength == 126) { + headerLength = 4; + if (available < headerLength) { + break; + } + payloadLength = (static_cast(frame[2]) << 8) | + static_cast(frame[3]); + if (payloadLength < 126) { // not minimally encoded + protocolError(); + return length; + } + } else if (encodedLength == 127) { + headerLength = 10; + if (available < headerLength) { + break; + } + if ((frame[2] & 0x80U) != 0) { // high bit of a 64-bit length is 0 + protocolError(); + return length; + } + payloadLength = 0; + for (size_t index = 0; index < 8; ++index) { + payloadLength = (payloadLength << 8) | frame[2 + index]; + } + if (payloadLength <= std::numeric_limits::max()) { + protocolError(); + return length; + } + } + + if (payloadLength > kMaxWebSocketMessage) { + messageTooBig(); + return length; + } + + constexpr size_t kMaskLength = 4; + const size_t frameLength = + headerLength + kMaskLength + static_cast(payloadLength); + if (available < frameLength) { + break; + } + + uint8_t *payload = frame + headerLength + kMaskLength; + const auto size = static_cast(payloadLength); + unmask(payload, size, frame + headerLength); + cursor += frameLength; + + if (control) { + if (!fin) { + protocolError(); + return length; + } + handleControlFrame(opcode, payload, size); + continue; + } + if (closeSent_) { + continue; // draining after our own close + } + + if (opcode == static_cast(Opcode::Continuation)) { + if (fragmentOpcode_ == 0) { + protocolError(); + return length; + } + if (size > kMaxWebSocketMessage - fragmentPayload_.size()) { + messageTooBig(); + return length; + } + append(fragmentPayload_, payload, size); + if (fin) { + finishFragmentedMessage(); + } + continue; + } + + if (opcode != static_cast(Opcode::Text) && + opcode != static_cast(Opcode::Binary)) { + protocolError(); + return length; + } + if (fragmentOpcode_ != 0) { // interleaved new message + protocolError(); + return length; + } + if (!fin) { + fragmentOpcode_ = opcode; + fragmentPayload_.assign(payload, payload + size); + continue; + } + if (opcode == static_cast(Opcode::Text) && + !validUtf8(payload, size)) { + invalidPayload(); + return length; + } + + if (owner != nullptr && size >= kZeroCopyThreshold) { + echoAsView(owner, frame, opcode, headerLength, size); + } else { + queueFrame(opcode, payload, size); + } + } + return cursor; + } + + void finishFragmentedMessage() { + if (fragmentOpcode_ == static_cast(Opcode::Text) && + !validUtf8(fragmentPayload_)) { + invalidPayload(); + return; + } + queueFrame(fragmentOpcode_, fragmentPayload_.data(), + fragmentPayload_.size()); + fragmentOpcode_ = 0; + fragmentPayload_.clear(); + fragmentPayload_.shrink_to_fit(); + } + + folly::IOBufQueue egress_{folly::IOBufQueue::cacheChainLength()}; + std::vector carry_; + std::vector fragmentPayload_; + uint8_t fragmentOpcode_{0}; + bool closeSent_{false}; + bool finished_{false}; +}; + +} // namespace httparena diff --git a/frameworks/proxygen/src/ClassicServer.cpp b/frameworks/proxygen/src/ClassicServer.cpp new file mode 100644 index 000000000..558e31413 --- /dev/null +++ b/frameworks/proxygen/src/ClassicServer.cpp @@ -0,0 +1,524 @@ +#include "ArenaCommon.h" +#include "ArenaHQServer.h" +#include "ArenaWebSocket.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using folly::SocketAddress; +using proxygen::HTTPMessage; +using proxygen::HTTPMethod; +using proxygen::HTTPServer; +using proxygen::ProxygenError; +using proxygen::RequestHandler; +using proxygen::RequestHandlerChain; +using proxygen::RequestHandlerFactory; +using proxygen::ResponseBuilder; +using proxygen::UpgradeProtocol; + +DEFINE_int32(http_port, 8080, "HTTP/1.1 and WebSocket port"); +DEFINE_int32(tls_port, 8081, "HTTP/1.1 over TLS port"); +DEFINE_int32(h2c_port, 8082, "HTTP/2 cleartext port"); +DEFINE_int32(h2_port, 8443, "HTTP/2 over TLS port"); +DEFINE_int32(h3_port, 8443, "HTTP/3 over QUIC port"); +DEFINE_string(ip, "::", "Address on which to listen"); +DEFINE_string(cert, "/certs/server.crt", "TLS certificate path"); +DEFINE_string(key, "/certs/server.key", "TLS private-key path"); +DEFINE_int32(threads, 0, "TCP I/O threads; 0 uses the available CPU count"); +DEFINE_int32(h3_threads, 0, "QUIC I/O threads; 0 uses the available CPU count"); + +namespace { + +using httparena::checkedAdd; +using httparena::checkedMultiply; +using httparena::contentType; +using httparena::kJsonPrefix; +using httparena::kMaxRequestBody; +using httparena::kStaticPrefix; +using httparena::loadDataset; +using httparena::parseInteger; +using httparena::queryValue; +using httparena::ContentEncoding; +using httparena::serveStatic; +using httparena::validWebSocketKey; + +class ArenaHandler final : public RequestHandler { +public: + // The dataset is owned by the handler factory, which outlives every handler + // it creates. Taking a reference rather than a shared_ptr copy keeps two + // atomic refcount updates on a globally shared cache line off the per-request + // path — one handler is allocated per request. + explicit ArenaHandler(const folly::dynamic &dataset) : dataset_(dataset) {} + + void onRequest(std::unique_ptr request) noexcept override { + const auto path = request->getPathAsStringPiece(); + const auto method = request->getMethod(); + + if (path == "/ws") { + route_ = Route::WebSocket; + const auto &headers = request->getHeaders(); + const auto &key = headers.getSingleOrEmpty("Sec-WebSocket-Key"); + const auto &version = headers.getSingleOrEmpty("Sec-WebSocket-Version"); + if (!method || *method != HTTPMethod::GET || + !request->isIngressWebsocketUpgrade() || + !validWebSocketKey(std::string_view(key.data(), key.size())) || + version != "13") { + ResponseBuilder(downstream_) + .status(426, "Upgrade Required") + .header("Content-Type", "text/plain") + .header("Sec-WebSocket-Version", "13") + .body("WebSocket upgrade required") + .sendWithEOM(); + responseFinished_ = true; + return; + } + + ResponseBuilder(downstream_) + .status(101, "Switching Protocols") + .setEgressWebsocketHeaders() + .send(); + websocketAccepted_ = true; + return; + } + + method_ = method.value_or(HTTPMethod::GET); + const std::string_view query(request->getQueryStringAsStringPiece().data(), + request->getQueryStringAsStringPiece().size()); + + if (path == "/baseline11" || path == "/baseline2") { + route_ = path == "/baseline11" ? Route::Baseline : Route::BaselineH2; + const auto a = queryValue(query, "a"); + const auto b = queryValue(query, "b"); + queryValid_ = a && b && parseInteger(*a, a_) && parseInteger(*b, b_); + return; + } + if (path.startsWith(kJsonPrefix)) { + route_ = Route::Json; + const std::string_view countText(path.data() + kJsonPrefix.size(), + path.size() - kJsonPrefix.size()); + int64_t count = 0; + const auto multiplierText = queryValue(query, "m"); + const bool multiplierValid = + !multiplierText ? (multiplier_ = 1, true) + : parseInteger(*multiplierText, multiplier_); + jsonValid_ = parseInteger(countText, count) && count >= 1 && + count <= 50 && multiplierValid; + if (jsonValid_) { + jsonCount_ = static_cast(count); + } + return; + } + if (path == "/upload") { + route_ = Route::Upload; + return; + } + if (path.startsWith(kStaticPrefix)) { + route_ = Route::Static; + staticName_.assign(path.data() + kStaticPrefix.size(), + path.size() - kStaticPrefix.size()); + const auto &encoding = + request->getHeaders().getSingleOrEmpty( + proxygen::HTTP_HEADER_ACCEPT_ENCODING); + acceptEncoding_.assign(encoding.data(), encoding.size()); + return; + } + if (path == "/pipeline") { + route_ = Route::Pipeline; + return; + } + route_ = Route::NotFound; + } + + void onBody(std::unique_ptr body) noexcept override { + if (!body || responseFinished_) { + return; + } + if (route_ == Route::Upload) { + const size_t bytes = body->computeChainDataLength(); + if (bytes > std::numeric_limits::max() - uploadBytes_) { + uploadValid_ = false; + } else { + uploadBytes_ += bytes; + } + return; + } + if (route_ == Route::WebSocket && websocketActive_) { + websocket_.onIngress(std::move(body)); + flushWebSocket(); + return; + } + if (route_ != Route::Baseline) { + return; + } + auto bytes = body->coalesce(); + if (requestBody_.size() + bytes.size() > kMaxRequestBody) { + bodyValid_ = false; + return; + } + requestBody_.append(reinterpret_cast(bytes.data()), + bytes.size()); + } + + void onUpgrade(UpgradeProtocol /*protocol*/) noexcept override { + if (route_ != Route::WebSocket || !websocketAccepted_) { + downstream_->sendAbort(); + return; + } + websocketActive_ = true; + } + + void onEOM() noexcept override { + if (responseFinished_) { + return; + } + switch (route_) { + case Route::WebSocket: + responseFinished_ = true; + downstream_->sendEOM(); + return; + case Route::NotFound: + sendText(404, "Not Found", "not found"); + return; + case Route::Pipeline: + if (method_ != HTTPMethod::GET) { + sendText(405, "Method Not Allowed", "method not allowed"); + } else { + sendText(200, "OK", "ok"); + } + return; + case Route::Baseline: + handleBaseline(true); + return; + case Route::BaselineH2: + handleBaseline(false); + return; + case Route::Json: + handleJson(); + return; + case Route::Upload: + if (method_ != HTTPMethod::POST) { + sendText(405, "Method Not Allowed", "method not allowed"); + } else if (!uploadValid_) { + sendText(400, "Bad Request", "upload too large"); + } else { + std::array digits; + const auto end = std::to_chars( + digits.data(), digits.data() + digits.size(), uploadBytes_); + sendText(200, "OK", + std::string_view( + digits.data(), + static_cast(end.ptr - digits.data()))); + } + return; + case Route::Static: + handleStatic(); + return; + } + } + + void requestComplete() noexcept override { delete this; } + + void onError(ProxygenError /*error*/) noexcept override { delete this; } + +private: + enum class Route { + NotFound, + Baseline, + BaselineH2, + Json, + Upload, + Static, + Pipeline, + WebSocket + }; + + void handleBaseline(bool allowPost) { + if (!queryValid_ || !bodyValid_) { + sendText(400, "Bad Request", "invalid integer"); + return; + } + if (method_ != HTTPMethod::GET && + (!allowPost || method_ != HTTPMethod::POST)) { + sendText(405, "Method Not Allowed", "method not allowed"); + return; + } + + int64_t sum = 0; + if (!checkedAdd(a_, b_, sum)) { + sendText(400, "Bad Request", "integer overflow"); + return; + } + if (method_ == HTTPMethod::POST) { + int64_t bodyValue = 0; + if (!parseInteger(requestBody_, bodyValue) || + !checkedAdd(sum, bodyValue, sum)) { + sendText(400, "Bad Request", "invalid integer"); + return; + } + } + std::array digits; + const auto end = std::to_chars(digits.data(), digits.data() + digits.size(), + sum); + sendText(200, "OK", + std::string_view(digits.data(), + static_cast(end.ptr - digits.data()))); + } + + void handleJson() { + if (method_ != HTTPMethod::GET) { + sendText(405, "Method Not Allowed", "method not allowed"); + return; + } + if (!jsonValid_ || jsonCount_ > dataset_.size()) { + sendText(400, "Bad Request", "invalid JSON parameters"); + return; + } + + try { + folly::dynamic items = folly::dynamic::array; + items.reserve(jsonCount_); + for (size_t index = 0; index < jsonCount_; ++index) { + folly::dynamic item = dataset_[index]; + int64_t subtotal = 0; + int64_t total = 0; + if (!checkedMultiply(item["price"].asInt(), item["quantity"].asInt(), + subtotal) || + !checkedMultiply(subtotal, multiplier_, total)) { + sendText(400, "Bad Request", "integer overflow"); + return; + } + item["total"] = total; + items.push_back(std::move(item)); + } + folly::dynamic response = folly::dynamic::object; + response["items"] = std::move(items); + response["count"] = static_cast(jsonCount_); + // fromString takes ownership of the serialized buffer; copyBuffer would + // memcpy up to ~30 KB per response and shows up directly in json-comp. + sendResponse(200, "OK", "application/json", + folly::IOBuf::fromString(folly::toJson(response))); + } catch (const std::exception &) { + sendText(500, "Internal Server Error", "JSON serialization failed"); + } + } + + void handleStatic() { + if (method_ != HTTPMethod::GET) { + sendText(405, "Method Not Allowed", "method not allowed"); + return; + } + if (staticName_.empty() || staticName_.find('/') != std::string::npos || + staticName_.find('\\') != std::string::npos || + staticName_.find("..") != std::string::npos) { + sendText(404, "Not Found", "not found"); + return; + } + auto asset = serveStatic(staticName_, acceptEncoding_); + if (!asset.body) { + sendText(404, "Not Found", "not found"); + return; + } + + responseFinished_ = true; + ResponseBuilder builder(downstream_); + builder.status(200, "OK") + .header(proxygen::HTTP_HEADER_CONTENT_TYPE, asset.contentType); + if (asset.encoding != ContentEncoding::Identity) { + builder.header(proxygen::HTTP_HEADER_CONTENT_ENCODING, + httparena::encodingToken(asset.encoding)); + // The same URL yields different bytes per Accept-Encoding. + builder.header(proxygen::HTTP_HEADER_VARY, "Accept-Encoding"); + } + builder.body(std::move(asset.body)).sendWithEOM(); + } + + void sendResponse(uint16_t status, const char *reason, std::string_view type, + std::unique_ptr body) { + responseFinished_ = true; + ResponseBuilder(downstream_) + .status(status, reason) + .header(proxygen::HTTP_HEADER_CONTENT_TYPE, type) + .body(std::move(body)) + .sendWithEOM(); + } + + void sendResponse(uint16_t status, const char *reason, std::string_view type, + std::string_view body) { + sendResponse(status, reason, type, folly::IOBuf::copyBuffer(body)); + } + + void sendText(uint16_t status, const char *reason, std::string_view body) { + sendResponse(status, reason, "text/plain", body); + } + + // Drains whatever the shared RFC 6455 codec produced for this read as a + // single egress write, then closes the stream if the codec is done. + void flushWebSocket() { + if (auto egress = websocket_.takeEgress()) { + downstream_->sendBody(std::move(egress)); + } + if (websocket_.finished() && !responseFinished_) { + responseFinished_ = true; + downstream_->sendEOM(); + } + } + + Route route_{Route::NotFound}; + const folly::dynamic &dataset_; + HTTPMethod method_{HTTPMethod::GET}; + int64_t a_{0}; + int64_t b_{0}; + int64_t multiplier_{1}; + size_t jsonCount_{0}; + size_t uploadBytes_{0}; + bool queryValid_{false}; + bool jsonValid_{false}; + bool uploadValid_{true}; + bool bodyValid_{true}; + bool websocketAccepted_{false}; + bool websocketActive_{false}; + bool responseFinished_{false}; + std::string requestBody_; + std::string staticName_; + std::string acceptEncoding_; + httparena::WebSocketEcho websocket_; +}; + +class ArenaHandlerFactory final : public RequestHandlerFactory { +public: + explicit ArenaHandlerFactory(std::shared_ptr dataset) + : dataset_(std::move(dataset)) {} + + void onServerStart(folly::EventBase * /*eventBase*/) noexcept override {} + + void onServerStop() noexcept override {} + + RequestHandler *onRequest(RequestHandler *, HTTPMessage *) noexcept override { + return new ArenaHandler(*dataset_); + } + +private: + std::shared_ptr dataset_; +}; + +wangle::SSLContextConfig h1TlsConfig() { + wangle::SSLContextConfig config; + config.isDefault = true; + config.clientVerification = + folly::SSLContext::VerifyClientCertificate::DO_NOT_REQUEST; + config.setCertificate(FLAGS_cert, FLAGS_key, ""); + config.setNextProtocols(std::list{"http/1.1"}); + return config; +} + +wangle::SSLContextConfig h2TlsConfig() { + wangle::SSLContextConfig config; + config.isDefault = true; + config.clientVerification = + folly::SSLContext::VerifyClientCertificate::DO_NOT_REQUEST; + config.setCertificate(FLAGS_cert, FLAGS_key, ""); + config.setNextProtocols(std::list{"h2"}); + return config; +} + +std::vector listenerConfigs() { + std::vector listeners; + listeners.emplace_back(SocketAddress(FLAGS_ip, FLAGS_http_port, true), + HTTPServer::Protocol::HTTP); + listeners.emplace_back(SocketAddress(FLAGS_ip, FLAGS_h2c_port, true), + HTTPServer::Protocol::HTTP2); + + HTTPServer::IPConfig tlsListener( + SocketAddress(FLAGS_ip, FLAGS_tls_port, true), + HTTPServer::Protocol::HTTP); + tlsListener.sslConfigs.push_back(h1TlsConfig()); + listeners.push_back(std::move(tlsListener)); + + HTTPServer::IPConfig h2Listener(SocketAddress(FLAGS_ip, FLAGS_h2_port, true), + HTTPServer::Protocol::HTTP2); + h2Listener.sslConfigs.push_back(h2TlsConfig()); + listeners.push_back(std::move(h2Listener)); + return listeners; +} + +} // namespace + +int main(int argc, char *argv[]) { + const folly::Init init(&argc, &argv, true); + + if (FLAGS_threads <= 0) { + FLAGS_threads = static_cast(folly::available_concurrency()); + } + if (FLAGS_h3_threads <= 0) { + FLAGS_h3_threads = static_cast(folly::available_concurrency()); + } + CHECK_GT(FLAGS_threads, 0); + CHECK_GT(FLAGS_h3_threads, 0); + + try { + auto dataset = loadDataset(); + + proxygen::HTTPServerOptions options; + options.threads = static_cast(FLAGS_threads); + options.idleTimeout = std::chrono::milliseconds(60000); + options.shutdownOn = {SIGINT, SIGTERM}; + // Nothing here speaks CONNECT; saying we do only stops HTTPServer from + // prepending RejectConnectFilterFactory to the per-request filter chain. + options.supportsConnect = true; + options.enableContentCompression = true; + // Only `json-comp` asks for a compressed response, and it is the one + // profile scored on compression ratio rather than raw rps. `static` sends + // `Accept-Encoding: br;q=1, gzip;q=0.8` too, and proxygen's default + // compressible set covers text/css, text/html and application/javascript — + // so every CSS/JS/HTML request was gzipping 8-200 KB on the event base for + // no scoring benefit (compression is explicitly optional for `static`). + // Restrict the set to the content type json-comp actually measures. + options.contentCompressionTypes = {"application/json"}; + options.initialReceiveWindow = 1U << 20; + options.receiveStreamWindowSize = 1U << 20; + options.receiveSessionWindowSize = 10U << 20; + options.maxConcurrentIncomingStreams = 1024; + options.handlerFactories = + RequestHandlerChain().addThen(dataset).build(); + + httparena::ArenaHQServer h3Server( + FLAGS_cert, FLAGS_key, static_cast(FLAGS_h3_threads), + [dataset](HTTPMessage *) -> proxygen::HTTPTransactionHandler * { + return new proxygen::RequestHandlerAdaptor( + new ArenaHandler(*dataset)); + }); + HTTPServer server(std::move(options)); + server.bind(listenerConfigs()); + h3Server.start(SocketAddress(FLAGS_ip, FLAGS_h3_port, true)); + server.start(); + h3Server.stop(); + } catch (const std::exception &error) { + std::cerr << "failed to start Proxygen HttpArena server: " << error.what() + << '\n'; + return 1; + } + return 0; +} diff --git a/frameworks/proxygen/src/CoroServer.cpp b/frameworks/proxygen/src/CoroServer.cpp new file mode 100644 index 000000000..e475f56e3 --- /dev/null +++ b/frameworks/proxygen/src/CoroServer.cpp @@ -0,0 +1,624 @@ +#include "ArenaCommon.h" +#include "ArenaWebSocket.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +DEFINE_int32(http_port, 8080, "HTTP/1.1 and WebSocket port"); +DEFINE_int32(tls_port, 8081, "HTTP/1.1 over TLS port"); +DEFINE_int32(h2c_port, 8082, "HTTP/2 cleartext port"); +DEFINE_int32(h2_port, 8443, "HTTP/2 over TLS port"); +DEFINE_int32(h3_port, 8443, "HTTP/3 over QUIC port"); +DEFINE_string(ip, "::", "Address on which to listen"); +DEFINE_string(cert, "/certs/server.crt", "TLS certificate path"); +DEFINE_string(key, "/certs/server.key", "TLS private-key path"); +DEFINE_int32(threads, 0, "TCP I/O threads; 0 uses the available CPU count"); +DEFINE_int32(h3_threads, 0, "QUIC I/O threads; 0 uses the available CPU count"); + +namespace { + +using proxygen::HTTPMessage; +using proxygen::HTTPMethod; +using proxygen::coro::HTTPBodyEvent; +using proxygen::coro::HTTPError; +using proxygen::coro::HTTPErrorCode; +using proxygen::coro::HTTPFixedSource; +using proxygen::coro::HTTPHandler; +using proxygen::coro::HTTPHeaderEvent; +using proxygen::coro::HTTPServer; +using proxygen::coro::HTTPSessionContextPtr; +using proxygen::coro::HTTPSource; +using proxygen::coro::HTTPSourceHolder; +using proxygen::coro::TimedBaton; + +using httparena::checkedAdd; +using httparena::checkedMultiply; +using httparena::contentType; +using httparena::kJsonPrefix; +using httparena::kMaxRequestBody; +using httparena::kStaticPrefix; +using httparena::loadDataset; +using httparena::parseInteger; +using httparena::queryValue; +using httparena::ContentEncoding; +using httparena::serveStatic; +using httparena::validWebSocketKey; + +HTTPFixedSource *makeResponse(uint16_t status, + std::string_view contentTypeValue, + std::unique_ptr body) { + auto *response = HTTPFixedSource::makeFixedResponse(status, std::move(body)); + response->msg_->getHeaders().set(proxygen::HTTP_HEADER_CONTENT_TYPE, + contentTypeValue); + return response; +} + +HTTPFixedSource *makeResponse(uint16_t status, + std::string_view contentTypeValue, + std::string_view body) { + return makeResponse(status, contentTypeValue, folly::IOBuf::copyBuffer(body)); +} + +HTTPFixedSource *makeTextResponse(uint16_t status, std::string_view body) { + return makeResponse(status, "text/plain", body); +} + +// Renders an integer response body without the std::to_string allocation. +HTTPFixedSource *makeIntResponse(uint16_t status, int64_t value) { + std::array digits; + const auto end = + std::to_chars(digits.data(), digits.data() + digits.size(), value); + return makeTextResponse( + status, std::string_view(digits.data(), + static_cast(end.ptr - digits.data()))); +} + +folly::coro::Task +readBodyEventNoSuspend(HTTPSourceHolder &source, + uint32_t max = std::numeric_limits::max()) { + while (true) { + auto event = co_await co_awaitTry(source.readBodyEvent(max)); + if (event.hasException()) { + co_yield folly::coro::co_error(std::move(event.exception())); + } + if (event->eventType == HTTPBodyEvent::SUSPEND) { + const auto status = co_await std::move(event->event.resume); + if (status == TimedBaton::Status::cancelled) { + co_yield folly::coro::co_error( + HTTPError(HTTPErrorCode::CORO_CANCELLED, "request read cancelled")); + } + continue; + } + co_return std::move(*event); + } +} + +struct RequestBody { + bool readOk{true}; + bool captureOk{true}; + size_t size{0}; + std::string captured; +}; + +folly::coro::Task readRequestBody(HTTPSourceHolder &source, + bool eom, size_t captureLimit) { + RequestBody result; + while (!eom) { + auto eventTry = co_await co_awaitTry(readBodyEventNoSuspend(source)); + if (eventTry.hasException()) { + result.readOk = false; + co_return result; + } + auto event = std::move(*eventTry); + eom = event.eom; + if (event.eventType == HTTPBodyEvent::PADDING) { + continue; + } + if (event.eventType != HTTPBodyEvent::BODY) { + result.readOk = false; + co_return result; + } + + const size_t bytes = event.event.body.chainLength(); + if (bytes > std::numeric_limits::max() - result.size) { + result.readOk = false; + co_return result; + } + result.size += bytes; + if (captureLimit == 0 || !result.captureOk) { + continue; + } + if (result.captured.size() > captureLimit || + bytes > captureLimit - result.captured.size()) { + result.captureOk = false; + continue; + } + auto body = event.event.body.move(); + if (body) { + const auto range = body->coalesce(); + result.captured.append(reinterpret_cast(range.data()), + range.size()); + } + } + co_return result; +} + +// Long-lived response source for an upgraded WebSocket connection: raw frames +// arrive as BODY events on the request source, and the echoed frames leave as +// BODY events on this one. Framing is the shared codec in ArenaWebSocket.h, so +// a read carrying N pipelined frames produces one egress event. +class WebSocketSource final : public HTTPSource { +public: + explicit WebSocketSource(HTTPSourceHolder request) + : request_(std::move(request)) { + setHeapAllocated(); + } + + folly::coro::Task readHeaderEvent() override { + auto response = std::make_unique(); + response->setHTTPVersion(1, 1); + response->setStatusCode(200); + response->setStatusMessage("OK"); + response->setEgressWebsocketUpgrade(); + HTTPHeaderEvent event(std::move(response), false); + auto guard = folly::makeGuard(lifetime(event)); + co_return event; + } + + folly::coro::Task + readBodyEvent(uint32_t max = std::numeric_limits::max()) override { + while (!websocket_.hasEgress() && !eom_) { + auto inputTry = co_await co_awaitTry(readBodyEventNoSuspend(request_)); + if (inputTry.hasException()) { + auto error = proxygen::coro::getHTTPError(inputTry); + auto guard = folly::makeGuard([this] { + if (heapAllocated_) { + delete this; + } + }); + co_yield folly::coro::co_error(std::move(error)); + } + + auto input = std::move(*inputTry); + if (input.eventType == HTTPBodyEvent::BODY) { + websocket_.onIngress(input.event.body.move()); + } + if (input.eom || websocket_.finished()) { + eom_ = true; + } + } + + auto body = websocket_.takeEgress(std::max(1, max)); + // Only signal EOM once everything the codec produced has been handed over. + const bool eom = eom_ && !websocket_.hasEgress(); + HTTPBodyEvent event(std::move(body), eom); + auto guard = folly::makeGuard(lifetime(event)); + co_return event; + } + + void stopReading(folly::Optional error = + folly::none) noexcept override { + if (request_) { + request_.stopReading(error); + } + if (heapAllocated_) { + delete this; + } + } + +private: + HTTPSourceHolder request_; + httparena::WebSocketEcho websocket_; + bool eom_{false}; +}; + + +// Only `json-comp` asks for a compressed response, and it is the one profile +// scored on compression ratio rather than raw rps. `static` sends +// `Accept-Encoding: br;q=1, gzip;q=0.8` too, so listing the static content +// types here would mean gzipping 8-200 KB per CSS/JS/HTML request on the event +// base for no scoring benefit (compression is explicitly optional for +// `static`). +proxygen::CompressionFilterUtils::FactoryOptions compressionOptions() { + proxygen::CompressionFilterUtils::FactoryOptions options; + options.compressibleContentTypes = + std::make_shared>( + std::set{"application/json"}); + // FactoryOptions defaults to 4; HTTPServerOptions (and so the classic entry) + // defaults to Z_DEFAULT_COMPRESSION. json-comp is scored on + // (minBpr/myBpr)^2, so the weaker ratio cost the coro entry ~17% of that + // profile's score for no throughput gain worth having. + options.zlibCompressionLevel = -1; + return options; +} + +class ArenaCoroHandler final : public HTTPHandler { +public: + explicit ArenaCoroHandler(std::shared_ptr dataset) + : dataset_(std::move(dataset)), items_(*dataset_) {} + + folly::coro::Task + handleRequest(folly::EventBase * /*eventBase*/, + HTTPSessionContextPtr /*session*/, + HTTPSourceHolder requestSource) override { + auto headerTry = co_await co_awaitTry(requestSource.readHeaderEvent()); + if (headerTry.hasException()) { + co_return makeTextResponse(400, "bad request"); + } + auto header = std::move(*headerTry); + auto request = std::move(header.headers); + const bool requestEom = header.eom; + const auto method = request->getMethod().value_or(HTTPMethod::GET); + const auto pathPiece = request->getPathAsStringPiece(); + const std::string_view path(pathPiece.data(), pathPiece.size()); + const auto queryPiece = request->getQueryStringAsStringPiece(); + const std::string_view query(queryPiece.data(), queryPiece.size()); + + if (path == "/ws") { + const auto &headers = request->getHeaders(); + const auto &key = headers.getSingleOrEmpty("Sec-WebSocket-Key"); + const auto &version = headers.getSingleOrEmpty("Sec-WebSocket-Version"); + if (method != HTTPMethod::GET || !request->isIngressWebsocketUpgrade() || + !validWebSocketKey(std::string_view(key.data(), key.size())) || + version != "13") { + auto *response = makeTextResponse(426, "WebSocket upgrade required"); + response->msg_->getHeaders().set("Sec-WebSocket-Version", "13"); + co_return response; + } + co_return new WebSocketSource(std::move(requestSource)); + } + + if (path == "/baseline11" || path == "/baseline2") { + const bool allowPost = path == "/baseline11"; + if (method != HTTPMethod::GET && + (!allowPost || method != HTTPMethod::POST)) { + co_return makeTextResponse(405, "method not allowed"); + } + int64_t a = 0; + int64_t b = 0; + const auto aText = queryValue(query, "a"); + const auto bText = queryValue(query, "b"); + if (!aText || !bText || !parseInteger(*aText, a) || + !parseInteger(*bText, b)) { + co_return makeTextResponse(400, "invalid integer"); + } + int64_t sum = 0; + if (!checkedAdd(a, b, sum)) { + co_return makeTextResponse(400, "integer overflow"); + } + if (method == HTTPMethod::POST) { + auto body = co_await readRequestBody(requestSource, requestEom, + kMaxRequestBody); + int64_t bodyValue = 0; + if (!body.readOk || !body.captureOk || + !parseInteger(body.captured, bodyValue) || + !checkedAdd(sum, bodyValue, sum)) { + co_return makeTextResponse(400, "invalid integer"); + } + } + co_return makeIntResponse(200, sum); + } + + if (path == "/pipeline") { + if (method != HTTPMethod::GET) { + co_return makeTextResponse(405, "method not allowed"); + } + co_return makeTextResponse(200, "ok"); + } + + if (path.starts_with(kJsonPrefix)) { + if (method != HTTPMethod::GET) { + co_return makeTextResponse(405, "method not allowed"); + } + const std::string_view countText(path.data() + kJsonPrefix.size(), + path.size() - kJsonPrefix.size()); + int64_t count = 0; + int64_t multiplier = 1; + const auto multiplierText = queryValue(query, "m"); + if (!parseInteger(countText, count) || count < 1 || count > 50 || + (multiplierText && !parseInteger(*multiplierText, multiplier))) { + co_return makeTextResponse(400, "invalid JSON parameters"); + } + if (static_cast(count) > items_.size()) { + co_return makeTextResponse(400, "invalid JSON parameters"); + } + + try { + folly::dynamic items = folly::dynamic::array; + items.reserve(static_cast(count)); + for (int64_t index = 0; index < count; ++index) { + folly::dynamic item = items_[static_cast(index)]; + int64_t subtotal = 0; + int64_t total = 0; + if (!checkedMultiply(item["price"].asInt(), item["quantity"].asInt(), + subtotal) || + !checkedMultiply(subtotal, multiplier, total)) { + co_return makeTextResponse(400, "integer overflow"); + } + item["total"] = total; + items.push_back(std::move(item)); + } + folly::dynamic response = folly::dynamic::object; + response["items"] = std::move(items); + response["count"] = count; + co_return maybeCompress( + makeResponse(200, "application/json", + folly::IOBuf::fromString(folly::toJson(response))), + *request); + } catch (const std::exception &) { + co_return makeTextResponse(500, "JSON serialization failed"); + } + } + + if (path == "/upload") { + if (method != HTTPMethod::POST) { + co_return makeTextResponse(405, "method not allowed"); + } + auto body = co_await readRequestBody(requestSource, requestEom, 0); + if (!body.readOk) { + co_return makeTextResponse(400, "upload failed"); + } + co_return makeIntResponse(200, static_cast(body.size)); + } + + if (path.starts_with(kStaticPrefix)) { + if (method != HTTPMethod::GET) { + co_return makeTextResponse(405, "method not allowed"); + } + const std::string_view name = path.substr(kStaticPrefix.size()); + if (name.empty() || name.find('/') != std::string_view::npos || + name.find('\\') != std::string_view::npos || + name.find("..") != std::string_view::npos) { + co_return makeTextResponse(404, "not found"); + } + const auto &accept = request->getHeaders().getSingleOrEmpty( + proxygen::HTTP_HEADER_ACCEPT_ENCODING); + auto asset = + serveStatic(name, std::string_view(accept.data(), accept.size())); + if (!asset.body) { + co_return makeTextResponse(404, "not found"); + } + auto *response = + makeResponse(200, asset.contentType, std::move(asset.body)); + if (asset.encoding != ContentEncoding::Identity) { + auto &headers = response->msg_->getHeaders(); + headers.set(proxygen::HTTP_HEADER_CONTENT_ENCODING, + httparena::encodingToken(asset.encoding)); + headers.set(proxygen::HTTP_HEADER_VARY, "Accept-Encoding"); + } + co_return response; + } + + co_return makeTextResponse(404, "not found"); + } + +private: + // Wraps `response` in proxygen's coro CompressionFilter when this client + // actually negotiated an encoding we support. + // + // The alternative is registering ServerCompressionFilterFactory on + // HTTPServer::Config, but HTTPFilterFactoryHandler calls makeFilters() + // before the request headers have been read, so the factory cannot be + // conditional: every request — `baseline` included, where nothing is + // compressible — pays a SharedCtx, a VisitorFilter with a capturing lambda, + // a CompressionFilter and the surrounding coroutine frame. That measured at + // ~10% of baseline throughput. Attaching the filter here instead keeps the + // cost on the responses that are actually compressed. (The classic + // HTTPServer path has no such problem: CompressionFilterFactory::onRequest + // sees the request and returns the handler unwrapped when there is nothing + // to do.) + HTTPSourceHolder maybeCompress(HTTPFixedSource *response, + const HTTPMessage &request) { + auto params = std::make_shared< + folly::Optional>( + proxygen::CompressionFilterUtils::getFilterParams(request, + compression_)); + if (!params->hasValue()) { + return response; + } + auto *filter = + new proxygen::coro::CompressionFilter(response, std::move(params)); + filter->setHeapAllocated(); + return filter; + } + + std::shared_ptr dataset_; + // The handler is shared by every session and outlives all of them, so the + // hot path can dereference the dataset once here instead of per request. + const folly::dynamic &items_; + const proxygen::CompressionFilterUtils::FactoryOptions compression_{ + compressionOptions()}; +}; + +constexpr uint32_t kH2StreamWindow = 1U << 20; +constexpr size_t kH2ConnectionWindow = 10U << 20; +constexpr uint32_t kMaxConcurrentStreams = 1024; + +HTTPServer::SessionConfig makeSessionConfig() { + HTTPServer::SessionConfig session; + session.settings = { + {proxygen::SettingsId::MAX_HEADER_LIST_SIZE, 32 * 1024}, + {proxygen::SettingsId::HEADER_TABLE_SIZE, 4096}, + {proxygen::SettingsId::MAX_FRAME_SIZE, 16384}, + {proxygen::SettingsId::MAX_CONCURRENT_STREAMS, kMaxConcurrentStreams}, + {proxygen::SettingsId::INITIAL_WINDOW_SIZE, kH2StreamWindow}}; + session.streamFlowControl = kH2StreamWindow; + session.connFlowControl = kH2ConnectionWindow; + session.streamReadTimeout = std::chrono::seconds(60); + session.connIdleTimeout = std::chrono::seconds(60); + return session; +} + + +wangle::SSLContextConfig tlsConfig(std::list protocols) { + auto config = HTTPServer::getDefaultTLSConfig(); + config.setCertificate(FLAGS_cert, FLAGS_key, ""); + config.setNextProtocols(protocols); + return config; +} + +std::shared_ptr +acceptorConfig(const HTTPServer::Config &serverConfig, + std::string plaintextProtocol, + std::list tlsProtocols = {}) { + auto config = std::make_shared(); + *static_cast(config.get()) = + serverConfig.socketConfig; + config->sslContextConfigs.clear(); + if (!tlsProtocols.empty()) { + config->sslContextConfigs.push_back(tlsConfig(std::move(tlsProtocols))); + } + config->egressSettings = serverConfig.sessionConfig.settings; + config->transactionIdleTimeout = serverConfig.sessionConfig.streamReadTimeout; + config->initialReceiveWindow = kH2StreamWindow; + config->receiveStreamWindowSize = kH2StreamWindow; + config->receiveSessionWindowSize = kH2ConnectionWindow; + config->maxConcurrentIncomingStreams = kMaxConcurrentStreams; + config->plaintextProtocol = std::move(plaintextProtocol); + config->forceHTTP1_0_to_1_1 = true; + config->connectionIdleTimeout = serverConfig.sessionConfig.connIdleTimeout; + config->readBufNewAllocSize = serverConfig.sessionConfig.readBufNewAllocSize; + return config; +} + +HTTPServer::SocketAcceptorConfigFactoryFn tcpSocketFactory() { + return [](folly::EventBase &eventBase, + const HTTPServer::Config &serverConfig) { + std::vector listeners; + const auto addListener = [&](uint16_t port, std::string protocol, + std::list tlsProtocols = {}) { + folly::AsyncServerSocket::UniquePtr socket( + new folly::AsyncServerSocket(&eventBase)); + socket->bind(folly::SocketAddress(FLAGS_ip, port, true)); + socket->listen(serverConfig.socketConfig.acceptBacklog); + listeners.emplace_back(std::move(socket), + acceptorConfig(serverConfig, std::move(protocol), + std::move(tlsProtocols))); + }; + + addListener(static_cast(FLAGS_http_port), "http/1.1"); + addListener(static_cast(FLAGS_tls_port), "http/1.1", + {"http/1.1"}); + addListener(static_cast(FLAGS_h2c_port), "h2c"); + addListener(static_cast(FLAGS_h2_port), "", {"h2"}); + return listeners; + }; +} + +HTTPServer::Config tcpConfig(size_t threads) { + HTTPServer::Config config; + config.numIOThreads = threads; + config.shutdownOnSignals = {SIGINT, SIGTERM}; + config.sessionConfig = makeSessionConfig(); + return config; +} + +HTTPServer::Config quicConfig(size_t threads) { + HTTPServer::Config config; + config.socketConfig.bindAddress = + folly::SocketAddress(FLAGS_ip, FLAGS_h3_port, true); + config.socketConfig.sslContextConfigs.push_back(tlsConfig({"h3"})); + config.numIOThreads = threads; + config.shutdownOnSignals = {}; + config.sessionConfig = makeSessionConfig(); + config.quicConfig = HTTPServer::QuicConfig(); + config.quicConfig->supportedAlpns = {"h3"}; + auto &transport = config.quicConfig->transportSettings; + transport.maxNumPTOs = 1000; + transport.maxCwndInMss = quic::kLargeMaxCwndInMss; + transport.batchingMode = quic::QuicBatchingMode::BATCHING_MODE_GSO; + transport.maxBatchSize = 48; + transport.dataPathType = quic::DataPathType::ContinuousMemory; + transport.writeConnectionDataPacketsLimit = 48; + return config; +} + +} // namespace + +int main(int argc, char *argv[]) { + const folly::Init init(&argc, &argv, true); + + try { + auto dataset = loadDataset(); + const size_t threads = + FLAGS_threads <= 0 ? static_cast(folly::available_concurrency()) + : static_cast(FLAGS_threads); + const size_t h3Threads = + FLAGS_h3_threads <= 0 + ? static_cast(folly::available_concurrency()) + : static_cast(FLAGS_h3_threads); + auto handler = std::make_shared(std::move(dataset)); + + HTTPServer tcpServer(tcpConfig(threads), handler, tcpSocketFactory()); + HTTPServer h3Server(quicConfig(h3Threads), std::move(handler)); + + std::promise h3Ready; + auto h3ReadyFuture = h3Ready.get_future(); + std::thread h3Thread([&] { + try { + h3Server.start([&] { h3Ready.set_value(); }); + } catch (...) { + try { + h3Ready.set_exception(std::current_exception()); + } catch (const std::future_error &) { + } + } + }); + + try { + h3ReadyFuture.get(); + } catch (...) { + h3Thread.join(); + throw; + } + + try { + tcpServer.start(); + } catch (...) { + h3Server.forceStop(); + h3Thread.join(); + throw; + } + + h3Server.forceStop(); + h3Thread.join(); + } catch (const std::exception &error) { + std::cerr << "failed to start Proxygen coroutine HttpArena server: " + << error.what() << '\n'; + return 1; + } + return 0; +} diff --git a/site/data/frameworks.json b/site/data/frameworks.json index 32254a029..f0e2a330f 100644 --- a/site/data/frameworks.json +++ b/site/data/frameworks.json @@ -1272,6 +1272,20 @@ "response": true } }, + "proxygen-coro": { + "dir": "proxygen-coro", + "description": "Meta's Proxygen native coroutine HTTPServer and HTTPSource APIs across HTTP/1.1, HTTP/1.1 TLS, h2c, HTTP/2 TLS, HTTP/3 QUIC, and RFC 6455 WebSockets.", + "repo": "https://github.com/facebook/proxygen", + "type": "engine", + "engine": "proxygen" + }, + "proxygen": { + "dir": "proxygen", + "description": "Meta's Proxygen HTTP engine: HTTPServer for HTTP/1.1, HTTP/1.1 TLS, h2c, HTTP/2 TLS, and RFC 6455 WebSockets, plus the mvfst-backed HQ server for HTTP/3 over QUIC.", + "repo": "https://github.com/facebook/proxygen", + "type": "engine", + "engine": "proxygen" + }, "pyronova": { "dir": "pyronova", "description": "Pyronova \u2014 Python web framework with a Rust core (hyper + tokio + rustls + mimalloc) and PEP 684 sub-interpreter workers for true multi-core parallelism. Opt-in features: gzip/brotli compression, rustls TLS with h2/h1 ALPN, streaming body ingest, async Postgres via sqlx::PgPool. Handlers are standard Python functions routed via decorators.",