graph_d 1.3.2

//! Concurrent transaction demonstration with locking.

use graph_d::transaction::{IsolationLevel, LockableResource, TransactionManager};
use graph_d::Result;
use std::sync::{Arc, Barrier};
use std::thread;
use std::time::{Duration, Instant};

fn main() -> Result<()> {
    println!("=== Concurrent Transaction Demo ===\n");

    demo_basic_locking()?;
    demo_concurrent_reads()?;
    demo_deadlock_prevention()?;
    demo_lock_statistics()?;

    Ok(())
}

fn demo_basic_locking() -> Result<()> {
    println!("=== Basic Locking Demo ===");

    let tx_manager = TransactionManager::new(IsolationLevel::ReadCommitted);

    // Create two concurrent transactions
    let mut tx1 = tx_manager.begin_concurrent();
    let mut tx2 = tx_manager.begin_concurrent();

    let node_resource = LockableResource::Node(1);

    println!("Transaction {}: Acquiring read lock on Node 1", tx1.id());
    tx1.read_lock(node_resource)?;

    println!("Transaction {}: Acquiring read lock on Node 1", tx2.id());
    tx2.read_lock(node_resource)?; // Should succeed (read locks are shared)

    println!("Both transactions successfully acquired read locks");

    // Try to acquire write lock from tx1 (should work since we already have a lock)
    println!("Transaction {}: Acquiring write lock on Node 2", tx1.id());
    tx1.write_lock(LockableResource::Node(2))?;

    println!("Transaction {}: Committing", tx1.id());
    tx1.commit()?;

    println!("Transaction {}: Committing", tx2.id());
    tx2.commit()?;

    println!("Basic locking demo completed successfully!\n");
    Ok(())
}

fn demo_concurrent_reads() -> Result<()> {
    println!("=== Concurrent Reads Demo ===");

    let tx_manager = Arc::new(TransactionManager::new(IsolationLevel::ReadCommitted));
    let num_readers = 5;
    let barrier = Arc::new(Barrier::new(num_readers));

    let mut handles = Vec::new();

    for i in 0..num_readers {
        let tx_manager = Arc::clone(&tx_manager);
        let barrier = Arc::clone(&barrier);

        let handle = thread::spawn(move || -> Result<()> {
            let mut tx = tx_manager.begin_concurrent();
            let node_resource = LockableResource::Node(100);

            // Wait for all threads to be ready
            barrier.wait();

            let start = Instant::now();

            // Acquire read lock
            tx.read_lock(node_resource)?;
            println!("Reader {}: Acquired read lock in {:?}", i, start.elapsed());

            // Simulate some read work
            thread::sleep(Duration::from_millis(50));

            // Commit
            tx.commit()?;
            println!("Reader {i}: Released lock");

            Ok(())
        });

        handles.push(handle);
    }

    // Wait for all threads to complete
    for handle in handles {
        handle.join().unwrap()?;
    }

    println!("All {num_readers} readers completed successfully!\n");
    Ok(())
}

