blockworks-foundation
/
solana
mirror of https://github.com/blockworks-foundation/solana.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

less intrusive counters (#655) 

* less intrusive counters

* fixed arg

* tests

* comments
This commit is contained in:
anatoly yakovenko 2018-07-16 18:33:50 -07:00 committed by GitHub
parent f6eb85e7a3
commit 09f2d273c5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 43 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/bank.rs
View File

@ -6,7 +6,7 @@
extern crate libc; extern crate libc;
use chrono::prelude::*; use chrono::prelude::*;
use counter::{self, Counter}; use counter::Counter;
use entry::Entry; use entry::Entry;
use hash::Hash; use hash::Hash;
use itertools::Itertools; use itertools::Itertools;
@ -204,17 +204,9 @@ impl Bank {
let option = bals.get_mut(&tx.from); let option = bals.get_mut(&tx.from);
if option.is_none() { if option.is_none() {
if let Instruction::NewVote(_) = &tx.instruction { if let Instruction::NewVote(_) = &tx.instruction {
static mut COUNTER_VOTE_ACCOUNT_NOT_FOUND: Counter = create_counter!( inc_new_counter!("bank-appy_debits-vote_account_not_found", 1);
"bank-appy_debits-vote_account_not_found",
counter::DEFAULT_LOG_RATE
);
inc_counter!(COUNTER_VOTE_ACCOUNT_NOT_FOUND, 1);
} else { } else {
static mut COUNTER_ACCOUNT_NOT_FOUND: Counter = create_counter!( inc_new_counter!("bank-appy_debits-generic_account_not_found", 1);
"bank-appy_debits-generic_account_not_found",
counter::DEFAULT_LOG_RATE
);
inc_counter!(COUNTER_ACCOUNT_NOT_FOUND, 1);
} }
return Err(BankError::AccountNotFound(tx.from)); return Err(BankError::AccountNotFound(tx.from));
} }

3
src/banking_stage.rs
View File

@ -89,7 +89,6 @@ impl BankingStage {
mms.len(), mms.len(),
); );
let count = mms.iter().map(|x| x.1.len()).sum(); let count = mms.iter().map(|x| x.1.len()).sum();
static mut COUNTER: Counter = create_counter!("banking_stage_process_packets", 1);
let proc_start = Instant::now(); let proc_start = Instant::now();
for (msgs, vers) in mms { for (msgs, vers) in mms {
let transactions = Self::deserialize_transactions(&msgs.read().unwrap()); let transactions = Self::deserialize_transactions(&msgs.read().unwrap());
@ -125,7 +124,7 @@ impl BankingStage {
reqs_len, reqs_len,
(reqs_len as f32) / (total_time_s) (reqs_len as f32) / (total_time_s)
); );
inc_counter!(COUNTER, count); inc_new_counter!("banking_stage-process_packets", count);
Ok(()) Ok(())
} }
} }

19
src/counter.rs
View File

@ -4,7 +4,6 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use timing; use timing;
const INFLUX_RATE: usize = 100; const INFLUX_RATE: usize = 100;
pub const DEFAULT_LOG_RATE: usize = 10;
pub struct Counter { pub struct Counter {
pub name: &'static str, pub name: &'static str,
@ -34,6 +33,17 @@ macro_rules! inc_counter {
}; };
} }
macro_rules! inc_new_counter {
($name:expr, $count:expr) => {{
static mut INC_NEW_COUNTER: Counter = create_counter!($name, 10);
inc_counter!(INC_NEW_COUNTER, $count);
}};
($name:expr, $count:expr, $lograte:expr) => {{
static mut INC_NEW_COUNTER: Counter = create_counter!($name, $lograte);
inc_counter!(INC_NEW_COUNTER, $count);
}};
}
impl Counter { impl Counter {
pub fn inc(&mut self, events: usize) { pub fn inc(&mut self, events: usize) {
let counts = self.counts.fetch_add(events, Ordering::Relaxed); let counts = self.counts.fetch_add(events, Ordering::Relaxed);
@ -89,4 +99,11 @@ mod tests {
assert_eq!(COUNTER.lastlog.load(Ordering::Relaxed), 399); assert_eq!(COUNTER.lastlog.load(Ordering::Relaxed), 399);
} }
} }
#[test]
fn test_inc_new_counter() {
//make sure that macros are syntactically correct
//the variable is internal to the macro scope so there is no way to introspect it
inc_new_counter!("counter-1", 1);
inc_new_counter!("counter-2", 1, 2);
}
} }

