/// Runs a test group by spawning replicas as threads
///
/// # Parameters
/// - num_replicas            Number of replicas in the test group
/// - test                    p2p test callback that need to be invoked for each
///   replica
Runs a test group by spawning replicas as threads
Parameters
num_replicas            Number of replicas in the test group
test                    p2p test callback that need to be invoked for each
replica
Runs a test group by spawning replicas as threads Parameters num_replicas Number of replicas in the test group test p2p test callback that need to be invoked for each replica
pub fn spawn_replicas_as_threads(
    real_artifact_pool: bool,
    num_replicas: u16,
    test: impl FnOnce(&mut P2PTestContext) + Copy + Send + Sync + 'static,
) {
    // Create a directory inside of `std::env::temp_dir()`
    let temp_dir = Builder::new()
        .prefix("p2p_tests")
        .tempdir()
        .expect("Cannot create a test directory");
    println!("Exec Test in {:?}", temp_dir);
    let allocated_ports =
        allocate_ports("127.0.0.1", num_replicas).expect("Port allocation for test failed");
    let node_port_allocation: Vec<u16> = allocated_ports.iter().map(|np| np.port).collect();
    assert_eq!(num_replicas as usize, node_port_allocation.len());
    let node_port_allocation = Arc::new(node_port_allocation);
    // Setup the test directory once. Child replicas should not do the setup
    let test_synchronizer = P2PTestSynchronizer::new(
        temp_dir.path().to_owned(),
        node_test_id(0),
        num_replicas,
        node_port_allocation.clone(),
    );
    test_synchronizer
        .setup_test_group_directory()
        .expect("Failed To Setup test directory");
    // Build the registry for the test
    let data_provider =
        test_group_set_registry(subnet_test_id(P2P_SUBNET_ID_DEFAULT), node_port_allocation);
    // Keep this around until the end of the test, as it contains a guard that stops
    // async logging on drop.
    let logger = p2p_test_setup_logger();
    let log: ReplicaLogger = logger.root.into();
    // Build the thread network/transport if
    let transport_hub: Hub = Default::default();
    let hub_access: HubAccess = Arc::new(Mutex::new(transport_hub));
    for instance_id in 0..num_replicas {
        let thread_port = ThreadPort::new(
            node_test_id(instance_id as u64),
            hub_access.clone(),
            log.clone(),
        );
        hub_access
            .lock()
            .unwrap()
            .insert(node_test_id(instance_id as u64), thread_port);
    }
    let registry_client = Arc::new(RegistryClientImpl::new(data_provider, None));
    registry_client.fetch_and_start_polling().unwrap();
    let mut join_handles = Vec::new();
    for i in 0..num_replicas {
        let test = test;
        let test = Box::new(test);
        let replica_log = log.clone();
        let transport_hub = hub_access.lock().unwrap();
        let tp = transport_hub.get(&node_test_id(i as u64)).clone();
        let replica_registry = Arc::clone(&registry_client) as Arc<dyn RegistryClient>;
        let replica_config = ReplicaConfig {
            node_id: node_test_id(i as u64),
            subnet_id: subnet_test_id(P2P_SUBNET_ID_DEFAULT),
        };
        let mut replica_test_synchronizer = test_synchronizer.clone();
        replica_test_synchronizer.node_id = node_test_id(i as u64);
        let jh = std::thread::Builder::new()
            .name(format!("Thread Node {}", i))
            .spawn(move || {
                let rt = tokio::runtime::Runtime::new().unwrap();
                let _rt_guard = rt.enter();
                // Spawn System
                if real_artifact_pool {
                    execute_test(
                        i as u64,
                        replica_config,
                        replica_registry,
                        tp,
                        replica_test_synchronizer,
                        replica_log.clone(),
                        test,
                    );
                } else {
                    execute_test_chunking_pool(
                        i as u64,
                        replica_config,
                        replica_registry,
                        tp,
                        replica_test_synchronizer,
                        replica_log,
                        test,
                    );
                }
            })
            .unwrap();
        join_handles.push(jh);
    }
    for join_handle in join_handles {
        #[warn(unused_must_use)]
        assert!(join_handle.join().is_ok())
    }
    test_synchronizer
        .cleanup_test_group_directory()
        .expect("Failed To Setup test directory");
}
pub fn spawn_replicas_as_threads ( real_artifact_pool : bool , num_replicas : u16 , test : impl FnOnce ( & mut P2PTestContext ) + Copy + Send + Sync + ' static , ) { // Create a directory inside of `std::env::temp_dir()` let temp_dir = Builder :: new ( ) . prefix ( " " ) . tempdir ( ) . expect ( " " ) ; println ! ( " " temp_dir ) ; let allocated_ports = allocate_ports ( " " , num_replicas ) . expect ( " " ) ; let node_port_allocation : Vec < u16 > = allocated_ports . iter ( ) . map ( | np | np . port ) . collect ( ) ; assert_eq ! ( num_replicas as usize node_port_allocation len ( ) ) ; let node_port_allocation = Arc :: new ( node_port_allocation ) ; // Setup the test directory once. Child replicas should not do the setup let test_synchronizer = P2PTestSynchronizer :: new ( temp_dir . path ( ) . to_owned ( ) , node_test_id ( 0 ) , num_replicas , node_port_allocation . clone ( ) , ) ; test_synchronizer . setup_test_group_directory ( ) . expect ( " " ) ; // Build the registry for the test let data_provider = test_group_set_registry ( subnet_test_id ( P2P_SUBNET_ID_DEFAULT ) , node_port_allocation ) ; // Keep this around until the end of the test, as it contains a guard that stops // async logging on drop. let logger = p2p_test_setup_logger ( ) ; let log : ReplicaLogger = logger . root . into ( ) ; // Build the thread network/transport if let transport_hub : Hub = Default :: default ( ) ; let hub_access : HubAccess = Arc :: new ( Mutex :: new ( transport_hub ) ) ; for instance_id in 0 .. num_replicas { let thread_port = ThreadPort :: new ( node_test_id ( instance_id as u64 ) , hub_access . clone ( ) , log . clone ( ) , ) ; hub_access . lock ( ) . unwrap ( ) . insert ( node_test_id ( instance_id as u64 ) , thread_port ) ; } let registry_client = Arc :: new ( RegistryClientImpl :: new ( data_provider , None ) ) ; registry_client . fetch_and_start_polling ( ) . unwrap ( ) ; let mut join_handles = Vec :: new ( ) ; for i in 0 .. num_replicas { let test = test ; let test = Box :: new ( test ) ; let replica_log = log . clone ( ) ; let transport_hub = hub_access . lock ( ) . unwrap ( ) ; let tp = transport_hub . get ( & node_test_id ( i as u64 ) ) . clone ( ) ; let replica_registry = Arc :: clone ( & registry_client ) as Arc < dyn RegistryClient > ; let replica_config = ReplicaConfig { node_id : node_test_id ( i as u64 ) , subnet_id : subnet_test_id ( P2P_SUBNET_ID_DEFAULT ) , } ; let mut replica_test_synchronizer = test_synchronizer . clone ( ) ; replica_test_synchronizer . node_id = node_test_id ( i as u64 ) ; let jh = std :: thread :: Builder :: new ( ) . name ( format ! ( " " i ) ) . spawn ( move | | { let rt = tokio :: runtime :: Runtime :: new ( ) . unwrap ( ) ; let _rt_guard = rt . enter ( ) ; // Spawn System if real_artifact_pool { execute_test ( i as u64 , replica_config , replica_registry , tp , replica_test_synchronizer , replica_log . clone ( ) , test , ) ; } else { execute_test_chunking_pool ( i as u64 , replica_config , replica_registry , tp , replica_test_synchronizer , replica_log , test , ) ; } } ) . unwrap ( ) ; join_handles . push ( jh ) ; } for join_handle in join_handles { # [ warn ( unused_must_use ) ] assert ! ( join_handle join ( ) is_ok ( ) ) } test_synchronizer . cleanup_test_group_directory ( ) . expect ( " " ) ; }
Runs a test group by spawning replicas as threads
 Parameters
 num_replicas            Number of replicas in the test group
 test                    p2p test callback that need to be invoked for each
replica
Runs a test group by spawning replicas as threads Parameters num_replicas Number of replicas in the test group test p2p test callback that need to be invoked for each replica
real_artifact_pool
bool
num_replicas
u16
test
impl FnOnce(&mut P2PTestContext) + Copy + Send + Sync + 'static