Struct MultiTransaction 

pub struct MultiTransaction(/* private fields */);

Implementations

impl MultiTransaction

pub fn testing() -> Self

Examples found in repository

examples/oracle_performance.rs (line 33)

pub fn oracle_performance_benchmark() {
	println!("=== Oracle Performance Benchmark ===\n");

	// Test different transaction counts to show scaling behavior
	let test_sizes = vec![1000, 5000, 10000, 25000];

	for &num_txns in &test_sizes {
		println!("Testing with {} transactions...", num_txns);

		let engine = MultiTransaction::testing();

		let start = Instant::now();

		// Create transactions sequentially (worst case for O(N²) algorithm)
		for i in 0..num_txns {
			let mut tx = engine.begin_command().unwrap();

			let key = as_key!(format!("key_{}", i));
			let value = as_values!(format!("value_{}", i));

			tx.set(&key, value).unwrap();
			tx.commit().unwrap();
		}

		let duration = start.elapsed();
		let tps = num_txns as f64 / duration.as_secs_f64();

		println!("  {} transactions in {:?}", num_txns, duration);
		println!("  {:.0} TPS (transactions per second)", tps);
		println!("  {:.2} μs per transaction\n", duration.as_micros() as f64 / num_txns as f64);
	}
}

/// Benchmark concurrent performance
pub fn concurrent_oracle_benchmark() {
	println!("=== Concurrent Oracle Performance Benchmark ===\n");

	let test_configs = vec![
		(10, 1000), // 10 threads, 1000 txns each
		(50, 500),  // 50 threads, 500 txns each
		(100, 250), // 100 threads, 250 txns each
		(1000, 50), // 1000 threads, 50 txns each
	];

	for &(num_threads, txns_per_thread) in &test_configs {
		let total_txns = num_threads * txns_per_thread;
		println!(
			"Testing {} threads × {} transactions = {} total...",
			num_threads, txns_per_thread, total_txns
		);

		let engine = Arc::new(MultiTransaction::testing());
		let start = Instant::now();

		let mut handles = vec![];

		for thread_id in 0..num_threads {
			let engine_clone = engine.clone();
			let handle = spawn(move || {
				let base_key = thread_id * txns_per_thread;
				for i in 0..txns_per_thread {
					let mut tx = engine_clone.begin_command().unwrap();

					let key = as_key!(base_key + i);
					let value = as_values!(i);

					tx.set(&key, value).unwrap();
					tx.commit().unwrap();
				}
			});
			handles.push(handle);
		}

		for handle in handles {
			handle.join().expect("Task panicked");
		}

		let duration = start.elapsed();
		let tps = total_txns as f64 / duration.as_secs_f64();

		println!("  {} total transactions in {:?}", total_txns, duration);
		println!("  {:.0} TPS (transactions per second)", tps);
		println!("  {:.2} μs per transaction\n", duration.as_micros() as f64 / total_txns as f64);
	}
}

/// Benchmark with actual conflicts to test conflict detection performance
pub fn conflict_detection_benchmark() {
	println!("=== Conflict Detection Performance Benchmark ===\n");

	let engine = MultiTransaction::testing();

	// Pre-populate with some data to create realistic conflict scenarios
	for i in 0..1000 {
		let mut tx = engine.begin_command().unwrap();
		let key = as_key!(format!("shared_key_{}", i % 100)); // 100 different keys
		let value = as_values!(i);
		tx.set(&key, value).unwrap();
		tx.commit().unwrap();
	}

	println!("Pre-populated with 1000 transactions across 100 keys");

	// Now test conflict detection performance
	let num_conflict_txns = 10000;
	let start = Instant::now();
	let mut conflicts = 0;

	for i in 0..num_conflict_txns {
		let mut tx = engine.begin_command().unwrap();

		// Try to modify keys that might conflict
		let key = as_key!(format!("shared_key_{}", i % 100));
		let value = as_values!(i + 1000);

		tx.set(&key, value).unwrap();

		match tx.commit() {
			Ok(_) => {}
			Err(e) if e.code == "TXN_001" => {
				conflicts += 1;
			}
			Err(e) => panic!("Unexpected error: {:?}", e),
		};
	}

	let duration = start.elapsed();
	let tps = num_conflict_txns as f64 / duration.as_secs_f64();

	println!("  {} transactions with potential conflicts in {:?}", num_conflict_txns, duration);
	println!(
		"  {} actual conflicts detected ({:.1}%)",
		conflicts,
		conflicts as f64 / num_conflict_txns as f64 * 100.0
	);
	println!("  {:.0} TPS (transactions per second)", tps);
	println!("  {:.2} μs per transaction", duration.as_micros() as f64 / num_conflict_txns as f64);
}

impl MultiTransaction

pub fn new( store: MultiStore, single: SingleTransaction, event_bus: EventBus, actor_system: ActorSystem, metrics_clock: Clock, system_config: SystemConfig, ) -> Result<Self>

pub fn actor_system(&self) -> ActorSystem

Get the actor system

pub fn system_config(&self) -> SystemConfig

Get the shared system config from the oracle.

impl MultiTransaction

pub fn version(&self) -> Result<CommitVersion>

pub fn begin_query(&self) -> Result<MultiReadTransaction>

pub fn begin_query_at_version( &self, version: CommitVersion, ) -> Result<MultiReadTransaction>

Begin a query transaction at a specific version.

This is used for parallel query execution where multiple tasks need to read from the same snapshot (same CommitVersion) for consistency.

impl MultiTransaction

pub fn begin_command(&self) -> Result<MultiWriteTransaction>

Examples found in repository

