fix: record transaction outcomes and measured connection events

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 <resultDirectory>/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
 ---------------------------------------------------
 

build.xml

@@ -28,7 +28,7 @@
 
   <target name="dist" depends="compile" >
     <mkdir dir="${dist}"/>
-    <jar jarfile="${dist}/BenchmarkSQL-5.0.jar" basedir="${build}"/>
+    <jar jarfile="${dist}/BenchmarkSQL-5.0.jar" basedir="${build}" excludes="tests/**"/>
   </target>
 
   <target name="clean" description="clean up" >
@@ -37,4 +37,17 @@
     <delete dir="${logs}"/>
   </target>
 
+  <target name="test" depends="compile">
+    <mkdir dir="${build}/tests"/>
+    <javac srcdir="tests" destdir="${build}/tests" includeantruntime="false" encoding="UTF-8">
+      <classpath><pathelement location="${build}"/><path refid="classpath"/></classpath>
+    </javac>
+    <java classname="TransactionMeasurementTest" fork="true" failonerror="true">
+      <classpath><pathelement location="${build}"/><pathelement location="${build}/tests"/><path refid="classpath"/></classpath>
+    </java>
+    <java classname="ConnectionMeasurementTest" fork="true" failonerror="true">
+      <classpath><pathelement location="${build}"/><pathelement location="${build}/tests"/><path refid="classpath"/></classpath>
+    </java>
+  </target>
+
 </project>

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())

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<String, Long> 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)
 	    {

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()

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;
+    }
+}