From 8e55e0d5de940540befdffabeef3471d7196cf7c Mon Sep 17 00:00:00 2001 From: crazycs520 Date: Tue, 15 Sep 2026 04:41:48 +0800 Subject: [PATCH] fix: record transaction outcomes and measured connection events --- HOW-TO-RUN.txt | 15 +++ build.xml | 15 ++- run/summarizeRun.py | 130 ++++++++++++++++++++++++ src/client/CommitException.java | 9 -- src/client/jTPCC.java | 22 +++- src/client/jTPCCConnection.java | 6 +- src/client/jTPCCSocketFactory.java | 74 ++++++++++++++ src/client/jTPCCTData.java | 29 +++++- src/client/jTPCCTerminal.java | 58 +++++------ tests/ConnectionMeasurementTest.java | 111 ++++++++++++++++++++ tests/TransactionMeasurementTest.java | 141 ++++++++++++++++++++++++++ tests/test_measurement_window.py | 59 +++++++++++ 12 files changed, 618 insertions(+), 51 deletions(-) create mode 100644 run/summarizeRun.py delete mode 100644 src/client/CommitException.java create mode 100644 src/client/jTPCCSocketFactory.java create mode 100644 tests/ConnectionMeasurementTest.java create mode 100644 tests/TransactionMeasurementTest.java create mode 100644 tests/test_measurement_window.py diff --git a/HOW-TO-RUN.txt b/HOW-TO-RUN.txt index a0428e0..061fa43 100755 --- a/HOW-TO-RUN.txt +++ b/HOW-TO-RUN.txt @@ -1,4 +1,19 @@ +Fixed-window measurements +------------------------- + +Use resultDirectory to retain data/result.csv. Run warmup and measurement in one JVM: for a two-minute warmup and five-minute measurement set runMins=7 and runTxnsPerTerminal=0. Then run: + + python3 run/summarizeRun.py /data --warmup-seconds 120 --measurement-seconds 300 + +The summary selects transactions by completion time in [120s, 420s), counts successful NEW_ORDER completions for tpmC, and reports nearest-rank p95 for each transaction type. Errors and successful intentional rollbacks are separate outcomes. p95 includes all attempts in the window; it is never averaged across runs. DELIVERY_BG is reported separately and is excluded from the main transaction throughput to avoid counting Delivery twice. + +For MySQL Connector/J, set measureConnections=true to record authenticated physical sockets in data/connections.csv. This uses the driver's socket factory callback and adds no SQL to the transaction loop. Each terminal's initial connection is checked before starting. Terminal socket authentication after startup counts as a reconnect when its timestamp falls in the measurement window. Failed authentication, repeated authentication on the same socket, and background discovery sockets are excluded; authentication followed by a JDBC initialization failure is included. This metric counts physical connections, not successful JDBC session initialization. The URL must not override socketFactory. + +Missing or incomplete connection evidence yields reconnects=null and measurement_valid=false, never zero. A truncated transaction run is also invalid. The summary exits nonzero when its evidence is incomplete. Preserve the CSVs, run.properties, log, client JAR and driver versions for comparisons. + +Validation: ant test; python3 -m unittest discover -s tests -p 'test_*.py'. Rebuild with ant clean dist when replacing an existing JAR, so deleted classes cannot remain in the build directory. + Instructions for running BenchmarkSQL on PostgreSQL --------------------------------------------------- diff --git a/build.xml b/build.xml index 4895cee..ca43736 100755 --- a/build.xml +++ b/build.xml @@ -28,7 +28,7 @@ - + @@ -37,4 +37,17 @@ + + + + + + + + + + + + + diff --git a/run/summarizeRun.py b/run/summarizeRun.py new file mode 100644 index 0000000..bb91102 --- /dev/null +++ b/run/summarizeRun.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""Summarize one JVM's fixed completion window from BenchmarkSQL raw CSVs.""" + +import argparse +import csv +import json +import math +from collections import Counter, defaultdict +from pathlib import Path + + +TRANSACTIONS = ("NEW_ORDER", "PAYMENT", "ORDER_STATUS", "DELIVERY", "STOCK_LEVEL", "DELIVERY_BG") + + +def connection_window(path, start_ms, end_ms): + """Count authenticated new physical sockets; failed authentication is excluded.""" + if not path.exists(): + return None, "connections.csv is missing" + count = 0 + started = None + ended = None + reconnects = 0 + last_timestamp = 0 + with path.open(newline="", encoding="utf-8") as stream: + for row in csv.DictReader(stream): + timestamp = int(row["epoch_ms"]) + if timestamp < last_timestamp or ended is not None: + raise ValueError("connection events are out of order") + last_timestamp = timestamp + if row["event"] == "authenticated": + count += 1 + if started is not None and row["terminal"] and start_ms <= timestamp - started < end_ms: + reconnects += 1 + elif row["event"] == "start": + if started is not None or count == 0: + raise ValueError("invalid connection measurement start") + started = timestamp + elif row["event"] == "end": + ended = timestamp + else: + raise ValueError("unknown connection event") + if int(row["handshakes"]) != count: + raise ValueError("connection event sequence is incomplete") + if started is None or ended is None or ended - started < end_ms: + return None, "connection measurement does not cover the full window" + return reconnects, None + + +def percentile95(samples): + if not samples: + return None + samples.sort() + return samples[math.ceil(len(samples) * 0.95) - 1] + + +def summarize(directory, warmup_seconds, measurement_seconds): + if warmup_seconds < 0 or measurement_seconds <= 0: + raise ValueError("window must have nonnegative warmup and positive measurement duration") + start_ms = warmup_seconds * 1000 + end_ms = start_ms + measurement_seconds * 1000 + buckets = defaultdict(lambda: {"latency": [], "dblatency": [], "outcomes": Counter()}) + last_completion = 0 + run_ids = set() + with (directory / "result.csv").open(newline="", encoding="utf-8") as stream: + for row in csv.DictReader(stream): + elapsed = int(row["elapsed"]) + last_completion = max(last_completion, elapsed) + run_ids.add(row["run"]) + if row["ttype"] not in TRANSACTIONS: + raise ValueError("unknown transaction type: " + row["ttype"]) + if not start_ms <= elapsed < end_ms: + continue + error, rollback = int(row["error"]), int(row["rbk"]) + latency, dblatency = int(row["latency"]), int(row["dblatency"]) + if error not in (0, 1) or rollback not in (0, 1) or min(latency, dblatency) < 0: + raise ValueError("invalid transaction result") + bucket = buckets[row["ttype"]] + bucket["latency"].append(latency) + bucket["dblatency"].append(dblatency) + # A rollback failure is an error, even for an intentional rollback. + bucket["outcomes"]["errors" if error else "intentional_rollbacks" if rollback else "successful"] += 1 + if len(run_ids) != 1: + raise ValueError("result.csv must contain exactly one run") + problems = [] + if last_completion < end_ms: + problems.append("transaction results do not cover the full window") + reconnects, problem = connection_window(directory / "connections.csv", start_ms, end_ms) + if problem: + problems.append(problem) + per_type = {} + for name in TRANSACTIONS: + bucket = buckets[name] + per_type[name] = { + "attempts": len(bucket["latency"]), + "successful": bucket["outcomes"]["successful"], + "errors": bucket["outcomes"]["errors"], + "intentional_rollbacks": bucket["outcomes"]["intentional_rollbacks"], + "p95_ms": percentile95(bucket["latency"]), + "db_p95_ms": percentile95(bucket["dblatency"]), + } + successful = sum(per_type[name]["successful"] for name in TRANSACTIONS if name != "DELIVERY_BG") + return { + "run": next(iter(run_ids)), + "window": {"warmup_seconds": warmup_seconds, "measurement_seconds": measurement_seconds, + "selection": "completion time in [warmup, warmup + measurement)"}, + "measurement_valid": not problems, + "measurement_problems": problems, + "tpmC": per_type["NEW_ORDER"]["successful"] * 60 / measurement_seconds, + "successful_transactions_per_second": successful / measurement_seconds, + "errors": sum(row["errors"] for row in per_type.values()), + "intentional_rollbacks": sum(row["intentional_rollbacks"] for row in per_type.values()), + "reconnects": reconnects, + "reconnect_definition": "terminal sockets completing MySQL authentication during the window; excludes failed authentication and background discovery sockets", + "transactions": per_type, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("data_directory", type=Path) + parser.add_argument("--warmup-seconds", type=int, required=True) + parser.add_argument("--measurement-seconds", type=int, required=True) + args = parser.parse_args() + result = summarize(args.data_directory, args.warmup_seconds, args.measurement_seconds) + print(json.dumps(result, indent=2, allow_nan=False)) + return 0 if result["measurement_valid"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/client/CommitException.java b/src/client/CommitException.java deleted file mode 100644 index 317484c..0000000 --- a/src/client/CommitException.java +++ /dev/null @@ -1,9 +0,0 @@ -public class CommitException extends RuntimeException { - - private static final long serialVersionUID = 2135244094396431474L; - - @Override - public synchronized Throwable fillInStackTrace() { - return this; - } -} diff --git a/src/client/jTPCC.java b/src/client/jTPCC.java index 5e7dc9c..9e409d3 100644 --- a/src/client/jTPCC.java +++ b/src/client/jTPCC.java @@ -46,6 +46,7 @@ public class jTPCC implements jTPCCConfig private double tpmC; private jTPCCRandom rnd; private OSCollector osCollector = null; + private boolean measureConnections; private HashMap costPerWorkerload; public static void main(String args[]) @@ -112,6 +113,7 @@ public jTPCC() log.info("Term-00, "); String resultDirectory = getProp(ini, "resultDirectory"); + measureConnections = Boolean.parseBoolean(ini.getProperty("measureConnections", "false")); String osCollectorScript = getProp(ini, "osCollectorScript"); log.info("Term-00, "); @@ -129,6 +131,8 @@ else if (iDB.equals("mysql")) log.error("unknown database type '" + iDB + "'"); return; } + if (measureConnections && (dbType != DB_MYSQL || resultDirectory == null)) + throw new IllegalArgumentException("measureConnections requires db=mysql and resultDirectory"); if(Integer.parseInt(limPerMin) !=0){ limPerMin_Terminal = Integer.parseInt(limPerMin)/Integer.parseInt(iTerminals); @@ -281,6 +285,10 @@ else if (iDB.equals("mysql")) Properties dbProps = new Properties(); dbProps.setProperty("user", iUser); dbProps.setProperty("password", iPassword); + if (measureConnections) { + jTPCCSocketFactory.open(new File(resultDirName, "data/connections.csv")); + dbProps.setProperty("socketFactory", jTPCCSocketFactory.class.getName()); + } /* * Fine tuning of database conneciton parameters if needed. @@ -476,7 +484,9 @@ else if(newOrderWeightValue == 0 && paymentWeightValue == 0 && orderStatusWeight String terminalName = "Term-" + (i>=9 ? ""+(i+1) : "0"+(i+1)); Connection conn = null; printMessage("Creating database connection for " + terminalName + "..."); + long previousHandshakes = measureConnections ? jTPCCSocketFactory.count() : 0; conn = DriverManager.getConnection(database, dbProps); + if (measureConnections) jTPCCSocketFactory.verify(conn, previousHandshakes); conn.setAutoCommit(false); jTPCCTerminal terminal = new jTPCCTerminal @@ -512,6 +522,7 @@ else if(newOrderWeightValue == 0 && paymentWeightValue == 0 && orderStatusWeight // Create Terminals, Start Transactions sessionStart = getCurrentTime(); sessionStartTimestamp = System.currentTimeMillis(); + if (measureConnections) jTPCCSocketFactory.start(sessionStartTimestamp); sessionNextTimestamp = sessionStartTimestamp; if(sessionEndTargetTime != -1) sessionEndTargetTime += sessionStartTimestamp; @@ -546,7 +557,7 @@ else if(newOrderWeightValue == 0 && paymentWeightValue == 0 && orderStatusWeight printMessage("Starting all terminals..."); transactionCount = 1; for(int i = 0; i < terminals.length; i++) - (new Thread(terminals[i])).start(); + (new Thread(terminals[i], "benchmarksql-" + terminalNames[i])).start(); } @@ -612,6 +623,13 @@ public void signalTerminalEnded(jTPCCTerminal terminal, long countNewOrdersExecu { sessionEnd = getCurrentTime(); sessionEndTimestamp = System.currentTimeMillis(); + if (measureConnections) { + try { + jTPCCSocketFactory.finish(sessionEndTimestamp); + } catch (IOException e) { + log.error("Connection measurement could not be completed", e); + } + } sessionEndTargetTime = -1; printMessage("All terminals finished executing " + sessionEnd); endReport(); @@ -673,6 +691,8 @@ public void resultAppend(jTPCCTData term) { resultCSV.write(runID + "," + term.resultLine(sessionStartTimestamp)); + // Fatal transaction paths call System.exit; retain their final row. + if (term.hasError()) resultCSV.flush(); } catch (IOException e) { diff --git a/src/client/jTPCCConnection.java b/src/client/jTPCCConnection.java index bb36623..77d7b79 100644 --- a/src/client/jTPCCConnection.java +++ b/src/client/jTPCCConnection.java @@ -306,11 +306,7 @@ public jTPCCConnection(String connURL, Properties connProps, int dbType) public void commit() throws SQLException { - try { - dbConn.commit(); - } catch(SQLException e) { - throw new CommitException(); - } + dbConn.commit(); } public void rollback() diff --git a/src/client/jTPCCSocketFactory.java b/src/client/jTPCCSocketFactory.java new file mode 100644 index 0000000..13fe0ea --- /dev/null +++ b/src/client/jTPCCSocketFactory.java @@ -0,0 +1,74 @@ +import com.mysql.cj.jdbc.JdbcConnection; +import com.mysql.cj.protocol.StandardSocketFactory; +import java.io.BufferedWriter; +import java.io.File; +import java.io.IOException; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.sql.Connection; +import java.sql.SQLException; + +/** Records successful physical MySQL authentication, without issuing measurement SQL. */ +public class jTPCCSocketFactory extends StandardSocketFactory { + private static BufferedWriter events; + private static long handshakes; + private Socket recordedSocket; + + static synchronized void open(File file) throws IOException { + if (events != null) throw new IllegalStateException("connection measurement already open"); + events = Files.newBufferedWriter(file.toPath(), StandardCharsets.UTF_8); + handshakes = 0; + events.write("event,epoch_ms,handshakes,local_port,remote_address,terminal\n"); + events.flush(); + } + + static synchronized long count() { + return handshakes; + } + + static void verify(Connection connection, long previousCount) throws SQLException { + JdbcConnection mysql = connection.unwrap(JdbcConnection.class); + if (!jTPCCSocketFactory.class.getName().equals( + mysql.getPropertySet().getStringProperty("socketFactory").getValue()) + || count() <= previousCount) { + throw new SQLException("Connection measurement did not observe this terminal's authentication"); + } + } + + static synchronized void start(long timestamp) throws IOException { + write("start", timestamp, 0, ""); + } + + static synchronized void finish(long timestamp) throws IOException { + write("end", timestamp, 0, ""); + events.close(); + events = null; + } + + private static void write(String event, long timestamp, int localPort, String remote) throws IOException { + if (events == null) throw new IOException("Connection measurement output is unavailable"); + String thread = Thread.currentThread().getName(); + // TiDB's load-balancing driver also opens discovery sockets on its + // background executor. Preserve those events without counting them as + // terminal reconnects. Workload JDBC calls run on the named terminals. + String terminal = thread.startsWith("benchmarksql-Term-") ? thread : ""; + events.write(event + "," + timestamp + "," + handshakes + "," + localPort + "," + remote + "," + terminal + "\n"); + events.flush(); + } + + @Override + public void afterHandshake() throws IOException { + super.afterHandshake(); + // Connector/J also invokes this callback for changeUser on an existing + // socket. Count a socket once; a later successful physical reconnect + // has a new socket even when the server reuses its connection ID. + if (recordedSocket == rawSocket) return; + synchronized (jTPCCSocketFactory.class) { + handshakes++; + write("authenticated", System.currentTimeMillis(), rawSocket.getLocalPort(), + rawSocket.getInetAddress().getHostAddress() + ":" + rawSocket.getPort()); + } + recordedSocket = rawSocket; + } +} diff --git a/src/client/jTPCCTData.java b/src/client/jTPCCTData.java index 74872cb..741729f 100644 --- a/src/client/jTPCCTData.java +++ b/src/client/jTPCCTData.java @@ -89,6 +89,8 @@ public void execute(Logger log, jTPCCConnection db) if (transDue == 0) transDue = transStart; + try + { switch (transType) { case TT_NEW_ORDER: @@ -119,7 +121,27 @@ public void execute(Logger log, jTPCCConnection db) throw new Exception("Unknown transType " + transType); } - transEnd = System.currentTimeMillis(); + } + catch (Exception e) + { + if (transError == null) + transError = e.toString(); + throw e; + } + finally + { + transEnd = System.currentTimeMillis(); + } + } + + public boolean isCommittedNewOrder() + { + return transType == TT_NEW_ORDER && transEnd != 0 && !transRbk && transError == null; + } + + public boolean hasError() + { + return transError != null; } public void traceScreen(Logger log) @@ -568,6 +590,7 @@ else if (newOrder.ol_supply_w_id[ol_seq[y]] == newOrder.ol_supply_w_id[ol_seq[x] } catch (SQLException se) { + transError = "SQLSTATE=" + se.getSQLState() + " code=" + se.getErrorCode() + ": " + se.getMessage(); log.error("Unexpected SQLException in NEW_ORDER"); for (SQLException x = se; x != null; x = x.getNextException()) log.error(x.getMessage()); @@ -935,6 +958,7 @@ private void executePayment(Logger log, jTPCCConnection db) } catch (SQLException se) { + transError = "SQLSTATE=" + se.getSQLState() + " code=" + se.getErrorCode() + ": " + se.getMessage(); log.error("Unexpected SQLException in PAYMENT"); for (SQLException x = se; x != null; x = x.getNextException()) log.error(x.getMessage()); @@ -1232,6 +1256,7 @@ private void executeOrderStatus(Logger log, jTPCCConnection db) } catch (SQLException se) { + transError = "SQLSTATE=" + se.getSQLState() + " code=" + se.getErrorCode() + ": " + se.getMessage(); log.error("Unexpected SQLException in ORDER_STATUS"); for (SQLException x = se; x != null; x = x.getNextException()) log.error(x.getMessage()); @@ -1388,6 +1413,7 @@ private void executeStockLevel(Logger log, jTPCCConnection db) } catch (SQLException se) { + transError = "SQLSTATE=" + se.getSQLState() + " code=" + se.getErrorCode() + ": " + se.getMessage(); log.error("Unexpected SQLException in STOCK_LEVEL"); for (SQLException x = se; x != null; x = x.getNextException()) log.error(x.getMessage()); @@ -1731,6 +1757,7 @@ private void executeDeliveryBG(Logger log, jTPCCConnection db) } catch (SQLException se) { + transError = "SQLSTATE=" + se.getSQLState() + " code=" + se.getErrorCode() + ": " + se.getMessage(); log.error("Unexpected SQLException in DELIVERY_BG"); for (SQLException x = se; x != null; x = x.getNextException()) log.error(x.getMessage()); diff --git a/src/client/jTPCCTerminal.java b/src/client/jTPCCTerminal.java index e75e4c7..b51758e 100644 --- a/src/client/jTPCCTerminal.java +++ b/src/client/jTPCCTerminal.java @@ -157,14 +157,9 @@ private void executeTransactions(int numTransactions) { term.generatePayment(log, rnd, 0); term.traceScreen(log); - term.execute(log, db); - parent.resultAppend(term); + executeTransaction(term); term.traceScreen(log); } - catch (CommitException e) - { - continue; - } catch (Exception e) { log.fatal(e.getMessage()); @@ -183,14 +178,9 @@ else if(transactionType <= paymentWeight + stockLevelWeight) { term.generateStockLevel(log, rnd, 0); term.traceScreen(log); - term.execute(log, db); - parent.resultAppend(term); + executeTransaction(term); term.traceScreen(log); } - catch (CommitException e) - { - continue; - } catch (Exception e) { log.fatal(e.getMessage()); @@ -209,14 +199,9 @@ else if(transactionType <= paymentWeight + stockLevelWeight + orderStatusWeight) { term.generateOrderStatus(log, rnd, 0); term.traceScreen(log); - term.execute(log, db); - parent.resultAppend(term); + executeTransaction(term); term.traceScreen(log); } - catch (CommitException e) - { - continue; - } catch (Exception e) { log.fatal(e.getMessage()); @@ -235,8 +220,7 @@ else if(transactionType <= paymentWeight + stockLevelWeight + orderStatusWeight { term.generateDelivery(log, rnd, 0); term.traceScreen(log); - term.execute(log, db); - parent.resultAppend(term); + executeTransaction(term); term.traceScreen(log); /* @@ -246,16 +230,11 @@ else if(transactionType <= paymentWeight + stockLevelWeight + orderStatusWeight */ jTPCCTData bg = term.getDeliveryBG(); bg.traceScreen(log); - bg.execute(log, db); - parent.resultAppend(bg); + executeTransaction(bg); bg.traceScreen(log); skippedDeliveries = bg.getSkippedDeliveries(); } - catch (CommitException e) - { - continue; - } catch (Exception e) { log.fatal(e.getMessage()); @@ -274,14 +253,9 @@ else if(transactionType <= paymentWeight + stockLevelWeight + orderStatusWeight { term.generateNewOrder(log, rnd, 0); term.traceScreen(log); - term.execute(log, db); - parent.resultAppend(term); + executeTransaction(term); term.traceScreen(log); } - catch (CommitException e) - { - continue; - } catch (Exception e) { log.fatal(e.getMessage()); @@ -289,8 +263,11 @@ else if(transactionType <= paymentWeight + stockLevelWeight + orderStatusWeight System.exit(4); } transactionTypeName = "New-Order"; - newOrderCounter++; - newOrder = 1; + if (term.isCommittedNewOrder()) + { + newOrderCounter++; + newOrder = 1; + } } long transactionEnd = System.currentTimeMillis(); @@ -322,6 +299,19 @@ else if(transactionType <= paymentWeight + stockLevelWeight + orderStatusWeight } + // Record every execution attempt, including failures that terminate a terminal. + private void executeTransaction(jTPCCTData term) throws Exception + { + try + { + term.execute(log, db); + } + finally + { + parent.resultAppend(term); + } + } + private void error(String type) { log.error(terminalName + ", TERMINAL=" + terminalName + " TYPE=" + type + " COUNT=" + transactionCount); System.out.println(terminalName + ", TERMINAL=" + terminalName + " TYPE=" + type + " COUNT=" + transactionCount); diff --git a/tests/ConnectionMeasurementTest.java b/tests/ConnectionMeasurementTest.java new file mode 100644 index 0000000..441c5d2 --- /dev/null +++ b/tests/ConnectionMeasurementTest.java @@ -0,0 +1,111 @@ +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.util.List; +import java.util.concurrent.FutureTask; + +/** Drives Connector/J's real authentication callback over a loopback MySQL socket. */ +public class ConnectionMeasurementTest { + private static void packet(OutputStream stream, int sequence, byte[] payload) throws Exception { + stream.write(payload.length); + stream.write(payload.length >> 8); + stream.write(payload.length >> 16); + stream.write(sequence); + stream.write(payload); + stream.flush(); + } + + private static void readPacket(InputStream stream) throws Exception { + int length = stream.read(); + if (length < 0) throw new EOFException(); + length |= stream.read() << 8; + length |= stream.read() << 16; + if (stream.read() < 0) throw new EOFException(); + for (int i = 0; i < length; i++) if (stream.read() < 0) throw new EOFException(); + } + + private static byte[] greeting() throws Exception { + ByteArrayOutputStream data = new ByteArrayOutputStream(); + data.write(10); + data.write("8.0.33\0".getBytes(StandardCharsets.US_ASCII)); + data.write(new byte[] {1, 0, 0, 0}); + data.write("12345678\0".getBytes(StandardCharsets.US_ASCII)); + // PROTOCOL_41, SECURE_CONNECTION, PLUGIN_AUTH and LONG_PASSWORD. + data.write(new byte[] {1, (byte) 0x82, 45, 2, 0, 8, 0, 21}); + data.write(new byte[10]); + data.write("abcdefghijkl\0mysql_native_password\0".getBytes(StandardCharsets.US_ASCII)); + return data.toByteArray(); + } + + private static void connect(boolean authenticate) throws Exception { + try (ServerSocket listener = new ServerSocket(0, 1, java.net.InetAddress.getLoopbackAddress())) { + FutureTask server = new FutureTask<>(() -> { + try (Socket socket = listener.accept()) { + socket.setSoTimeout(5000); + InputStream input = socket.getInputStream(); + OutputStream output = socket.getOutputStream(); + packet(output, 0, greeting()); + readPacket(input); + if (authenticate) { + packet(output, 2, new byte[] {0, 0, 0, 2, 0, 0, 0}); + readPacket(input); + // Authentication has completed, but stop JDBC setup. + packet(output, 1, "\377\323\004#HY000fixture stops after authentication".getBytes(StandardCharsets.ISO_8859_1)); + } else { + packet(output, 2, "\377\025\004#28000authentication denied".getBytes(StandardCharsets.ISO_8859_1)); + } + } + return null; + }); + Thread thread = new Thread(server, "mysql-handshake-fixture"); + thread.setDaemon(true); + thread.start(); + try { + DriverManager.getConnection("jdbc:mysql://127.0.0.1:" + listener.getLocalPort() + + "/?useSSL=false&connectTimeout=5000&socketTimeout=5000&socketFactory=jTPCCSocketFactory", "root", ""); + throw new AssertionError("fixture must stop before JDBC setup completes"); + } catch (SQLException expected) { + if (expected.getErrorCode() != (authenticate ? 1235 : 1045)) throw expected; + } + server.get(10, java.util.concurrent.TimeUnit.SECONDS); + } + } + + public static void main(String[] args) throws Exception { + Class.forName("com.mysql.cj.jdbc.Driver"); + Path output = Files.createTempFile("benchmarksql-connections-", ".csv"); + try { + jTPCCSocketFactory.open(output.toFile()); + connect(false); + if (jTPCCSocketFactory.count() != 0) throw new AssertionError("failed auth counted"); + connect(true); + if (jTPCCSocketFactory.count() != 1) throw new AssertionError("initial auth missing"); + jTPCCSocketFactory.start(System.currentTimeMillis()); + String originalName = Thread.currentThread().getName(); + try { + Thread.currentThread().setName("benchmarksql-Term-01"); + connect(true); + } finally { + Thread.currentThread().setName(originalName); + } + if (jTPCCSocketFactory.count() != 2) throw new AssertionError("new physical auth missing"); + jTPCCSocketFactory.finish(System.currentTimeMillis()); + List rows = Files.readAllLines(output, StandardCharsets.UTF_8); + if (rows.size() != 5 || !rows.get(2).startsWith("start,") || !rows.get(4).startsWith("end,")) + throw new AssertionError(rows); + if (!rows.get(3).endsWith(",benchmarksql-Term-01") || !rows.get(1).endsWith(",")) + throw new AssertionError("terminal and background sockets must be distinguishable"); + System.out.println("Connection measurement: real successful and failed authentication boundaries passed"); + } finally { + Files.deleteIfExists(output); + } + } +} diff --git a/tests/TransactionMeasurementTest.java b/tests/TransactionMeasurementTest.java new file mode 100644 index 0000000..b95e17c --- /dev/null +++ b/tests/TransactionMeasurementTest.java @@ -0,0 +1,141 @@ +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Proxy; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import org.apache.log4j.Logger; + +/** Exercises transaction result rows against JDBC success and failure boundaries. */ +public class TransactionMeasurementTest { + private static final Logger LOG = Logger.getLogger(TransactionMeasurementTest.class); + + private static T jdbc(Class type, InvocationHandler handler) { + return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class[] {type}, handler)); + } + + private static final class Database { + final SQLException failure = new SQLException("injected failure", "40001", 1213); + boolean failStatement; + boolean failCommit; + boolean failRollback; + int commits; + int rollbacks; + + Connection connection() { + return jdbc(Connection.class, (proxy, method, args) -> { + switch (method.getName()) { + case "prepareStatement": return statement(); + case "commit": + if (failCommit) throw failure; + commits++; + return null; + case "rollback": + if (failRollback) throw new SQLException("lost connection", "08006"); + rollbacks++; + return null; + default: throw new AssertionError(method); + } + }); + } + + PreparedStatement statement() { + return jdbc(PreparedStatement.class, (proxy, method, args) -> { + String name = method.getName(); + if (name.startsWith("set") || name.equals("addBatch") || name.equals("clearBatch")) return null; + if (failStatement && name.startsWith("execute")) throw failure; + if (name.equals("executeUpdate")) return 1; + if (name.equals("executeBatch")) return new int[] {1}; + if (name.equals("executeQuery")) { + boolean[] returned = {false}; + return jdbc(ResultSet.class, (result, getter, columns) -> { + switch (getter.getName()) { + case "next": + if (returned[0]) return false; + returned[0] = true; + return true; + case "getInt": return 1; + case "getDouble": return 1.0; + case "getString": return "GC"; + case "getTimestamp": return new Timestamp(1); + case "close": return null; + default: throw new AssertionError(getter); + } + }); + } + throw new AssertionError(method); + }); + } + } + + private static final class Input extends jTPCCRandom { + boolean intentionalRollback; + + @Override + public int nextInt(int low, int high) { + if (low == 1 && high == 100) return intentionalRollback ? 1 : 50; + return low; + } + + @Override + public int getItemID() { return 1; } + } + + private static String[] executeNewOrder(Input input, Database database, boolean fatal) throws Exception { + jTPCCTData transaction = new jTPCCTData(); + transaction.setNumWarehouses(1); + transaction.setWarehouse(1); + transaction.setDistrict(1); + transaction.generateNewOrder(LOG, input, 0); + long start = System.currentTimeMillis(); + try { + transaction.execute(LOG, new jTPCCConnection(database.connection(), jTPCCConfig.DB_MYSQL)); + if (fatal) throw new AssertionError("rollback failure must propagate"); + } catch (Exception error) { + if (!fatal) throw error; + } + String[] row = transaction.resultLine(start).trim().split(","); + if (row.length != 7 || Long.parseLong(row[0]) < 0 || Long.parseLong(row[2]) < 0) { + throw new AssertionError("every attempt needs completed timing and one full CSV row"); + } + if (!row[3].equals("NEW_ORDER")) throw new AssertionError("transaction type lost"); + return row; + } + + private static void outcome(String[] row, String rollback, String error) { + if (!row[4].equals(rollback) || !row[6].equals(error)) { + throw new AssertionError("wrong rollback/error classification: " + String.join(",", row)); + } + } + + public static void main(String[] args) throws Exception { + Input input = new Input(); + outcome(executeNewOrder(input, new Database(), false), "0", "0"); + input.intentionalRollback = true; + outcome(executeNewOrder(input, new Database(), false), "1", "0"); + input.intentionalRollback = false; + + Database statementFailure = new Database(); + statementFailure.failStatement = true; + outcome(executeNewOrder(input, statementFailure, false), "0", "1"); + + Database commitFailure = new Database(); + commitFailure.failCommit = true; + outcome(executeNewOrder(input, commitFailure, false), "0", "1"); + + Database lostConnection = new Database(); + lostConnection.failStatement = true; + lostConnection.failRollback = true; + outcome(executeNewOrder(input, lostConnection, true), "0", "1"); + + jTPCCConnection connection = new jTPCCConnection(commitFailure.connection(), jTPCCConfig.DB_MYSQL); + try { + connection.commit(); + throw new AssertionError("commit failure was suppressed"); + } catch (SQLException error) { + if (error != commitFailure.failure) throw new AssertionError("JDBC cause was replaced"); + } + System.out.println("Transaction measurement tests passed"); + } +} diff --git a/tests/test_measurement_window.py b/tests/test_measurement_window.py new file mode 100644 index 0000000..a928921 --- /dev/null +++ b/tests/test_measurement_window.py @@ -0,0 +1,59 @@ +import csv +import importlib.util +from pathlib import Path +import tempfile +import unittest + +spec = importlib.util.spec_from_file_location("summary", Path(__file__).resolve().parents[1] / "run/summarizeRun.py") +summary = importlib.util.module_from_spec(spec) +spec.loader.exec_module(summary) + + +class MeasurementWindowTest(unittest.TestCase): + def test_window_percentiles_and_outcomes(self): + with tempfile.TemporaryDirectory() as temporary: + directory = Path(temporary) + with (directory / "result.csv").open("w", newline="") as stream: + writer = csv.writer(stream) + writer.writerow(["run", "elapsed", "latency", "dblatency", "ttype", "rbk", "dskipped", "error"]) + writer.writerow([1, 119999, 999, 999, "NEW_ORDER", 0, 0, 0]) + for i in range(20): + writer.writerow([1, 120000 + i, i + 1, i + 1, "NEW_ORDER", 0, 0, 0]) + writer.writerow([1, 130000, 1, 1, "NEW_ORDER", 1, 0, 0]) + writer.writerow([1, 140000, 1, 1, "NEW_ORDER", 1, 0, 1]) + writer.writerow([1, 419999, 5, 5, "PAYMENT", 0, 0, 1]) + writer.writerow([1, 420000, 999, 999, "NEW_ORDER", 0, 0, 0]) + (directory / "connections.csv").write_text( + "event,epoch_ms,handshakes,local_port,remote_address,terminal\n" + "authenticated,999,1,1,127.0.0.1:4000,\n" + "start,1000,1,0,,\n" + "authenticated,120999,2,2,127.0.0.1:4000,benchmarksql-Term-01\n" + "authenticated,121000,3,3,127.0.0.1:4000,benchmarksql-Term-01\n" + "authenticated,200000,4,4,127.0.0.1:4000,\n" + "authenticated,421000,5,5,127.0.0.1:4000,benchmarksql-Term-01\n" + "end,421001,5,0,,\n", encoding="utf-8") + result = summary.summarize(directory, 120, 300) + self.assertTrue(result["measurement_valid"]) + self.assertEqual(result["tpmC"], 4) + self.assertEqual(result["errors"], 2) + self.assertEqual(result["intentional_rollbacks"], 1) + self.assertEqual(result["reconnects"], 1) + self.assertEqual(result["transactions"]["NEW_ORDER"]["p95_ms"], 19) + self.assertIsNone(result["transactions"]["DELIVERY_BG"]["p95_ms"]) + (directory / "connections.csv").unlink() + unmeasured = summary.summarize(directory, 120, 300) + self.assertFalse(unmeasured["measurement_valid"]) + self.assertIsNone(unmeasured["reconnects"]) + + def test_truncated_connection_measurement_is_not_zero(self): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) / "connections.csv" + path.write_text("event,epoch_ms,handshakes,local_port,remote_address,terminal\n" + "authenticated,1,1,1,127.0.0.1:4000,\nstart,2,1,0,,\n", encoding="utf-8") + count, problem = summary.connection_window(path, 120000, 420000) + self.assertIsNone(count) + self.assertIn("full window", problem) + + +if __name__ == "__main__": + unittest.main()