/// Send requests to AC async, wait for responses from AC.
/// Return #accepted TXNs and submission duration.
Send requests to AC async, wait for responses from AC.
Return #accepted TXNs and submission duration.
Send requests to AC async wait for responses from AC . Return #accepted TXNs and submission duration .
pub fn submit_txns(&mut self, txn_reqs: &[SubmitTransactionRequest]) -> (usize, u128) {
        let txn_req_chunks = divide_items(txn_reqs, self.clients.len());
        let now = time::Instant::now();
        // Zip txn_req_chunks with clients: when first iter returns none,
        // zip will short-circuit and next will not be called on the second iter.
        let children: Vec<thread::JoinHandle<_>> = txn_req_chunks
            .zip(self.clients.iter().cycle())
            .map(|(chunk, client)| {
                let local_chunk = Vec::from(chunk);
                let local_client = Arc::clone(client);
                let stagger_range_ms = self.stagger_range_ms;
                // Spawn threads with corresponding client.
                thread::spawn(
                    // Dispatch TXN requests to client and submit, return the list of responses
                    // that are accepted by AC, and how long the client is delayed.
                    move || -> (Vec<ProtoSubmitTransactionResponse>, u16) {
                        let delay_duration_ms = Self::stagger_client(stagger_range_ms);
                        info!(
                            "Dispatch a chunk of {} requests to client and start to submit after staggered {} ms.",
                            local_chunk.len(),
                            delay_duration_ms,
                        );
                        (submit_and_wait_txn_requests(&local_client, &local_chunk), delay_duration_ms)
                    },
                )
            })
            .collect();
        // Wait for threads and gather reponses.
        // TODO: Group response by error type and report staticstics.
        let mut txn_resps: Vec<ProtoSubmitTransactionResponse> = vec![];
        let mut delay_duration_ms = self.stagger_range_ms;
        for child in children {
            let resp_tuple = child.join().expect("failed to join a request thread");
            txn_resps.extend(resp_tuple.0.into_iter());
            // Start counting time as soon as the first client starts to submit TXNs.
            delay_duration_ms = std::cmp::min(delay_duration_ms, resp_tuple.1);
        }
        let mut request_duration_ms = now.elapsed().as_millis();
        // Calling stagger_client() should ensure delay duration strictly < self.stagger_range_ms.
        if delay_duration_ms < self.stagger_range_ms {
            request_duration_ms -= u128::from(delay_duration_ms);
        }
        info!(
            "Submitted and accepted {} TXNs within {} ms.",
            txn_resps.len(),
            request_duration_ms,
        );
        (txn_resps.len(), request_duration_ms)
    }
pub fn submit_txns ( & mut self , txn_reqs : & [ SubmitTransactionRequest ] ) -> ( usize , u128 ) { let txn_req_chunks = divide_items ( txn_reqs , self . clients . len ( ) ) ; let now = time :: Instant :: now ( ) ; // Zip txn_req_chunks with clients: when first iter returns none, // zip will short-circuit and next will not be called on the second iter. let children : Vec < thread :: JoinHandle < _ > > = txn_req_chunks . zip ( self . clients . iter ( ) . cycle ( ) ) . map ( | ( chunk , client ) | { let local_chunk = Vec :: from ( chunk ) ; let local_client = Arc :: clone ( client ) ; let stagger_range_ms = self . stagger_range_ms ; // Spawn threads with corresponding client. thread :: spawn ( // Dispatch TXN requests to client and submit, return the list of responses // that are accepted by AC, and how long the client is delayed. move | | -> ( Vec < ProtoSubmitTransactionResponse > , u16 ) { let delay_duration_ms = Self :: stagger_client ( stagger_range_ms ) ; info ! ( " " local_chunk len ( ) delay_duration_ms ) ; ( submit_and_wait_txn_requests ( & local_client , & local_chunk ) , delay_duration_ms ) } , ) } ) . collect ( ) ; // Wait for threads and gather reponses. // TODO: Group response by error type and report staticstics. let mut txn_resps : Vec < ProtoSubmitTransactionResponse > = vec ! [ ] ; let mut delay_duration_ms = self . stagger_range_ms ; for child in children { let resp_tuple = child . join ( ) . expect ( " " ) ; txn_resps . extend ( resp_tuple . 0 . into_iter ( ) ) ; // Start counting time as soon as the first client starts to submit TXNs. delay_duration_ms = std :: cmp :: min ( delay_duration_ms , resp_tuple . 1 ) ; } let mut request_duration_ms = now . elapsed ( ) . as_millis ( ) ; // Calling stagger_client() should ensure delay duration strictly < self.stagger_range_ms. if delay_duration_ms < self . stagger_range_ms { request_duration_ms -= u128 :: from ( delay_duration_ms ) ; } info ! ( " " txn_resps len ( ) request_duration_ms ) ; ( txn_resps . len ( ) , request_duration_ms ) }
Send requests to AC async, wait for responses from AC.
Send requests to AC async wait for responses from AC .
&mut self
txn_reqs
&[SubmitTransactionRequest]