examples/oracle_performance.rs (line 39)

pub fn oracle_performance_benchmark() {
	println!("=== Oracle Performance Benchmark ===\n");

	// Test different transaction counts to show scaling behavior
	let test_sizes = vec![1000, 5000, 10000, 25000];

	for &num_txns in &test_sizes {
		println!("Testing with {} transactions...", num_txns);

		let engine = MultiTransaction::testing();

		let start = Instant::now();

		// Create transactions sequentially (worst case for O(N²) algorithm)
		for i in 0..num_txns {
			let mut tx = engine.begin_command().unwrap();

			let key = as_key!(format!("key_{}", i));
			let value = as_values!(format!("value_{}", i));

			tx.set(&key, value).unwrap();
			tx.commit().unwrap();
		}

		let duration = start.elapsed();
		let tps = num_txns as f64 / duration.as_secs_f64();

		println!("  {} transactions in {:?}", num_txns, duration);
		println!("  {:.0} TPS (transactions per second)", tps);
		println!("  {:.2} μs per transaction\n", duration.as_micros() as f64 / num_txns as f64);
	}
}

/// Benchmark concurrent performance
pub fn concurrent_oracle_benchmark() {
	println!("=== Concurrent Oracle Performance Benchmark ===\n");

	let test_configs = vec![
		(10, 1000), // 10 threads, 1000 txns each
		(50, 500),  // 50 threads, 500 txns each
		(100, 250), // 100 threads, 250 txns each
		(1000, 50), // 1000 threads, 50 txns each
	];

	for &(num_threads, txns_per_thread) in &test_configs {
		let total_txns = num_threads * txns_per_thread;
		println!(
			"Testing {} threads × {} transactions = {} total...",
			num_threads, txns_per_thread, total_txns
		);

		let engine = Arc::new(MultiTransaction::testing());
		let start = Instant::now();

		let mut handles = vec![];

		for thread_id in 0..num_threads {
			let engine_clone = engine.clone();
			let handle = spawn(move || {
				let base_key = thread_id * txns_per_thread;
				for i in 0..txns_per_thread {
					let mut tx = engine_clone.begin_command().unwrap();

					let key = as_key!(base_key + i);
					let value = as_values!(i);

					tx.set(&key, value).unwrap();
					tx.commit().unwrap();
				}
			});
			handles.push(handle);
		}

		for handle in handles {
			handle.join().expect("Task panicked");
		}

		let duration = start.elapsed();
		let tps = total_txns as f64 / duration.as_secs_f64();

		println!("  {} total transactions in {:?}", total_txns, duration);
		println!("  {:.0} TPS (transactions per second)", tps);
		println!("  {:.2} μs per transaction\n", duration.as_micros() as f64 / total_txns as f64);
	}
}

/// Benchmark with actual conflicts to test conflict detection performance
pub fn conflict_detection_benchmark() {
	println!("=== Conflict Detection Performance Benchmark ===\n");

	let engine = MultiTransaction::testing();

	// Pre-populate with some data to create realistic conflict scenarios
	for i in 0..1000 {
		let mut tx = engine.begin_command().unwrap();
		let key = as_key!(format!("shared_key_{}", i % 100)); // 100 different keys
		let value = as_values!(i);
		tx.set(&key, value).unwrap();
		tx.commit().unwrap();
	}

	println!("Pre-populated with 1000 transactions across 100 keys");

	// Now test conflict detection performance
	let num_conflict_txns = 10000;
	let start = Instant::now();
	let mut conflicts = 0;

	for i in 0..num_conflict_txns {
		let mut tx = engine.begin_command().unwrap();

		// Try to modify keys that might conflict
		let key = as_key!(format!("shared_key_{}", i % 100));
		let value = as_values!(i + 1000);

		tx.set(&key, value).unwrap();

		match tx.commit() {
			Ok(_) => {}
			Err(e) if e.code == "TXN_001" => {
				conflicts += 1;
			}
			Err(e) => panic!("Unexpected error: {:?}", e),
		};
	}

	let duration = start.elapsed();
	let tps = num_conflict_txns as f64 / duration.as_secs_f64();

	println!("  {} transactions with potential conflicts in {:?}", num_conflict_txns, duration);
	println!(
		"  {} actual conflicts detected ({:.1}%)",
		conflicts,
		conflicts as f64 / num_conflict_txns as f64 * 100.0
	);
	println!("  {:.0} TPS (transactions per second)", tps);
	println!("  {:.2} μs per transaction", duration.as_micros() as f64 / num_conflict_txns as f64);
}

impl MultiTransaction

pub fn get( &self, key: &EncodedKey, version: CommitVersion, ) -> Result<Option<Committed>>

pub fn contains_key( &self, key: &EncodedKey, version: CommitVersion, ) -> Result<bool>

pub fn store(&self) -> &MultiStore

Get a reference to the underlying transaction store.

impl MultiTransaction

pub fn current_version(&self) -> Result<CommitVersion>

Get the current version from the transaction manager

pub fn done_until(&self) -> CommitVersion

Returns the highest version where ALL prior versions have completed. This is useful for CDC polling to know the safe upper bound for fetching CDC events - all events up to this version are guaranteed to be in storage.

pub fn wait_for_mark_timeout( &self, version: CommitVersion, timeout: Duration, ) -> bool

Wait for the watermark to reach the given version with a timeout. Returns true if the watermark reached the target, false if timeout occurred.

Trait Implementations

impl Clone for MultiTransaction

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Deref for MultiTransaction

type Target = Inner

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl WithEventBus for MultiTransaction

fn event_bus(&self) -> &EventBus