Struct MultiWriteTransaction 

pub struct MultiWriteTransaction { /* private fields */ }

Implementations

impl MultiWriteTransaction

pub fn new(engine: MultiTransaction) -> Result<Self>

impl MultiWriteTransaction

pub fn id(&self) -> TransactionId

pub fn version(&self) -> CommitVersion

pub fn base_version(&self) -> CommitVersion

pub fn stamp_source(&mut self, source: SourceVersion)

pub fn read_as_of_version_exclusive(&mut self, version: CommitVersion)

pub fn read_as_of_version_inclusive( &mut self, version: CommitVersion, ) -> Result<()>

pub fn pending_writes(&self) -> &PendingWrites

pub fn conflicts(&self) -> &ConflictManager

pub fn mark_preexisting<K: Into<TaggedKey> + Clone>(&mut self, key: &K)

pub fn preexisting_keys(&self) -> &BTreeSet<TaggedKey>

impl MultiWriteTransaction

pub fn savepoint(&self) -> WriteSavepoint

pub fn restore_savepoint(&mut self, sp: WriteSavepoint)

impl MultiWriteTransaction

pub fn marker(&mut self) -> Marker<'_>

pub fn marker_with_pending_writes(&mut self) -> (Marker<'_>, &PendingWrites)

pub fn mark_read(&mut self, k: &TaggedKey)

pub fn mark_write(&mut self, k: &TaggedKey)

pub fn reserve_writes(&mut self, additional: usize)

impl MultiWriteTransaction

pub fn set<K: Into<TaggedKey> + Clone>( &mut self, key: &K, bytes: impl Into<EncodedBytes>, ) -> Result<()>

Examples found in repository

examples/oracle_performance.rs (line 57)

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

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

		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(vec![]).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);
	}
}

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

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

	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(vec![]).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);
	}
}

pub fn conflict_detection_benchmark() {
	println!("=== Conflict Detection Performance Benchmark ===\n");

	let engine = MultiTransaction::testing();

	for i in 0..1000 {
		let mut tx = engine.begin_command().unwrap();
		let key = as_key!(format!("shared_key_{}", i % 100));
		let value = as_values!(i);
		tx.set(&key, value).unwrap();
		tx.commit(vec![]).unwrap();
	}

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

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

		let key = as_key!(format!("shared_key_{}", i % 100));
		let value = as_values!(i + 1000);

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

		match tx.commit(vec![]) {
			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);
}

pub fn remove_with_pre<K: Into<TaggedKey> + Clone>( &mut self, key: &K, pre: EncodedBytes, ) -> Result<()>

pub fn remove<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<()>

pub fn remove_unobserved<K: Into<TaggedKey> + Clone>( &mut self, key: &K, ) -> Result<()>

pub fn remove_unobserved_with_pre<K: Into<TaggedKey> + Clone>( &mut self, key: &K, pre: EncodedBytes, ) -> Result<()>

pub fn remove_silent<K: Into<TaggedKey> + Clone>( &mut self, key: &K, ) -> Result<()>

pub fn rollback(&mut self) -> Result<()>

pub fn contains<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<bool>

pub fn get<K: Into<TaggedKey> + Clone>( &mut self, key: &K, ) -> Result<Option<TransactionValue>>

pub fn get_committed<K: Into<TaggedKey> + Clone>( &mut self, key: &K, ) -> Result<Option<TransactionValue>>

impl MultiWriteTransaction

pub fn commit(&mut self, flow_changes: Vec<Change>) -> Result<CommitVersion>

Examples found in repository

examples/oracle_performance.rs (line 58)

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

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

		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(vec![]).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);
	}
}

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

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

	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(vec![]).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);
	}
}

pub fn conflict_detection_benchmark() {
	println!("=== Conflict Detection Performance Benchmark ===\n");

	let engine = MultiTransaction::testing();

	for i in 0..1000 {
		let mut tx = engine.begin_command().unwrap();
		let key = as_key!(format!("shared_key_{}", i % 100));
		let value = as_values!(i);
		tx.set(&key, value).unwrap();
		tx.commit(vec![]).unwrap();
	}

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

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

		let key = as_key!(format!("shared_key_{}", i % 100));
		let value = as_values!(i + 1000);

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

		match tx.commit(vec![]) {
			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 MultiWriteTransaction

pub fn discard(&mut self)

pub fn is_discard(&self) -> bool

impl MultiWriteTransaction

pub fn prefix( &mut self, prefix: &EncodedKey, ) -> Result<MultiVersionBatch<TaggedKey>>

pub fn prefix_rev( &mut self, prefix: &EncodedKey, ) -> Result<MultiVersionBatch<TaggedKey>>

pub fn range( &mut self, range: TaggedKeyBoundRange, scope: RangeScope, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>

pub fn range_row( &mut self, storage: StorageId, start: Bound<StorageRowKey>, end: Bound<StorageRowKey>, scope: RangeScope, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow<StorageRowKey>>> + Send + '_>

pub fn range_partitioned_row( &mut self, storage: StorageId, start: Bound<StoragePartitionedRowKey>, end: Bound<StoragePartitionedRowKey>, scope: RangeScope, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow<StoragePartitionedRowKey>>> + Send + '_>

pub fn range_persistence( &mut self, range: TaggedKeyBoundRange, scope: RangeScope, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>

pub fn range_rev( &mut self, range: TaggedKeyBoundRange, scope: RangeScope, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>

pub fn range_rev_persistence( &mut self, range: TaggedKeyBoundRange, scope: RangeScope, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>

Trait Implementations

impl Drop for MultiWriteTransaction

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.