pub async fn run_worker(
    opt: &DbOpt,
    worker_id: u32,
    num_devices: u32,
    num_measurements: u32,
) -> Result<usize> {
    let batch_size = opt.batch_size;
    let num_metrics = opt.num_metrics;
    let (tx, mut rx) = mpsc::channel::<Vec<Measurement>>(100);

    thread::spawn(move || {
        let mut measurements =
            MeasurementIterator::new(worker_id, num_devices, num_metrics, num_measurements);
        let mut num_sent = 0;
        let send_until = num_devices * num_measurements;

        while num_sent < send_until {
            let mut data = Vec::with_capacity(batch_size);
            for m in &mut measurements {
                data.push(m);
                num_sent += 1;

                if data.len() == batch_size || num_sent == send_until {
                    break;
                }
            }

            if tx.blocking_send(data).is_err() {
                break; // Reader disconnected
            }
        }
    });

    let mut num_written = 0;
    let mut db = Db::connect(opt).await?;
    let command = command::Command::new(opt, &db.client).await?;

    while let Some(data) = rx.recv().await {
        num_written += if opt.dry_run {
            data.len()
        } else {
            command.execute(&mut db.client, data).await?
        };
    }

    Ok(num_written)
}
/// Run worker writes devices * measurements rows.
Run worker writes devices * measurements rows.
Run worker writes devices * measurements rows .
pub async fn run_worker(
    opt: &DbOpt,
    worker_id: u32,
    num_devices: u32,
    num_measurements: u32,
) -> Result<usize> {
    let batch_size = opt.batch_size;
    let num_metrics = opt.num_metrics;
    let (tx, mut rx) = mpsc::channel::<Vec<Measurement>>(100);
    thread::spawn(move || {
        let mut measurements =
            MeasurementIterator::new(worker_id, num_devices, num_metrics, num_measurements);
        let mut num_sent = 0;
        let send_until = num_devices * num_measurements;
        while num_sent < send_until {
            let mut data = Vec::with_capacity(batch_size);
            for m in &mut measurements {
                data.push(m);
                num_sent += 1;
                if data.len() == batch_size || num_sent == send_until {
                    break;
                }
            }
            if tx.blocking_send(data).is_err() {
                break; // Reader disconnected
            }
        }
    });
    let mut num_written = 0;
    let mut db = Db::connect(opt).await?;
    let command = command::Command::new(opt, &db.client).await?;
    while let Some(data) = rx.recv().await {
        num_written += if opt.dry_run {
            data.len()
        } else {
            command.execute(&mut db.client, data).await?
        };
    }
    Ok(num_written)
}
pub async fn run_worker ( opt : & DbOpt , worker_id : u32 , num_devices : u32 , num_measurements : u32 , ) -> Result < usize > { let batch_size = opt . batch_size ; let num_metrics = opt . num_metrics ; let ( tx , mut rx ) = mpsc :: channel :: < Vec < Measurement > > ( 100 ) ; thread :: spawn ( move | | { let mut measurements = MeasurementIterator :: new ( worker_id , num_devices , num_metrics , num_measurements ) ; let mut num_sent = 0 ; let send_until = num_devices * num_measurements ; while num_sent < send_until { let mut data = Vec :: with_capacity ( batch_size ) ; for m in & mut measurements { data . push ( m ) ; num_sent += 1 ; if data . len ( ) == batch_size || num_sent == send_until { break ; } } if tx . blocking_send ( data ) . is_err ( ) { break ; // Reader disconnected } } } ) ; let mut num_written = 0 ; let mut db = Db :: connect ( opt ) . await ? ; let command = command :: Command :: new ( opt , & db . client ) . await ? ; while let Some ( data ) = rx . recv ( ) . await { num_written += if opt . dry_run { data . len ( ) } else { command . execute ( & mut db . client , data ) . await ? } ; } Ok ( num_written ) }
Run worker writes devices * measurements rows.
Run worker writes devices * measurements rows .
opt
&DbOpt
worker_id
u32
num_devices
u32
num_measurements
u32
opt
&DbOpt
worker_id
u32
num_devices
u32
num_measurements
u32