Compare the HTTP host with drogon and by CPU time

A drogon 1.9 service with the axum baseline's contract joins the benchmark
(memory, fdatasync and crash-safe modes, three IO threads), built from
Debian's packages unpacked under .deps without installing them. drogon
listens once per event loop, so the service picks a free port itself:
with port 0 every loop got a port of its own and the load reached one.

The load generator has one core and saturates near 150k answers a second,
so the harness now also reports answers per second of server CPU time
(all threads). Over three rotated rounds: small GETs 151.6k/s on three
shards (drogon 145.3k, axum 120.9k) and 128k per CPU-second on one thread
(drogon 65k, axum 62k); 1 MiB GETs 2,794/s (axum 1,372, drogon 708);
crash-safe PUTs 5.1-5.8 times axum and 13-30 times drogon; peak RSS
5.0 MiB against 14.4 and 76.5 MiB. --workloads is now a regular expression.

5 files changed, 219 insertions, 10 deletions

diff --git a/benchmarks/http/README.md b/benchmarks/http/README.md
index e26174b..b253dad 100644
--- a/benchmarks/http/README.md
+++ b/benchmarks/http/README.md
@@ -1,10 +1,12 @@
 # HTTP 벤치마크
 
-Pure의 문서 서버들과 같은 계약의 axum 서비스를 같은 기기·코어 배정·부하 생성기로 잰다([075](../../docs/research/075-http-host-against-axum.md)).
+Pure의 문서 서버들과 같은 계약의 axum(Rust)·drogon(C++) 서비스를 같은 기기·코어 배정·부하 생성기로 잰다
+([075](../../docs/research/075-http-host-against-axum.md)).
 
 ```sh
 cc -O2 -std=c17 -o load load.c
 (cd axum-baseline && cargo build --release)
+drogon-baseline/build.sh   # Debian의 drogon 패키지를 설치하지 않고 .deps에 풀어 빌드한다
 make -C ../../prototypes/connections build/pure-serve
 make -C ../../prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-connections
 aup build ../../../aup/examples/editor.aup -o editor.pmod
diff --git a/benchmarks/http/drogon-baseline/.gitignore b/benchmarks/http/drogon-baseline/.gitignore
new file mode 100644
index 0000000..ec322a0
--- /dev/null
+++ b/benchmarks/http/drogon-baseline/.gitignore
@@ -0,0 +1,2 @@
+.deps/
+drogon-baseline
diff --git a/benchmarks/http/drogon-baseline/build.sh b/benchmarks/http/drogon-baseline/build.sh
new file mode 100755
index 0000000..342f388
--- /dev/null
+++ b/benchmarks/http/drogon-baseline/build.sh
@@ -0,0 +1,15 @@
+#!/bin/sh
+# Build the drogon reference service against Debian's drogon without
+# installing anything: the packages are downloaded and unpacked under .deps
+# (an old-style rpath, so drogon finds its own dependencies there too).
+set -e
+cd "$(dirname "$0")"
+LIB=.deps/usr/lib/$(dpkg-architecture -qDEB_HOST_MULTIARCH)
+if [ ! -d "$LIB" ]; then
+    mkdir -p .deps/debs
+    (cd .deps/debs && apt-get download libdrogon-dev libdrogon1t64 libtrantor-dev libtrantor1 libjsoncpp-dev \
+        libhiredis1.1.0 libyaml-cpp0.8 && for deb in *.deb; do dpkg -x "$deb" ..; done)
+fi
+g++ -O2 -std=c++17 -I.deps/usr/include -I.deps/usr/include/jsoncpp -o drogon-baseline main.cc \
+    -L"$LIB" -Wl,--disable-new-dtags,-rpath,"$PWD/$LIB" -Wl,-rpath-link,"$LIB" -ldrogon -ltrantor -l:libjsoncpp.so.26 \
+    -lssl -lcrypto -lz -luuid -lpthread
diff --git a/benchmarks/http/drogon-baseline/main.cc b/benchmarks/http/drogon-baseline/main.cc
new file mode 100644
index 0000000..3151084
--- /dev/null
+++ b/benchmarks/http/drogon-baseline/main.cc
@@ -0,0 +1,172 @@
+// Reference service for Pure's HTTP benchmarks in drogon (C++), with the
+// contract of axum-baseline (see its src/main.rs): GET/PUT /objects/{key},
+// If-Match / If-None-Match: *, a PUT body of u32le start, u32le remove, then
+// the bytes to insert; the reply is the inserted bytes; ETag is the revision.
+//
+// drogon-baseline --port P [--durable DIR | --atomic DIR]
+// --durable writes the document and fdatasyncs before answering (not crash
+// safe); --atomic writes key.tmp, fsyncs it, renames it over the key and
+// fsyncs the directory. Handlers block their event loop while they sync, as
+// a straightforward drogon handler would.
+#include <arpa/inet.h>
+#include <drogon/drogon.h>
+#include <fcntl.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#include <cstdio>
+#include <cstring>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <unordered_map>
+
+using namespace drogon;
+
+struct Document {
+    std::mutex lock;
+    std::string bytes;
+    uint64_t revision = 0;
+};
+
+static std::mutex documents_lock;
+static std::unordered_map<std::string, std::shared_ptr<Document>> documents;
+static std::string durable;
+static bool atomic_writes = false;
+
+static bool valid_key(const std::string &key) {
+    if (key.empty() || key.size() > 64) return false;
+    for (char c : key)
+        if (!isalnum((unsigned char)c) && c != '-' && c != '_') return false;
+    return true;
+}
+
+static std::shared_ptr<Document> document(const std::string &key) {
+    std::lock_guard<std::mutex> guard(documents_lock);
+    auto &slot = documents[key];
+    if (!slot) slot = std::make_shared<Document>();
+    return slot;
+}
+
+static HttpResponsePtr tagged(HttpStatusCode status, uint64_t revision, const char *error, std::string body) {
+    auto response = HttpResponse::newHttpResponse();
+    response->setStatusCode(status);
+    response->setContentTypeCode(CT_APPLICATION_OCTET_STREAM);
+    response->addHeader("X-Pure-Error", error);
+    if (revision) response->addHeader("ETag", "\"" + std::to_string(revision) + "\"");
+    response->setBody(std::move(body));
+    return response;
+}
+
+static bool write_all(int fd, const std::string &bytes) {
+    for (size_t at = 0; at < bytes.size();) {
+        ssize_t n = write(fd, bytes.data() + at, bytes.size() - at);
+        if (n <= 0) return false;
+        at += (size_t)n;
+    }
+    return true;
+}
+
+static bool persist(const std::string &key, const std::string &bytes) {
+    std::string path = durable + "/" + key;
+    if (!atomic_writes) {
+        int fd = open(path.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0644);
+        bool ok = fd >= 0 && write_all(fd, bytes) && fdatasync(fd) == 0;
+        if (fd >= 0) close(fd);
+        return ok;
+    }
+    std::string temporary = path + ".tmp";
+    int fd = open(temporary.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0644);
+    bool ok = fd >= 0 && write_all(fd, bytes) && fsync(fd) == 0;
+    if (fd >= 0) close(fd);
+    ok = ok && rename(temporary.c_str(), path.c_str()) == 0;
+    int directory = ok ? open(durable.c_str(), O_RDONLY | O_DIRECTORY) : -1;
+    ok = ok && directory >= 0 && fsync(directory) == 0;
+    if (directory >= 0) close(directory);
+    return ok;
+}
+
+static HttpResponsePtr show(const std::string &key) {
+    if (!valid_key(key)) return tagged(k400BadRequest, 0, "request", "");
+    auto doc = document(key);
+    std::lock_guard<std::mutex> guard(doc->lock);
+    if (!doc->revision) return tagged(k404NotFound, 0, "not_found", "");
+    return tagged(k200OK, doc->revision, "none", doc->bytes);
+}
+
+static HttpResponsePtr edit(const HttpRequestPtr &request, const std::string &key) {
+    const std::string &match = request->getHeader("if-match"), &none = request->getHeader("if-none-match");
+    std::string_view body = request->body();
+    uint64_t expected = 0;
+    bool known = false;
+    if (!match.empty() && none.empty() && match.size() > 2 && match.front() == '"' && match.back() == '"') {
+        char *end = nullptr;
+        std::string digits = match.substr(1, match.size() - 2);
+        expected = strtoull(digits.c_str(), &end, 10);
+        known = !digits.empty() && end && !*end;
+    } else if (match.empty() && none == "*") known = true;
+    if (!known || !valid_key(key) || body.size() < 8) return tagged(k400BadRequest, 0, "request", "");
+    uint32_t start, remove;
+    memcpy(&start, body.data(), 4);
+    memcpy(&remove, body.data() + 4, 4);
+    auto doc = document(key);
+    std::lock_guard<std::mutex> guard(doc->lock);
+    if (doc->revision != expected) return tagged(k412PreconditionFailed, doc->revision, "conflict", "");
+    if (start > doc->bytes.size() || remove > doc->bytes.size() - start)
+        return tagged(k413RequestEntityTooLarge, 0, "limit", "");
+    std::string_view insert = body.substr(8);
+    doc->bytes.replace(start, remove, insert.data(), insert.size());
+    if (!durable.empty() && !persist(key, doc->bytes)) return tagged(k500InternalServerError, 0, "storage", "");
+    ++doc->revision;
+    return tagged(k200OK, doc->revision, "none", std::string(insert));
+}
+
+/* drogon listens once per event loop (SO_REUSEPORT): port 0 would give each
+ * loop a port of its own, so a free port is chosen first. */
+static uint16_t free_port() {
+    int fd = socket(AF_INET, SOCK_STREAM, 0);
+    sockaddr_in address{};
+    address.sin_family = AF_INET;
+    address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    socklen_t length = sizeof(address);
+    uint16_t port = 0;
+    if (fd >= 0 && !bind(fd, (sockaddr *)&address, sizeof(address)) && !getsockname(fd, (sockaddr *)&address, &length))
+        port = ntohs(address.sin_port);
+    if (fd >= 0) close(fd);
+    return port;
+}
+
+int main(int argc, char **argv) {
+    uint16_t port = 8769;
+    for (int i = 1; i + 1 < argc; i += 2) {
+        std::string flag = argv[i];
+        if (flag == "--port") port = (uint16_t)std::stoi(argv[i + 1]);
+        else if (flag == "--durable") durable = argv[i + 1];
+        else if (flag == "--atomic") durable = argv[i + 1], atomic_writes = true;
+        else {
+            fprintf(stderr, "unknown argument %s\n", argv[i]);
+            return 2;
+        }
+    }
+    if (!durable.empty()) mkdir(durable.c_str(), 0755);
+    app().registerHandler(
+        "/objects/{key}",
+        [](const HttpRequestPtr &request, std::function<void(const HttpResponsePtr &)> &&callback,
+           const std::string &key) {
+            callback(request->method() == Get ? show(key) : edit(request, key));
+        },
+        {Get, Put});
+    if (!port) port = free_port();
+    app().addListener("127.0.0.1", port);
+    app().setThreadNum(3); // the benchmark gives servers three cores
+    app().setClientMaxBodySize(64 << 20);
+    app().setClientMaxMemoryBodySize(64 << 20);
+    app().setLogLevel(trantor::Logger::kWarn);
+    app().registerBeginningAdvice([] {
+        auto listeners = app().getListeners();
+        printf("{\"port\":%u}\n", listeners.empty() ? 0u : (unsigned)listeners[0].toPort());
+        fflush(stdout);
+    });
+    app().run();
+}
diff --git a/benchmarks/http/run.py b/benchmarks/http/run.py
index f9fdbc0..91356f9 100644
--- a/benchmarks/http/run.py
+++ b/benchmarks/http/run.py
@@ -1,11 +1,14 @@
 #!/usr/bin/env python3
-"""HTTP benchmarks: Pure's document servers against an axum service with the
-same contract, on the same machine, with the same load generator.
+"""HTTP benchmarks: Pure's document servers against axum (Rust) and drogon (C++)
+services with the same contract, on the same machine, with the same load
+generator.
 
 Servers are pinned to CPUs 0-2 and the load to CPU 3; stores live under
 --directory (use the device you care about: fsync costs are the device's).
 Prints one JSON object per (server, workload, round) and a summary table of
-medians over rounds.
+medians over rounds. The load generator has one core, so near its limit the
+servers' CPU time tells them apart: 2xx/cpu-s is answers per second of server
+CPU (all threads, user and system).
 
 Durable writes are only comparable on a device in the same state. An SD card
 slows down under sustained syncs, so the order of runs alone can change a
@@ -17,6 +20,7 @@ next to each result.
 import argparse
 import json
 import os
+import re
 import select
 import shutil
 import signal
@@ -42,6 +46,12 @@ def start(args):
     return process, port
 
 
+def cpu_seconds(pid):
+    """User and system time of every thread of the process."""
+    fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()
+    return (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK')
+
+
 def rss_kib(pid, field):
     for line in Path(f'/proc/{pid}/status').read_text().splitlines():
         if line.startswith(field):
@@ -91,12 +101,13 @@ def main():
     parser.add_argument('--directory', type=Path, default=Path('/var/tmp'))
     parser.add_argument('--seconds', type=float, default=5)
     parser.add_argument('--only', default='')
-    parser.add_argument('--workloads', default='', help='run only workloads whose label starts with one of these')
+    parser.add_argument('--workloads', default='', help='run only workloads whose label matches this regular expression')
     parser.add_argument('--rounds', type=int, default=1, help='repeat every server, rotating their order')
     parser.add_argument('--pause', type=float, default=0, help='seconds to rest the device between runs')
     parser.add_argument('--budget', type=int, default=64 << 20, help="Pure's payload budget in bytes")
     args = parser.parse_args()
     axum = HERE / 'axum-baseline/target/release/axum-baseline'
+    drogon = HERE / 'drogon-baseline/drogon-baseline'
     base = Path(tempfile.mkdtemp(dir=args.directory, prefix='pure-http-bench-'))
     servers = {
         'pure-program': lambda d: [PURE / 'build/tree/pure-connections', '--program', PURE / 'examples/editor.pure',
@@ -108,6 +119,9 @@ def main():
         'axum-memory': lambda d: [axum, '--port', '0'],
         'axum-durable': lambda d: [axum, '--port', '0', '--durable', d],
         'axum-atomic': lambda d: [axum, '--port', '0', '--atomic', d],
+        'drogon-memory': lambda d: [drogon, '--port', '0'],
+        'drogon-durable': lambda d: [drogon, '--port', '0', '--durable', d],
+        'drogon-atomic': lambda d: [drogon, '--port', '0', '--atomic', d],
     }
     workloads = [
         ('get 64B, 64 conn, close', dict(connections=64, method='get', keys=64, mode='close')),
@@ -117,7 +131,7 @@ def main():
         ('put 16B, 8 conn, keep-alive', dict(connections=8, method='put', keys=8, mode='keep-alive')),
     ]
     if args.workloads:
-        workloads = [w for w in workloads if w[0].startswith(tuple(args.workloads.split(',')))]
+        workloads = [w for w in workloads if re.search(args.workloads, w[0])]
     chosen = [name for name in servers if not args.only or name in args.only.split(',')]
     results = []
     try:
@@ -136,10 +150,13 @@ def main():
                         # the 1 MiB document is k0 for this workload
                         put(port, 'k0', b'B' * (1 << 20), 1)
                     probe = probe_ms(base) if workload['method'] == 'put' else None
+                    used = cpu_seconds(process.pid)
                     measured = load(port, workload['connections'], args.seconds, workload['method'],
                                     workload['keys'], workload['mode'])
+                    used = cpu_seconds(process.pid) - used
                     measured.update(server=name, workload=label, round=round_, probe_ms=probe,
-                                    rss_kib=rss_kib(process.pid, 'VmRSS:'),
+                                    rss_kib=rss_kib(process.pid, 'VmRSS:'), cpu_seconds=round(used, 2),
+                                    ok_per_cpu_second=round(measured['2xx'] / used) if used else 0,
                                     ok_per_second=round(measured['2xx'] / args.seconds), idle_rss_kib=idle)
                     results.append(measured)
                     print(json.dumps(measured), flush=True)
@@ -155,8 +172,8 @@ def main():
     median = lambda rows, field: statistics.median(r[field] for r in rows)
     print()
     print(f"medians over {args.rounds} round(s)")
-    print(f"{'workload':32s} {'server':14s} {'2xx/s':>9s} {'p50 us':>8s} {'p99 us':>8s} {'errors':>7s} "
-          f"{'non-2xx':>8s} {'probe ms':>9s}")
+    print(f"{'workload':32s} {'server':14s} {'2xx/s':>9s} {'2xx/cpu-s':>10s} {'p50 us':>8s} {'p99 us':>8s} "
+          f"{'errors':>7s} {'non-2xx':>8s} {'probe ms':>9s}")
     for label, _ in workloads:
         for name in chosen:
             rows = [r for r in results if r['server'] == name and r['workload'] == label]
@@ -164,7 +181,8 @@ def main():
                 continue
             probes = [r['probe_ms'] for r in rows if r['probe_ms'] is not None]
             probe = f"{min(probes):.1f}-{max(probes):.1f}" if probes else '-'
-            print(f"{label:32s} {name:14s} {median(rows, 'ok_per_second'):9.0f} {median(rows, 'p50_us'):8.0f} "
+            print(f"{label:32s} {name:14s} {median(rows, 'ok_per_second'):9.0f} "
+                  f"{median(rows, 'ok_per_cpu_second'):10.0f} {median(rows, 'p50_us'):8.0f} "
                   f"{median(rows, 'p99_us'):8.0f} {sum(r['errors'] for r in rows):7d} "
                   f"{sum(r['4xx'] + r['5xx'] for r in rows):8d} {probe:>9s}")
     print()