* count entries processed by Bank * initialize windows with initial height of Entries
This commit is contained in:
parent
3966eb5374
commit
1919ec247b
15
src/bank.rs
15
src/bank.rs
|
@ -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 {
|
||||
|
@ -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
|
||||
|
@ -296,11 +302,13 @@ 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?;
|
||||
|
@ -308,7 +316,7 @@ impl Bank {
|
|||
}
|
||||
self.register_entry_id(&entry.id);
|
||||
}
|
||||
Ok(())
|
||||
Ok(self.entry_count())
|
||||
}
|
||||
|
||||
/// Process a Witness Signature. Any payment plans waiting on this signature
|
||||
|
@ -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)]
|
||||
|
|
|
@ -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...");
|
||||
|
||||
|
|
|
@ -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) {
|
||||
|
@ -816,6 +817,7 @@ mod test {
|
|||
r_reader,
|
||||
s_window,
|
||||
s_retransmit,
|
||||
0,
|
||||
);
|
||||
let (s_responder, r_responder) = channel();
|
||||
let t_responder = responder(
|
||||
|
|
|
@ -86,6 +86,7 @@ impl Tvu {
|
|||
exit.clone(),
|
||||
blob_recycler.clone(),
|
||||
fetch_stage.blob_receiver,
|
||||
bank.entry_count(),
|
||||
);
|
||||
|
||||
let replicate_stage =
|
||||
|
|
|
@ -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();
|
||||
|
||||
|
@ -41,6 +42,7 @@ impl WindowStage {
|
|||
fetch_stage_receiver,
|
||||
blob_sender,
|
||||
retransmit_sender,
|
||||
entry_count,
|
||||
);
|
||||
let thread_hdls = vec![t_retransmit, t_window];
|
||||
|
||||
|
|
Loading…
Reference in New Issue