Как поправить код для отдельного потока, чтобы обрабатывать данные по заданному времени?

Question

[TicSo]

Приведенный ниже код позволяет вводить в консоли текст для его хранения
или удаления в/из вектора v_txt_main
Его данные передаются в отдельный поток и также выводятся в консоли.

Не получается сделать следующее:
Нужно выводить эти данные (в отдельном потоке в консоли) раз в 10 секунд,
но предварительно получать их сюда `let job = rx.recv();` только на девятой секунде,
т.е. за секунду до вывода данных в отдельном потоке в консоли.

Таким образом буду выводить актуальные данные вектора, давая ~ 9 сек. на его изменение.
Заданные интервалы (10 и 1 секунды), - тестовые и будут уменьшены, поэтому скорость кода важна. Как делал:

(время dt_start, от которого нужен шаг в 10 сек. завёл так)

fn main() {

   let dt_start: DateTime<chrono::Utc> = Utc::now();
   println!("---\n{:?}\n---\n", &dt_start);
   
   let mut v_txt_main: Vec<String> = Vec::new();
   let (tx_01, rx_01):(Sender<DateTime<chrono::Utc>>, Receiver<_>) = mpsc::channel();
   let (tx_02, rx_02):(Sender<Vec<String>>, Receiver<_>) = mpsc::channel();

   thread::spawn(move || {
      let job_01: Result<chrono::DateTime<chrono::Utc>, std::sync::mpsc::RecvError> = rx_01.recv();
      println!("dt_start to thread: {:?}", &job_01);

      let mut job_02: Result<Vec<String>, std::sync::mpsc::RecvError>;
      loop {
        job_02 = rx_02.recv();
        for v in job_02.unwrap().iter() {
         println!("from thread: {:?} -> {:?} -> {:?} -> {:?}", &v, Utc::now(), Utc::now().timestamp_subsec_millis(), &job_01.unwrap());
        }
        sleep(Duration::from_millis(10));
      }
   });

   
   let _ = tx_01.send(dt_start);

   loop {
     let mut h_add_del_txt: HashMap<u8, String> = HashMap::new();
     let console: (u8, String) = f01(); 
     // println!("Принято с консоли: {:?}", &console);

     if console.0 == 1 {
      h_add_del_txt.insert(1, console.1);
     } else if console.0 == 2 {
      h_add_del_txt.insert(2, console.1);
     } else {  
      // ...
     }

     for (k, v) in &h_add_del_txt {
      if v_txt_main.iter().any(|n| *n == *v) {
         
         if *k == 2 {
            v_txt_main.retain(|a| *a != *v);
         } else {
            // ...
         }
      } else {
         if *k == 1 {
            v_txt_main.push(v.clone());
         } else {
            // ...
         }     
      }
     }

     let _ = tx_02.send(v_txt_main.clone());
     sleep(Duration::from_millis(20));
   }
}

Оборачивал код отдельного потока в циклы (loop) и по делению времени без остатка
пытался решить вопрос ... , - не получилось.

код

// [dependencies]
// chrono = "0.4.41"

use std::io;
use std::collections::HashMap;
//
use std::thread;
use std::thread::sleep;
use std::time::Duration;
use std::sync::mpsc;
//
use std::sync::mpsc::Sender;
use std::sync::mpsc::Receiver;
//
use chrono::prelude::*;

fn main() {

   let mut v_txt_main: Vec<String> = Vec::new();
   let (tx, rx):(Sender<Vec<String>>, Receiver<_>) = mpsc::channel();

   thread::spawn(move || {
      loop {
        let job = rx.recv();
        for v in job.unwrap().iter() {
         println!("from thread: {:?} -> {:?} -> {:?}", &v, Utc::now(), Utc::now().timestamp_subsec_millis());
        }
        sleep(Duration::from_millis(10));
      }
   });


   loop {
     let mut h_add_del_txt: HashMap<u8, String> = HashMap::new();
     let console: (u8, String) = f01(); 
     // println!("Принято с консоли: {:?}", &console);

     if console.0 == 1 {
      h_add_del_txt.insert(1, console.1);
     } else if console.0 == 2 {
      h_add_del_txt.insert(2, console.1);
     } else {  
      // ...
     }

     for (k, v) in &h_add_del_txt {
      if v_txt_main.iter().any(|n| *n == *v) {
         
         if *k == 2 {
            v_txt_main.retain(|a| *a != *v);
         } else {
            // ...
         }
      } else {
         if *k == 1 {
            v_txt_main.push(v.clone());
         } else {
            // ...
         }     
      }
     }

     let _ = tx.send(v_txt_main.clone());
     sleep(Duration::from_millis(20));
   }
}


fn f01() -> (u8, String) {

   println!("\nВведите вариант (1 или 2): \n 1 - для добавления `ТЕКСТА`\n или \n 2 - для удаления `ТЕКСТА`\n");   
   let mut flag = String::new();
   io::stdin().read_line(&mut flag).expect("Failed to read line");
   let flag: u8 = flag.trim().parse().expect("Please type a number!");

   if flag == 1 {
     println!("Внесите ТЕКСТ на добавление:");   
     let mut g = String::new();
     io::stdin().read_line(&mut g).expect("Failed to read line");
     let g: String = g.trim().parse().expect("Please type ...");

     return (flag, g);
   } else if flag == 2 {
     println!("Внесите ТЕКСТ на удаление:");   
     let mut g = String::new();
     io::stdin().read_line(&mut g).expect("Failed to read line");
     let g: String = g.trim().parse().expect("Please type a ...");
     return (flag, g);
   } else {
     return (0, "Error_...".to_string());
   }     
}

Answer 1

[Дмитрий Беляев]

Во-первых, следует понять, что канал по своей природе - это очередь.
То есть, если пользователь успеет за отведённое время сделать 5 изменений, то наш дочерний поток получит 5 копий вектора, а судя по задаче там ожидается только актуальная последняя копия.
Гораздо логичнее просто пошарить между потоками единственный вектор обёрнутый в Arc<RwLock<...>> или Arc<Mutex<...>>, заодно и памяти меньше сожрём и не будем греть проц бесполезными аллокациями памяти и копированием вектора.

Но если всё же задача другая, то у Receiver канала помимо метода recv, который блокирует поток до получения сообщения, есть ещё методы try_recv (не блокирует поток вообще) и recv_timeout (блокирует, но не дольше переданного таймаута).
https://doc.rust-lang.org/std/sync/mpsc/struct.Rec...
https://doc.rust-lang.org/std/sync/mpsc/struct.Rec...

P.S.

Duration::from_millis(10)

тут явно не 10 секунд...

P.S.S. Перформанс сия кода ушёл в отставку:
В коде есть лишние итерации по вектору.
В коде куча лишних аллокаций и копирования памяти.
Всю задачу можно вообще решить на одном потоке и оно будет работать быстрее ибо про синк кэшей проца тоже не надо забывать.

Ну и так по мелочи:

if console.0 == 1 {
    h_add_del_txt.insert(1, console.1);
} else if console.0 == 2 {
     h_add_del_txt.insert(2, console.1);
} else {  
     // ...
}

match, enum вместо u8

et mut g = String::new();
     io::stdin().read_line(&mut g).expect("Failed to read line");
     let g: String = g.trim().parse().expect("Please type a ...");

WTF?

Ну и паниковать на каждый чих так себе идея...

Comments

[TicSo]

Спасибо за помощь, Дмитрий.