From 2b44c4504a90f76e686b26cb63f3d9c6de0dfb1c Mon Sep 17 00:00:00 2001 From: Greg Fitzgerald Date: Fri, 7 Sep 2018 13:38:48 -0600 Subject: [PATCH] Use WindowUtil for more idiomatic code --- src/broadcast_stage.rs | 9 ++------- src/window.rs | 17 ++++------------- 2 files changed, 6 insertions(+), 20 deletions(-) diff --git a/src/broadcast_stage.rs b/src/broadcast_stage.rs index 1c69f703a..6af68bd93 100644 --- a/src/broadcast_stage.rs +++ b/src/broadcast_stage.rs @@ -16,7 +16,7 @@ use std::sync::{Arc, RwLock}; use std::thread::{self, Builder, JoinHandle}; use std::time::Duration; use streamer::BlobReceiver; -use window::{self, SharedWindow, WindowIndex, WINDOW_SIZE}; +use window::{self, SharedWindow, WindowIndex, WindowUtil, WINDOW_SIZE}; fn broadcast( node_info: &NodeInfo, @@ -42,12 +42,7 @@ fn broadcast( // break them up into window-sized chunks to process let blobs_chunked = blobs_vec.chunks(WINDOW_SIZE as usize).map(|x| x.to_vec()); - if log_enabled!(Level::Trace) { - trace!( - "{}", - window::print_window(&window.read().unwrap(), &id, *receive_index) - ); - } + trace!("{}", window.read().unwrap().print(&id, *receive_index)); for mut blobs in blobs_chunked { let blobs_len = blobs.len(); diff --git a/src/window.rs b/src/window.rs index fae26c5e5..bc5650b1a 100644 --- a/src/window.rs +++ b/src/window.rs @@ -198,7 +198,7 @@ fn repair_window( let num_peers = crdt.read().unwrap().table.len() as u64; 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 .into_iter() .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); - process_blob( - &mut window.write().unwrap(), + window.write().unwrap().process_blob( id, b, pix, @@ -538,7 +537,7 @@ fn recv_window( ); } if log_enabled!(Level::Trace) { - trace!("{}", print_window(&window.read().unwrap(), id, *consumed)); + trace!("{}", window.read().unwrap().print(id, *consumed)); trace!( "{}: consumed: {} received: {} sending consume.len: {} pixs: {:?} took {} ms", id, @@ -728,15 +727,7 @@ pub fn window_service( } let mut window = window.write().unwrap(); - let reqs = repair_window( - &mut window, - &crdt, - &recycler, - &id, - times, - consumed, - received, - ); + let reqs = window.repair(&crdt, &recycler, &id, times, consumed, received); for (to, req) in reqs { repair_socket.send_to(&req, to).unwrap_or_else(|e| { info!("{} repair req send_to({}) error {:?}", id, to, e);