/// Starts a local replica runtime and installs the nns canisters.
/// The runtime is kept alive for the duration of the test.
Starts a local replica runtime and installs the nns canisters.
The runtime is kept alive for the duration of the test.
Starts a local replica runtime and installs the nns canisters . The runtime is kept alive for the duration of the test .
fn initialize(&mut self, state: &FuzzState) {
        println!("Initializing LocalNnsFuzzDriver");
        let ops_per_batch = (self.params.total_num_ops / self.params.num_batches) as usize;
        let (ops_sender, ops_receiver) = bounded(ops_per_batch);
        let (results_sender, results_receiver) = bounded(ops_per_batch);
        self.operation_sender = Some(ops_sender);
        self.results_receiver = Some(results_receiver);
        let state = state.clone();
        let economics = self.params.network_economics.clone();
        self.sender_thread = Some(thread::spawn(move || {
            local_test_on_nns_subnet(move |runtime| async move {
                println!("Initializing the NNS.");
                let mut governance = GovernanceCanisterInitPayloadBuilder::new().proto;
                governance.economics = Some(economics);
                // Note that we're introducing some indeterminism by allowing the ledger
                // to receive a list of accounts that is different each time (since it
                // uses a HashMap). Something to dig into if we see indet problems.
                let mut ledger_init_state = HashMap::new();
                for account in state.accounts.values() {
                    ledger_init_state.insert(account.clone().into(), account.balance);
                }
                for neuron in state
                    .neurons
                    .values()
                    .filter(|n| n.id_in_governance.is_some())
                {
                    governance
                        .neurons
                        .insert(neuron.id_in_governance.unwrap(), neuron.clone().into());
                }
                let ledger_init_args = LedgerCanisterInitPayload::new(
                    GOVERNANCE_CANISTER_ID.into(),
                    ledger_init_state,
                    None,
                    None,
                    None,
                    HashSet::new(),
                );
                let nns_init_payload = NnsInitPayloadsBuilder::new()
                    .with_governance_proto(governance)
                    .with_ledger_init_state(ledger_init_args)
                    .build();
                // Get a static reference to the runtime that we can pass to operation execution
                // threads below.
                let runtime = Box::new(runtime);
                let runtime: &'static canister_test::Runtime = Box::leak(runtime);
                // Surroung the canisters struct in Arc so that we can pass it to operation
                // execution threads below.
                let nns_canisters = Arc::new(NnsCanisters::set_up(runtime, nns_init_payload).await);
                println!("NNS initialized. Starting to send operations.");
                let mut operations = Vec::new();
                while let Ok(operation) = ops_receiver.recv() {
                    operations.push(operation);
                    if operations.len() < ops_per_batch {
                        continue;
                    }
                    let mut join_handles = Vec::new();
                    for operation_group in Self::split_parallelizable_operations(&operations) {
                        let nns_canisters_handle = nns_canisters.clone();
                        join_handles.push(tokio::runtime::Handle::current().spawn(async move {
                            let mut results = Vec::new();
                            for operation in operation_group {
                                println!(
                                    "Thread: {:?} Executing operation: {:?}",
                                    thread::current().id(),
                                    operation
                                );
                                results.push(
                                    Self::handle_operation(operation, nns_canisters_handle.clone())
                                        .await,
                                )
                            }
                            results
                        }));
                    }
                    println!("Waiting for latest batch to complete.");
                    // Wait on all the ops before proceeding.
                    let result_groups = futures::future::join_all(join_handles.into_iter()).await;
                    for result_group in result_groups {
                        match result_group {
                            Ok(results) => {
                                for result in results {
                                    results_sender
                                        .send(result)
                                        .expect("Can't send result, channel closed.");
                                }
                            }
                            Err(join_error) => {
                                // If we had an error, return, which will cause the results
                                // receiver to fail.
                                return Err(format!(
                                    "Error waiting on a results operation: {:?}",
                                    join_error
                                ));
                            }
                        }
                    }
                    operations.clear();
                }
                println!("All operations sent. Execution complete.");
                runtime.stop();
                Ok(())
            });
        }));
    }
