Make calculation in parallel

src/value_estimation_team/indicators/rsi.rs

@@ -7,7 +7,9 @@ use csv::{DeserializeRecordsIter, StringRecord};
 use serde::Deserialize;
 use sqlx::FromRow;
 use std::f64::NAN;
-use tokio::{fs::*, io::AsyncWriteExt, time::*};
+use futures::future::try_join_all;
+use std::sync::Arc;
+use tokio::{fs::*, io::AsyncWriteExt, sync::Mutex, time::*};
 
 #[derive(Clone, Debug)]
 pub struct RsiData {
@@ -28,11 +30,25 @@ impl RsiData {
 pub async fn rsi(
     rsi_number: usize,
     input_rt_data: &Vec<(String, Vec<RealtimePriceData>)>,
-    output_rsi_data: &mut Vec<(String, Vec<RsiData>)>,
     filtered_symbols: &Vec<(String, i64)>,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+) -> Result<Vec<(String, Vec<RsiData>)>, Box<dyn std::error::Error + Send + Sync>> {
     let instant = Instant::now();
 
+    let mut rsi_data_wrapper: Vec<(String, Vec<RsiData>)> = Vec::new();
+    let mut rsi_data_wrapper_arc = Arc::new(Mutex::new(rsi_data_wrapper));
+    
+    let mut task_vec = Vec::new();
+    for element in filtered_symbols {
+
+        let element_c = element.clone();
+        let rsi_data_wrapper_arc_c = Arc::clone(&rsi_data_wrapper_arc);
+        let symbol_search_result = input_rt_data.iter().position(|x| x.0 == *element_c.0);
+        match symbol_search_result {
+            Some(T) => {
+                let input_rt_data_c = input_rt_data.clone();
+                task_vec.push(tokio::spawn(async move {
+                    let mut rsi_data_vec: Vec<RsiData> = Vec::new();
+                    let mut rsi_data = RsiData::new();
                     let mut read_data_vec: Vec<RealtimePriceData> = Vec::new();
                     let mut read_price_buffer: Vec<RealtimePriceData> = Vec::new();
                     let mut prev_price: f64 = 0.0;
@@ -42,22 +58,12 @@ pub async fn rsi(
                     let mut rsi: f64 = 0.0;
                     let mut last_close_time = 0;
 
-    let mut rsi_data_wrapper: Vec<(String, Vec<RsiData>)> = Vec::new();
+                    if input_rt_data_c[T].1.len() < rsi_number + 1 {
-    let mut rsi_data_vec: Vec<RsiData> = Vec::new();
-    let mut rsi_data = RsiData::new();
-    for symbol in filtered_symbols {
-        read_data_vec.clear();
-        read_price_buffer.clear();
-        rsi_data_vec.clear();
-        let symbol_search_result = input_rt_data.iter().position(|x| x.0 == *symbol.0);
-        match symbol_search_result {
-            Some(T) => {
-                if input_rt_data[T].1.len() < rsi_number + 1 {
                         rsi_data.rsi_value = 0.0;
                         rsi_data.close_time = 0;
                         rsi_data_vec.push(rsi_data.clone());
                     } else {
-                    read_data_vec = input_rt_data[T].1.clone();
+                        read_data_vec = input_rt_data_c[T].1.clone();
                         if read_data_vec.len() >= (150 + rsi_number) as usize {
                             read_data_vec.reverse();
                             read_data_vec.truncate((150 + rsi_number) as usize);
@@ -138,14 +144,15 @@ pub async fn rsi(
                             rsi_data.close_time = buffer.last().unwrap().close_time;
                             rsi_data_vec.push(rsi_data.clone());
                         }
-                    rsi_data_wrapper.push((symbol.0.clone(), rsi_data_vec.clone()));
+                        let mut rsi_data_wrapper_lock = rsi_data_wrapper_arc_c.lock().await;
+                        rsi_data_wrapper_lock.push((element_c.0.clone(), rsi_data_vec.clone()));
                     }
+                }));
             }
             None => {}
         }
     }
-    *output_rsi_data = rsi_data_wrapper;
+    try_join_all(task_vec).await?;
-    // println!("                                              indicators/rsi{} 완료 elapsed:{:.2}s", rsi_number, instant.elapsed().as_secs_f32());
+    let a = rsi_data_wrapper_arc.lock().await.to_owned();
-
+    Ok(a)
-    Ok(())
 }