Skip to content

Commit

Permalink
add a clock to validator windows (part 3 of solana-labs#309)
Browse files Browse the repository at this point in the history
 * count entries processed by Bank
 * initialize windows with initial height of Entries
  • Loading branch information
rob-solana committed Jun 25, 2018
1 parent c22ef50 commit 4f377dd
Show file tree
Hide file tree
Showing 5 changed files with 23 additions and 6 deletions.
15 changes: 13 additions & 2 deletions src/bank.rs
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,11 @@ pub struct Bank {
/// The number of transactions the bank has processed without error since the
/// start of the ledger.
transaction_count: AtomicUsize,

/// The number of Entries the bank has processed without error since start
/// of the ledger, i.e. poor-man's network synchronization
/// TODO: upgrade to U64 when stable?
entry_count: AtomicUsize,
}

impl Bank {
Expand All @@ -100,6 +105,7 @@ impl Bank {
time_sources: RwLock::new(HashSet::new()),
last_time: RwLock::new(Utc.timestamp(0, 0)),
transaction_count: AtomicUsize::new(0),
entry_count: AtomicUsize::new(0),
};
bank.apply_payment(deposit, &mut bank.balances.write().unwrap());
bank
Expand Down Expand Up @@ -296,19 +302,21 @@ impl Bank {
}

/// Process an ordered list of entries.
pub fn process_entries<I>(&self, entries: I) -> Result<()>
pub fn process_entries<I>(&self, entries: I) -> Result<usize>
where
I: IntoIterator<Item = Entry>,
{
for entry in entries {
self.entry_count.fetch_add(1, Ordering::Relaxed);

if !entry.transactions.is_empty() {
for result in self.process_transactions(entry.transactions) {
result?;
}
}
self.register_entry_id(&entry.id);
}
Ok(())
Ok(self.entry_count())
}

/// Process a Witness Signature. Any payment plans waiting on this signature
Expand Down Expand Up @@ -422,6 +430,9 @@ impl Bank {
pub fn transaction_count(&self) -> usize {
self.transaction_count.load(Ordering::Relaxed)
}
pub fn entry_count(&self) -> usize {
self.entry_count.load(Ordering::Relaxed)
}
}

#[cfg(test)]
Expand Down
3 changes: 2 additions & 1 deletion src/bin/fullnode.rs
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,8 @@ fn main() {
bank.register_entry_id(&entry1.id);

eprintln!("processing entries...");
bank.process_entries(entries).expect("process_entries");
let num_entries = bank.process_entries(entries).expect("process_entries");
eprintln!("processed {} entries...", num_entries);

eprintln!("creating networking stack...");

Expand Down
8 changes: 5 additions & 3 deletions src/streamer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -395,13 +395,14 @@ pub fn window(
r: BlobReceiver,
s: BlobSender,
retransmit: BlobSender,
entry_count: usize,
) -> JoinHandle<()> {
Builder::new()
.name("solana-window".to_string())
.spawn(move || {
let mut consumed = 0;
let mut received = 0;
let mut last = 0;
let mut consumed = entry_count;
let mut received = entry_count;
let mut last = entry_count;
let mut times = 0;
loop {
if exit.load(Ordering::Relaxed) {
Expand Down Expand Up @@ -816,6 +817,7 @@ mod test {
r_reader,
s_window,
s_retransmit,
0,
);
let (s_responder, r_responder) = channel();
let t_responder = responder(
Expand Down
1 change: 1 addition & 0 deletions src/tvu.rs
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ impl Tvu {
exit.clone(),
blob_recycler.clone(),
fetch_stage.blob_receiver,
bank.entry_count(),
);

let replicate_stage =
Expand Down
2 changes: 2 additions & 0 deletions src/window_stage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ impl WindowStage {
exit: Arc<AtomicBool>,
blob_recycler: packet::BlobRecycler,
fetch_stage_receiver: streamer::BlobReceiver,
entry_count: usize,
) -> Self {
let (retransmit_sender, retransmit_receiver) = channel();

Expand All @@ -41,6 +42,7 @@ impl WindowStage {
fetch_stage_receiver,
blob_sender,
retransmit_sender,
entry_count,
);
let thread_hdls = vec![t_retransmit, t_window];

Expand Down

0 comments on commit 4f377dd

Please sign in to comment.