From fc0867773e9ad71f35f45f0067d44be4da3784d7 Mon Sep 17 00:00:00 2001 From: jadit19 Date: Sun, 19 Jul 2026 16:45:03 +0530 Subject: [PATCH] Add benchmarking suite and update .gitignore - Introduced a new benchmarking suite for comparing the Expresso framework against other HTTP frameworks. - Added README.md for benchmark suite documentation. - Implemented scripts for running benchmarks and comparing results. - Updated .gitignore to exclude benchmark results and cache files. - Added shared fixtures and server implementations for Node.js and Express.js mirrors. --- .gitignore | 7 + benchmark/README.md | 116 +++++ benchmark/scripts/compare.py | 134 ++++++ benchmark/scripts/frameworks.json | 88 ++++ benchmark/scripts/run_benchmarks.py | 471 ++++++++++++++++++++ benchmark/servers/express/package.json | 14 + benchmark/servers/express/server.js | 80 ++++ benchmark/servers/node-http/package.json | 10 + benchmark/servers/node-http/server.js | 186 ++++++++ benchmark/servers/shared/about.js | 57 +++ benchmark/servers/shared/headers.js | 16 + benchmark/servers/shared/mime.js | 19 + benchmark/servers/shared/package.json | 7 + benchmark/servers/shared/zip.js | 103 +++++ include/expresso/core/io/connection.h | 69 +++ include/expresso/core/io/poller.h | 47 ++ include/expresso/core/io/socket.h | 15 + include/expresso/core/server.h | 55 ++- include/expresso/enums/network_event_type.h | 14 + include/expresso/helpers/request.h | 13 + include/expresso/helpers/response.h | 8 +- include/expresso/messages/file_region.h | 21 + include/expresso/messages/response.h | 23 +- src/core/io/connection.cpp | 167 +++++++ src/core/io/poller_epoll.cpp | 70 +++ src/core/io/poller_kqueue.cpp | 87 ++++ src/core/io/socket.cpp | 15 + src/core/server.cpp | 331 ++++++++++---- src/helpers/request.cpp | 29 ++ src/helpers/response.cpp | 50 +-- src/messages/response.cpp | 116 ++--- src/middleware/static_serve.cpp | 2 + 32 files changed, 2248 insertions(+), 192 deletions(-) create mode 100644 benchmark/README.md create mode 100644 benchmark/scripts/compare.py create mode 100644 benchmark/scripts/frameworks.json create mode 100644 benchmark/scripts/run_benchmarks.py create mode 100644 benchmark/servers/express/package.json create mode 100644 benchmark/servers/express/server.js create mode 100644 benchmark/servers/node-http/package.json create mode 100644 benchmark/servers/node-http/server.js create mode 100644 benchmark/servers/shared/about.js create mode 100644 benchmark/servers/shared/headers.js create mode 100644 benchmark/servers/shared/mime.js create mode 100644 benchmark/servers/shared/package.json create mode 100644 benchmark/servers/shared/zip.js create mode 100644 include/expresso/core/io/connection.h create mode 100644 include/expresso/core/io/poller.h create mode 100644 include/expresso/core/io/socket.h create mode 100644 include/expresso/enums/network_event_type.h create mode 100644 include/expresso/helpers/request.h create mode 100644 include/expresso/messages/file_region.h create mode 100644 src/core/io/connection.cpp create mode 100644 src/core/io/poller_epoll.cpp create mode 100644 src/core/io/poller_kqueue.cpp create mode 100644 src/core/io/socket.cpp create mode 100644 src/helpers/request.cpp diff --git a/.gitignore b/.gitignore index 8b69501..33243bd 100644 --- a/.gitignore +++ b/.gitignore @@ -36,6 +36,13 @@ .idea build cmake-build-debug +.cache + +# Benchmarking +node_modules +benchmark/results +__pycache__ +**/package-lock.json # Secrets *.env \ No newline at end of file diff --git a/benchmark/README.md b/benchmark/README.md new file mode 100644 index 0000000..f28d08f --- /dev/null +++ b/benchmark/README.md @@ -0,0 +1,116 @@ +# Expresso benchmark suite + +Compares expresso against other HTTP frameworks and runtimes using servers +that all mirror `example/main.cpp` — same routes, same payloads, same +headers — so the measurements compare frameworks, not fixtures. + +## Layout + +``` +benchmark/ +├── servers/ +│ ├── shared/ # framework-agnostic fixtures (about payload, zip +│ │ # builder, mime map, parity headers) +│ ├── node-http/ # dependency-free node:http mirror (runs on node/bun/deno) +│ └── express/ # express.js mirror (pinned deps, npm ci) +├── scripts/ +│ ├── frameworks.json # registry of frameworks, scenarios, load settings +│ ├── run_benchmarks.py # orchestrator (python3 stdlib only) +│ └── compare.py # renders markdown reports from result JSON +└── results/ # timestamped output, gitignored +``` + +## Prerequisites + +- [wrk](https://github.com/wg/wrk) (`brew install wrk` / `apt install wrk`) +- python3 (stdlib only, no pip packages) +- node ≥ 20 for the JavaScript servers; bun and deno are optional and are + skipped automatically when not installed +- the expresso example built: `cmake -B build && cmake --build build` + +## Running + +```sh +cd benchmark/scripts +python3 run_benchmarks.py # full run, all available frameworks +python3 run_benchmarks.py --quick # 2s smoke run +python3 run_benchmarks.py --frameworks expresso,express --scenarios health +python3 run_benchmarks.py --mode keepalive # see caveats below +``` + +Each run writes `benchmark/results/_/` containing +`results.json` (raw trials + environment metadata), `comparison.md`, and a +log per server. Re-render or inspect old runs with +`python3 compare.py ../results//results.json`. + +## Methodology and reproducibility + +- **Connection-close mode by default.** Expresso does not implement + keep-alive yet, so every response carries `connection: close`. All + frameworks are therefore driven with `Connection: close` so everyone pays + the same per-request TCP setup cost. `--mode keepalive` exists for + measuring the JavaScript servers against each other, but expresso numbers + in that mode are still close-mode numbers — don't compare across. +- **TIME_WAIT draining.** Close-mode runs burn one ephemeral port per + request; on macOS especially, back-to-back runs exhaust ports and produce + garbage numbers (connect errors, 100x throughput collapse). The harness + waits for TIME_WAIT sockets to drop below a threshold between trials. +- **The macOS close-mode wall (why trials are short on macOS).** Whichever + TCP peer closes first holds the TIME_WAIT. When the *client* (wrk) wins + the FIN race, that ephemeral port is pinned for 2×MSL and cannot be reused + toward the same server — and macOS only has ~16k ephemeral ports. A server + that closes even slightly late loses the FIN race often (measured: ~47% of + connections for node vs ~16% for expresso), so a long trial saturates the + client port space mid-run and throughput collapses: the same node server + measured 32.5k req/s in a 2s trial but 8.5k in an 8s trial from an + identical clean start. This is a client-side artifact, not server + performance. The harness therefore defaults to 2s × 5 trials on macOS + (`load.platform_overrides` in frameworks.json) so trials finish before the + wall; Linux keeps 8s × 3 (wider port range, `tcp_tw_reuse`). For + authoritative sustained close-mode numbers, run the suite on Linux. +- **Warmup + median.** Every scenario gets a warmup run, then N timed trials + (default 3); the report shows the median. Trials with socket connect + errors, timeouts, or non-2xx responses are discarded and retried. +- **Environment stamping.** Results embed the git commit (with dirty flag), + CPU, core count, OS, and tool versions. Compare runs only when those match. +- **Pinned dependencies.** The express server pins exact versions and + installs with `npm ci` against the committed lockfile. +- Benchmark on an idle machine, on AC power, and prefer comparing numbers + from the same run over numbers from different days. + +## Scenarios + +| Name | Route | What it measures | +|---|---|---| +| health | `GET /health` | per-request framework overhead | +| about-json | `GET /about` | ~1 KB JSON serialization | +| static-png | `GET /coffee.png` | file serving (expresso uses sendfile) | +| post-print | `POST /print` | body parsing + JSON echo | + +`/download` (zip) and `/pictures` (directory listing) are implemented by +every server for API parity but not timed: the zip route's output size +differs per archiver, which would make transfer-bound comparisons +meaningless. + +## Adding a framework + +1. Create `benchmark/servers//` with a server that mirrors + `example/main.cpp`. Reuse `servers/shared/` for the about payload, mime + types, parity headers, and the zip builder — responses must stay + byte-comparable. The server must read `PORT` (and optionally `ASSETS`) + from the environment. +2. Add an entry to `frameworks.json`: `name`, `cwd` (repo-root relative), + `start` command, optional `probe` (binary that must exist, else the + framework is skipped), optional `install`/`installed_marker` and `build`. + +Adding another runtime for an existing server (like bun/deno for node-http) +is just another registry entry with a different `start` command. + +## Known caveats + +- Header parity is approximate: each framework adds its own incidental + headers (`X-Powered-By`, `Date`, etags), so tiny-response scenarios have a + few dozen bytes of variance. +- The JavaScript mirrors log `POST /print` bodies like the example does; + all server stdout goes to the per-run log files so console throughput + doesn't skew anyone. diff --git a/benchmark/scripts/compare.py b/benchmark/scripts/compare.py new file mode 100644 index 0000000..291848d --- /dev/null +++ b/benchmark/scripts/compare.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +"""Render a markdown comparison from benchmark result JSON files. + +Used automatically by run_benchmarks.py at the end of a run, and usable +standalone to re-render or diff older runs: + + python3 compare.py ../results/20260719T120000Z_close/results.json +""" + +import argparse +import json +from pathlib import Path + + +def _scenario_names(results): + names = [] + for framework in results["frameworks"].values(): + for name in framework.get("scenarios", {}): + if name not in names: + names.append(name) + return names + + +def _framework_names(results): + return list(results["frameworks"].keys()) + + +def _median_rate(results, framework, scenario): + data = ( + results["frameworks"] + .get(framework, {}) + .get("scenarios", {}) + .get(scenario, {}) + ) + rate = data.get("requests_per_second") + return rate["median"] if rate else None + + +def render_markdown(results, baseline=None): + metadata = results["metadata"] + frameworks = _framework_names(results) + scenarios = _scenario_names(results) + if baseline not in frameworks: + baseline = frameworks[0] if frameworks else None + + lines = ["# Benchmark comparison", ""] + lines.append(f"- **Date:** {metadata['timestamp_utc']}") + lines.append(f"- **Commit:** {metadata['git_commit']}") + lines.append(f"- **Machine:** {metadata['cpu']} ({metadata['logical_cores']} cores), {metadata['platform']}") + load = metadata["load"] + lines.append( + f"- **Load:** wrk -t{load['threads']} -c{load['connections']} " + f"-d{load['duration_seconds']}s, {load['trials']} trial(s), " + f"connection mode: {load['connection_mode']}" + ) + versions = ", ".join( + f"{tool} {version}" + for tool, version in metadata["tool_versions"].items() + if version + ) + lines.append(f"- **Tools:** {versions}") + lines.append("") + + lines.append("## Throughput (median requests/sec, higher is better)") + lines.append("") + header = ["Scenario"] + [ + f"{name}{' (baseline)' if name == baseline else ''}" for name in frameworks + ] + lines.append("| " + " | ".join(header) + " |") + lines.append("|" + "---|" * len(header)) + for scenario in scenarios: + base_rate = _median_rate(results, baseline, scenario) + row = [scenario] + for framework in frameworks: + rate = _median_rate(results, framework, scenario) + if rate is None: + skipped = results["frameworks"][framework].get("skipped") + row.append("skipped" if skipped else "-") + elif base_rate and framework != baseline: + row.append(f"{rate:,.0f} ({rate / base_rate:.2f}x)") + else: + row.append(f"{rate:,.0f}") + lines.append("| " + " | ".join(row) + " |") + lines.append("") + + lines.append("## Latency (median of trials, milliseconds)") + lines.append("") + lines.append("| Scenario | " + " | ".join(frameworks) + " |") + lines.append("|" + "---|" * (len(frameworks) + 1)) + for scenario in scenarios: + row = [scenario] + for framework in frameworks: + data = ( + results["frameworks"][framework] + .get("scenarios", {}) + .get(scenario, {}) + ) + if "latency_avg_ms" in data: + row.append( + f"avg {data['latency_avg_ms']}, p99 {data['latency_p99_ms']}" + ) + else: + row.append("-") + lines.append("| " + " | ".join(row) + " |") + lines.append("") + + skipped = { + name: framework["skipped"] + for name, framework in results["frameworks"].items() + if "skipped" in framework + } + if skipped: + lines.append("## Skipped") + lines.append("") + for name, reason in skipped.items(): + lines.append(f"- **{name}**: {reason}") + lines.append("") + + return "\n".join(lines) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("results", nargs="+", help="results.json file(s) to render") + parser.add_argument("--baseline", default="node-http") + args = parser.parse_args() + + for results_path in args.results: + results = json.loads(Path(results_path).read_text()) + print(render_markdown(results, baseline=args.baseline)) + + +if __name__ == "__main__": + main() diff --git a/benchmark/scripts/frameworks.json b/benchmark/scripts/frameworks.json new file mode 100644 index 0000000..0149e72 --- /dev/null +++ b/benchmark/scripts/frameworks.json @@ -0,0 +1,88 @@ +{ + "port": 3210, + "load": { + "threads": 4, + "connections": 64, + "duration_seconds": 8, + "warmup_seconds": 2, + "trials": 3, + "connection_mode": "close", + "time_wait_threshold": 2000, + "time_wait_max_wait_seconds": 60, + "platform_overrides": { + "darwin": { + "duration_seconds": 2, + "warmup_seconds": 1, + "trials": 5 + } + } + }, + "frameworks": [ + { + "name": "expresso", + "language": "C++", + "probe": null, + "cwd": "build", + "build": ["cmake", "--build", ".", "-j"], + "start": ["./server"] + }, + { + "name": "node-http", + "language": "JavaScript (node)", + "probe": "node", + "cwd": "benchmark/servers/node-http", + "start": ["node", "server.js"] + }, + { + "name": "express", + "language": "JavaScript (node)", + "probe": "node", + "cwd": "benchmark/servers/express", + "install": ["npm", "ci"], + "installed_marker": "node_modules", + "start": ["node", "server.js"] + }, + { + "name": "node-http-bun", + "language": "JavaScript (bun)", + "probe": "bun", + "cwd": "benchmark/servers/node-http", + "start": ["bun", "server.js"] + }, + { + "name": "node-http-deno", + "language": "JavaScript (deno)", + "probe": "deno", + "cwd": "benchmark/servers/node-http", + "start": ["deno", "run", "--allow-net", "--allow-read", "--allow-env", "server.js"] + } + ], + "scenarios": [ + { + "name": "health", + "description": "Tiny plain-text response; measures per-request framework overhead", + "method": "GET", + "path": "/health" + }, + { + "name": "about-json", + "description": "~1 KB JSON serialization", + "method": "GET", + "path": "/about" + }, + { + "name": "static-png", + "description": "1.4 MB static file; measures the file-serving path", + "method": "GET", + "path": "/coffee.png" + }, + { + "name": "post-print", + "description": "JSON body parse + echo", + "method": "POST", + "path": "/print", + "body": "{\"benchmark\": true, \"suite\": \"expresso\"}", + "content_type": "application/json" + } + ] +} diff --git a/benchmark/scripts/run_benchmarks.py b/benchmark/scripts/run_benchmarks.py new file mode 100644 index 0000000..2d9faf5 --- /dev/null +++ b/benchmark/scripts/run_benchmarks.py @@ -0,0 +1,471 @@ +#!/usr/bin/env python3 +"""Run the expresso benchmark suite across every available framework. + +Starts each server from benchmark/scripts/frameworks.json in turn, drives it +with wrk, and writes a timestamped JSON result plus a markdown comparison to +benchmark/results/. Python stdlib only; the sole external tool is wrk. + +Reproducibility measures baked in: + - connection-close mode by default (expresso does not keep connections + alive yet, so keep-alive would compare different workloads) + - warmup run before timing, median over N trials + - waits for TIME_WAIT sockets to drain between runs so ephemeral-port + exhaustion cannot corrupt close-mode numbers + - a trial with connect errors or non-2xx responses is invalid and retried + - every result embeds the git commit, hardware, and tool versions +""" + +import argparse +import datetime +import json +import platform +import re +import shutil +import socket +import statistics +import subprocess +import sys +import time +import urllib.error +import urllib.request +from pathlib import Path + +import compare + +SCRIPTS_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPTS_DIR.parents[1] +RESULTS_DIR = REPO_ROOT / "benchmark" / "results" + +READY_TIMEOUT_SECONDS = 15 +STOP_TIMEOUT_SECONDS = 5 + + +def log(message): + print(f"[bench] {message}", flush=True) + + +def run_command(command, cwd=None, timeout=None): + return subprocess.run( + command, cwd=cwd, capture_output=True, text=True, timeout=timeout + ) + + +def tool_version(command): + if shutil.which(command[0]) is None: + return None + try: + result = run_command(command, timeout=10) + output = (result.stdout or result.stderr).strip() + return output.splitlines()[0] if output else "" + except (OSError, subprocess.TimeoutExpired): + return None + + +def cpu_model(): + if platform.system() == "Darwin": + result = run_command(["sysctl", "-n", "machdep.cpu.brand_string"]) + return result.stdout.strip() + try: + for line in Path("/proc/cpuinfo").read_text().splitlines(): + if line.startswith("model name"): + return line.split(":", 1)[1].strip() + except OSError: + pass + return platform.processor() + + +def git_revision(): + result = run_command(["git", "rev-parse", "HEAD"], cwd=REPO_ROOT) + revision = result.stdout.strip() + dirty = run_command(["git", "status", "--porcelain"], cwd=REPO_ROOT).stdout.strip() + return f"{revision}{'-dirty' if dirty else ''}" if revision else "unknown" + + +def collect_metadata(config): + import os + + return { + "timestamp_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "git_commit": git_revision(), + "platform": platform.platform(), + "cpu": cpu_model(), + "logical_cores": os.cpu_count(), + "load": config["load"], + "tool_versions": { + "wrk": tool_version(["wrk", "--version"]), + "node": tool_version(["node", "--version"]), + "bun": tool_version(["bun", "--version"]), + "deno": tool_version(["deno", "--version"]), + "python": platform.python_version(), + }, + } + + +def time_wait_count(): + if platform.system() == "Darwin": + result = run_command(["netstat", "-an", "-p", "tcp"]) + else: + result = run_command(["ss", "-tan"]) + return sum("TIME_WAIT" in line for line in result.stdout.splitlines()) + + +def wait_for_time_wait_drain(load): + threshold = load["time_wait_threshold"] + deadline = time.monotonic() + load["time_wait_max_wait_seconds"] + count = time_wait_count() + while count > threshold and time.monotonic() < deadline: + log(f"waiting for {count} TIME_WAIT sockets to drain below {threshold}...") + time.sleep(3) + count = time_wait_count() + if count > threshold: + log(f"warning: proceeding with {count} TIME_WAIT sockets still open") + + +def port_in_use(port): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + return probe.connect_ex(("127.0.0.1", port)) == 0 + + +def wait_until_ready(port): + deadline = time.monotonic() + READY_TIMEOUT_SECONDS + url = f"http://127.0.0.1:{port}/health" + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(url, timeout=2) as response: + if response.status == 200: + return True + except (urllib.error.URLError, ConnectionError, TimeoutError): + time.sleep(0.1) + return False + + +class ServerProcess: + def __init__(self, framework, port, log_path): + self.framework = framework + self.port = port + self.log_path = log_path + self.process = None + + def __enter__(self): + import os + + if port_in_use(self.port): + raise RuntimeError( + f"port {self.port} is already in use; stop the other process first" + ) + env = dict(os.environ, PORT=str(self.port)) + self.log_file = open(self.log_path, "w") + self.process = subprocess.Popen( + self.framework["start"], + cwd=REPO_ROOT / self.framework["cwd"], + env=env, + stdout=self.log_file, + stderr=subprocess.STDOUT, + ) + if not wait_until_ready(self.port): + self.__exit__(None, None, None) + raise RuntimeError( + f"{self.framework['name']} did not become ready on port " + f"{self.port}; see {self.log_path}" + ) + return self + + def __exit__(self, *_): + if self.process is not None: + self.process.terminate() + try: + self.process.wait(STOP_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait() + self.log_file.close() + deadline = time.monotonic() + STOP_TIMEOUT_SECONDS + while port_in_use(self.port) and time.monotonic() < deadline: + time.sleep(0.1) + + +LATENCY_UNITS = {"us": 0.001, "ms": 1.0, "s": 1000.0, "m": 60000.0} + + +def parse_latency_ms(text): + match = re.fullmatch(r"([\d.]+)(us|ms|s|m)", text) + if match is None: + return None + return round(float(match.group(1)) * LATENCY_UNITS[match.group(2)], 3) + + +def parse_wrk_output(output): + parsed = { + "total_requests": 0, + "requests_per_second": None, + "latency_avg_ms": None, + "latency_p99_ms": None, + "transfer_per_second": None, + "non_2xx_responses": 0, + "socket_errors": {"connect": 0, "read": 0, "write": 0, "timeout": 0}, + } + match = re.search(r"(\d+) requests in", output) + if match: + parsed["total_requests"] = int(match.group(1)) + match = re.search(r"Requests/sec:\s+([\d.]+)", output) + if match: + parsed["requests_per_second"] = float(match.group(1)) + match = re.search(r"Transfer/sec:\s+([\d.]+\w+)", output) + if match: + parsed["transfer_per_second"] = match.group(1) + match = re.search(r"^\s*Latency\s+([\d.]+(?:us|ms|s|m))", output, re.MULTILINE) + if match: + parsed["latency_avg_ms"] = parse_latency_ms(match.group(1)) + match = re.search(r"^\s*99%\s+([\d.]+(?:us|ms|s|m))", output, re.MULTILINE) + if match: + parsed["latency_p99_ms"] = parse_latency_ms(match.group(1)) + match = re.search(r"Non-2xx or 3xx responses:\s+(\d+)", output) + if match: + parsed["non_2xx_responses"] = int(match.group(1)) + match = re.search( + r"Socket errors: connect (\d+), read (\d+), write (\d+), timeout (\d+)", output + ) + if match: + parsed["socket_errors"] = { + "connect": int(match.group(1)), + "read": int(match.group(2)), + "write": int(match.group(3)), + "timeout": int(match.group(4)), + } + return parsed + + +def write_wrk_script(scenario, close_mode, directory): + """wrk needs a lua script for anything beyond a plain GET.""" + lines = [] + if scenario["method"] != "GET": + lines.append(f'wrk.method = "{scenario["method"]}"') + if scenario.get("body"): + lines.append(f"wrk.body = {json.dumps(scenario['body'])}") + if scenario.get("content_type"): + lines.append(f'wrk.headers["Content-Type"] = "{scenario["content_type"]}"') + if close_mode: + lines.append('wrk.headers["Connection"] = "close"') + if not lines: + return None + script_path = directory / f"{scenario['name']}.lua" + script_path.write_text("\n".join(lines) + "\n") + return script_path + + +def run_wrk(scenario, port, load, duration, script_dir): + close_mode = load["connection_mode"] == "close" + command = [ + "wrk", + f"-t{load['threads']}", + f"-c{load['connections']}", + f"-d{duration}s", + "--latency", + ] + script = write_wrk_script(scenario, close_mode, script_dir) + if script is not None: + command += ["-s", str(script)] + elif close_mode: + command += ["-H", "Connection: close"] + command.append(f"http://127.0.0.1:{port}{scenario['path']}") + + result = run_command(command, timeout=duration + 30) + return result.stdout + + +def trial_is_valid(parsed): + # A handful of connect errors among tens of thousands of connections is + # loopback noise; a large fraction means port exhaustion corrupted the + # trial. Non-2xx responses are never acceptable. + error_budget = max(5, parsed["total_requests"] * 0.005) + errors = parsed["socket_errors"]["connect"] + parsed["socket_errors"]["timeout"] + return ( + parsed["requests_per_second"] is not None + and parsed["non_2xx_responses"] == 0 + and errors <= error_budget + ) + + +def run_scenario(scenario, port, load, script_dir): + log(f" scenario {scenario['name']}: warmup {load['warmup_seconds']}s") + run_wrk(scenario, port, load, load["warmup_seconds"], script_dir) + + trials = [] + attempts_left = load["trials"] + 2 # allow a couple of invalid-trial retries + while len(trials) < load["trials"] and attempts_left > 0: + attempts_left -= 1 + if load["connection_mode"] == "close": + wait_for_time_wait_drain(load) + output = run_wrk(scenario, port, load, load["duration_seconds"], script_dir) + parsed = parse_wrk_output(output) + parsed["raw_output"] = output + parsed["valid"] = trial_is_valid(parsed) + if parsed["valid"]: + trials.append(parsed) + log(f" trial {len(trials)}: {parsed['requests_per_second']:.0f} req/s") + else: + log(" invalid trial (errors or non-2xx responses), retrying") + trials_note = parsed + trials_note.pop("raw_output", None) + log(f" details: {json.dumps(trials_note)}") + + if not trials: + return {"error": "no valid trials", "trials": []} + + rates = [trial["requests_per_second"] for trial in trials] + return { + "requests_per_second": { + "median": round(statistics.median(rates), 2), + "min": min(rates), + "max": max(rates), + }, + "latency_avg_ms": statistics.median( + trial["latency_avg_ms"] for trial in trials + ), + "latency_p99_ms": statistics.median( + trial["latency_p99_ms"] for trial in trials + ), + "trials": trials, + } + + +def prepare_framework(framework): + probe = framework.get("probe") + if probe is not None and shutil.which(probe) is None: + return f"runtime '{probe}' not installed" + + cwd = REPO_ROOT / framework["cwd"] + if not cwd.is_dir(): + return f"directory {framework['cwd']} does not exist" + + marker = framework.get("installed_marker") + if framework.get("install") and (marker is None or not (cwd / marker).exists()): + log(f" installing dependencies: {' '.join(framework['install'])}") + result = run_command(framework["install"], cwd=cwd, timeout=300) + if result.returncode != 0: + return f"install failed: {result.stderr.strip()[-500:]}" + + if framework.get("build"): + log(f" building: {' '.join(framework['build'])}") + result = run_command(framework["build"], cwd=cwd, timeout=600) + if result.returncode != 0: + return f"build failed: {result.stderr.strip()[-500:]}" + + return None + + +def parse_arguments(): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--config", default=str(SCRIPTS_DIR / "frameworks.json")) + parser.add_argument("--frameworks", help="comma-separated subset to run") + parser.add_argument("--scenarios", help="comma-separated subset to run") + parser.add_argument("--mode", choices=["close", "keepalive"]) + parser.add_argument("--trials", type=int) + parser.add_argument("--duration", type=int, help="seconds per trial") + parser.add_argument("--threads", type=int) + parser.add_argument("--connections", type=int) + parser.add_argument("--port", type=int) + parser.add_argument("--baseline", default="node-http") + parser.add_argument("--out", default=str(RESULTS_DIR)) + parser.add_argument( + "--quick", action="store_true", help="1 trial of 2s each, for smoke tests" + ) + return parser.parse_args() + + +def apply_overrides(config, args): + load = config["load"] + # On macOS loopback, long close-mode trials exhaust the ~16k ephemeral + # ports with client-held TIME_WAIT entries and throughput collapses + # mid-trial (see README: "The macOS close-mode wall"). Short trials that + # finish before the wall, repeated more often, measure honestly. + platform_overrides = load.pop("platform_overrides", {}) + load.update(platform_overrides.get(platform.system().lower(), {})) + if args.quick: + load.update({"trials": 1, "duration_seconds": 2, "warmup_seconds": 1}) + if args.mode: + load["connection_mode"] = args.mode + if args.trials: + load["trials"] = args.trials + if args.duration: + load["duration_seconds"] = args.duration + if args.threads: + load["threads"] = args.threads + if args.connections: + load["connections"] = args.connections + if args.port: + config["port"] = args.port + + def keep(items, csv): + if csv is None: + return items + wanted = [name.strip() for name in csv.split(",")] + unknown = set(wanted) - {item["name"] for item in items} + if unknown: + sys.exit(f"unknown names: {', '.join(sorted(unknown))}") + return [item for item in items if item["name"] in wanted] + + config["frameworks"] = keep(config["frameworks"], args.frameworks) + config["scenarios"] = keep(config["scenarios"], args.scenarios) + return config + + +def main(): + args = parse_arguments() + if shutil.which("wrk") is None: + sys.exit("wrk is required: https://github.com/wg/wrk (brew install wrk)") + + config = apply_overrides(json.loads(Path(args.config).read_text()), args) + port = config["port"] + load = config["load"] + + out_dir = Path(args.out) + out_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ") + run_dir = out_dir / f"{stamp}_{load['connection_mode']}" + run_dir.mkdir() + + results = {"metadata": collect_metadata(config), "frameworks": {}} + + for framework in config["frameworks"]: + name = framework["name"] + log(f"framework {name}") + skip_reason = prepare_framework(framework) + if skip_reason is not None: + log(f" skipped: {skip_reason}") + results["frameworks"][name] = {"skipped": skip_reason} + continue + + try: + with ServerProcess(framework, port, run_dir / f"{name}.log"): + scenarios = {} + for scenario in config["scenarios"]: + scenarios[scenario["name"]] = run_scenario( + scenario, port, load, run_dir + ) + results["frameworks"][name] = { + "language": framework.get("language"), + "scenarios": scenarios, + } + except RuntimeError as error: + log(f" failed: {error}") + results["frameworks"][name] = {"skipped": str(error)} + + if load["connection_mode"] == "close": + wait_for_time_wait_drain(load) + + json_path = run_dir / "results.json" + json_path.write_text(json.dumps(results, indent=2)) + + markdown = compare.render_markdown(results, baseline=args.baseline) + markdown_path = run_dir / "comparison.md" + markdown_path.write_text(markdown) + + print() + print(markdown) + log(f"results: {json_path}") + log(f"report: {markdown_path}") + + +if __name__ == "__main__": + main() diff --git a/benchmark/servers/express/package.json b/benchmark/servers/express/package.json new file mode 100644 index 0000000..4d53e96 --- /dev/null +++ b/benchmark/servers/express/package.json @@ -0,0 +1,14 @@ +{ + "name": "@expresso-bench/express", + "version": "1.0.0", + "private": true, + "type": "module", + "description": "Express.js mirror of example/main.cpp for benchmarking", + "scripts": { + "start": "node server.js" + }, + "dependencies": { + "cookie-parser": "1.4.7", + "express": "5.1.0" + } +} diff --git a/benchmark/servers/express/server.js b/benchmark/servers/express/server.js new file mode 100644 index 0000000..24e1e75 --- /dev/null +++ b/benchmark/servers/express/server.js @@ -0,0 +1,80 @@ +// Express.js mirror of example/main.cpp. Uses express idiomatically +// (express.static, res.json, middleware chain) so the benchmark reflects what +// a real express user would deploy, while payloads and headers come from the +// shared fixtures for byte-comparable responses. + +import fs from "node:fs"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +import cookieParser from "cookie-parser"; +import express from "express"; + +import { aboutPayload } from "../shared/about.js"; +import { applyParityHeaders } from "../shared/headers.js"; +import { zipDirectory } from "../shared/zip.js"; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const ASSETS = path.resolve(process.env.ASSETS ?? path.join(here, "../../../assets")); +const PICTURES = path.join(ASSETS, "github"); +const PORT = Number(process.env.PORT ?? 8000); + +const app = express(); + +// Mirrors the example's Cors and Cacher middlewares +app.use((req, res, next) => { + applyParityHeaders(res); + next(); +}); +app.use(cookieParser()); +app.use(express.json()); + +app.get("/health", (req, res) => { + res.status(200).type("text/plain").send("Running healthy\r\n"); +}); + +app.get("/about", (req, res) => { + res.status(200).json(aboutPayload); +}); + +app.post("/print", (req, res) => { + console.log(JSON.stringify(req.body, null, 2)); + res.status(200).json({ + data: req.body, + timestamp: new Date().toUTCString(), + printed: true, + }); +}); + +app.get("/download", (req, res) => { + const zip = zipDirectory(ASSETS); + res + .status(200) + .set({ + "content-type": "application/zip", + "content-disposition": 'inline; filename="assets.zip"', + }) + .send(zip); +}); + +// Mirrors the listing-enabled StaticServe mounted at /pictures +app.get(["/pictures", "/pictures/"], (req, res) => { + const items = fs + .readdirSync(PICTURES) + .sort() + .map((entry) => `
  • ${entry}
    • `) + .join(""); + res.status(200).type("text/html").send(`
      ${items}
    `); +}); +app.use("/pictures", express.static(PICTURES)); + +// Mirrors the root StaticServe of ../assets +app.use(express.static(ASSETS)); + +app.use((req, res) => { + res.status(404).type("text/plain").send("Not Found"); +}); + +app.listen(PORT, () => { + console.log(`express mirror listening on port ${PORT}`); +}); diff --git a/benchmark/servers/node-http/package.json b/benchmark/servers/node-http/package.json new file mode 100644 index 0000000..59ce094 --- /dev/null +++ b/benchmark/servers/node-http/package.json @@ -0,0 +1,10 @@ +{ + "name": "@expresso-bench/node-http", + "version": "1.0.0", + "private": true, + "type": "module", + "description": "Dependency-free node:http mirror of example/main.cpp for benchmarking", + "scripts": { + "start": "node server.js" + } +} diff --git a/benchmark/servers/node-http/server.js b/benchmark/servers/node-http/server.js new file mode 100644 index 0000000..db459eb --- /dev/null +++ b/benchmark/servers/node-http/server.js @@ -0,0 +1,186 @@ +// Plain node:http mirror of example/main.cpp. Routes, payloads, and headers +// match the expresso example so load tests compare frameworks, not fixtures. +// Runs unchanged under node, bun, and deno (node-compat). + +import fs from "node:fs"; +import http from "node:http"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +import { aboutPayload } from "../shared/about.js"; +import { applyParityHeaders } from "../shared/headers.js"; +import { mimeType } from "../shared/mime.js"; +import { zipDirectory } from "../shared/zip.js"; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const ASSETS = path.resolve(process.env.ASSETS ?? path.join(here, "../../../assets")); +const PICTURES = path.join(ASSETS, "github"); +const PORT = Number(process.env.PORT ?? 8000); +const MAX_BODY_SIZE = 1024 * 1024; + +function sendText(res, statusCode, body, contentType = "text/plain") { + res.writeHead(statusCode, { + "content-type": contentType, + "content-length": Buffer.byteLength(body), + }); + res.end(body); +} + +function sendJson(res, statusCode, payload) { + sendText(res, statusCode, JSON.stringify(payload), "application/json"); +} + +// Resolves a URL path against a root directory, refusing escapes above it. +function resolveWithin(root, urlPath) { + const resolved = path.resolve(root, "." + path.posix.normalize(urlPath)); + if (resolved !== root && !resolved.startsWith(root + path.sep)) { + return null; + } + return resolved; +} + +function serveFile(req, res, filePath) { + let stats; + try { + stats = fs.statSync(filePath); + } catch { + return sendText(res, 404, "Not Found"); + } + if (stats.isDirectory()) { + filePath = path.join(filePath, "index.html"); + try { + stats = fs.statSync(filePath); + } catch { + return sendText(res, 404, "Not Found"); + } + } + + const fileName = path.basename(filePath); + res.setHeader("content-type", mimeType(fileName)); + res.setHeader("content-disposition", `inline; filename="${fileName}"`); + res.setHeader("accept-ranges", "bytes"); + + let start = 0; + let end = stats.size - 1; + let statusCode = 200; + const range = /^bytes=(\d*)-(\d*)$/.exec(req.headers.range ?? ""); + if (range && (range[1] !== "" || range[2] !== "")) { + start = range[1] === "" ? 0 : Number(range[1]); + end = range[2] === "" ? stats.size - 1 : Number(range[2]); + if (start >= stats.size || end >= stats.size || start > end) { + return sendText(res, 416, "Invalid Range"); + } + statusCode = 206; + res.setHeader("content-range", `bytes ${start}-${end}/${stats.size}`); + } + + res.writeHead(statusCode, { "content-length": end - start + 1 }); + fs.createReadStream(filePath, { start, end }).pipe(res); +} + +function serveListing(res, directory, urlPrefix) { + let entries; + try { + entries = fs.readdirSync(directory).sort(); + } catch { + return sendText(res, 404, "Not Found"); + } + const items = entries + .map((entry) => `
  • ${entry}
    • `) + .join(""); + sendText(res, 200, `
      ${items}
    `, "text/html"); +} + +function readBody(req) { + return new Promise((resolve, reject) => { + const chunks = []; + let size = 0; + req.on("data", (chunk) => { + size += chunk.length; + if (size > MAX_BODY_SIZE) { + reject(new Error("Body too large")); + req.destroy(); + return; + } + chunks.push(chunk); + }); + req.on("end", () => resolve(Buffer.concat(chunks).toString())); + req.on("error", reject); + }); +} + +async function handlePrint(req, res) { + let body; + try { + const raw = await readBody(req); + body = + req.headers["content-type"] === "application/json" && raw !== "" + ? JSON.parse(raw) + : raw; + } catch { + return sendText(res, 400, "Bad Request"); + } + + console.log(JSON.stringify(body, null, 2)); + sendJson(res, 200, { + data: body, + timestamp: new Date().toUTCString(), + printed: true, + }); +} + +const server = http.createServer((req, res) => { + applyParityHeaders(res); + + // Mirrors expresso's CookieParser middleware + req.cookies = Object.fromEntries( + (req.headers.cookie ?? "") + .split(";") + .map((cookie) => cookie.trim().split("=")) + .filter((pair) => pair.length === 2), + ); + + const urlPath = decodeURIComponent(new URL(req.url, "http://localhost").pathname); + + if (req.method === "GET" && urlPath === "/health") { + return sendText(res, 200, "Running healthy\r\n"); + } + if (req.method === "GET" && urlPath === "/about") { + return sendJson(res, 200, aboutPayload); + } + if (req.method === "POST" && urlPath === "/print") { + return handlePrint(req, res); + } + if (req.method === "GET" && urlPath === "/download") { + const zip = zipDirectory(ASSETS); + res.writeHead(200, { + "content-type": "application/zip", + "content-disposition": 'inline; filename="assets.zip"', + "content-length": zip.length, + }); + return res.end(zip); + } + if (req.method === "GET" && (urlPath === "/pictures" || urlPath === "/pictures/")) { + return serveListing(res, PICTURES, "/pictures"); + } + if (req.method === "GET" && urlPath.startsWith("/pictures/")) { + const filePath = resolveWithin(PICTURES, urlPath.slice("/pictures".length)); + if (filePath === null) { + return sendText(res, 404, "Not Found"); + } + return serveFile(req, res, filePath); + } + if (req.method === "GET") { + const filePath = resolveWithin(ASSETS, urlPath); + if (filePath === null) { + return sendText(res, 404, "Not Found"); + } + return serveFile(req, res, filePath); + } + + return sendText(res, 404, "Not Found"); +}); + +server.listen(PORT, () => { + console.log(`node-http mirror listening on port ${PORT}`); +}); diff --git a/benchmark/servers/shared/about.js b/benchmark/servers/shared/about.js new file mode 100644 index 0000000..7c4001f --- /dev/null +++ b/benchmark/servers/shared/about.js @@ -0,0 +1,57 @@ +// Mirrors the payload built by `about()` in example/main.cpp so every +// framework serves an identical /about response. + +const submodule = (name, repository, work, location) => ({ + name, + repository, + work, + location, +}); + +export const aboutPayload = { + framework: "Expresso", + repository: "https://github.com/coding-cpp/expresso", + language: "C++", + creator: { + name: "Adit Jain", + github: "https://github.com/jadit19", + }, + submodules: [ + submodule( + "Logger", + "https://github.com/coding-cpp/logger", + "Logging library for C++", + "lib/logger", + ), + submodule( + "Nexus", + "https://github.com/coding-cpp/nexus", + "Thread pool library for C++", + "lib/nexus", + ), + submodule( + "JSON", + "https://github.com/coding-cpp/json", + "JSON library for C++", + "lib/json", + ), + submodule( + "Brewtils", + "https://github.com/coding-cpp/brewtils", + "Utilities library for C++", + "lib/brewtils", + ), + submodule( + "Zippuccino", + "https://github.com/coding-cpp/zippuccino", + "Zipping library for C++", + "lib/zippuccino", + ), + submodule( + "Mochios", + "https://github.com/coding-cpp/mochios", + "HTTP request library for C++", + "lib/mochios", + ), + ], +}; diff --git a/benchmark/servers/shared/headers.js b/benchmark/servers/shared/headers.js new file mode 100644 index 0000000..6f24b08 --- /dev/null +++ b/benchmark/servers/shared/headers.js @@ -0,0 +1,16 @@ +// Approximates the headers the example expresso app attaches on every +// response through its Cors and Cacher middlewares, so response sizes stay +// comparable across frameworks. Kept in one place on purpose: change it here +// and every benchmark server stays in sync. + +export const parityHeaders = { + "access-control-allow-origin": "*", + "access-control-allow-credentials": "true", + "cache-control": "max-age=3600", +}; + +export function applyParityHeaders(res) { + for (const [name, value] of Object.entries(parityHeaders)) { + res.setHeader(name, value); + } +} diff --git a/benchmark/servers/shared/mime.js b/benchmark/servers/shared/mime.js new file mode 100644 index 0000000..ea41159 --- /dev/null +++ b/benchmark/servers/shared/mime.js @@ -0,0 +1,19 @@ +const MIME_TYPES = { + ".css": "text/css", + ".html": "text/html", + ".ico": "image/x-icon", + ".js": "application/javascript", + ".json": "application/json", + ".png": "image/png", + ".svg": "image/svg+xml", + ".txt": "text/plain", + ".zip": "application/zip", +}; + +export function mimeType(fileName) { + const dot = fileName.lastIndexOf("."); + if (dot < 0) { + return "application/octet-stream"; + } + return MIME_TYPES[fileName.slice(dot).toLowerCase()] ?? "application/octet-stream"; +} diff --git a/benchmark/servers/shared/package.json b/benchmark/servers/shared/package.json new file mode 100644 index 0000000..9b1a693 --- /dev/null +++ b/benchmark/servers/shared/package.json @@ -0,0 +1,7 @@ +{ + "name": "@expresso-bench/shared", + "version": "1.0.0", + "private": true, + "type": "module", + "description": "Framework-agnostic fixtures shared by all benchmark servers so responses stay byte-comparable" +} diff --git a/benchmark/servers/shared/zip.js b/benchmark/servers/shared/zip.js new file mode 100644 index 0000000..eaccaa4 --- /dev/null +++ b/benchmark/servers/shared/zip.js @@ -0,0 +1,103 @@ +// Dependency-free, store-only (no compression) ZIP builder used to mirror +// expresso's /download route. Store-only keeps it small and deterministic; +// the benchmark suite never times this route, it exists for API parity. + +import fs from "node:fs"; +import path from "node:path"; + +const CRC_TABLE = (() => { + const table = new Uint32Array(256); + for (let i = 0; i < 256; i++) { + let crc = i; + for (let j = 0; j < 8; j++) { + crc = crc & 1 ? 0xedb88320 ^ (crc >>> 1) : crc >>> 1; + } + table[i] = crc >>> 0; + } + return table; +})(); + +function crc32(buffer) { + let crc = 0xffffffff; + for (const byte of buffer) { + crc = CRC_TABLE[(crc ^ byte) & 0xff] ^ (crc >>> 8); + } + return (crc ^ 0xffffffff) >>> 0; +} + +function listFiles(directory) { + const files = []; + for (const entry of fs.readdirSync(directory, { withFileTypes: true })) { + const fullPath = path.join(directory, entry.name); + if (entry.isDirectory()) { + files.push(...listFiles(fullPath)); + } else if (entry.isFile()) { + files.push(fullPath); + } + } + return files.sort(); +} + +// Builds a ZIP of every file under `rootDir`, with entry names rooted at the +// directory's basename (zipping /x/assets yields entries like assets/app.js). +export function zipDirectory(rootDir) { + const root = path.resolve(rootDir); + const prefix = path.basename(root); + const localParts = []; + const centralParts = []; + let offset = 0; + + for (const filePath of listFiles(root)) { + const data = fs.readFileSync(filePath); + const name = Buffer.from( + path.join(prefix, path.relative(root, filePath)).split(path.sep).join("/"), + ); + const crc = crc32(data); + + const local = Buffer.alloc(30); + local.writeUInt32LE(0x04034b50, 0); // local file header signature + local.writeUInt16LE(20, 4); // version needed + local.writeUInt16LE(0, 6); // flags + local.writeUInt16LE(0, 8); // method: store + local.writeUInt32LE(0, 10); // dos time/date + local.writeUInt32LE(crc, 14); + local.writeUInt32LE(data.length, 18); // compressed size + local.writeUInt32LE(data.length, 22); // uncompressed size + local.writeUInt16LE(name.length, 26); + local.writeUInt16LE(0, 28); // extra length + localParts.push(local, name, data); + + const central = Buffer.alloc(46); + central.writeUInt32LE(0x02014b50, 0); // central directory signature + central.writeUInt16LE(20, 4); // version made by + central.writeUInt16LE(20, 6); // version needed + central.writeUInt16LE(0, 8); // flags + central.writeUInt16LE(0, 10); // method: store + central.writeUInt32LE(0, 12); // dos time/date + central.writeUInt32LE(crc, 16); + central.writeUInt32LE(data.length, 20); + central.writeUInt32LE(data.length, 24); + central.writeUInt16LE(name.length, 28); + central.writeUInt32LE(0, 30); // extra + comment lengths + central.writeUInt32LE(0, 34); // disk number + internal attrs + central.writeUInt32LE(0, 38); // external attrs + central.writeUInt32LE(offset, 42); // local header offset + centralParts.push(central, name); + + offset += local.length + name.length + data.length; + } + + const centralSize = centralParts.reduce((sum, part) => sum + part.length, 0); + const fileCount = centralParts.length / 2; + const eocd = Buffer.alloc(22); + eocd.writeUInt32LE(0x06054b50, 0); // end of central directory signature + eocd.writeUInt16LE(0, 4); // disk numbers + eocd.writeUInt16LE(0, 6); + eocd.writeUInt16LE(fileCount, 8); + eocd.writeUInt16LE(fileCount, 10); + eocd.writeUInt32LE(centralSize, 12); + eocd.writeUInt32LE(offset, 16); // central directory offset + eocd.writeUInt16LE(0, 20); // comment length + + return Buffer.concat([...localParts, ...centralParts, eocd]); +} diff --git a/include/expresso/core/io/connection.h b/include/expresso/core/io/connection.h new file mode 100644 index 0000000..6c8e7c2 --- /dev/null +++ b/include/expresso/core/io/connection.h @@ -0,0 +1,69 @@ +#pragma once + +#include +#include + +#include + +namespace expresso { + +namespace core { + +namespace io { + +// One client connection moving through its lifecycle: +// READING (event loop) -> PROCESSING (worker thread) -> WRITING (event loop). +// The event loop and a worker never own a connection at the same time, so no +// synchronization is needed on its members. +class Connection { +public: + enum class Phase { + READING, + PROCESSING, + WRITING, + }; + + enum class IOResult { + PENDING, + COMPLETE, + CLOSED, + }; + +private: + int socket; + Phase phase; + + std::string readBuffer; + std::string writeBuffer; + size_t bytesWritten; + expresso::messages::FileRegion file; + std::chrono::steady_clock::time_point lastActivity; + + IOResult flushBuffer(); + IOResult flushFile(); + +public: + explicit Connection(int socket) noexcept(false); + ~Connection(); + + Connection(const Connection&) = delete; + Connection& operator=(const Connection&) = delete; + + int getSocket() const; + Phase getPhase() const; + void setPhase(Phase phase); + + const std::string& getRequest() const; + void setResponse(std::string&& data, expresso::messages::FileRegion file); + + bool idleFor(std::chrono::seconds duration) const; + + IOResult read(); + IOResult write(); +}; + +} // namespace io + +} // namespace core + +} // namespace expresso diff --git a/include/expresso/core/io/poller.h b/include/expresso/core/io/poller.h new file mode 100644 index 0000000..ede9de0 --- /dev/null +++ b/include/expresso/core/io/poller.h @@ -0,0 +1,47 @@ +#pragma once + +#include + +#include + +#if !defined(__linux__) && !defined(__APPLE__) +#error "expresso only supports linux (epoll) and macOS (kqueue)" +#endif + +namespace expresso { + +namespace core { + +namespace io { + +struct Event { + int fd; + expresso::enums::NetworkEventType type; +}; + +// Thin wrapper over the platform readiness API: epoll on linux, kqueue on +// macOS. The backend is picked at compile time (poller_epoll.cpp or +// poller_kqueue.cpp). +class Poller { +private: + int pollerFd; + + static constexpr int MAX_EVENTS = 64; + +public: + Poller() noexcept(false); + ~Poller(); + + Poller(const Poller&) = delete; + Poller& operator=(const Poller&) = delete; + + void add(int fd, expresso::enums::NetworkEventType type) noexcept(false); + void remove(int fd); + std::vector wait(int timeoutMs = -1); +}; + +} // namespace io + +} // namespace core + +} // namespace expresso diff --git a/include/expresso/core/io/socket.h b/include/expresso/core/io/socket.h new file mode 100644 index 0000000..db6d542 --- /dev/null +++ b/include/expresso/core/io/socket.h @@ -0,0 +1,15 @@ +#pragma once + +namespace expresso { + +namespace core { + +namespace io { + +void setNonBlocking(int fd) noexcept(false); + +} // namespace io + +} // namespace core + +} // namespace expresso diff --git a/include/expresso/core/server.h b/include/expresso/core/server.h index b40cd18..0e44d9d 100644 --- a/include/expresso/core/server.h +++ b/include/expresso/core/server.h @@ -1,11 +1,21 @@ #pragma once +#include #include +#include +#include +#include +#include +#include #include +#include +#include #include #include +#include +#include #include #include #include @@ -15,21 +25,56 @@ namespace expresso { namespace core { +// Reactor-style server: a single event loop thread performs all socket i/o +// through non-blocking sockets and the platform poller, while request +// handling runs on the thread pool. Workers hand finished responses back to +// the event loop through `processedConnections` and the wakeup pipe, so no +// thread ever blocks on a client. class Server : public Router { private: - int socket; + int serverSocket; size_t maxConnections; struct sockaddr_in address; + static constexpr std::chrono::seconds IDLE_TIMEOUT{60}; + static constexpr int POLL_INTERVAL_MS = 1000; + + expresso::core::io::Poller poller; + std::unordered_map> + connections; + + // Sockets currently owned by a worker thread. Only the event loop touches + // this set; the idle sweep must skip these connections because their state + // is being written by the worker. + std::unordered_set processingConnections; + + int wakeupPipe[2]; + std::queue processedConnections; + std::mutex processedConnectionsMutex; + + // Declared last so its destructor joins the workers while the members they + // use are still alive + nexus::pool threadPool; + mochios::enums::method getMethodFromString( const std::string& method) noexcept(false); void setupMiddlewares(); + void runEventLoop(); void acceptConnections(); - void handleConnection(int clientSocket) noexcept(false); + void handleReadable( + std::shared_ptr connection); + void handleWritable( + std::shared_ptr connection); + void processRequest( + std::shared_ptr connection); + void flushProcessedConnections(); + void closeIdleConnections(); + void closeConnection( + std::shared_ptr connection); - expresso::messages::Request makeRequest(std::string& request) noexcept(false); - nexus::pool threadPool; + expresso::messages::Request makeRequest( + const std::string& request) noexcept(false); public: Server(size_t maxConnections = SOMAXCONN, @@ -41,4 +86,4 @@ class Server : public Router { } // namespace core -} // namespace expresso \ No newline at end of file +} // namespace expresso diff --git a/include/expresso/enums/network_event_type.h b/include/expresso/enums/network_event_type.h new file mode 100644 index 0000000..f807639 --- /dev/null +++ b/include/expresso/enums/network_event_type.h @@ -0,0 +1,14 @@ +#pragma once + +namespace expresso { + +namespace enums { + +enum class NetworkEventType { + READ, + WRITE, +}; + +} // namespace enums + +} // namespace expresso diff --git a/include/expresso/helpers/request.h b/include/expresso/helpers/request.h new file mode 100644 index 0000000..3dc0155 --- /dev/null +++ b/include/expresso/helpers/request.h @@ -0,0 +1,13 @@ +#pragma once + +#include + +namespace expresso { + +namespace helpers { + +bool isRequestComplete(const std::string& request); + +} // namespace helpers + +} // namespace expresso diff --git a/include/expresso/helpers/response.h b/include/expresso/helpers/response.h index 24b63f7..7b4360c 100644 --- a/include/expresso/helpers/response.h +++ b/include/expresso/helpers/response.h @@ -1,10 +1,8 @@ #pragma once -#include #include #include -#include #include namespace expresso { @@ -17,10 +15,10 @@ std::string getAvailableFile(const std::string& path); const std::string generateETag(const std::string& data); -bool sendChunkedData(const int& socket, const std::string& data); +std::string makeChunkedData(const std::string& data); -bool sendFileInChunks(const int& socket, const std::string& path); +std::string readFileAsChunked(const std::string& path); } // namespace helpers -} // namespace expresso \ No newline at end of file +} // namespace expresso diff --git a/include/expresso/messages/file_region.h b/include/expresso/messages/file_region.h new file mode 100644 index 0000000..106bb75 --- /dev/null +++ b/include/expresso/messages/file_region.h @@ -0,0 +1,21 @@ +#pragma once + +#include + +namespace expresso { + +namespace messages { + +// A slice of an open file that still has to be streamed to a client. The +// owner of a FileRegion is responsible for closing `fd`. +struct FileRegion { + int fd = -1; + off_t offset = 0; + off_t remaining = 0; + + bool isValid() const { return this->fd >= 0; } +}; + +} // namespace messages + +} // namespace expresso diff --git a/include/expresso/messages/response.h b/include/expresso/messages/response.h index d776a12..f81cf9f 100644 --- a/include/expresso/messages/response.h +++ b/include/expresso/messages/response.h @@ -1,11 +1,11 @@ #pragma once -#include -#include +#include +#include +#include #include #include -#include #include #include #include @@ -13,25 +13,28 @@ #include #include +#include namespace expresso { namespace messages { +// Responses are serialized into an in-memory payload (plus an optional file +// region for zero-copy transfers) which the server's event loop writes to +// the client socket. A Response never touches the socket itself. class Response : public mochios::messages::Response { private: bool hasEnded; - int socket; - std::string message; + std::string data; + FileRegion file; std::vector cookies; - void sendToClient(); - void sendHeaders(); + void serializeHeaders(); public: - Response(int clientSocket); + Response(); ~Response(); void set(std::string headerName, std::string headerValue); @@ -50,9 +53,11 @@ class Response : public mochios::messages::Response { void sendInvalidRange(); void end(); + std::string takeData(); + FileRegion takeFileRegion(); const void print() const override; }; } // namespace messages -} // namespace expresso \ No newline at end of file +} // namespace expresso diff --git a/src/core/io/connection.cpp b/src/core/io/connection.cpp new file mode 100644 index 0000000..83b6192 --- /dev/null +++ b/src/core/io/connection.cpp @@ -0,0 +1,167 @@ +#include +#include +#include +#include + +#ifdef __linux__ +#include +#endif + +#include +#include + +namespace { +constexpr size_t READ_BUFFER_SIZE = 4096; +} + +expresso::core::io::Connection::Connection(int socket) noexcept(false) + : socket(socket), phase(Phase::READING), bytesWritten(0), + lastActivity(std::chrono::steady_clock::now()) { + expresso::core::io::setNonBlocking(socket); + + return; +} + +expresso::core::io::Connection::~Connection() { + if (this->file.isValid()) { + close(this->file.fd); + } + close(this->socket); + + return; +} + +int expresso::core::io::Connection::getSocket() const { return this->socket; } + +expresso::core::io::Connection::Phase +expresso::core::io::Connection::getPhase() const { + return this->phase; +} + +void expresso::core::io::Connection::setPhase(Phase phase) { + this->phase = phase; + + return; +} + +const std::string& expresso::core::io::Connection::getRequest() const { + return this->readBuffer; +} + +void expresso::core::io::Connection::setResponse( + std::string&& data, expresso::messages::FileRegion file) { + this->writeBuffer = std::move(data); + this->bytesWritten = 0; + this->file = file; + this->lastActivity = std::chrono::steady_clock::now(); + this->phase = Phase::WRITING; + + return; +} + +bool expresso::core::io::Connection::idleFor( + std::chrono::seconds duration) const { + return std::chrono::steady_clock::now() - this->lastActivity > duration; +} + +expresso::core::io::Connection::IOResult +expresso::core::io::Connection::read() { + this->lastActivity = std::chrono::steady_clock::now(); + char buffer[READ_BUFFER_SIZE]; + + while (true) { + ssize_t bytesRead = recv(this->socket, buffer, sizeof(buffer), 0); + if (bytesRead > 0) { + this->readBuffer.append(buffer, bytesRead); + continue; + } + if (bytesRead == 0) { + return IOResult::CLOSED; + } + if (errno == EINTR) { + continue; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) { + return IOResult::PENDING; + } + return IOResult::CLOSED; + } +} + +expresso::core::io::Connection::IOResult +expresso::core::io::Connection::write() { + this->lastActivity = std::chrono::steady_clock::now(); + IOResult result = this->flushBuffer(); + if (result != IOResult::COMPLETE) { + return result; + } + + return this->flushFile(); +} + +expresso::core::io::Connection::IOResult +expresso::core::io::Connection::flushBuffer() { + while (this->bytesWritten < this->writeBuffer.size()) { + ssize_t bytesSent = + send(this->socket, this->writeBuffer.data() + this->bytesWritten, + this->writeBuffer.size() - this->bytesWritten, 0); + if (bytesSent < 0) { + if (errno == EINTR) { + continue; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) { + return IOResult::PENDING; + } + return IOResult::CLOSED; + } + this->bytesWritten += bytesSent; + } + + return IOResult::COMPLETE; +} + +expresso::core::io::Connection::IOResult +expresso::core::io::Connection::flushFile() { + while (this->file.isValid() && this->file.remaining > 0) { +#ifdef __linux__ + ssize_t bytesSent = + sendfile(this->socket, this->file.fd, &this->file.offset, + static_cast(this->file.remaining)); + if (bytesSent < 0) { + if (errno == EINTR) { + continue; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) { + return IOResult::PENDING; + } + return IOResult::CLOSED; + } + if (bytesSent == 0) { + // The file is shorter than promised; the response cannot be completed + return IOResult::CLOSED; + } + this->file.remaining -= bytesSent; +#else + off_t bytesSent = this->file.remaining; + int status = sendfile(this->file.fd, this->socket, this->file.offset, + &bytesSent, nullptr, 0); + this->file.offset += bytesSent; + this->file.remaining -= bytesSent; + if (status < 0) { + if (errno == EINTR) { + continue; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) { + return IOResult::PENDING; + } + return IOResult::CLOSED; + } + if (bytesSent == 0 && this->file.remaining > 0) { + // The file is shorter than promised; the response cannot be completed + return IOResult::CLOSED; + } +#endif + } + + return IOResult::COMPLETE; +} diff --git a/src/core/io/poller_epoll.cpp b/src/core/io/poller_epoll.cpp new file mode 100644 index 0000000..366ccec --- /dev/null +++ b/src/core/io/poller_epoll.cpp @@ -0,0 +1,70 @@ +#ifdef __linux__ + +#include +#include + +#include + +#include + +expresso::core::io::Poller::Poller() noexcept(false) { + this->pollerFd = epoll_create1(0); + if (this->pollerFd < 0) { + logger::error("Unable to create epoll instance!", + "expresso::core::io::Poller::Poller()"); + } + + return; +} + +expresso::core::io::Poller::~Poller() { + if (this->pollerFd >= 0) { + close(this->pollerFd); + } + + return; +} + +void expresso::core::io::Poller::add( + int fd, expresso::enums::NetworkEventType type) noexcept(false) { + struct epoll_event event; + event.data.fd = fd; + event.events = + type == expresso::enums::NetworkEventType::READ ? EPOLLIN : EPOLLOUT; + if (epoll_ctl(this->pollerFd, EPOLL_CTL_ADD, fd, &event) < 0) { + logger::error("Unable to watch fd " + std::to_string(fd) + "!", + "void expresso::core::io::Poller::add(int fd, " + "expresso::enums::NetworkEventType type)"); + } + + return; +} + +void expresso::core::io::Poller::remove(int fd) { + epoll_ctl(this->pollerFd, EPOLL_CTL_DEL, fd, nullptr); + + return; +} + +std::vector +expresso::core::io::Poller::wait(int timeoutMs) { + struct epoll_event events[Poller::MAX_EVENTS]; + int count = epoll_wait(this->pollerFd, events, Poller::MAX_EVENTS, timeoutMs); + + std::vector result; + if (count < 0) { + return result; + } + + result.reserve(count); + for (int i = 0; i < count; i++) { + expresso::enums::NetworkEventType type = + events[i].events & EPOLLOUT ? expresso::enums::NetworkEventType::WRITE + : expresso::enums::NetworkEventType::READ; + result.push_back({events[i].data.fd, type}); + } + + return result; +} + +#endif diff --git a/src/core/io/poller_kqueue.cpp b/src/core/io/poller_kqueue.cpp new file mode 100644 index 0000000..7313866 --- /dev/null +++ b/src/core/io/poller_kqueue.cpp @@ -0,0 +1,87 @@ +#ifdef __APPLE__ + +#include +#include +#include + +#include + +#include + +expresso::core::io::Poller::Poller() noexcept(false) { + this->pollerFd = kqueue(); + if (this->pollerFd < 0) { + logger::error("Unable to create kqueue instance!", + "expresso::core::io::Poller::Poller()"); + } + + return; +} + +expresso::core::io::Poller::~Poller() { + if (this->pollerFd >= 0) { + close(this->pollerFd); + } + + return; +} + +void expresso::core::io::Poller::add( + int fd, expresso::enums::NetworkEventType type) noexcept(false) { + struct kevent event; + int filter = type == expresso::enums::NetworkEventType::READ ? EVFILT_READ + : EVFILT_WRITE; + EV_SET(&event, fd, filter, EV_ADD, 0, 0, nullptr); + if (kevent(this->pollerFd, &event, 1, nullptr, 0, nullptr) < 0) { + logger::error("Unable to watch fd " + std::to_string(fd) + "!", + "void expresso::core::io::Poller::add(int fd, " + "expresso::enums::NetworkEventType type)"); + } + + return; +} + +void expresso::core::io::Poller::remove(int fd) { + // A connection is only ever watched for one filter at a time; deleting + // both tolerates removal from any state, and a missing filter is harmless. + struct kevent events[2]; + EV_SET(&events[0], fd, EVFILT_READ, EV_DELETE, 0, 0, nullptr); + EV_SET(&events[1], fd, EVFILT_WRITE, EV_DELETE, 0, 0, nullptr); + kevent(this->pollerFd, &events[0], 1, nullptr, 0, nullptr); + kevent(this->pollerFd, &events[1], 1, nullptr, 0, nullptr); + + return; +} + +std::vector +expresso::core::io::Poller::wait(int timeoutMs) { + struct timespec timeout; + struct timespec* timeoutPtr = nullptr; + if (timeoutMs >= 0) { + timeout.tv_sec = timeoutMs / 1000; + timeout.tv_nsec = (timeoutMs % 1000) * 1000000L; + timeoutPtr = &timeout; + } + + struct kevent events[Poller::MAX_EVENTS]; + int count = kevent(this->pollerFd, nullptr, 0, events, Poller::MAX_EVENTS, + timeoutPtr); + + std::vector result; + if (count < 0) { + return result; + } + + result.reserve(count); + for (int i = 0; i < count; i++) { + expresso::enums::NetworkEventType type = + events[i].filter == EVFILT_WRITE + ? expresso::enums::NetworkEventType::WRITE + : expresso::enums::NetworkEventType::READ; + result.push_back({static_cast(events[i].ident), type}); + } + + return result; +} + +#endif diff --git a/src/core/io/socket.cpp b/src/core/io/socket.cpp new file mode 100644 index 0000000..afa2ef8 --- /dev/null +++ b/src/core/io/socket.cpp @@ -0,0 +1,15 @@ +#include + +#include + +#include + +void expresso::core::io::setNonBlocking(int fd) noexcept(false) { + int flags = fcntl(fd, F_GETFL, 0); + if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) { + logger::error("Unable to make fd " + std::to_string(fd) + " non-blocking!", + "void expresso::core::io::setNonBlocking(int fd)"); + } + + return; +} diff --git a/src/core/server.cpp b/src/core/server.cpp index 578955a..f5d17d0 100644 --- a/src/core/server.cpp +++ b/src/core/server.cpp @@ -1,4 +1,12 @@ +#include +#include + +#include +#include + +#include #include +#include expresso::core::Server::Server(size_t maxConnections, size_t maxThreads) : maxConnections(maxConnections), threadPool(maxThreads) { @@ -11,11 +19,23 @@ expresso::core::Server::Server(size_t maxConnections, size_t maxThreads) "." + std::to_string(EXPRESSO_VERSION_MINOR) + "." + std::to_string(EXPRESSO_VERSION_PATCH)); - this->socket = brewtils::sys::socket(AF_INET, SOCK_STREAM, 0); - if (this->socket < 0) { + this->serverSocket = brewtils::sys::socket(AF_INET, SOCK_STREAM, 0); + if (this->serverSocket < 0) { logger::error("Socket not created!", "expresso::core::Server::Server()"); } + int reuseAddress = 1; + setsockopt(this->serverSocket, SOL_SOCKET, SO_REUSEADDR, &reuseAddress, + sizeof(reuseAddress)); + expresso::core::io::setNonBlocking(this->serverSocket); + + if (pipe(this->wakeupPipe) < 0) { + logger::error("Wakeup pipe not created!", + "expresso::core::Server::Server()"); + } + expresso::core::io::setNonBlocking(this->wakeupPipe[0]); + expresso::core::io::setNonBlocking(this->wakeupPipe[1]); + this->address.sin_family = AF_INET; this->address.sin_addr.s_addr = INADDR_ANY; this->setupMiddlewares(); @@ -24,9 +44,11 @@ expresso::core::Server::Server(size_t maxConnections, size_t maxThreads) } expresso::core::Server::~Server() { - if (this->socket > 0) { - close(this->socket); + if (this->serverSocket > 0) { + close(this->serverSocket); } + close(this->wakeupPipe[0]); + close(this->wakeupPipe[1]); return; } @@ -36,21 +58,21 @@ void expresso::core::Server::listen(int port, std::function callback) { pid_t pid = getpid(); logger::info("Server started with PID " + std::to_string(pid)); - if (brewtils::sys::bind(this->socket, (struct sockaddr*)&this->address, + if (brewtils::sys::bind(this->serverSocket, (struct sockaddr*)&this->address, sizeof(this->address)) < 0) { logger::error("Unable to bind socket!", - "void expresso::core::Server::run(int port)"); + "void expresso::core::Server::listen(int port)"); } - if (brewtils::sys::listen(this->socket, this->maxConnections) < 0) { + if (brewtils::sys::listen(this->serverSocket, this->maxConnections) < 0) { logger::error("Unable to listen on socket!", - "void expresso::core::Server::run(int port)"); + "void expresso::core::Server::listen(int port)"); } if (callback != nullptr) { callback(); } - this->acceptConnections(); + this->runEventLoop(); return; } @@ -66,7 +88,8 @@ mochios::enums::method expresso::core::Server::getMethodFromString( else if (method == "HEAD") return mochios::enums::method::HEAD; else logger::error("Unsupported HTTP method: " + method, - "expresso::core::Server::getMethodFromString(std::string &method) noexcept(false)"); + "expresso::core::Server::getMethodFromString(const " + "std::string &method) noexcept(false)"); } void expresso::core::Server::setupMiddlewares() { @@ -74,107 +97,243 @@ void expresso::core::Server::setupMiddlewares() { this->use(std::make_unique()); } +void expresso::core::Server::runEventLoop() { + this->poller.add(this->serverSocket, + expresso::enums::NetworkEventType::READ); + this->poller.add(this->wakeupPipe[0], + expresso::enums::NetworkEventType::READ); + + std::chrono::steady_clock::time_point lastSweep = + std::chrono::steady_clock::now(); + + while (true) { + for (const expresso::core::io::Event& event : + this->poller.wait(Server::POLL_INTERVAL_MS)) { + if (event.fd == this->serverSocket) { + this->acceptConnections(); + continue; + } + if (event.fd == this->wakeupPipe[0]) { + this->flushProcessedConnections(); + continue; + } + + std::unordered_map< + int, std::shared_ptr>::iterator it = + this->connections.find(event.fd); + if (it == this->connections.end()) { + continue; + } + if (event.type == expresso::enums::NetworkEventType::READ) { + this->handleReadable(it->second); + } else { + this->handleWritable(it->second); + } + } + + // Drain regardless of wakeup-pipe events so a lost wakeup can only delay + // a response by one poll interval, never strand it + this->flushProcessedConnections(); + + std::chrono::steady_clock::time_point now = + std::chrono::steady_clock::now(); + if (now - lastSweep >= std::chrono::seconds(1)) { + this->closeIdleConnections(); + lastSweep = now; + } + } + + return; +} + void expresso::core::Server::acceptConnections() { while (true) { struct sockaddr_in clientAddress; socklen_t clientAddressLength = sizeof(clientAddress); - int clientSocket = accept(this->socket, (struct sockaddr*)&clientAddress, + int clientSocket = accept(this->serverSocket, + (struct sockaddr*)&clientAddress, &clientAddressLength); if (clientSocket < 0) { - logger::error("Client connection not accepted!", - "void expresso::core::Server::acceptConnections()"); + if (errno != EAGAIN && errno != EWOULDBLOCK) { + logger::error("Client connection not accepted!"); + } return; } - this->threadPool.enqueue( - [this, clientSocket]() { this->handleConnection(clientSocket); }); + // A setup failure must not unwind the event loop; drop the client and + // keep serving + try { + std::shared_ptr connection = + std::make_shared(clientSocket); + this->connections[clientSocket] = connection; + this->poller.add(clientSocket, expresso::enums::NetworkEventType::READ); + } catch (const std::exception& e) { + logger::error(e.what()); + // If the connection made it into the map, erasing it closes the socket + // through the destructor; otherwise the raw fd is still ours to close + if (this->connections.erase(clientSocket) == 0) { + close(clientSocket); + } + } } return; } -void expresso::core::Server::handleConnection( - int clientSocket) noexcept(false) { - constexpr size_t bufferSize = 4096; - std::vector charRequest; - charRequest.resize(bufferSize, '\0'); - size_t totalBytesRead = 0; +void expresso::core::Server::handleReadable( + std::shared_ptr connection) { + if (connection->getPhase() != + expresso::core::io::Connection::Phase::READING) { + return; + } - while (true) { - // sanity check - if (charRequest.size() - totalBytesRead < bufferSize) { - charRequest.resize(charRequest.size() + bufferSize, '\0'); - } + expresso::core::io::Connection::IOResult result = connection->read(); + if (result == expresso::core::io::Connection::IOResult::CLOSED) { + this->closeConnection(connection); + return; + } - ssize_t bytesRead = brewtils::sys::recv( - clientSocket, - charRequest.data() + totalBytesRead, - bufferSize, - 0); - - // miraculous happening - if (bytesRead < 0) { - close(clientSocket); - logger::error("Failed to receive data from client!", - "void expresso::core::Server::handleConnection(int clientSocket) noexcept(false)"); - return; - } + if (!expresso::helpers::isRequestComplete(connection->getRequest())) { + return; + } + + // The worker owns the connection from here on; stop watching it until the + // response is ready to be written + this->poller.remove(connection->getSocket()); + this->processingConnections.insert(connection->getSocket()); + connection->setPhase(expresso::core::io::Connection::Phase::PROCESSING); + this->threadPool.enqueue( + [this, connection]() { this->processRequest(connection); }); + + return; +} - // if client closed connection - if (bytesRead == 0) { - break; +void expresso::core::Server::processRequest( + std::shared_ptr connection) { + expresso::messages::Response res; + + try { + expresso::messages::Request req = + this->makeRequest(connection->getRequest()); + req.res = &res; + this->handleRequest(req, res); + } catch (const std::exception& e) { + logger::error(e.what()); + res.status(expresso::enums::STATUS_CODE::BAD_REQUEST).send("Bad Request"); + } + + res.end(); + connection->setResponse(res.takeData(), res.takeFileRegion()); + + { + std::lock_guard lock(this->processedConnectionsMutex); + this->processedConnections.push(connection->getSocket()); + } + char wakeupSignal = 1; + if (write(this->wakeupPipe[1], &wakeupSignal, 1) < 0) { + // The pipe is full, so a wakeup is already pending and the event loop + // will drain the whole queue + } + + return; +} + +void expresso::core::Server::flushProcessedConnections() { + char buffer[64]; + while (read(this->wakeupPipe[0], buffer, sizeof(buffer)) > 0) { + } + + std::queue ready; + { + std::lock_guard lock(this->processedConnectionsMutex); + std::swap(ready, this->processedConnections); + } + + while (!ready.empty()) { + std::unordered_map< + int, std::shared_ptr>::iterator it = + this->connections.find(ready.front()); + this->processingConnections.erase(ready.front()); + ready.pop(); + if (it != this->connections.end()) { + try { + this->poller.add(it->first, expresso::enums::NetworkEventType::WRITE); + } catch (const std::exception& e) { + logger::error(e.what()); + this->closeConnection(it->second); + } } + } - totalBytesRead += bytesRead; + return; +} - // if end of headers reached - if (std::string(charRequest.data(), totalBytesRead).find("\r\n\r\n") != - std::string::npos) { - break; +void expresso::core::Server::closeIdleConnections() { + std::vector> idle; + for (const std::pair< + const int, std::shared_ptr>& entry : + this->connections) { + if (this->processingConnections.find(entry.first) != + this->processingConnections.end()) { + continue; + } + if (entry.second->idleFor(Server::IDLE_TIMEOUT)) { + idle.push_back(entry.second); } } - charRequest.resize(totalBytesRead); - std::string request(charRequest.data(), totalBytesRead); - if (totalBytesRead == 0 || request.empty()) { - close(clientSocket); - return; + for (const std::shared_ptr& connection : + idle) { + this->closeConnection(connection); } - expresso::messages::Response* res = - new expresso::messages::Response(clientSocket); + return; +} - try { - expresso::messages::Request req = this->makeRequest(request); - req.res = res; - this->handleRequest(req, *res); - delete res; - } catch (const std::exception& e) { - logger::error( - e.what(), - "void expresso::core::Server::handleConnection(int clientSocket)"); - res->status(expresso::enums::STATUS_CODE::BAD_REQUEST).send("Bad Request"); +void expresso::core::Server::handleWritable( + std::shared_ptr connection) { + if (connection->getPhase() != + expresso::core::io::Connection::Phase::WRITING) { + return; } - close(clientSocket); + expresso::core::io::Connection::IOResult result = connection->write(); + if (result == expresso::core::io::Connection::IOResult::PENDING) { + return; + } + this->closeConnection(connection); + return; } -expresso::messages::Request -expresso::core::Server::makeRequest(std::string& request) noexcept(false) { +void expresso::core::Server::closeConnection( + std::shared_ptr connection) { + this->poller.remove(connection->getSocket()); + this->connections.erase(connection->getSocket()); + + return; +} + +expresso::messages::Request expresso::core::Server::makeRequest( + const std::string& request) noexcept(false) { std::string line; std::istringstream stream(request); std::getline(stream, line); std::vector parts = brewtils::string::split(line, " "); + if (parts.size() < 3) { + logger::error("Malformed request line: " + line, + "expresso::core::Server::makeRequest(const std::string " + "&request) noexcept(false)"); + } const std::string method = brewtils::string::upper(parts[0]); expresso::messages::Request req(parts[1]); req.method = this->getMethodFromString(method); req.httpVersion = parts[2]; if (req.httpVersion.substr(0, 5) != "HTTP/") { logger::error("Invalid HTTP version: " + req.httpVersion, - "expresso::core::Server::makeRequest(std::string &request) " - "noexcept(false)"); + "expresso::core::Server::makeRequest(const std::string " + "&request) noexcept(false)"); } req.tempPath = req.path.substr(1, req.path.size()); @@ -230,7 +389,8 @@ expresso::core::Server::makeRequest(std::string& request) noexcept(false) { // Fixing the tempPath req.tempPath = req.tempPath.substr(0, req.tempPath.find('?', 0)); - if (req.tempPath[req.tempPath.size() - 1] == '/') { + if (!req.tempPath.empty() && + req.tempPath[req.tempPath.size() - 1] == '/') { req.tempPath = req.tempPath.substr(0, req.tempPath.size() - 1); } @@ -253,14 +413,28 @@ expresso::core::Server::makeRequest(std::string& request) noexcept(false) { std::string value; req.body = json::object(std::map()); for (const std::string& str : parts) { - key = brewtils::url::decode(brewtils::string::split(str, "=")[0]); - value = brewtils::url::decode(brewtils::string::split(str, "=")[1]); + std::vector pair = brewtils::string::split(str, "="); + if (pair.size() != 2) { + continue; + } + key = brewtils::url::decode(pair[0]); + value = brewtils::url::decode(pair[1]); req.body[key] = json::object(value); } } else if (brewtils::string::split(contentType, ";")[0] == "multipart/form-data") { - std::string delimiter = brewtils::string::split( - brewtils::string::split(contentType, ";")[1], "=")[1]; + std::vector typeParts = + brewtils::string::split(contentType, ";"); + if (typeParts.size() < 2) { + return req; + } + std::vector boundaryParts = + brewtils::string::split(typeParts[1], "="); + if (boundaryParts.size() < 2) { + return req; + } + + std::string delimiter = boundaryParts[1]; std::vector parts = brewtils::string::split(body, delimiter); std::vector data; std::string key; @@ -272,7 +446,12 @@ expresso::core::Server::makeRequest(std::string& request) noexcept(false) { if (data.size() == 2) { key = brewtils::string::split(data[1], "\r\n")[0]; key = key.substr(0, key.size() - 1); - value = brewtils::string::split(data[1], "\r\n\r\n")[1]; + std::vector valueParts = + brewtils::string::split(data[1], "\r\n\r\n"); + if (valueParts.size() < 2) { + continue; + } + value = valueParts[1]; value = brewtils::string::trim(value.substr(0, value.size() - 3)); req.body[key] = json::object(value); } @@ -280,4 +459,4 @@ expresso::core::Server::makeRequest(std::string& request) noexcept(false) { } return req; -} \ No newline at end of file +} diff --git a/src/helpers/request.cpp b/src/helpers/request.cpp new file mode 100644 index 0000000..6af6071 --- /dev/null +++ b/src/helpers/request.cpp @@ -0,0 +1,29 @@ +#include + +#include + +bool expresso::helpers::isRequestComplete(const std::string& request) { + size_t headersEnd = request.find("\r\n\r\n"); + if (headersEnd == std::string::npos) { + return false; + } + + size_t bodyStart = headersEnd + 4; + std::string headers = + brewtils::string::lower(request.substr(0, headersEnd + 2)); + + size_t contentLength = 0; + size_t position = headers.find("\r\ncontent-length:"); + if (position != std::string::npos) { + size_t valueStart = headers.find(':', position) + 1; + size_t valueEnd = headers.find('\r', valueStart); + try { + contentLength = std::stoull(brewtils::string::trim( + headers.substr(valueStart, valueEnd - valueStart))); + } catch (const std::exception&) { + contentLength = 0; + } + } + + return request.size() - bodyStart >= contentLength; +} diff --git a/src/helpers/response.cpp b/src/helpers/response.cpp index d519832..bd356b3 100644 --- a/src/helpers/response.cpp +++ b/src/helpers/response.cpp @@ -1,3 +1,9 @@ +#include +#include +#include + +#include + #include std::string expresso::helpers::getAvailableFile(const std::string& path) { @@ -26,52 +32,36 @@ const std::string expresso::helpers::generateETag(const std::string& data) { return etag.str(); } -bool expresso::helpers::sendChunkedData(const int& socket, - const std::string& data) { - std::ostringstream dataSizeHex; - dataSizeHex << std::hex << data.length(); - std::string dataSize = dataSizeHex.str() + "\r\n"; - - if (brewtils::sys::send(socket, dataSize.c_str(), dataSize.length(), 0) == - -1) { - return false; - } - if (brewtils::sys::send(socket, data.c_str(), data.length(), 0) == -1) { - return false; - } - if (brewtils::sys::send(socket, "\r\n", 2, 0) == -1) { - return false; - } - return true; +std::string expresso::helpers::makeChunkedData(const std::string& data) { + std::ostringstream chunk; + chunk << std::hex << data.length() << "\r\n" << data << "\r\n"; + return chunk.str(); } -bool expresso::helpers::sendFileInChunks(const int& socket, - const std::string& path) { +std::string expresso::helpers::readFileAsChunked(const std::string& path) { + std::string chunks; std::fstream file(path, std::ios::in | std::ios::binary); char buffer[expresso::helpers::CHUNK_SIZE]; try { - std::streamsize bytesRead = 0; while (true) { file.read(buffer, expresso::helpers::CHUNK_SIZE); - bytesRead = file.gcount(); + std::streamsize bytesRead = file.gcount(); if (bytesRead == 0) { break; } - - if (!expresso::helpers::sendChunkedData(socket, - std::string(buffer, bytesRead))) { - return false; - } + chunks += expresso::helpers::makeChunkedData( + std::string(buffer, static_cast(bytesRead))); } } catch (const std::exception& e) { - logger::error(e.what(), "bool sendFileInChunks(const int &socket, const " + logger::error(e.what(), + "std::string expresso::helpers::readFileAsChunked(const " "std::string &path)"); - return false; } + if (file.is_open()) { file.close(); } - return true; -} \ No newline at end of file + return chunks; +} diff --git a/src/messages/response.cpp b/src/messages/response.cpp index 295929f..84d746e 100644 --- a/src/messages/response.cpp +++ b/src/messages/response.cpp @@ -1,8 +1,10 @@ +#include +#include + #include #include -expresso::messages::Response::Response(int clientSocket) - : hasEnded(false), socket(clientSocket), message("") { +expresso::messages::Response::Response() : hasEnded(false), message("") { this->set("connection", "close"); this->statusCode = expresso::enums::STATUS_CODE::OK; @@ -10,8 +12,8 @@ expresso::messages::Response::Response(int clientSocket) } expresso::messages::Response::~Response() { - if (!this->hasEnded) { - this->sendToClient(); + if (this->file.isValid()) { + close(this->file.fd); } for (expresso::messages::Cookie* cookie : this->cookies) { @@ -77,6 +79,10 @@ expresso::messages::Response::json(json::object response) { void expresso::messages::Response::sendFile(const std::string& path, int64_t start, int64_t end) { + if (this->hasEnded) { + return; + } + std::string availableFile = expresso::helpers::getAvailableFile(path); if (availableFile.empty() || !brewtils::os::file::exists(availableFile)) { return this->sendNotFound(); @@ -112,43 +118,32 @@ void expresso::messages::Response::sendFile(const std::string& path, std::to_string(end) + "/" + std::to_string(fileSize)); } - this->sendHeaders(); - std::ifstream file(availableFile, std::ios::binary); - try { - file.seekg(start, std::ios::beg); - char buffer[expresso::helpers::CHUNK_SIZE]; - while (file.read(buffer, expresso::helpers::CHUNK_SIZE)) { - if (brewtils::sys::send(this->socket, buffer, - expresso::helpers::CHUNK_SIZE, 0) == -1) { - this->hasEnded = true; - break; - } - } - if (file.gcount() > 0) { - if (brewtils::sys::send(this->socket, buffer, file.gcount(), 0) == -1) { - this->hasEnded = true; - } - } - } catch (const std::exception& e) { - logger::error(e.what(), "void expresso::messages::Response::sendFile(const " - "std::string &path, int64_t start, int64_t end)"); - } - if (file.is_open()) { - file.close(); + int fileFd = open(availableFile.c_str(), O_RDONLY); + if (fileFd < 0) { + return this->sendNotFound(); } + this->serializeHeaders(); + this->file = {fileFd, static_cast(start), + static_cast(end - start + 1)}; + this->hasEnded = true; + return; } void expresso::messages::Response::sendFiles(const std::set& paths, const std::string& zipFileName) { + if (this->hasEnded) { + return; + } + this->headers.erase("content-length"); this->set("transfer-encoding", "chunked"); this->set("content-type", brewtils::os::file::getMimeType(zipFileName)); this->set("content-disposition", "inline; filename=\"" + zipFileName + "\""); this->set("accept-ranges", "bytes"); - this->sendHeaders(); + this->serializeHeaders(); zippuccino::Zipper zipper; for (const std::string& path : paths) { @@ -158,25 +153,12 @@ void expresso::messages::Response::sendFiles(const std::set& paths, try { while (!zipper.isFinished()) { - if (!expresso::helpers::sendChunkedData(this->socket, - zipper.getHeader())) { - this->hasEnded = true; - return; - } - std::string currentFile = zipper.getCurrentFile(); - if (!expresso::helpers::sendFileInChunks(this->socket, currentFile)) { - this->hasEnded = true; - return; - } - } - - if (!expresso::helpers::sendChunkedData(this->socket, zipper.getFooter())) { - this->hasEnded = true; - return; - } - if (brewtils::sys::send(this->socket, "0\r\n\r\n", 5, 0) == -1) { - this->hasEnded = true; + this->data += expresso::helpers::makeChunkedData(zipper.getHeader()); + this->data += + expresso::helpers::readFileAsChunked(zipper.getCurrentFile()); } + this->data += expresso::helpers::makeChunkedData(zipper.getFooter()); + this->data += "0\r\n\r\n"; } catch (const std::exception& e) { logger::error(e.what(), "void expresso::messages::Response::sendFiles(const " @@ -184,6 +166,8 @@ void expresso::messages::Response::sendFiles(const std::set& paths, "&zipFileName)"); } + this->hasEnded = true; + return; } @@ -204,12 +188,25 @@ void expresso::messages::Response::sendInvalidRange() { void expresso::messages::Response::end() { if (!this->hasEnded) { - this->sendToClient(); + this->set("content-length", std::to_string(this->message.length())); + this->serializeHeaders(); + this->data += this->message; + this->hasEnded = true; } return; } +std::string expresso::messages::Response::takeData() { + return std::move(this->data); +} + +expresso::messages::FileRegion expresso::messages::Response::takeFileRegion() { + FileRegion region = this->file; + this->file = FileRegion(); + return region; +} + const void expresso::messages::Response::print() const { logger::info("Response: "); logger::info(" statusCode: " + std::to_string(this->statusCode)); @@ -223,31 +220,16 @@ const void expresso::messages::Response::print() const { return; } -void expresso::messages::Response::sendToClient() { - this->set("content-length", std::to_string(this->message.length())); - this->sendHeaders(); - if (this->hasEnded) { - return; - } - brewtils::sys::send(this->socket, this->message.c_str(), - this->message.length(), 0); - this->hasEnded = true; - - return; -} - -void expresso::messages::Response::sendHeaders() { +void expresso::messages::Response::serializeHeaders() { std::string header = "HTTP/1.1 " + std::to_string(this->statusCode) + "\r\n"; - for (std::pair it : this->headers) { + for (const std::pair& it : this->headers) { header += it.first + ": " + it.second + "\r\n"; } for (Cookie* cookie : this->cookies) { header += "set-cookie: " + cookie->serialize() + "\r\n"; } header += "\r\n"; - if (brewtils::sys::send(this->socket, header.c_str(), header.length(), 0) == - -1) { - this->hasEnded = true; - } + this->data += header; + return; -} \ No newline at end of file +} diff --git a/src/middleware/static_serve.cpp b/src/middleware/static_serve.cpp index 74302f5..32205d8 100644 --- a/src/middleware/static_serve.cpp +++ b/src/middleware/static_serve.cpp @@ -1,3 +1,5 @@ +#include + #include #include