fn client(&self, message_size: Byte, message_count: usize, rps: Option<u32>) {
        // Spawn the response handler thread
        let request_map = Arc::new(Mutex::new(HashMap::new()));
        let request_map_cl = request_map.clone();
        let receiver_cl = self.receiver.clone();
        let log_cl = self.log.clone();
        let handle = std::thread::spawn(move || {
            Self::client_response_handler(
                message_size,
                message_count,
                request_map_cl,
                receiver_cl,
                log_cl,
            );
        });

        let mut rate_limiter: Box<dyn RateLimiter> = match rps {
            Some(r) => Box::new(RateLimiterImpl::new(r)),
            None => Box::new(NoopRateLimiter {}),
        };

        // Send the requests
        let start = Instant::now();
        let mut qfull = 0;
        let mut last_update = Instant::now();
        for i in 1..=message_count {
            rate_limiter.wait();
            let id = i as u64;
            let request = TestMessage::Request(Request {
                id,
                payload_size: message_size.get_bytes() as usize,
            });

            request_map.lock().unwrap().insert(id, Instant::now());
            qfull += self.send_message(request);
            if last_update.elapsed().as_secs() > STATUS_UPDATE_SEC || (i % 1000) == 0 {
                info!(
                    self.log,
                    "Client: sent request[{}/{}], qfull = {}", i, message_count, qfull
                );
                last_update = Instant::now();
            }
        }
        info!(
            self.log,
            "Client: requests done, sent {} requests (message size {}, total {}) in {:?}",
            message_count,
            Self::format_bytes(message_size.get_bytes() as usize),
            Self::format_bytes(message_count * (message_size.get_bytes() as usize)),
            start.elapsed(),
        );
        info!(
            self.log,
            "Client: requested bandwidth = {}/sec, rps = {:?}, qfull = {}",
            Self::bandwidth(
                message_count * (message_size.get_bytes() as usize),
                start.elapsed()
            ),
            rps,
            qfull,
        );

        // Wait for all the responses
        handle.join().unwrap();
    }
// The client side to send the messages
The client side to send the messages
The client side to send the messages
fn client(&self, message_size: Byte, message_count: usize, rps: Option<u32>) {
        // Spawn the response handler thread
        let request_map = Arc::new(Mutex::new(HashMap::new()));
        let request_map_cl = request_map.clone();
        let receiver_cl = self.receiver.clone();
        let log_cl = self.log.clone();
        let handle = std::thread::spawn(move || {
            Self::client_response_handler(
                message_size,
                message_count,
                request_map_cl,
                receiver_cl,
                log_cl,
            );
        });
        let mut rate_limiter: Box<dyn RateLimiter> = match rps {
            Some(r) => Box::new(RateLimiterImpl::new(r)),
            None => Box::new(NoopRateLimiter {}),
        };
        // Send the requests
        let start = Instant::now();
        let mut qfull = 0;
        let mut last_update = Instant::now();
        for i in 1..=message_count {
            rate_limiter.wait();
            let id = i as u64;
            let request = TestMessage::Request(Request {
                id,
                payload_size: message_size.get_bytes() as usize,
            });
            request_map.lock().unwrap().insert(id, Instant::now());
            qfull += self.send_message(request);
            if last_update.elapsed().as_secs() > STATUS_UPDATE_SEC || (i % 1000) == 0 {
                info!(
                    self.log,
                    "Client: sent request[{}/{}], qfull = {}", i, message_count, qfull
                );
                last_update = Instant::now();
            }
        }
        info!(
            self.log,
            "Client: requests done, sent {} requests (message size {}, total {}) in {:?}",
            message_count,
            Self::format_bytes(message_size.get_bytes() as usize),
            Self::format_bytes(message_count * (message_size.get_bytes() as usize)),
            start.elapsed(),
        );
        info!(
            self.log,
            "Client: requested bandwidth = {}/sec, rps = {:?}, qfull = {}",
            Self::bandwidth(
                message_count * (message_size.get_bytes() as usize),
                start.elapsed()
            ),
            rps,
            qfull,
        );
        // Wait for all the responses
        handle.join().unwrap();
    }
fn client ( & self , message_size : Byte , message_count : usize , rps : Option < u32 > ) { // Spawn the response handler thread let request_map = Arc :: new ( Mutex :: new ( HashMap :: new ( ) ) ) ; let request_map_cl = request_map . clone ( ) ; let receiver_cl = self . receiver . clone ( ) ; let log_cl = self . log . clone ( ) ; let handle = std :: thread :: spawn ( move | | { Self :: client_response_handler ( message_size , message_count , request_map_cl , receiver_cl , log_cl , ) ; } ) ; let mut rate_limiter : Box < dyn RateLimiter > = match rps { Some ( r ) => Box :: new ( RateLimiterImpl :: new ( r ) ) , None => Box :: new ( NoopRateLimiter { } ) , } ; // Send the requests let start = Instant :: now ( ) ; let mut qfull = 0 ; let mut last_update = Instant :: now ( ) ; for i in 1 ..= message_count { rate_limiter . wait ( ) ; let id = i as u64 ; let request = TestMessage :: Request ( Request { id , payload_size : message_size . get_bytes ( ) as usize , } ) ; request_map . lock ( ) . unwrap ( ) . insert ( id , Instant :: now ( ) ) ; qfull += self . send_message ( request ) ; if last_update . elapsed ( ) . as_secs ( ) > STATUS_UPDATE_SEC || ( i % 1000 ) == 0 { info ! ( self log " " i message_count qfull ) ; last_update = Instant :: now ( ) ; } } info ! ( self log " " message_count Self format_bytes ( message_size get_bytes ( ) as usize ) Self format_bytes ( message_count ( message_size get_bytes ( ) as usize ) ) start elapsed ( ) ) ; info ! ( self log " " Self bandwidth ( message_count ( message_size get_bytes ( ) as usize ) start elapsed ( ) ) rps qfull ) ; // Wait for all the responses handle . join ( ) . unwrap ( ) ; }
The client side to send the messages
The client side to send the messages
&self
message_size
Byte
message_count
usize
rps
Option<u32>
&self
message_size
Byte
message_count
usize
rps
Option<u32>