27
src/crdt.rs
View File

@ -37,8 +37,6 @@ use streamer::{BlobReceiver, BlobSender, Window};
use timing::timestamp; use timing::timestamp;
use transaction::Vote; use transaction::Vote;
const LOG_RATE: usize = 10;
/// milliseconds we sleep for between gossip requests /// milliseconds we sleep for between gossip requests
const GOSSIP_SLEEP_MILLIS: u64 = 100; const GOSSIP_SLEEP_MILLIS: u64 = 100;
const GOSSIP_PURGE_MILLIS: u64 = 15000; const GOSSIP_PURGE_MILLIS: u64 = 15000;
@ -348,8 +346,7 @@ impl Crdt {
} }
} }
pub fn insert_votes(&mut self, votes: &[(PublicKey, Vote, Hash)]) { pub fn insert_votes(&mut self, votes: &[(PublicKey, Vote, Hash)]) {
static mut COUNTER_VOTE: Counter = create_counter!("crdt-vote-count", LOG_RATE); inc_new_counter!("crdt-vote-count", votes.len());
inc_counter!(COUNTER_VOTE, votes.len());
if !votes.is_empty() { if !votes.is_empty() {
info!("{:x}: INSERTING VOTES {}", self.debug_id(), votes.len()); info!("{:x}: INSERTING VOTES {}", self.debug_id(), votes.len());
} }
@ -373,8 +370,7 @@ impl Crdt {
self.update_index += 1; self.update_index += 1;
let _ = self.table.insert(v.id, v.clone()); let _ = self.table.insert(v.id, v.clone());
let _ = self.local.insert(v.id, self.update_index); let _ = self.local.insert(v.id, self.update_index);
static mut COUNTER_UPDATE: Counter = create_counter!("crdt-update-count", LOG_RATE); inc_new_counter!("crdt-update-count", 1);
inc_counter!(COUNTER_UPDATE, 1);
} else { } else {
trace!( trace!(
"{:x}: INSERT FAILED data: {:x} new.version: {} me.version: {}", "{:x}: INSERT FAILED data: {:x} new.version: {} me.version: {}",
@ -446,8 +442,7 @@ impl Crdt {
}) })
.collect(); .collect();
static mut COUNTER_PURGE: Counter = create_counter!("crdt-purge-count", LOG_RATE); inc_new_counter!("crdt-purge-count", dead_ids.len());
inc_counter!(COUNTER_PURGE, dead_ids.len());
for id in &dead_ids { for id in &dead_ids {
self.alive.remove(id); self.alive.remove(id);
@ -898,24 +893,18 @@ impl Crdt {
outblob.meta.set_addr(&from.contact_info.tvu_window); outblob.meta.set_addr(&from.contact_info.tvu_window);
outblob.set_id(sender_id).expect("blob set_id"); outblob.set_id(sender_id).expect("blob set_id");
} }
static mut COUNTER_REQ_WINDOW_PASS: Counter = inc_new_counter!("crdt-window-request-pass", 1);
create_counter!("crdt-window-request-pass", LOG_RATE);
inc_counter!(COUNTER_REQ_WINDOW_PASS, 1);
return Some(out); return Some(out);
} else { } else {
static mut COUNTER_REQ_WINDOW_OUTSIDE: Counter = inc_new_counter!("crdt-window-request-outside", 1);
create_counter!("crdt-window-request-outside", LOG_RATE);
inc_counter!(COUNTER_REQ_WINDOW_OUTSIDE, 1);
info!( info!(
"requested ix {} != blob_ix {}, outside window!", "requested ix {} != blob_ix {}, outside window!",
ix, blob_ix ix, blob_ix
); );
} }
} else { } else {
static mut COUNTER_REQ_WINDOW_FAIL: Counter = inc_new_counter!("crdt-window-request-fail", 1);
create_counter!("crdt-window-request-fail", LOG_RATE);
inc_counter!(COUNTER_REQ_WINDOW_FAIL, 1);
assert!(window.read().unwrap()[pos].is_none()); assert!(window.read().unwrap()[pos].is_none());
info!( info!(
"{:x}: failed RequestWindowIndex {:x} {} {}", "{:x}: failed RequestWindowIndex {:x} {} {}",
@ -991,9 +980,7 @@ impl Crdt {
//TODO verify from is signed //TODO verify from is signed
obj.write().unwrap().insert(&from); obj.write().unwrap().insert(&from);
let me = obj.read().unwrap().my_data().clone(); let me = obj.read().unwrap().my_data().clone();
static mut COUNTER_REQ_WINDOW: Counter = inc_new_counter!("crdt-window-request-recv", 1);
create_counter!("crdt-window-request-recv", LOG_RATE);
inc_counter!(COUNTER_REQ_WINDOW, 1);
trace!( trace!(
"{:x}:received RequestWindowIndex {:x} {} ", "{:x}:received RequestWindowIndex {:x} {} ",
me.debug_id(), me.debug_id(),

4
src/packet.rs
View File

@ -19,7 +19,6 @@ pub type SharedBlobs = VecDeque<SharedBlob>;
pub type PacketRecycler = Recycler<Packets>; pub type PacketRecycler = Recycler<Packets>;
pub type BlobRecycler = Recycler<Blob>; pub type BlobRecycler = Recycler<Blob>;
const LOG_RATE: usize = 10;
pub const NUM_PACKETS: usize = 1024 * 8; pub const NUM_PACKETS: usize = 1024 * 8;
pub const BLOB_SIZE: usize = 64 * 1024; pub const BLOB_SIZE: usize = 64 * 1024;
pub const BLOB_DATA_SIZE: usize = BLOB_SIZE - BLOB_HEADER_SIZE; pub const BLOB_DATA_SIZE: usize = BLOB_SIZE - BLOB_HEADER_SIZE;
@ -188,7 +187,6 @@ impl<T: Default> Recycler<T> {
impl Packets { impl Packets {
fn run_read_from(&mut self, socket: &UdpSocket) -> Result<usize> { fn run_read_from(&mut self, socket: &UdpSocket) -> Result<usize> {
static mut COUNTER: Counter = create_counter!("packets", LOG_RATE);
self.packets.resize(NUM_PACKETS, Packet::default()); self.packets.resize(NUM_PACKETS, Packet::default());
let mut i = 0; let mut i = 0;
//DOCUMENTED SIDE-EFFECT //DOCUMENTED SIDE-EFFECT
@ -203,7 +201,7 @@ impl Packets {
trace!("receiving on {}", socket.local_addr().unwrap()); trace!("receiving on {}", socket.local_addr().unwrap());
match socket.recv_from(&mut p.data) { match socket.recv_from(&mut p.data) {
Err(_) if i > 0 => { Err(_) if i > 0 => {
inc_counter!(COUNTER, i); inc_new_counter!("packets-recv_count", 1);
debug!("got {:?} messages on {}", i, socket.local_addr().unwrap()); debug!("got {:?} messages on {}", i, socket.local_addr().unwrap());
break; break;
} }

20
src/replicate_stage.rs
View File

@ -2,7 +2,7 @@
use bank::Bank; use bank::Bank;
use bincode::serialize; use bincode::serialize;
use counter::{Counter, DEFAULT_LOG_RATE}; use counter::Counter;
use crdt::Crdt; use crdt::Crdt;
use ledger; use ledger;
use packet::BlobRecycler; use packet::BlobRecycler;
@ -52,14 +52,10 @@ impl ReplicateStage {
let mut wcrdt = crdt.write().unwrap(); let mut wcrdt = crdt.write().unwrap();
wcrdt.insert_votes(&votes); wcrdt.insert_votes(&votes);
}; };
{ inc_new_counter!(
static mut COUNTER_REPLICATE: Counter = "replicate-transactions",
create_counter!("replicate-transactions", DEFAULT_LOG_RATE); entries.iter().map(|x| x.transactions.len()).sum()
inc_counter!( );
COUNTER_REPLICATE,
entries.iter().map(|x| x.transactions.len()).sum()
);
}
let res = bank.process_entries(entries); let res = bank.process_entries(entries);
if res.is_err() { if res.is_err() {
error!("process_entries {} {:?}", blobs_len, res); error!("process_entries {} {:?}", blobs_len, res);
@ -84,11 +80,7 @@ impl ReplicateStage {
blob.meta.set_addr(&addr); blob.meta.set_addr(&addr);
blob.meta.size = len; blob.meta.size = len;
} }
{ inc_new_counter!("replicate-vote_sent", 1);
static mut COUNTER_REPLICATE_VOTE: Counter =
create_counter!("replicate-vote_sent", DEFAULT_LOG_RATE);
inc_counter!(COUNTER_REPLICATE_VOTE, 1);
}
*last_vote = now; *last_vote = now;
vote_blob_sender.send(VecDeque::from(vec![shared_blob]))?; vote_blob_sender.send(VecDeque::from(vec![shared_blob]))?;

6
src/sigverify.rs
View File

@ -77,7 +77,6 @@ fn batch_size(batches: &[SharedPackets]) -> usize {
#[cfg(not(feature = "cuda"))] #[cfg(not(feature = "cuda"))]
pub fn ed25519_verify(batches: &Vec<SharedPackets>) -> Vec<Vec<u8>> { pub fn ed25519_verify(batches: &Vec<SharedPackets>) -> Vec<Vec<u8>> {
use rayon::prelude::*; use rayon::prelude::*;
static mut COUNTER: Counter = create_counter!("ed25519_verify", 1);
let count = batch_size(batches); let count = batch_size(batches);
info!("CPU ECDSA for {}", batch_size(batches)); info!("CPU ECDSA for {}", batch_size(batches));
let rv = batches let rv = batches
@ -91,7 +90,7 @@ pub fn ed25519_verify(batches: &Vec<SharedPackets>) -> Vec<Vec<u8>> {
.collect() .collect()
}) })
.collect(); .collect();
inc_counter!(COUNTER, count); inc_new_counter!("ed25519_verify", count);
rv rv
} }
@ -109,7 +108,6 @@ pub fn init() {
#[cfg(feature = "cuda")] #[cfg(feature = "cuda")]
pub fn ed25519_verify(batches: &Vec<SharedPackets>) -> Vec<Vec<u8>> { pub fn ed25519_verify(batches: &Vec<SharedPackets>) -> Vec<Vec<u8>> {
use packet::PACKET_DATA_SIZE; use packet::PACKET_DATA_SIZE;
static mut COUNTER: Counter = create_counter!("ed25519_verify_cuda", 1);
let count = batch_size(batches); let count = batch_size(batches);
info!("CUDA ECDSA for {}", batch_size(batches)); info!("CUDA ECDSA for {}", batch_size(batches));
let mut out = Vec::new(); let mut out = Vec::new();
@ -169,7 +167,7 @@ pub fn ed25519_verify(batches: &Vec<SharedPackets>) -> Vec<Vec<u8>> {
num += 1; num += 1;
} }
} }
inc_counter!(COUNTER, count); inc_new_counter!("ed25519_verify", count);
rvs rvs
} }

20
src/streamer.rs
View File

@ -18,7 +18,6 @@ use std::sync::{Arc, RwLock};
use std::thread::{Builder, JoinHandle}; use std::thread::{Builder, JoinHandle};
use std::time::Duration; use std::time::Duration;
const LOG_RATE: usize = 10;
pub const WINDOW_SIZE: u64 = 2 * 1024; pub const WINDOW_SIZE: u64 = 2 * 1024;
pub type PacketReceiver = Receiver<SharedPackets>; pub type PacketReceiver = Receiver<SharedPackets>;
pub type PacketSender = Sender<SharedPackets>; pub type PacketSender = Sender<SharedPackets>;
@ -224,9 +223,7 @@ fn repair_window(
let reqs = find_next_missing(locked_window, crdt, consumed, received)?; let reqs = find_next_missing(locked_window, crdt, consumed, received)?;
trace!("{:x}: repair_window missing: {}", debug_id, reqs.len()); trace!("{:x}: repair_window missing: {}", debug_id, reqs.len());
if !reqs.is_empty() { if !reqs.is_empty() {
static mut COUNTER_REPAIR: Counter = inc_new_counter!("streamer-repair_window-repair", reqs.len());
create_counter!("streamer-repair_window-repair", LOG_RATE);
inc_counter!(COUNTER_REPAIR, reqs.len());
debug!( debug!(
"{:x}: repair_window counter times: {} consumed: {} received: {} missing: {}", "{:x}: repair_window counter times: {} consumed: {} received: {} missing: {}",
debug_id, debug_id,
@ -301,9 +298,7 @@ fn retransmit_all_leader_blocks(
*received, *received,
retransmit_queue.len(), retransmit_queue.len(),
); );
static mut COUNTER_RETRANSMIT: Counter = inc_new_counter!("streamer-recv_window-retransmit", retransmit_queue.len());
create_counter!("streamer-recv_window-retransmit", LOG_RATE);
inc_counter!(COUNTER_RETRANSMIT, retransmit_queue.len());
retransmit.send(retransmit_queue)?; retransmit.send(retransmit_queue)?;
} }
Ok(()) Ok(())
@ -413,8 +408,7 @@ fn recv_window(
while let Ok(mut nq) = r.try_recv() { while let Ok(mut nq) = r.try_recv() {
dq.append(&mut nq) dq.append(&mut nq)
} }
static mut COUNTER_RECV: Counter = create_counter!("streamer-recv_window-recv", LOG_RATE); inc_new_counter!("streamer-recv_window-recv", dq.len());
inc_counter!(COUNTER_RECV, dq.len());
debug!( debug!(
"{:x}: RECV_WINDOW {} {}: got packets {}", "{:x}: RECV_WINDOW {} {}: got packets {}",
debug_id, debug_id,
@ -480,9 +474,7 @@ fn recv_window(
consume_queue.len(), consume_queue.len(),
); );
trace!("sending consume_queue.len: {}", consume_queue.len()); trace!("sending consume_queue.len: {}", consume_queue.len());
static mut COUNTER_CONSUME: Counter = inc_new_counter!("streamer-recv_window-consume", consume_queue.len());
create_counter!("streamer-recv_window-consume", LOG_RATE);
inc_counter!(COUNTER_CONSUME, consume_queue.len());
s.send(consume_queue)?; s.send(consume_queue)?;
} }
Ok(()) Ok(())
@ -647,9 +639,7 @@ fn broadcast(
// Index the blobs // Index the blobs
Crdt::index_blobs(&me, &blobs, receive_index)?; Crdt::index_blobs(&me, &blobs, receive_index)?;
// keep the cache of blobs that are broadcast // keep the cache of blobs that are broadcast
static mut COUNTER_BROADCAST: Counter = inc_new_counter!("streamer-broadcast-sent", blobs.len());
create_counter!("streamer-broadcast-sent", LOG_RATE);
inc_counter!(COUNTER_BROADCAST, blobs.len());
{ {
let mut win = window.write().unwrap(); let mut win = window.write().unwrap();
assert!(blobs.len() <= win.len()); assert!(blobs.len() <= win.len());