|
| 1 | +/* |
| 2 | + * Copyright 2017-2026 Lenses.io Ltd |
| 3 | + * |
| 4 | + * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | + * you may not use this file except in compliance with the License. |
| 6 | + * You may obtain a copy of the License at |
| 7 | + * |
| 8 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | + * |
| 10 | + * Unless required by applicable law or agreed to in writing, software |
| 11 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | + * See the License for the specific language governing permissions and |
| 14 | + * limitations under the License. |
| 15 | + */ |
| 16 | +package io.lenses.streamreactor.connect.benchmarks |
| 17 | + |
| 18 | +/** |
| 19 | + * Outcome of a single throughput run (one sink, one iteration of one scenario). |
| 20 | + * |
| 21 | + * @param sink label identifying which sink/config this run measured, e.g. "HTTP" or "GCS". |
| 22 | + * @param scenario short human-readable description of the sweep point, e.g. "latency=400ms". |
| 23 | + * @param records total number of records successfully drained. |
| 24 | + * @param elapsedNanos wall-clock time from the first `put()` call to full drain (all records |
| 25 | + * acknowledged/committed). |
| 26 | + * @param networkOps number of latency-charged network round-trips performed. For HTTP this |
| 27 | + * is the number of HTTP requests; for GCS this is every latency-charged |
| 28 | + * storage call (upload + temp-file move/delete + index/lock bookkeeping), |
| 29 | + * NOT just the upload count -- see [[flushes]]. |
| 30 | + * @param flushes number of flushes/uploaded files. For HTTP this equals [[networkOps]] |
| 31 | + * (one request per batch). For GCS this is the `uploadFile` count (one |
| 32 | + * data file per flush), which with exactly-once enabled is far smaller |
| 33 | + * than [[networkOps]] because the commit chain makes several extra |
| 34 | + * latency-charged calls per flush. |
| 35 | + * @param heapUsedDeltaBytes approximate heap growth attributable to the run (best-effort; see |
| 36 | + * [[HeapSampler]]). |
| 37 | + */ |
| 38 | +final case class BenchResult( |
| 39 | + sink: String, |
| 40 | + scenario: String, |
| 41 | + records: Long, |
| 42 | + elapsedNanos: Long, |
| 43 | + networkOps: Long, |
| 44 | + flushes: Long, |
| 45 | + heapUsedDeltaBytes: Long, |
| 46 | +) { |
| 47 | + def elapsedMillis: Double = elapsedNanos / 1e6 |
| 48 | + |
| 49 | + def recordsPerSec: Double = if (elapsedNanos <= 0) 0.0 else records * 1e9 / elapsedNanos.toDouble |
| 50 | + |
| 51 | + /** Records per latency-charged network round-trip -- the fair, apples-to-apples amortisation. */ |
| 52 | + def avgRecordsPerOp: Double = if (networkOps <= 0) 0.0 else records.toDouble / networkOps.toDouble |
| 53 | + |
| 54 | + def avgOpLatencyMillis: Double = if (networkOps <= 0) 0.0 else elapsedMillis / networkOps.toDouble |
| 55 | +} |
| 56 | + |
| 57 | +/** |
| 58 | + * A scenario point measured across `iterations` repeats (after any discarded warm-up runs), |
| 59 | + * summarised as median/min/max `recordsPerSec` to absorb JIT warm-up and GC noise from a single |
| 60 | + * shared JVM process. Non-timing columns (records, ops, rec/op) are deterministic given fixed |
| 61 | + * inputs, so `sample` (the last measured iteration) is used to report those directly rather than |
| 62 | + * aggregating them. |
| 63 | + * |
| 64 | + * @param family groups scenario points that should be compared against each other across sinks, |
| 65 | + * e.g. "cpu-ceiling", "latency=400ms", "size=2048B". Used by |
| 66 | + * [[ResultsTablePrinter.printDeviation]] to line up the GCS and HTTP rows for the |
| 67 | + * same sweep point. |
| 68 | + * @param variant distinguishes multiple configurations run within the same sink/family, e.g. |
| 69 | + * "http batch=1500" vs "http batch=10000". |
| 70 | + */ |
| 71 | +final case class AggregatedResult( |
| 72 | + sink: String, |
| 73 | + family: String, |
| 74 | + variant: String, |
| 75 | + iterations: Int, |
| 76 | + medianRecordsPerSec: Double, |
| 77 | + minRecordsPerSec: Double, |
| 78 | + maxRecordsPerSec: Double, |
| 79 | + sample: BenchResult, |
| 80 | +) |
| 81 | + |
| 82 | +object BenchAggregator { |
| 83 | + |
| 84 | + /** `runs` must be non-empty measured iterations (warm-up iterations must already be excluded). */ |
| 85 | + def aggregate(sink: String, family: String, variant: String, runs: Seq[BenchResult]): AggregatedResult = { |
| 86 | + require(runs.nonEmpty, "aggregate() requires at least one measured iteration") |
| 87 | + val sortedRates = runs.map(_.recordsPerSec).sorted |
| 88 | + AggregatedResult( |
| 89 | + sink = sink, |
| 90 | + family = family, |
| 91 | + variant = variant, |
| 92 | + iterations = runs.size, |
| 93 | + medianRecordsPerSec = median(sortedRates), |
| 94 | + minRecordsPerSec = sortedRates.head, |
| 95 | + maxRecordsPerSec = sortedRates.last, |
| 96 | + sample = runs.last, |
| 97 | + ) |
| 98 | + } |
| 99 | + |
| 100 | + private def median(sorted: Seq[Double]): Double = { |
| 101 | + val n = sorted.size |
| 102 | + if (n % 2 == 1) sorted(n / 2) else (sorted(n / 2 - 1) + sorted(n / 2)) / 2.0 |
| 103 | + } |
| 104 | +} |
| 105 | + |
| 106 | +/** |
| 107 | + * Renders [[AggregatedResult]]s as plain-text tables: the per-scenario measurements, and a |
| 108 | + * separate GCS-vs-HTTP deviation table computed by grouping on `family`. |
| 109 | + */ |
| 110 | +object ResultsTablePrinter { |
| 111 | + |
| 112 | + def printAggregated(results: Seq[AggregatedResult]): String = { |
| 113 | + val header = |
| 114 | + f"${"Family"}%-24s ${"Variant"}%-34s ${"Sink"}%-5s ${"Iter"}%5s ${"Median rec/s"}%13s ${"Min rec/s"}%11s ${"Max rec/s"}%11s ${"Records"}%9s ${"Ops"}%6s ${"Flushes"}%8s ${"Rec/Op"}%9s ${"Avg Op(ms)"}%11s" |
| 115 | + val separator = "-" * header.length |
| 116 | + val rows = results.map { a => |
| 117 | + f"${a.family}%-24s ${a.variant}%-34s ${a.sink}%-5s ${a.iterations}%5d ${a.medianRecordsPerSec}%13.1f ${a.minRecordsPerSec}%11.1f ${a.maxRecordsPerSec}%11.1f ${a.sample.records}%9d ${a.sample.networkOps}%6d ${a.sample.flushes}%8d ${a.sample.avgRecordsPerOp}%9.1f ${a.sample.avgOpLatencyMillis}%11.3f" |
| 118 | + } |
| 119 | + (Seq(header, separator) ++ rows).mkString("\n") |
| 120 | + } |
| 121 | + |
| 122 | + /** |
| 123 | + * For every `family` that has a canonical GCS row, prints how each non-GCS (HTTP) row in that |
| 124 | + * same family compares to it, as a throughput ratio. The canonical GCS row is the default, |
| 125 | + * exactly-once-enabled configuration (the production default); families may additionally carry |
| 126 | + * an `eo=off` GCS variant, which is intentionally excluded from the ratio here so the comparison |
| 127 | + * stays against the production-default GCS behaviour (the `eo=off` numbers are shown in the main |
| 128 | + * results table). Families with no GCS row (e.g. the logging-cost HTTP-only comparison) are |
| 129 | + * skipped -- that comparison is HTTP-vs-HTTP, not GCS-vs-HTTP. |
| 130 | + */ |
| 131 | + def printDeviation(results: Seq[AggregatedResult]): String = { |
| 132 | + val byFamily = results.groupBy(_.family) |
| 133 | + val header = f"${"Family"}%-24s ${"GCS median rec/s"}%18s ${"HTTP variant"}%-34s ${"HTTP median rec/s"}%19s ${"GCS/HTTP ratio"}%15s" |
| 134 | + val separator = "-" * header.length |
| 135 | + val rows = byFamily.toSeq.sortBy(_._1).flatMap { |
| 136 | + case (family, rows) => |
| 137 | + val httpRows = rows.filter(_.sink == "HTTP") |
| 138 | + // Canonical GCS row = the exactly-once-enabled (production default) variant, i.e. not the |
| 139 | + // "eo=off" contrast variant. |
| 140 | + val gcsRows = rows.filter(row => row.sink == "GCS" && !row.variant.contains("eo=off")) |
| 141 | + if (gcsRows.size != 1 || httpRows.isEmpty) { |
| 142 | + Seq.empty |
| 143 | + } else { |
| 144 | + val gcs = gcsRows.head |
| 145 | + httpRows.map { http => |
| 146 | + val ratio = if (http.medianRecordsPerSec <= 0) Double.PositiveInfinity else gcs.medianRecordsPerSec / http.medianRecordsPerSec |
| 147 | + f"${family}%-24s ${gcs.medianRecordsPerSec}%18.1f ${http.variant}%-34s ${http.medianRecordsPerSec}%19.1f ${ratio}%14.2fx" |
| 148 | + } |
| 149 | + } |
| 150 | + } |
| 151 | + (Seq(header, separator) ++ rows).mkString("\n") |
| 152 | + } |
| 153 | +} |
| 154 | + |
| 155 | +/** |
| 156 | + * Best-effort heap usage sampling around a benchmark run. Not a substitute for a real profiler: |
| 157 | + * intended only to corroborate large, order-of-magnitude differences, not to make precise |
| 158 | + * per-record allocation claims. |
| 159 | + */ |
| 160 | +object HeapSampler { |
| 161 | + |
| 162 | + /** Requests a full GC and reads used heap. Best-effort: JVMs are not required to honour System.gc(). */ |
| 163 | + def usedHeapBytes(): Long = { |
| 164 | + val runtime = Runtime.getRuntime |
| 165 | + System.gc() |
| 166 | + Thread.sleep(50) |
| 167 | + System.gc() |
| 168 | + runtime.totalMemory() - runtime.freeMemory() |
| 169 | + } |
| 170 | + |
| 171 | + def measure[A](block: => A): (A, Long, Long) = { |
| 172 | + val before = usedHeapBytes() |
| 173 | + val result = block |
| 174 | + val after = usedHeapBytes() |
| 175 | + (result, before, after) |
| 176 | + } |
| 177 | +} |
0 commit comments