fn demo_deadlock_prevention() -> Result<()> {
    println!("=== Deadlock Prevention Demo ===");

    let tx_manager = Arc::new(TransactionManager::new(IsolationLevel::ReadCommitted));
    let barrier = Arc::new(Barrier::new(2));

    let tx_manager1 = Arc::clone(&tx_manager);
    let tx_manager2 = Arc::clone(&tx_manager);
    let barrier1 = Arc::clone(&barrier);
    let barrier2 = Arc::clone(&barrier);

    // Transaction 1: Try to lock Node 1, then Node 2
    let handle1 = thread::spawn(move || -> Result<()> {
        let mut tx = tx_manager1.begin_concurrent();

        println!("Transaction {}: Acquiring write lock on Node 1", tx.id());
        tx.write_lock(LockableResource::Node(1))?;

        // Wait for both transactions to acquire their first locks
        barrier1.wait();

        // Small delay to increase chance of deadlock scenario
        thread::sleep(Duration::from_millis(100));

        println!(
            "Transaction {}: Trying to acquire write lock on Node 2",
            tx.id()
        );
        match tx.write_lock(LockableResource::Node(2)) {
            Ok(_) => {
                println!("Transaction {}: Successfully acquired both locks", tx.id());
                tx.commit()?;
            }
            Err(e) => {
                println!(
                    "Transaction {}: Failed to acquire second lock: {}",
                    tx.id(),
                    e
                );
                tx.rollback()?;
                return Err(e);
            }
        }

        Ok(())
    });

    // Transaction 2: Try to lock Node 2, then Node 1
    let handle2 = thread::spawn(move || -> Result<()> {
        let mut tx = tx_manager2.begin_concurrent();

        println!("Transaction {}: Acquiring write lock on Node 2", tx.id());
        tx.write_lock(LockableResource::Node(2))?;

        // Wait for both transactions to acquire their first locks
        barrier2.wait();

        // Small delay to increase chance of deadlock scenario
        thread::sleep(Duration::from_millis(100));

        println!(
            "Transaction {}: Trying to acquire write lock on Node 1",
            tx.id()
        );
        match tx.write_lock(LockableResource::Node(1)) {
            Ok(_) => {
                println!("Transaction {}: Successfully acquired both locks", tx.id());
                tx.commit()?;
            }
            Err(e) => {
                println!(
                    "Transaction {}: Failed to acquire second lock: {}",
                    tx.id(),
                    e
                );
                tx.rollback()?;
                return Err(e);
            }
        }

        Ok(())
    });

    // Wait for both threads
    let result1 = handle1.join().unwrap();
    let result2 = handle2.join().unwrap();

    // At least one should succeed, demonstrating deadlock prevention
    match (result1, result2) {
        (Ok(_), Ok(_)) => println!("Both transactions completed successfully (no deadlock)"),
        (Ok(_), Err(_)) => println!("Transaction 1 succeeded, Transaction 2 was aborted"),
        (Err(_), Ok(_)) => println!("Transaction 2 succeeded, Transaction 1 was aborted"),
        (Err(e1), Err(e2)) => {
            println!("Both transactions failed - this might indicate a deadlock:");
            println!("  Transaction 1 error: {e1}");
            println!("  Transaction 2 error: {e2}");
        }
    }

    println!("Deadlock prevention demo completed!\n");
    Ok(())
}

fn demo_lock_statistics() -> Result<()> {
    println!("=== Lock Statistics Demo ===");

    let tx_manager = TransactionManager::new(IsolationLevel::ReadCommitted);

    // Initial statistics
    let stats = tx_manager.lock_statistics();
    println!("Initial lock statistics:");
    println!("  Active locks: {}", stats.total_active_locks);
    println!("  Waiting requests: {}", stats.waiting_requests);
    println!("  Locked resources: {}", stats.locked_resources);

    // Create some transactions and acquire locks
    let mut tx1 = tx_manager.begin_concurrent();
    let mut tx2 = tx_manager.begin_concurrent();
    let mut tx3 = tx_manager.begin_concurrent();

    tx1.read_lock(LockableResource::Node(1))?;
    tx1.write_lock(LockableResource::Node(2))?;
    tx2.read_lock(LockableResource::Node(1))?; // Shared with tx1
    tx3.read_lock(LockableResource::Node(3))?;

    let stats = tx_manager.lock_statistics();
    println!("\nAfter acquiring locks:");
    println!("  Active locks: {}", stats.total_active_locks);
    println!("  Waiting requests: {}", stats.waiting_requests);
    println!("  Locked resources: {}", stats.locked_resources);

    // Commit transactions
    tx1.commit()?;
    tx2.commit()?;
    tx3.commit()?;

    let stats = tx_manager.lock_statistics();
    println!("\nAfter committing all transactions:");
    println!("  Active locks: {}", stats.total_active_locks);
    println!("  Waiting requests: {}", stats.waiting_requests);
    println!("  Locked resources: {}", stats.locked_resources);

    println!("Lock statistics demo completed!\n");
    Ok(())
}

#[allow(dead_code)]
fn demo_transaction_isolation() -> Result<()> {
    println!("=== Transaction Isolation Demo ===");

    // Test different isolation levels
    let isolation_levels = vec![
        IsolationLevel::ReadUncommitted,
        IsolationLevel::ReadCommitted,
        IsolationLevel::RepeatableRead,
        IsolationLevel::Serializable,
    ];

    for isolation_level in isolation_levels {
        println!("Testing isolation level: {isolation_level:?}");

        let tx_manager = TransactionManager::new(isolation_level);
        let tx = tx_manager.begin_concurrent_with_isolation(isolation_level);

        println!(
            "  Created transaction {} with isolation {:?}",
            tx.id(),
            tx.transaction.isolation_level()
        );

        // In a real implementation, different isolation levels would
        // affect locking behavior and visibility of uncommitted changes
    }

    println!("Transaction isolation demo completed!\n");
    Ok(())
}