How to read all messages from a channel

[erelde]

@fritshoogland

 pub fn read_statements_into_vector(
+    hostname_port_vec: &[&str],
-    hostname_port_vec: &Vec<&str>,
     parallel: usize
 ) -> Vec<StoredStatements>
 {
     let pool = rayon::ThreadPoolBuilder::new().num_threads(parallel).build().unwrap();
-    let (tx, rx) = channel();
     pool.scope(|s| {
+        let (tx, rx) = channel();
         for hostname_port in hostname_port_vec {
             let tx = tx.clone();
             s.spawn(move |_| {
                 let detail_snapshot_time = Local::now();
                 let statements = read_statements(&hostname_port);
                 tx.send((hostname_port, detail_snapshot_time, statements)).expect("channel will be waiting in the pool");
             });
         }
+        drop(tx);
+        let mut stored_statements: Vec<StoredStatements> = Vec::new();
+        for (hostname_port, detail_snapshot_time, statements) in rx {
+            add_to_statements_vector(statements, hostname_port, detail_snapshot_time, &mut stored_statements);
+        }
+        stored_statements
+    })
-    });
-    let mut stored_statements: Vec<StoredStatements> = Vec::new();
-    for (hostname_port, detail_snapshot_time, statements) in rx {
-        add_to_statements_vector(statements, hostname_port, detail_snapshot_time, -&mut stored_statements);
-    }
-   stored_statements
 }

Again, by putting the collecting loop outside the pool.scope you are effectively collecting every items into your channel and then recollecting them into a Vec.
The thread with the collecting loop (using for .. in or iterator methods) should live inside the thread pool's scope.

And that :

let mut stored_statements: Vec<StoredStatements> = Vec::new();
for (hostname_port, detail_snapshot_time, statements) in rx {
      add_to_statements_vector(statements, hostname_port, detail_snapshot_time, &mut stored_statements);
}
stored_statements

Would be better expressed in Rust with :

rx.into_iter()
    .map(|(hostname_port, detail_snapshot_time, statements)| make_statement(statements, hostname_port, detail_snapshot_time)
    .collect();

Which avoids the temporary mutable borrow.

See this playground for a few improvements you could make.

[fritshoogland]

I do see and understand that by not providing a complete example I am making things complicated. Again, thank you for your effort.

read_statements() returns a vector of struct Statements, which are multiple rows from a single host. It can be hundreds of statements for each host.
The statements do not have any time or host indication with them, so what this function does is add that data, for each row into a new vector which has these fields added. And they all need to end up in the same vector for all hosts.

so each invocation of read_statements returns a Vec, which then is needs to be processed to Vec, which contains the same data, with some additional fields, which is supposed to collect all the Statements from all the hosts:

pub struct Queries {
    pub query: String,
    pub calls: i64,
    pub total_time: f64,
    pub min_time: f64,
    pub max_time: f64,
    pub mean_time: f64,
    pub stddev_time: f64,
    pub rows: i64,

pub struct StoredStatements {
    pub hostname_port: String,
    pub timestamp: DateTime<Local>,
    pub query: String,
    pub calls: i64,
    pub total_time: f64,
    pub min_time: f64,
    pub max_time: f64,
    pub mean_time: f64,
    pub stddev_time: f64,
    pub rows: i64,
}

The read_statements_into_vector() function returns Vec

The reason for telling is that I don't understand how the make_statement() function could work, because it seems to take and return a single statement.

[erelde]

Does this playground fit more your actual code ?

Note that the functions marked "stub" are only there to show examples, they could do anything.

[riking]

@fritshoogland

read_statements() returns a vector of struct Statements, which are multiple rows from a single host. It can be hundreds of statements for each host.

The general solution to this is to use .flat_map(|i| [1, 2, i+3]).collect() // => 1, 2, 3, 1, 2, 4 instead of .map(|i| i+3) // => 3, 4, 5.

Conceptually, flat_map peels back one layer from the containers you return from the closure.