fn initialize ( & mut self , state : & FuzzState ) { println ! ( " " ) ; let ops_per_batch = ( self . params . total_num_ops / self . params . num_batches ) as usize ; let ( ops_sender , ops_receiver ) = bounded ( ops_per_batch ) ; let ( results_sender , results_receiver ) = bounded ( ops_per_batch ) ; self . operation_sender = Some ( ops_sender ) ; self . results_receiver = Some ( results_receiver ) ; let state = state . clone ( ) ; let economics = self . params . network_economics . clone ( ) ; self . sender_thread = Some ( thread :: spawn ( move | | { local_test_on_nns_subnet ( move | runtime | async move { println ! ( " " ) ; let mut governance = GovernanceCanisterInitPayloadBuilder :: new ( ) . proto ; governance . economics = Some ( economics ) ; // Note that we're introducing some indeterminism by allowing the ledger // to receive a list of accounts that is different each time (since it // uses a HashMap). Something to dig into if we see indet problems. let mut ledger_init_state = HashMap :: new ( ) ; for account in state . accounts . values ( ) { ledger_init_state . insert ( account . clone ( ) . into ( ) , account . balance ) ; } for neuron in state . neurons . values ( ) . filter ( | n | n . id_in_governance . is_some ( ) ) { governance . neurons . insert ( neuron . id_in_governance . unwrap ( ) , neuron . clone ( ) . into ( ) ) ; } let ledger_init_args = LedgerCanisterInitPayload :: new ( GOVERNANCE_CANISTER_ID . into ( ) , ledger_init_state , None , None , None , HashSet :: new ( ) , ) ; let nns_init_payload = NnsInitPayloadsBuilder :: new ( ) . with_governance_proto ( governance ) . with_ledger_init_state ( ledger_init_args ) . build ( ) ; // Get a static reference to the runtime that we can pass to operation execution // threads below. let runtime = Box :: new ( runtime ) ; let runtime : & ' static canister_test :: Runtime = Box :: leak ( runtime ) ; // Surroung the canisters struct in Arc so that we can pass it to operation // execution threads below. let nns_canisters = Arc :: new ( NnsCanisters :: set_up ( runtime , nns_init_payload ) . await ) ; println ! ( " " ) ; let mut operations = Vec :: new ( ) ; while let Ok ( operation ) = ops_receiver . recv ( ) { operations . push ( operation ) ; if operations . len ( ) < ops_per_batch { continue ; } let mut join_handles = Vec :: new ( ) ; for operation_group in Self :: split_parallelizable_operations ( & operations ) { let nns_canisters_handle = nns_canisters . clone ( ) ; join_handles . push ( tokio :: runtime :: Handle :: current ( ) . spawn ( async move { let mut results = Vec :: new ( ) ; for operation in operation_group { println ! ( " " thread current ( ) id ( ) operation ) ; results . push ( Self :: handle_operation ( operation , nns_canisters_handle . clone ( ) ) . await , ) } results } ) ) ; } println ! ( " " ) ; // Wait on all the ops before proceeding. let result_groups = futures :: future :: join_all ( join_handles . into_iter ( ) ) . await ; for result_group in result_groups { match result_group { Ok ( results ) => { for result in results { results_sender . send ( result ) . expect ( " " ) ; } } Err ( join_error ) => { // If we had an error, return, which will cause the results // receiver to fail. return Err ( format ! ( " " join_error ) ) ; } } } operations . clear ( ) ; } println ! ( " " ) ; runtime . stop ( ) ; Ok ( ( ) ) } ) ; } ) ) ; }
Starts a local replica runtime and installs the nns canisters.
Starts a local replica runtime and installs the nns canisters .
&mut self
state
&FuzzState