gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Use WindowUtil for more idiomatic code

Browse Source
This commit is contained in:
Greg Fitzgerald
2018-09-07 13:38:48 -06:00
parent d2c9beb843
commit 2b44c4504a
2 changed files with 6 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/broadcast_stage.rs
View File

@@ -16,7 +16,7 @@ use std::sync::{Arc, RwLock};
use std::thread::{self, Builder, JoinHandle}; use std::thread::{self, Builder, JoinHandle};
use std::time::Duration; use std::time::Duration;
use streamer::BlobReceiver; use streamer::BlobReceiver;
use window::{self, SharedWindow, WindowIndex, WINDOW_SIZE}; use window::{self, SharedWindow, WindowIndex, WindowUtil, WINDOW_SIZE};
fn broadcast( fn broadcast(
node_info: &NodeInfo, node_info: &NodeInfo,
@@ -42,12 +42,7 @@ fn broadcast(
// break them up into window-sized chunks to process // break them up into window-sized chunks to process
let blobs_chunked = blobs_vec.chunks(WINDOW_SIZE as usize).map(|x| x.to_vec()); let blobs_chunked = blobs_vec.chunks(WINDOW_SIZE as usize).map(|x| x.to_vec());
if log_enabled!(Level::Trace) { trace!("{}", window.read().unwrap().print(&id, *receive_index));
trace!(
"{}",
window::print_window(&window.read().unwrap(), &id, *receive_index)
);
}
for mut blobs in blobs_chunked { for mut blobs in blobs_chunked {
let blobs_len = blobs.len(); let blobs_len = blobs.len();

17
src/window.rs
View File

@@ -198,7 +198,7 @@ fn repair_window(
let num_peers = crdt.read().unwrap().table.len() as u64; let num_peers = crdt.read().unwrap().table.len() as u64;
let highest_lost = calculate_highest_lost_blob_index(num_peers, consumed, received); let highest_lost = calculate_highest_lost_blob_index(num_peers, consumed, received);
let idxs = clear_window_slots(window, recycler, consumed, highest_lost); let idxs = window.clear_slots(recycler, consumed, highest_lost);
let reqs: Vec<_> = idxs let reqs: Vec<_> = idxs
.into_iter() .into_iter()
.filter_map(|pix| crdt.read().unwrap().window_index_request(pix).ok()) .filter_map(|pix| crdt.read().unwrap().window_index_request(pix).ok())
@@ -525,8 +525,7 @@ fn recv_window(
trace!("{} window pix: {} size: {}", id, pix, meta_size); trace!("{} window pix: {} size: {}", id, pix, meta_size);
process_blob( window.write().unwrap().process_blob(
&mut window.write().unwrap(),
id, id,
b, b,
pix, pix,
@@ -538,7 +537,7 @@ fn recv_window(
); );
} }
if log_enabled!(Level::Trace) { if log_enabled!(Level::Trace) {
trace!("{}", print_window(&window.read().unwrap(), id, *consumed)); trace!("{}", window.read().unwrap().print(id, *consumed));
trace!( trace!(
"{}: consumed: {} received: {} sending consume.len: {} pixs: {:?} took {} ms", "{}: consumed: {} received: {} sending consume.len: {} pixs: {:?} took {} ms",
id, id,
@@ -728,15 +727,7 @@ pub fn window_service(
} }
let mut window = window.write().unwrap(); let mut window = window.write().unwrap();
let reqs = repair_window( let reqs = window.repair(&crdt, &recycler, &id, times, consumed, received);
&mut window,
&crdt,
&recycler,
&id,
times,
consumed,
received,
);
for (to, req) in reqs { for (to, req) in reqs {
repair_socket.send_to(&req, to).unwrap_or_else(|e| { repair_socket.send_to(&req, to).unwrap_or_else(|e| {
info!("{} repair req send_to({}) error {:?}", id, to, e); info!("{} repair req send_to({}) error {:?}", id, to, e);