benchmark/benchmark-core.js - iotdb-client-nodejs - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 /**
  * Benchmark Core Engine
  *
  * Core benchmark engine for managing test execution, client pools,
  * and performance metrics collection. Supports concurrent operations
  * with detailed latency and throughput tracking.
  */

 const { performance } = require("perf_hooks");

 /**
  * Performance metrics collector
  */
 class MetricsCollector {
   constructor(config) {
     this.config = config;
     this.reset();
   }

   reset() {
     this.operations = [];
     this.errors = [];
     this.startTime = null;
     this.endTime = null;
     this.operationCount = 0;
     this.successCount = 0;
     this.failureCount = 0;
     this.totalDataPoints = 0;
   }

   start() {
     this.startTime = performance.now();
   }

   recordOperation(latencyMs, dataPoints, success = true, error = null) {
     this.operationCount++;
     this.operations.push(latencyMs);

     if (success) {
       this.successCount++;
       this.totalDataPoints += dataPoints;
     } else {
       this.failureCount++;
       if (error) {
         this.errors.push(error);
       }
     }
   }

   end() {
     this.endTime = performance.now();
   }

   getDuration() {
     return this.endTime - this.startTime;
   }

   /**
    * Calculate percentile
    * @param {number} percentile - Percentile (0-100)
    * @returns {number} Percentile value
    */
   getPercentile(percentile) {
     if (this.operations.length === 0) return 0;

     const sorted = [...this.operations].sort((a, b) => a - b);
     const index = Math.ceil((percentile / 100) * sorted.length) - 1;
     return sorted[Math.max(0, index)];
   }

   /**
    * Get comprehensive statistics
    * @returns {Object} Statistics object
    */
   getStats() {
     const durationMs = this.getDuration();
     const durationSec = durationMs / 1000;

     const stats = {
       // Time
       duration_ms: durationMs.toFixed(2),
       duration_sec: durationSec.toFixed(2),

       // Operations
       total_operations: this.operationCount,
       successful_operations: this.successCount,
       failed_operations: this.failureCount,
       success_rate:
         ((this.successCount / this.operationCount) * 100).toFixed(2) + "%",

       // Data points
       total_data_points: this.totalDataPoints,

       // Throughput
       operations_per_sec: (this.operationCount / durationSec).toFixed(2),
       points_per_sec: (this.totalDataPoints / durationSec).toFixed(2),

       // Latency (ms)
       latency: {
         min:
           this.operations.length > 0
             ? Math.min(...this.operations).toFixed(2)
             : 0,
         max:
           this.operations.length > 0
             ? Math.max(...this.operations).toFixed(2)
             : 0,
         avg:
           this.operations.length > 0
             ? (
                 this.operations.reduce((a, b) => a + b, 0) /
                 this.operations.length
               ).toFixed(2)
             : 0,
       },
     };

     // Add percentiles if enabled
     if (this.config.ENABLE_DETAILED_METRICS && this.operations.length > 0) {
       stats.latency.p50 = this.getPercentile(50).toFixed(2);
       stats.latency.p90 = this.getPercentile(90).toFixed(2);
       stats.latency.p95 = this.getPercentile(95).toFixed(2);
       stats.latency.p99 = this.getPercentile(99).toFixed(2);
     }

     return stats;
   }

   /**
    * Print statistics summary
    */
   printStats(label = "Benchmark Results") {
     const stats = this.getStats();

     console.log("\n" + "=".repeat(80));
     console.log(label.toUpperCase());
     console.log("=".repeat(80));

     console.log("\n[Execution Time]");
     console.log(
       `  Duration:              ${stats.duration_sec}s (${stats.duration_ms}ms)`,
     );

     console.log("\n[Operations]");
     console.log(`  Total Operations:      ${stats.total_operations}`);
     console.log(`  Successful:            ${stats.successful_operations}`);
     console.log(`  Failed:                ${stats.failed_operations}`);
     console.log(`  Success Rate:          ${stats.success_rate}`);

     console.log("\n[Data Points]");
     console.log(
       `  Total Points Written:  ${stats.total_data_points.toLocaleString()}`,
     );

     console.log("\n[Throughput]");
     console.log(`  Operations/sec:        ${stats.operations_per_sec}`);
     console.log(
       `  Points/sec:            ${parseFloat(stats.points_per_sec).toLocaleString()}`,
     );

     console.log("\n[Latency (ms)]");
     console.log(`  Min:                   ${stats.latency.min}ms`);
     console.log(`  Max:                   ${stats.latency.max}ms`);
     console.log(`  Average:               ${stats.latency.avg}ms`);

     if (this.config.ENABLE_DETAILED_METRICS && stats.latency.p50) {
       console.log(`  P50 (Median):          ${stats.latency.p50}ms`);
       console.log(`  P90:                   ${stats.latency.p90}ms`);
       console.log(`  P95:                   ${stats.latency.p95}ms`);
       console.log(`  P99:                   ${stats.latency.p99}ms`);
     }

     if (this.failureCount > 0 && this.errors.length > 0) {
       console.log("\n[Error Samples]");
       const sampleErrors = this.errors.slice(0, 5);
       sampleErrors.forEach((err, idx) => {
         console.log(`  ${idx + 1}. ${err.message || err}`);
       });
       if (this.errors.length > 5) {
         console.log(`  ... and ${this.errors.length - 5} more errors`);
       }
     }

     console.log("\n" + "=".repeat(80));
   }
 }

 /**
  * Progress reporter for long-running tests
  */
 class ProgressReporter {
   constructor(config, metricsCollector) {
     this.config = config;
     this.metrics = metricsCollector;
     this.lastReportTime = performance.now();
     this.lastOperationCount = 0;
     this.intervalId = null;
   }

   start() {
     if (this.config.REPORT_INTERVAL > 0) {
       this.intervalId = setInterval(() => {
         this.report();
       }, this.config.REPORT_INTERVAL);
     }
   }

   stop() {
     if (this.intervalId) {
       clearInterval(this.intervalId);
       this.intervalId = null;
     }
   }

   report() {
     const now = performance.now();
     const elapsed = now - this.lastReportTime;
     const operations = this.metrics.operationCount - this.lastOperationCount;
     const opsPerSec = (operations / elapsed) * 1000;
     const totalPoints = this.metrics.totalDataPoints;

     console.log(
       `[Progress] Operations: ${this.metrics.operationCount}, ` +
         `Rate: ${opsPerSec.toFixed(2)} ops/s, ` +
         `Total Points: ${totalPoints.toLocaleString()}`,
     );

     this.lastReportTime = now;
     this.lastOperationCount = this.metrics.operationCount;
   }
 }

 /**
  * Execute write operations concurrently with device-session binding
  * @param {Object} pool - IoTDB session pool
  * @param {Array} workload - Array of work items (should be ordered by device)
  * @param {number} poolSize - Session pool size
  * @param {MetricsCollector} metrics - Metrics collector
  * @param {Function} executor - Async function to execute each work item (pool, work, session)
  */
 async function executeConcurrentWithBinding(
   pool,
   workload,
   poolSize,
   metrics,
   executor,
 ) {
   // Get sessions from pool and bind to devices
   const sessions = [];
   for (let i = 0; i < poolSize; i++) {
     sessions.push(await pool.getSession());
   }

   console.log(
     `[Device-Session Binding] Bound ${sessions.length} sessions to devices`,
   );

   try {
     const workers = [];

     // Each worker handles workload items for devices bound to its session
     // Workload is pre-ordered, so we can partition it evenly
     const workPerSession = Math.ceil(workload.length / poolSize);

     const worker = async (sessionIdx) => {
       const session = sessions[sessionIdx];
       const startIdx = sessionIdx * workPerSession;
       const endIdx = Math.min(startIdx + workPerSession, workload.length);

       for (let i = startIdx; i < endIdx; i++) {
         const work = workload[i];
         const startTime = performance.now();

         try {
           const dataPoints = await executor(pool, work, session);
           const latency = performance.now() - startTime;
           metrics.recordOperation(latency, dataPoints, true);
         } catch (error) {
           const latency = performance.now() - startTime;
           metrics.recordOperation(latency, 0, false, error);
         }
       }
     };

     // Start workers, one per session
     for (let i = 0; i < poolSize; i++) {
       workers.push(worker(i));
     }

     // Wait for all workers to complete
     await Promise.all(workers);
   } finally {
     // Release all sessions back to pool
     for (const session of sessions) {
       pool.releaseSession(session);
     }
     console.log(
       `[Device-Session Binding] Released ${sessions.length} sessions`,
     );
   }
 }

 /**
  * Execute write operations concurrently with pre-acquired sessions
  * Each worker acquires a dedicated session at the start and uses it for all operations,
  * which significantly improves performance by avoiding session acquisition overhead per operation.
  *
  * @param {Object} pool - IoTDB session pool
  * @param {Array} workload - Array of work items
  * @param {number} concurrency - Number of concurrent workers
  * @param {MetricsCollector} metrics - Metrics collector
  * @param {Function} executor - Async function to execute each work item: (pool, work, session) => Promise<dataPoints>
  *                              The session parameter is the pre-acquired dedicated session for the worker.
  *                              Executors should use this session directly when provided (not null) for optimal performance.
  *                              If executor doesn't need the session, it can use pool methods which will handle session management.
  */
 async function executeConcurrent(
   pool,
   workload,
   concurrency,
   metrics,
   executor,
 ) {
   // Pre-acquire sessions for all workers to enable true concurrent execution
   const actualConcurrency = Math.min(concurrency, workload.length);

   // Acquire all sessions in parallel
   const sessions = await Promise.all(
     Array.from({ length: actualConcurrency }, () => pool.getSession()),
   );

   console.log(
     `[Concurrent Execution] Pre-acquired ${sessions.length} sessions for ${actualConcurrency} workers`,
   );

   let workIndex = 0;
   const workers = [];

   try {
     // Worker function - uses dedicated session for all its operations
     const worker = async (workerSession) => {
       while (workIndex < workload.length) {
         const index = workIndex++;
         if (index >= workload.length) break;

         const work = workload[index];
         const startTime = performance.now();

         try {
           // Pass the pre-acquired session to executor for direct use
           const dataPoints = await executor(pool, work, workerSession);
           const latency = performance.now() - startTime;
           metrics.recordOperation(latency, dataPoints, true);
         } catch (error) {
           const latency = performance.now() - startTime;
           metrics.recordOperation(latency, 0, false, error);
         }
       }
     };

     // Start workers, each with its own dedicated session
     for (let i = 0; i < actualConcurrency; i++) {
       workers.push(worker(sessions[i]));
     }

     // Wait for all workers to complete
     await Promise.all(workers);
   } finally {
     // Release all sessions back to pool
     for (const session of sessions) {
       pool.releaseSession(session);
     }
     console.log(`[Concurrent Execution] Released ${sessions.length} sessions`);
   }
 }

 /**
  * Run benchmark test
  * @param {Object} pool - IoTDB session pool
  * @param {Object} testData - Test data
  * @param {Object} config - Configuration
  * @param {Function} executor - Async function to execute work items
  * @param {Function} workloadGenerator - Function to generate workload
  * @returns {Object} Test results
  */
 async function runBenchmark(
   pool,
   testData,
   config,
   executor,
   workloadGenerator,
 ) {
   console.log("\n" + "=".repeat(80));
   console.log("STARTING BENCHMARK TEST");
   console.log("=".repeat(80));

   const metrics = new MetricsCollector(config);
   const reporter = new ProgressReporter(config, metrics);

   // Generate workload
   console.log("\nGenerating workload...");
   const workload = workloadGenerator(testData, config);
   console.log(`Workload generated: ${workload.length} operations`);

   // Warmup phase
   if (config.WARMUP_ROUNDS > 0) {
     console.log(
       `\n[Warmup Phase] Running ${config.WARMUP_ROUNDS} warmup rounds...`,
     );
     for (let round = 0; round < config.WARMUP_ROUNDS; round++) {
       const warmupMetrics = new MetricsCollector(config);
       warmupMetrics.start();

       await executeConcurrent(
         pool,
         workload.slice(0, Math.min(10, workload.length)),
         Math.min(config.CLIENT_NUMBER, 2),
         warmupMetrics,
         executor,
       );

       warmupMetrics.end();
       console.log(
         `  Warmup round ${round + 1}/${config.WARMUP_ROUNDS} completed`,
       );
     }
   }

   // Main test phase
   if (config.ENABLE_DEVICE_SESSION_BINDING) {
     console.log(
       `\n[Test Phase] Running benchmark with device-session binding (${config.POOL_MAX_SIZE} sessions)...`,
     );
   } else {
     console.log(
       `\n[Test Phase] Running benchmark with ${config.CLIENT_NUMBER} concurrent clients...`,
     );
   }

   metrics.start();
   reporter.start();

   try {
     if (config.ENABLE_DEVICE_SESSION_BINDING) {
       await executeConcurrentWithBinding(
         pool,
         workload,
         config.POOL_MAX_SIZE,
         metrics,
         executor,
       );
     } else {
       await executeConcurrent(
         pool,
         workload,
         config.CLIENT_NUMBER,
         metrics,
         executor,
       );
     }
   } finally {
     reporter.stop();
     metrics.end();
   }

   // Print results
   metrics.printStats("Benchmark Results");

   return metrics.getStats();
 }

 /**
  * Execute batch insert operations using insertTabletsParallel
  * This is optimized for high throughput by pre-acquiring sessions
  * and distributing tablets across workers.
  *
  * @param {Object} pool - IoTDB session pool
  * @param {Array} tablets - Array of tablet objects ready for insertion
  * @param {number} concurrency - Number of concurrent workers
  * @param {MetricsCollector} metrics - Metrics collector
  * @returns {Promise<void>}
  */
 async function executeBatchInsert(pool, tablets, concurrency, metrics) {
   const batchSize = Math.ceil(tablets.length / Math.ceil(tablets.length / 100)); // ~100 tablets per batch
   const batches = [];

   for (let i = 0; i < tablets.length; i += batchSize) {
     batches.push(tablets.slice(i, i + batchSize));
   }

   console.log(`[Batch Insert] ${tablets.length} tablets in ${batches.length} batches, concurrency: ${concurrency}`);

   const actualConcurrency = Math.min(concurrency, batches.length);

   // Pre-acquire sessions
   const sessions = await Promise.all(
     Array.from({ length: actualConcurrency }, () => pool.getSession()),
   );

   console.log(`[Batch Insert] Pre-acquired ${sessions.length} sessions`);

   let batchIndex = 0;

   try {
     const workers = sessions.map(async (session) => {
       while (batchIndex < batches.length) {
         const idx = batchIndex++;
         if (idx >= batches.length) break;

         const batch = batches[idx];
         const startTime = performance.now();

         try {
           // Insert all tablets in this batch sequentially using the same session
           for (const tablet of batch) {
             await session.insertTablet(tablet);
           }
           const latency = performance.now() - startTime;
           // Calculate total data points in this batch
           const dataPoints = batch.reduce((sum, t) => {
             const rows = t.timestamps.length;
             const cols = t.measurements?.length || t.columnNames?.length || 0;
             return sum + rows * cols;
           }, 0);
           metrics.recordOperation(latency, dataPoints, true);
         } catch (error) {
           const latency = performance.now() - startTime;
           metrics.recordOperation(latency, 0, false, error);
         }
       }
     });

     await Promise.all(workers);
   } finally {
     for (const session of sessions) {
       pool.releaseSession(session);
     }
     console.log(`[Batch Insert] Released ${sessions.length} sessions`);
   }
 }

 /**
  * Run benchmark using batch insert mode (optimized for high throughput)
  * @param {Object} pool - IoTDB session pool
  * @param {Array} tablets - Pre-built tablet array
  * @param {Object} config - Configuration
  * @returns {Object} Test results
  */
 async function runBatchBenchmark(pool, tablets, config) {
   console.log("\n" + "=".repeat(80));
   console.log("STARTING BATCH BENCHMARK TEST");
   console.log("=".repeat(80));

   const metrics = new MetricsCollector(config);
   const reporter = new ProgressReporter(config, metrics);

   console.log(`\nTotal tablets to insert: ${tablets.length}`);

   // Warmup phase
   if (config.WARMUP_ROUNDS > 0) {
     console.log(
       `\n[Warmup Phase] Running ${config.WARMUP_ROUNDS} warmup rounds...`,
     );
     const warmupTablets = tablets.slice(0, Math.min(20, tablets.length));
     for (let round = 0; round < config.WARMUP_ROUNDS; round++) {
       const warmupMetrics = new MetricsCollector(config);
       warmupMetrics.start();

       await executeBatchInsert(
         pool,
         warmupTablets,
         Math.min(config.CLIENT_NUMBER, 2),
         warmupMetrics,
       );

       warmupMetrics.end();
       console.log(
         `  Warmup round ${round + 1}/${config.WARMUP_ROUNDS} completed`,
       );
     }
   }

   // Main test phase
   console.log(
     `\n[Test Phase] Running batch benchmark with ${config.CLIENT_NUMBER} concurrent clients...`,
   );

   metrics.start();
   reporter.start();

   try {
     await executeBatchInsert(pool, tablets, config.CLIENT_NUMBER, metrics);
   } finally {
     reporter.stop();
     metrics.end();
   }

   // Print results
   metrics.printStats("Batch Benchmark Results");

   return metrics.getStats();
 }

 module.exports = {
   MetricsCollector,
   ProgressReporter,
   executeConcurrent,
   executeConcurrentWithBinding,
   executeBatchInsert,
   runBenchmark,
   runBatchBenchmark,
 };
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	/**
	* Benchmark Core Engine
	*
	* Core benchmark engine for managing test execution, client pools,
	* and performance metrics collection. Supports concurrent operations
	* with detailed latency and throughput tracking.
	*/

	const { performance } = require("perf_hooks");

	/**
	* Performance metrics collector
	*/
	class MetricsCollector {
	constructor(config) {
	this.config = config;
	this.reset();
	}

	reset() {
	this.operations = [];
	this.errors = [];
	this.startTime = null;
	this.endTime = null;
	this.operationCount = 0;
	this.successCount = 0;
	this.failureCount = 0;
	this.totalDataPoints = 0;
	}

	start() {
	this.startTime = performance.now();
	}

	recordOperation(latencyMs, dataPoints, success = true, error = null) {
	this.operationCount++;
	this.operations.push(latencyMs);

	if (success) {
	this.successCount++;
	this.totalDataPoints += dataPoints;
	} else {
	this.failureCount++;
	if (error) {
	this.errors.push(error);
	}
	}
	}

	end() {
	this.endTime = performance.now();
	}

	getDuration() {
	return this.endTime - this.startTime;
	}

	/**
	* Calculate percentile
	* @param {number} percentile - Percentile (0-100)
	* @returns {number} Percentile value
	*/
	getPercentile(percentile) {
	if (this.operations.length === 0) return 0;

	const sorted = [...this.operations].sort((a, b) => a - b);
	const index = Math.ceil((percentile / 100) * sorted.length) - 1;
	return sorted[Math.max(0, index)];
	}

	/**
	* Get comprehensive statistics
	* @returns {Object} Statistics object
	*/
	getStats() {
	const durationMs = this.getDuration();
	const durationSec = durationMs / 1000;

	const stats = {
	// Time
	duration_ms: durationMs.toFixed(2),
	duration_sec: durationSec.toFixed(2),

	// Operations
	total_operations: this.operationCount,
	successful_operations: this.successCount,
	failed_operations: this.failureCount,
	success_rate:
	((this.successCount / this.operationCount) * 100).toFixed(2) + "%",

	// Data points
	total_data_points: this.totalDataPoints,

	// Throughput
	operations_per_sec: (this.operationCount / durationSec).toFixed(2),
	points_per_sec: (this.totalDataPoints / durationSec).toFixed(2),

	// Latency (ms)
	latency: {
	min:
	this.operations.length > 0
	? Math.min(...this.operations).toFixed(2)
	: 0,
	max:
	this.operations.length > 0
	? Math.max(...this.operations).toFixed(2)
	: 0,
	avg:
	this.operations.length > 0
	? (
	this.operations.reduce((a, b) => a + b, 0) /
	this.operations.length
	).toFixed(2)
	: 0,
	},
	};

	// Add percentiles if enabled
	if (this.config.ENABLE_DETAILED_METRICS && this.operations.length > 0) {
	stats.latency.p50 = this.getPercentile(50).toFixed(2);
	stats.latency.p90 = this.getPercentile(90).toFixed(2);
	stats.latency.p95 = this.getPercentile(95).toFixed(2);
	stats.latency.p99 = this.getPercentile(99).toFixed(2);
	}

	return stats;
	}

	/**
	* Print statistics summary
	*/
	printStats(label = "Benchmark Results") {
	const stats = this.getStats();

	console.log("\n" + "=".repeat(80));
	console.log(label.toUpperCase());
	console.log("=".repeat(80));

	console.log("\n[Execution Time]");
	console.log(
	` Duration: ${stats.duration_sec}s (${stats.duration_ms}ms)`,
	);

	console.log("\n[Operations]");
	console.log(` Total Operations: ${stats.total_operations}`);
	console.log(` Successful: ${stats.successful_operations}`);
	console.log(` Failed: ${stats.failed_operations}`);
	console.log(` Success Rate: ${stats.success_rate}`);

	console.log("\n[Data Points]");
	console.log(
	` Total Points Written: ${stats.total_data_points.toLocaleString()}`,
	);

	console.log("\n[Throughput]");
	console.log(` Operations/sec: ${stats.operations_per_sec}`);
	console.log(
	` Points/sec: ${parseFloat(stats.points_per_sec).toLocaleString()}`,
	);

	console.log("\n[Latency (ms)]");
	console.log(` Min: ${stats.latency.min}ms`);
	console.log(` Max: ${stats.latency.max}ms`);
	console.log(` Average: ${stats.latency.avg}ms`);

	if (this.config.ENABLE_DETAILED_METRICS && stats.latency.p50) {
	console.log(` P50 (Median): ${stats.latency.p50}ms`);
	console.log(` P90: ${stats.latency.p90}ms`);
	console.log(` P95: ${stats.latency.p95}ms`);
	console.log(` P99: ${stats.latency.p99}ms`);
	}

	if (this.failureCount > 0 && this.errors.length > 0) {
	console.log("\n[Error Samples]");
	const sampleErrors = this.errors.slice(0, 5);
	sampleErrors.forEach((err, idx) => {
	console.log(` ${idx + 1}. ${err.message \|\| err}`);
	});
	if (this.errors.length > 5) {
	console.log(` ... and ${this.errors.length - 5} more errors`);
	}
	}

	console.log("\n" + "=".repeat(80));
	}
	}

	/**
	* Progress reporter for long-running tests
	*/
	class ProgressReporter {
	constructor(config, metricsCollector) {
	this.config = config;
	this.metrics = metricsCollector;
	this.lastReportTime = performance.now();
	this.lastOperationCount = 0;
	this.intervalId = null;
	}

	start() {
	if (this.config.REPORT_INTERVAL > 0) {
	this.intervalId = setInterval(() => {
	this.report();
	}, this.config.REPORT_INTERVAL);
	}
	}

	stop() {
	if (this.intervalId) {
	clearInterval(this.intervalId);
	this.intervalId = null;
	}
	}

	report() {
	const now = performance.now();
	const elapsed = now - this.lastReportTime;
	const operations = this.metrics.operationCount - this.lastOperationCount;
	const opsPerSec = (operations / elapsed) * 1000;
	const totalPoints = this.metrics.totalDataPoints;

	console.log(
	`[Progress] Operations: ${this.metrics.operationCount}, ` +
	`Rate: ${opsPerSec.toFixed(2)} ops/s, ` +
	`Total Points: ${totalPoints.toLocaleString()}`,
	);

	this.lastReportTime = now;
	this.lastOperationCount = this.metrics.operationCount;
	}
	}

	/**
	* Execute write operations concurrently with device-session binding
	* @param {Object} pool - IoTDB session pool
	* @param {Array} workload - Array of work items (should be ordered by device)
	* @param {number} poolSize - Session pool size
	* @param {MetricsCollector} metrics - Metrics collector
	* @param {Function} executor - Async function to execute each work item (pool, work, session)
	*/
	async function executeConcurrentWithBinding(
	pool,
	workload,
	poolSize,
	metrics,
	executor,
	) {
	// Get sessions from pool and bind to devices
	const sessions = [];
	for (let i = 0; i < poolSize; i++) {
	sessions.push(await pool.getSession());
	}

	console.log(
	`[Device-Session Binding] Bound ${sessions.length} sessions to devices`,
	);

	try {
	const workers = [];

	// Each worker handles workload items for devices bound to its session
	// Workload is pre-ordered, so we can partition it evenly
	const workPerSession = Math.ceil(workload.length / poolSize);

	const worker = async (sessionIdx) => {
	const session = sessions[sessionIdx];
	const startIdx = sessionIdx * workPerSession;
	const endIdx = Math.min(startIdx + workPerSession, workload.length);

	for (let i = startIdx; i < endIdx; i++) {
	const work = workload[i];
	const startTime = performance.now();

	try {
	const dataPoints = await executor(pool, work, session);
	const latency = performance.now() - startTime;
	metrics.recordOperation(latency, dataPoints, true);
	} catch (error) {
	const latency = performance.now() - startTime;
	metrics.recordOperation(latency, 0, false, error);
	}
	}
	};

	// Start workers, one per session
	for (let i = 0; i < poolSize; i++) {
	workers.push(worker(i));
	}

	// Wait for all workers to complete
	await Promise.all(workers);
	} finally {
	// Release all sessions back to pool
	for (const session of sessions) {
	pool.releaseSession(session);
	}
	console.log(
	`[Device-Session Binding] Released ${sessions.length} sessions`,
	);
	}
	}

	/**
	* Execute write operations concurrently with pre-acquired sessions
	* Each worker acquires a dedicated session at the start and uses it for all operations,
	* which significantly improves performance by avoiding session acquisition overhead per operation.
	*
	* @param {Object} pool - IoTDB session pool
	* @param {Array} workload - Array of work items
	* @param {number} concurrency - Number of concurrent workers
	* @param {MetricsCollector} metrics - Metrics collector
	* @param {Function} executor - Async function to execute each work item: (pool, work, session) => Promise<dataPoints>
	* The session parameter is the pre-acquired dedicated session for the worker.
	* Executors should use this session directly when provided (not null) for optimal performance.
	* If executor doesn't need the session, it can use pool methods which will handle session management.
	*/
	async function executeConcurrent(
	pool,
	workload,
	concurrency,
	metrics,
	executor,
	) {
	// Pre-acquire sessions for all workers to enable true concurrent execution
	const actualConcurrency = Math.min(concurrency, workload.length);

	// Acquire all sessions in parallel
	const sessions = await Promise.all(
	Array.from({ length: actualConcurrency }, () => pool.getSession()),
	);

	console.log(
	`[Concurrent Execution] Pre-acquired ${sessions.length} sessions for ${actualConcurrency} workers`,
	);

	let workIndex = 0;
	const workers = [];

	try {
	// Worker function - uses dedicated session for all its operations
	const worker = async (workerSession) => {
	while (workIndex < workload.length) {
	const index = workIndex++;
	if (index >= workload.length) break;

	const work = workload[index];
	const startTime = performance.now();

	try {
	// Pass the pre-acquired session to executor for direct use
	const dataPoints = await executor(pool, work, workerSession);
	const latency = performance.now() - startTime;
	metrics.recordOperation(latency, dataPoints, true);
	} catch (error) {
	const latency = performance.now() - startTime;
	metrics.recordOperation(latency, 0, false, error);
	}
	}
	};

	// Start workers, each with its own dedicated session
	for (let i = 0; i < actualConcurrency; i++) {
	workers.push(worker(sessions[i]));
	}

	// Wait for all workers to complete
	await Promise.all(workers);
	} finally {
	// Release all sessions back to pool
	for (const session of sessions) {
	pool.releaseSession(session);
	}
	console.log(`[Concurrent Execution] Released ${sessions.length} sessions`);
	}
	}

	/**
	* Run benchmark test
	* @param {Object} pool - IoTDB session pool
	* @param {Object} testData - Test data
	* @param {Object} config - Configuration
	* @param {Function} executor - Async function to execute work items
	* @param {Function} workloadGenerator - Function to generate workload
	* @returns {Object} Test results
	*/
	async function runBenchmark(
	pool,
	testData,
	config,
	executor,
	workloadGenerator,
	) {
	console.log("\n" + "=".repeat(80));
	console.log("STARTING BENCHMARK TEST");
	console.log("=".repeat(80));

	const metrics = new MetricsCollector(config);
	const reporter = new ProgressReporter(config, metrics);

	// Generate workload
	console.log("\nGenerating workload...");
	const workload = workloadGenerator(testData, config);
	console.log(`Workload generated: ${workload.length} operations`);

	// Warmup phase
	if (config.WARMUP_ROUNDS > 0) {
	console.log(
	`\n[Warmup Phase] Running ${config.WARMUP_ROUNDS} warmup rounds...`,
	);
	for (let round = 0; round < config.WARMUP_ROUNDS; round++) {
	const warmupMetrics = new MetricsCollector(config);
	warmupMetrics.start();

	await executeConcurrent(
	pool,
	workload.slice(0, Math.min(10, workload.length)),
	Math.min(config.CLIENT_NUMBER, 2),
	warmupMetrics,
	executor,
	);

	warmupMetrics.end();
	console.log(
	` Warmup round ${round + 1}/${config.WARMUP_ROUNDS} completed`,
	);
	}
	}

	// Main test phase
	if (config.ENABLE_DEVICE_SESSION_BINDING) {
	console.log(
	`\n[Test Phase] Running benchmark with device-session binding (${config.POOL_MAX_SIZE} sessions)...`,
	);
	} else {
	console.log(
	`\n[Test Phase] Running benchmark with ${config.CLIENT_NUMBER} concurrent clients...`,
	);
	}

	metrics.start();
	reporter.start();

	try {
	if (config.ENABLE_DEVICE_SESSION_BINDING) {
	await executeConcurrentWithBinding(
	pool,
	workload,
	config.POOL_MAX_SIZE,
	metrics,
	executor,
	);
	} else {
	await executeConcurrent(
	pool,
	workload,
	config.CLIENT_NUMBER,
	metrics,
	executor,
	);
	}
	} finally {
	reporter.stop();
	metrics.end();
	}

	// Print results
	metrics.printStats("Benchmark Results");

	return metrics.getStats();
	}

	/**
	* Execute batch insert operations using insertTabletsParallel
	* This is optimized for high throughput by pre-acquiring sessions
	* and distributing tablets across workers.
	*
	* @param {Object} pool - IoTDB session pool
	* @param {Array} tablets - Array of tablet objects ready for insertion
	* @param {number} concurrency - Number of concurrent workers
	* @param {MetricsCollector} metrics - Metrics collector
	* @returns {Promise<void>}
	*/
	async function executeBatchInsert(pool, tablets, concurrency, metrics) {
	const batchSize = Math.ceil(tablets.length / Math.ceil(tablets.length / 100)); // ~100 tablets per batch
	const batches = [];

	for (let i = 0; i < tablets.length; i += batchSize) {
	batches.push(tablets.slice(i, i + batchSize));
	}

	console.log(`[Batch Insert] ${tablets.length} tablets in ${batches.length} batches, concurrency: ${concurrency}`);

	const actualConcurrency = Math.min(concurrency, batches.length);

	// Pre-acquire sessions
	const sessions = await Promise.all(
	Array.from({ length: actualConcurrency }, () => pool.getSession()),
	);

	console.log(`[Batch Insert] Pre-acquired ${sessions.length} sessions`);

	let batchIndex = 0;

	try {
	const workers = sessions.map(async (session) => {
	while (batchIndex < batches.length) {
	const idx = batchIndex++;
	if (idx >= batches.length) break;

	const batch = batches[idx];
	const startTime = performance.now();

	try {
	// Insert all tablets in this batch sequentially using the same session
	for (const tablet of batch) {
	await session.insertTablet(tablet);
	}
	const latency = performance.now() - startTime;
	// Calculate total data points in this batch
	const dataPoints = batch.reduce((sum, t) => {
	const rows = t.timestamps.length;
	const cols = t.measurements?.length \|\| t.columnNames?.length \|\| 0;
	return sum + rows * cols;
	}, 0);
	metrics.recordOperation(latency, dataPoints, true);
	} catch (error) {
	const latency = performance.now() - startTime;
	metrics.recordOperation(latency, 0, false, error);
	}
	}
	});

	await Promise.all(workers);
	} finally {
	for (const session of sessions) {
	pool.releaseSession(session);
	}
	console.log(`[Batch Insert] Released ${sessions.length} sessions`);
	}
	}

	/**
	* Run benchmark using batch insert mode (optimized for high throughput)
	* @param {Object} pool - IoTDB session pool
	* @param {Array} tablets - Pre-built tablet array
	* @param {Object} config - Configuration
	* @returns {Object} Test results
	*/
	async function runBatchBenchmark(pool, tablets, config) {
	console.log("\n" + "=".repeat(80));
	console.log("STARTING BATCH BENCHMARK TEST");
	console.log("=".repeat(80));

	const metrics = new MetricsCollector(config);
	const reporter = new ProgressReporter(config, metrics);

	console.log(`\nTotal tablets to insert: ${tablets.length}`);

	// Warmup phase
	if (config.WARMUP_ROUNDS > 0) {
	console.log(
	`\n[Warmup Phase] Running ${config.WARMUP_ROUNDS} warmup rounds...`,
	);
	const warmupTablets = tablets.slice(0, Math.min(20, tablets.length));
	for (let round = 0; round < config.WARMUP_ROUNDS; round++) {
	const warmupMetrics = new MetricsCollector(config);
	warmupMetrics.start();

	await executeBatchInsert(
	pool,
	warmupTablets,
	Math.min(config.CLIENT_NUMBER, 2),
	warmupMetrics,
	);

	warmupMetrics.end();
	console.log(
	` Warmup round ${round + 1}/${config.WARMUP_ROUNDS} completed`,
	);
	}
	}

	// Main test phase
	console.log(
	`\n[Test Phase] Running batch benchmark with ${config.CLIENT_NUMBER} concurrent clients...`,
	);

	metrics.start();
	reporter.start();

	try {
	await executeBatchInsert(pool, tablets, config.CLIENT_NUMBER, metrics);
	} finally {
	reporter.stop();
	metrics.end();
	}

	// Print results
	metrics.printStats("Batch Benchmark Results");

	return metrics.getStats();
	}

	module.exports = {
	MetricsCollector,
	ProgressReporter,
	executeConcurrent,
	executeConcurrentWithBinding,
	executeBatchInsert,
	runBenchmark,
	runBatchBenchmark,
	};