cargo fmt fixups
This commit is contained in:
parent
3e36e6dcf8
commit
f2b1a04bca
|
@ -12,7 +12,7 @@ _() {
|
||||||
}
|
}
|
||||||
|
|
||||||
_ rustup component add rustfmt-preview
|
_ rustup component add rustfmt-preview
|
||||||
_ cargo fmt -- --write-mode=check
|
_ cargo fmt -- --check
|
||||||
_ cargo build --verbose
|
_ cargo build --verbose
|
||||||
_ cargo test --verbose
|
_ cargo test --verbose
|
||||||
_ cargo bench --verbose
|
_ cargo bench --verbose
|
||||||
|
|
33
src/bank.rs
33
src/bank.rs
|
@ -150,7 +150,8 @@ impl Bank {
|
||||||
|
|
||||||
/// Forget the given `signature` with `last_id` because the transaction was rejected.
|
/// Forget the given `signature` with `last_id` because the transaction was rejected.
|
||||||
fn forget_signature_with_last_id(&self, signature: &Signature, last_id: &Hash) {
|
fn forget_signature_with_last_id(&self, signature: &Signature, last_id: &Hash) {
|
||||||
if let Some(entry) = self.last_ids_sigs
|
if let Some(entry) = self
|
||||||
|
.last_ids_sigs
|
||||||
.write()
|
.write()
|
||||||
.expect("'last_ids' read lock in forget_signature_with_last_id")
|
.expect("'last_ids' read lock in forget_signature_with_last_id")
|
||||||
.get_mut(last_id)
|
.get_mut(last_id)
|
||||||
|
@ -167,7 +168,8 @@ impl Bank {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn reserve_signature_with_last_id(&self, signature: &Signature, last_id: &Hash) -> Result<()> {
|
fn reserve_signature_with_last_id(&self, signature: &Signature, last_id: &Hash) -> Result<()> {
|
||||||
if let Some(entry) = self.last_ids_sigs
|
if let Some(entry) = self
|
||||||
|
.last_ids_sigs
|
||||||
.write()
|
.write()
|
||||||
.expect("'last_ids' read lock in reserve_signature_with_last_id")
|
.expect("'last_ids' read lock in reserve_signature_with_last_id")
|
||||||
.get_mut(last_id)
|
.get_mut(last_id)
|
||||||
|
@ -198,10 +200,12 @@ impl Bank {
|
||||||
/// the oldest ones once its internal cache is full. Once boot, the
|
/// the oldest ones once its internal cache is full. Once boot, the
|
||||||
/// bank will reject transactions using that `last_id`.
|
/// bank will reject transactions using that `last_id`.
|
||||||
pub fn register_entry_id(&self, last_id: &Hash) {
|
pub fn register_entry_id(&self, last_id: &Hash) {
|
||||||
let mut last_ids = self.last_ids
|
let mut last_ids = self
|
||||||
|
.last_ids
|
||||||
.write()
|
.write()
|
||||||
.expect("'last_ids' write lock in register_entry_id");
|
.expect("'last_ids' write lock in register_entry_id");
|
||||||
let mut last_ids_sigs = self.last_ids_sigs
|
let mut last_ids_sigs = self
|
||||||
|
.last_ids_sigs
|
||||||
.write()
|
.write()
|
||||||
.expect("last_ids_sigs write lock");
|
.expect("last_ids_sigs write lock");
|
||||||
if last_ids.len() >= MAX_ENTRY_IDS {
|
if last_ids.len() >= MAX_ENTRY_IDS {
|
||||||
|
@ -262,7 +266,8 @@ impl Bank {
|
||||||
if let Some(payment) = plan.final_payment() {
|
if let Some(payment) = plan.final_payment() {
|
||||||
self.apply_payment(&payment, balances);
|
self.apply_payment(&payment, balances);
|
||||||
} else {
|
} else {
|
||||||
let mut pending = self.pending
|
let mut pending = self
|
||||||
|
.pending
|
||||||
.write()
|
.write()
|
||||||
.expect("'pending' write lock in apply_credits");
|
.expect("'pending' write lock in apply_credits");
|
||||||
pending.insert(tx.sig, plan);
|
pending.insert(tx.sig, plan);
|
||||||
|
@ -298,7 +303,8 @@ impl Bank {
|
||||||
debug!("processing Transactions {}", txs.len());
|
debug!("processing Transactions {}", txs.len());
|
||||||
let txs_len = txs.len();
|
let txs_len = txs.len();
|
||||||
let now = Instant::now();
|
let now = Instant::now();
|
||||||
let results: Vec<_> = txs.into_iter()
|
let results: Vec<_> = txs
|
||||||
|
.into_iter()
|
||||||
.map(|tx| self.apply_debits(&tx, bals).map(|_| tx))
|
.map(|tx| self.apply_debits(&tx, bals).map(|_| tx))
|
||||||
.collect(); // Calling collect() here forces all debits to complete before moving on.
|
.collect(); // Calling collect() here forces all debits to complete before moving on.
|
||||||
|
|
||||||
|
@ -460,7 +466,8 @@ impl Bank {
|
||||||
/// Process a Witness Signature. Any payment plans waiting on this signature
|
/// Process a Witness Signature. Any payment plans waiting on this signature
|
||||||
/// will progress one step.
|
/// will progress one step.
|
||||||
fn apply_signature(&self, from: PublicKey, tx_sig: Signature) -> Result<()> {
|
fn apply_signature(&self, from: PublicKey, tx_sig: Signature) -> Result<()> {
|
||||||
if let Occupied(mut e) = self.pending
|
if let Occupied(mut e) = self
|
||||||
|
.pending
|
||||||
.write()
|
.write()
|
||||||
.expect("write() in apply_signature")
|
.expect("write() in apply_signature")
|
||||||
.entry(tx_sig)
|
.entry(tx_sig)
|
||||||
|
@ -483,7 +490,8 @@ impl Bank {
|
||||||
|
|
||||||
// Hold 'pending' write lock until the end of this function. Otherwise another thread can
|
// Hold 'pending' write lock until the end of this function. Otherwise another thread can
|
||||||
// double-spend if it enters before the modified plan is removed from 'pending'.
|
// double-spend if it enters before the modified plan is removed from 'pending'.
|
||||||
let mut pending = self.pending
|
let mut pending = self
|
||||||
|
.pending
|
||||||
.write()
|
.write()
|
||||||
.expect("'pending' write lock in apply_timestamp");
|
.expect("'pending' write lock in apply_timestamp");
|
||||||
for (key, plan) in pending.iter_mut() {
|
for (key, plan) in pending.iter_mut() {
|
||||||
|
@ -532,7 +540,8 @@ impl Bank {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn get_balance(&self, pubkey: &PublicKey) -> i64 {
|
pub fn get_balance(&self, pubkey: &PublicKey) -> i64 {
|
||||||
let bals = self.balances
|
let bals = self
|
||||||
|
.balances
|
||||||
.read()
|
.read()
|
||||||
.expect("'balances' read lock in get_balance");
|
.expect("'balances' read lock in get_balance");
|
||||||
bals.get(pubkey).cloned().unwrap_or(0)
|
bals.get(pubkey).cloned().unwrap_or(0)
|
||||||
|
@ -543,7 +552,8 @@ impl Bank {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn has_signature(&self, signature: &Signature) -> bool {
|
pub fn has_signature(&self, signature: &Signature) -> bool {
|
||||||
let last_ids_sigs = self.last_ids_sigs
|
let last_ids_sigs = self
|
||||||
|
.last_ids_sigs
|
||||||
.read()
|
.read()
|
||||||
.expect("'last_ids_sigs' read lock");
|
.expect("'last_ids_sigs' read lock");
|
||||||
for (_hash, signatures) in last_ids_sigs.iter() {
|
for (_hash, signatures) in last_ids_sigs.iter() {
|
||||||
|
@ -674,7 +684,8 @@ mod tests {
|
||||||
let bank = Bank::new(&mint);
|
let bank = Bank::new(&mint);
|
||||||
let pubkey = KeyPair::new().pubkey();
|
let pubkey = KeyPair::new().pubkey();
|
||||||
let dt = Utc::now();
|
let dt = Utc::now();
|
||||||
let sig = bank.transfer_on_date(1, &mint.keypair(), pubkey, dt, mint.last_id())
|
let sig = bank
|
||||||
|
.transfer_on_date(1, &mint.keypair(), pubkey, dt, mint.last_id())
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
// Assert the debit counts as a transaction.
|
// Assert the debit counts as a transaction.
|
||||||
|
|
24
src/crdt.rs
24
src/crdt.rs
|
@ -473,7 +473,8 @@ impl Crdt {
|
||||||
}
|
}
|
||||||
let leader_id = self.leader_data().unwrap().id;
|
let leader_id = self.leader_data().unwrap().id;
|
||||||
let limit = GOSSIP_PURGE_MILLIS;
|
let limit = GOSSIP_PURGE_MILLIS;
|
||||||
let dead_ids: Vec<PublicKey> = self.alive
|
let dead_ids: Vec<PublicKey> = self
|
||||||
|
.alive
|
||||||
.iter()
|
.iter()
|
||||||
.filter_map(|(&k, v)| {
|
.filter_map(|(&k, v)| {
|
||||||
if k != self.me && (now - v) > limit {
|
if k != self.me && (now - v) > limit {
|
||||||
|
@ -521,7 +522,8 @@ impl Crdt {
|
||||||
let live: Vec<_> = self.alive.iter().collect();
|
let live: Vec<_> = self.alive.iter().collect();
|
||||||
//thread_rng().shuffle(&mut live);
|
//thread_rng().shuffle(&mut live);
|
||||||
let me = &self.table[&self.me];
|
let me = &self.table[&self.me];
|
||||||
let cloned_table: Vec<NodeInfo> = live.iter()
|
let cloned_table: Vec<NodeInfo> = live
|
||||||
|
.iter()
|
||||||
.map(|x| &self.table[x.0])
|
.map(|x| &self.table[x.0])
|
||||||
.filter(|v| {
|
.filter(|v| {
|
||||||
if me.id == v.id {
|
if me.id == v.id {
|
||||||
|
@ -733,7 +735,8 @@ impl Crdt {
|
||||||
|
|
||||||
fn get_updates_since(&self, v: u64) -> (PublicKey, u64, Vec<NodeInfo>) {
|
fn get_updates_since(&self, v: u64) -> (PublicKey, u64, Vec<NodeInfo>) {
|
||||||
//trace!("get updates since {}", v);
|
//trace!("get updates since {}", v);
|
||||||
let data = self.table
|
let data = self
|
||||||
|
.table
|
||||||
.values()
|
.values()
|
||||||
.filter(|x| x.id != PublicKey::default() && self.local[&x.id] > v)
|
.filter(|x| x.id != PublicKey::default() && self.local[&x.id] > v)
|
||||||
.cloned()
|
.cloned()
|
||||||
|
@ -744,7 +747,8 @@ impl Crdt {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn window_index_request(&self, ix: u64) -> Result<(SocketAddr, Vec<u8>)> {
|
pub fn window_index_request(&self, ix: u64) -> Result<(SocketAddr, Vec<u8>)> {
|
||||||
let valid: Vec<_> = self.table
|
let valid: Vec<_> = self
|
||||||
|
.table
|
||||||
.values()
|
.values()
|
||||||
.filter(|r| r.id != self.me && Self::is_valid_address(r.contact_info.tvu_window))
|
.filter(|r| r.id != self.me && Self::is_valid_address(r.contact_info.tvu_window))
|
||||||
.collect();
|
.collect();
|
||||||
|
@ -764,7 +768,8 @@ impl Crdt {
|
||||||
/// * A - Address to send to
|
/// * A - Address to send to
|
||||||
/// * B - RequestUpdates protocol message
|
/// * B - RequestUpdates protocol message
|
||||||
fn gossip_request(&self) -> Result<(SocketAddr, Protocol)> {
|
fn gossip_request(&self) -> Result<(SocketAddr, Protocol)> {
|
||||||
let options: Vec<_> = self.table
|
let options: Vec<_> = self
|
||||||
|
.table
|
||||||
.values()
|
.values()
|
||||||
.filter(|v| {
|
.filter(|v| {
|
||||||
v.id != self.me
|
v.id != self.me
|
||||||
|
@ -826,7 +831,8 @@ impl Crdt {
|
||||||
|
|
||||||
// Lock the object only to do this operation and not for any longer
|
// Lock the object only to do this operation and not for any longer
|
||||||
// especially not when doing the `sock.send_to`
|
// especially not when doing the `sock.send_to`
|
||||||
let (remote_gossip_addr, req) = obj.read()
|
let (remote_gossip_addr, req) = obj
|
||||||
|
.read()
|
||||||
.expect("'obj' read lock in fn run_gossip")
|
.expect("'obj' read lock in fn run_gossip")
|
||||||
.gossip_request()?;
|
.gossip_request()?;
|
||||||
|
|
||||||
|
@ -902,7 +908,8 @@ impl Crdt {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
let liveness_entry = self.external_liveness
|
let liveness_entry = self
|
||||||
|
.external_liveness
|
||||||
.entry(*pk)
|
.entry(*pk)
|
||||||
.or_insert_with(HashMap::new);
|
.or_insert_with(HashMap::new);
|
||||||
let peer_index = *liveness_entry.entry(from).or_insert(*external_remote_index);
|
let peer_index = *liveness_entry.entry(from).or_insert(*external_remote_index);
|
||||||
|
@ -1135,7 +1142,8 @@ impl Crdt {
|
||||||
while let Ok(mut more) = requests_receiver.try_recv() {
|
while let Ok(mut more) = requests_receiver.try_recv() {
|
||||||
reqs.append(&mut more);
|
reqs.append(&mut more);
|
||||||
}
|
}
|
||||||
let resp: VecDeque<_> = reqs.iter()
|
let resp: VecDeque<_> = reqs
|
||||||
|
.iter()
|
||||||
.filter_map(|b| Self::handle_blob(obj, window, blob_recycler, &b.read().unwrap()))
|
.filter_map(|b| Self::handle_blob(obj, window, blob_recycler, &b.read().unwrap()))
|
||||||
.collect();
|
.collect();
|
||||||
response_sender.send(resp)?;
|
response_sender.send(resp)?;
|
||||||
|
|
|
@ -168,7 +168,8 @@ impl<T: Default> Clone for Recycler<T> {
|
||||||
impl<T: Default> Recycler<T> {
|
impl<T: Default> Recycler<T> {
|
||||||
pub fn allocate(&self) -> Arc<RwLock<T>> {
|
pub fn allocate(&self) -> Arc<RwLock<T>> {
|
||||||
let mut gc = self.gc.lock().expect("recycler lock in pb fn allocate");
|
let mut gc = self.gc.lock().expect("recycler lock in pb fn allocate");
|
||||||
let x = gc.pop()
|
let x = gc
|
||||||
|
.pop()
|
||||||
.unwrap_or_else(|| Arc::new(RwLock::new(Default::default())));
|
.unwrap_or_else(|| Arc::new(RwLock::new(Default::default())));
|
||||||
|
|
||||||
// Only return the item if this recycler is the last reference to it.
|
// Only return the item if this recycler is the last reference to it.
|
||||||
|
|
|
@ -53,7 +53,8 @@ fn recv_loop(
|
||||||
loop {
|
loop {
|
||||||
let msgs = re.allocate();
|
let msgs = re.allocate();
|
||||||
loop {
|
loop {
|
||||||
let result = msgs.write()
|
let result = msgs
|
||||||
|
.write()
|
||||||
.expect("write lock in fn recv_loop")
|
.expect("write lock in fn recv_loop")
|
||||||
.recv_from(sock);
|
.recv_from(sock);
|
||||||
match result {
|
match result {
|
||||||
|
@ -351,7 +352,8 @@ fn process_blob(
|
||||||
let w = (pix % WINDOW_SIZE) as usize;
|
let w = (pix % WINDOW_SIZE) as usize;
|
||||||
|
|
||||||
let is_coding = {
|
let is_coding = {
|
||||||
let blob_r = blob.read()
|
let blob_r = blob
|
||||||
|
.read()
|
||||||
.expect("blob read lock for flogs streamer::window");
|
.expect("blob read lock for flogs streamer::window");
|
||||||
blob_r.is_coding()
|
blob_r.is_coding()
|
||||||
};
|
};
|
||||||
|
@ -448,7 +450,8 @@ fn recv_window(
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let timer = Duration::from_millis(200);
|
let timer = Duration::from_millis(200);
|
||||||
let mut dq = r.recv_timeout(timer)?;
|
let mut dq = r.recv_timeout(timer)?;
|
||||||
let maybe_leader: Option<NodeInfo> = crdt.read()
|
let maybe_leader: Option<NodeInfo> = crdt
|
||||||
|
.read()
|
||||||
.expect("'crdt' read lock in fn recv_window")
|
.expect("'crdt' read lock in fn recv_window")
|
||||||
.leader_data()
|
.leader_data()
|
||||||
.cloned();
|
.cloned();
|
||||||
|
|
|
@ -100,7 +100,8 @@ pub fn send_leader_vote(
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let now = timing::timestamp();
|
let now = timing::timestamp();
|
||||||
if now - *last_vote > VOTE_TIMEOUT_MS {
|
if now - *last_vote > VOTE_TIMEOUT_MS {
|
||||||
let ids: Vec<_> = crdt.read()
|
let ids: Vec<_> = crdt
|
||||||
|
.read()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.table
|
.table
|
||||||
.values()
|
.values()
|
||||||
|
|
|
@ -263,7 +263,8 @@ fn test_external_liveness_table() {
|
||||||
c4.write().unwrap().insert(&c1_data);
|
c4.write().unwrap().insert(&c1_data);
|
||||||
c4.write().unwrap().set_leader(c1_data.id);
|
c4.write().unwrap().set_leader(c1_data.id);
|
||||||
for _ in 0..30 {
|
for _ in 0..30 {
|
||||||
done = c1.read()
|
done = c1
|
||||||
|
.read()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.get_external_liveness_entry(&c4_id)
|
.get_external_liveness_entry(&c4_id)
|
||||||
.is_none();
|
.is_none();
|
||||||
|
|
|
@ -393,14 +393,16 @@ fn test_multi_node_dynamic_network() {
|
||||||
logger::setup();
|
logger::setup();
|
||||||
let key = "SOLANA_DYNAMIC_NODES";
|
let key = "SOLANA_DYNAMIC_NODES";
|
||||||
let num_nodes: usize = match env::var(key) {
|
let num_nodes: usize = match env::var(key) {
|
||||||
Ok(val) => val.parse()
|
Ok(val) => val
|
||||||
|
.parse()
|
||||||
.expect(&format!("env var {} is not parse-able as usize", key)),
|
.expect(&format!("env var {} is not parse-able as usize", key)),
|
||||||
Err(_) => 60,
|
Err(_) => 60,
|
||||||
};
|
};
|
||||||
|
|
||||||
let purge_key = "SOLANA_DYNAMIC_NODES_PURGE_LAG";
|
let purge_key = "SOLANA_DYNAMIC_NODES_PURGE_LAG";
|
||||||
let purge_lag: usize = match env::var(purge_key) {
|
let purge_lag: usize = match env::var(purge_key) {
|
||||||
Ok(val) => val.parse()
|
Ok(val) => val
|
||||||
|
.parse()
|
||||||
.expect(&format!("env var {} is not parse-able as usize", purge_key)),
|
.expect(&format!("env var {} is not parse-able as usize", purge_key)),
|
||||||
Err(_) => std::usize::MAX,
|
Err(_) => std::usize::MAX,
|
||||||
};
|
};
|
||||||
|
|
Loading…
Reference in New Issue