machankura/mempool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor rust code for style

Browse Source
This commit is contained in:
Mononaut 2023-06-23 20:41:39 -04:00
parent 1d51b01bd1
commit 0e00881826
No known key found for this signature in database
GPG Key ID: A3F058E41374C04E
5 changed files with 480 additions and 425 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
backend/rust-gbt/src/audit_transaction.rs
View File

@ -1,4 +1,8 @@
use std::{collections::{HashSet}, hash::{Hash, Hasher}, cmp::Ordering}; use std::{
cmp::Ordering,
collections::HashSet,
hash::{Hash, Hasher},
};
#[derive(Clone)] #[derive(Clone)]
pub struct AuditTransaction { pub struct AuditTransaction {
@ -10,7 +14,7 @@ pub struct AuditTransaction {
pub effective_fee_per_vsize: f64, pub effective_fee_per_vsize: f64,
pub dependency_rate: f64, pub dependency_rate: f64,
pub inputs: Vec<u32>, pub inputs: Vec<u32>,
pub is_relatives_set: bool, pub relatives_set_flag: bool,
pub ancestors: HashSet<u32>, pub ancestors: HashSet<u32>,
pub children: HashSet<u32>, pub children: HashSet<u32>,
pub ancestor_fee: u64, pub ancestor_fee: u64,

120
backend/rust-gbt/src/gbt.rs
View File

@ -1,13 +1,15 @@
use priority_queue::PriorityQueue;
use std::cmp::Ordering; use std::cmp::Ordering;
use std::collections::{HashMap, HashSet, VecDeque}; use std::collections::{HashMap, HashSet, VecDeque};
use std::f64::INFINITY; use std::f64::INFINITY;
use priority_queue::PriorityQueue;
use crate::thread_transaction::{ThreadTransaction}; use crate::audit_transaction::AuditTransaction;
use crate::audit_transaction::{AuditTransaction}; use crate::thread_transaction::ThreadTransaction;
const BLOCK_WEIGHT_UNITS: u32 = 4_000_000; const BLOCK_WEIGHT_UNITS: u32 = 4_000_000;
const BLOCK_SIGOPS: u32 = 80_000; const BLOCK_SIGOPS: u32 = 80_000;
const BLOCK_RESERVED_WEIGHT: u32 = 4_000;
const MAX_BLOCKS: usize = 8;
struct TxPriority { struct TxPriority {
uid: u32, uid: u32,
@ -34,8 +36,10 @@ impl Ord for TxPriority {
} }
} }
pub fn gbt(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Vec<u32>>, Vec<(u32, f64)>, Vec<Vec<u32>>) { pub fn gbt(
return make_block_templates(mempool); mempool: &mut HashMap<u32, ThreadTransaction>,
) -> Option<(Vec<Vec<u32>>, Vec<(u32, f64)>, Vec<Vec<u32>>)> {
make_block_templates(mempool)
} }
/* /*
@ -43,7 +47,9 @@ pub fn gbt(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Vec<u32>>, Vec<
* (see BlockAssembler in https://github.com/bitcoin/bitcoin/blob/master/src/node/miner.cpp) * (see BlockAssembler in https://github.com/bitcoin/bitcoin/blob/master/src/node/miner.cpp)
* Ported from https://github.com/mempool/mempool/blob/master/backend/src/api/tx-selection-worker.ts * Ported from https://github.com/mempool/mempool/blob/master/backend/src/api/tx-selection-worker.ts
*/ */
fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Vec<u32>>, Vec<(u32, f64)>, Vec<Vec<u32>>) { fn make_block_templates(
mempool: &mut HashMap<u32, ThreadTransaction>,
) -> Option<(Vec<Vec<u32>>, Vec<(u32, f64)>, Vec<Vec<u32>>)> {
let mut audit_pool: HashMap<u32, AuditTransaction> = HashMap::new(); let mut audit_pool: HashMap<u32, AuditTransaction> = HashMap::new();
let mut mempool_array: VecDeque<u32> = VecDeque::new(); let mut mempool_array: VecDeque<u32> = VecDeque::new();
let mut cluster_array: Vec<Vec<u32>> = Vec::new(); let mut cluster_array: Vec<Vec<u32>> = Vec::new();
@ -59,7 +65,7 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
effective_fee_per_vsize: tx.effective_fee_per_vsize, effective_fee_per_vsize: tx.effective_fee_per_vsize,
dependency_rate: INFINITY, dependency_rate: INFINITY,
inputs: tx.inputs.clone(), inputs: tx.inputs.clone(),
is_relatives_set: false, relatives_set_flag: false,
ancestors: HashSet::new(), ancestors: HashSet::new(),
children: HashSet::new(), children: HashSet::new(),
ancestor_fee: tx.fee, ancestor_fee: tx.fee,
@ -89,7 +95,7 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
// Build blocks by greedily choosing the highest feerate package // Build blocks by greedily choosing the highest feerate package
// (i.e. the package rooted in the transaction with the best ancestor score) // (i.e. the package rooted in the transaction with the best ancestor score)
let mut blocks: Vec<Vec<u32>> = Vec::new(); let mut blocks: Vec<Vec<u32>> = Vec::new();
let mut block_weight: u32 = 4000; let mut block_weight: u32 = BLOCK_RESERVED_WEIGHT;
let mut block_sigops: u32 = 0; let mut block_sigops: u32 = 0;
let mut transactions: Vec<u32> = Vec::new(); let mut transactions: Vec<u32> = Vec::new();
let mut modified: PriorityQueue<u32, TxPriority> = PriorityQueue::new(); let mut modified: PriorityQueue<u32, TxPriority> = PriorityQueue::new();
@ -98,30 +104,34 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
while mempool_array.len() > 0 || !modified.is_empty() { while mempool_array.len() > 0 || !modified.is_empty() {
let next_txid: u32; let next_txid: u32;
if modified.is_empty() { if modified.is_empty() {
next_txid = mempool_array.pop_front().unwrap(); next_txid = mempool_array.pop_front()?;
} else if mempool_array.len() == 0 { } else if mempool_array.len() == 0 {
next_txid = modified.pop().unwrap().0; next_txid = modified.pop()?.0;
} else { } else {
let next_array_txid = mempool_array.front().unwrap(); let next_array_txid = mempool_array.front()?;
let next_modified_txid = modified.peek().unwrap().0; let next_modified_txid = modified.peek()?.0;
let array_tx: &AuditTransaction = audit_pool.get(next_array_txid).unwrap(); let array_tx: &AuditTransaction = audit_pool.get(next_array_txid)?;
let modified_tx: &AuditTransaction = audit_pool.get(next_modified_txid).unwrap(); let modified_tx: &AuditTransaction = audit_pool.get(next_modified_txid)?;
match array_tx.cmp(&modified_tx) { match array_tx.cmp(&modified_tx) {
std::cmp::Ordering::Equal | std::cmp::Ordering::Greater => { std::cmp::Ordering::Equal | std::cmp::Ordering::Greater => {
next_txid = mempool_array.pop_front().unwrap(); next_txid = mempool_array.pop_front()?;
} }
std::cmp::Ordering::Less => { std::cmp::Ordering::Less => {
next_txid = modified.pop().unwrap().0; next_txid = modified.pop()?.0;
} }
} }
} }
let next_tx: AuditTransaction = audit_pool.get(&next_txid).unwrap().clone();
let next_tx = audit_pool.get(&next_txid)?;
if next_tx.used { if next_tx.used {
continue; continue;
} }
if blocks.len() < 7 && ((block_weight + next_tx.ancestor_weight >= BLOCK_WEIGHT_UNITS) || (block_sigops + next_tx.ancestor_sigops > BLOCK_SIGOPS)) { if blocks.len() < (MAX_BLOCKS - 1)
&& ((block_weight + next_tx.ancestor_weight >= BLOCK_WEIGHT_UNITS)
|| (block_sigops + next_tx.ancestor_sigops > BLOCK_SIGOPS))
{
// hold this package in an overflow list while we check for smaller options // hold this package in an overflow list while we check for smaller options
overflow.push(next_txid); overflow.push(next_txid);
failures += 1; failures += 1;
@ -129,8 +139,8 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
let mut package: Vec<(u32, usize, u32)> = Vec::new(); let mut package: Vec<(u32, usize, u32)> = Vec::new();
let mut cluster: Vec<u32> = Vec::new(); let mut cluster: Vec<u32> = Vec::new();
let is_cluster: bool = next_tx.ancestors.len() > 0; let is_cluster: bool = next_tx.ancestors.len() > 0;
package.push((next_tx.uid, next_tx.ancestors.len(), next_tx.weight)); package.push((next_txid, next_tx.ancestors.len(), next_tx.weight));
cluster.push(next_tx.uid); cluster.push(next_txid);
for ancestor_id in &next_tx.ancestors { for ancestor_id in &next_tx.ancestors {
if let Some(ancestor) = audit_pool.get(ancestor_id) { if let Some(ancestor) = audit_pool.get(ancestor_id) {
package.push((*ancestor_id, ancestor.ancestors.len(), ancestor.weight)); package.push((*ancestor_id, ancestor.ancestors.len(), ancestor.weight));
@ -143,7 +153,9 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
cluster_array.push(cluster); cluster_array.push(cluster);
} }
let cluster_rate = next_tx.dependency_rate.min(next_tx.ancestor_fee as f64 / (next_tx.ancestor_weight as f64 / 4.0)); let cluster_rate = next_tx
.dependency_rate
.min(next_tx.ancestor_fee as f64 / (next_tx.ancestor_weight as f64 / 4.0));
for package_entry in &package { for package_entry in &package {
if let Some(tx) = audit_pool.get_mut(&package_entry.0) { if let Some(tx) = audit_pool.get_mut(&package_entry.0) {
@ -156,16 +168,22 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
block_weight += tx.weight; block_weight += tx.weight;
block_sigops += tx.sigops; block_sigops += tx.sigops;
} }
update_descendants(package_entry.0, &mut audit_pool, &mut modified, cluster_rate); update_descendants(
package_entry.0,
&mut audit_pool,
&mut modified,
cluster_rate,
);
} }
failures = 0; failures = 0;
} }
// this block is full // this block is full
let exceeded_package_tries = failures > 1000 && block_weight > (BLOCK_WEIGHT_UNITS - 4000); let exceeded_package_tries =
failures > 1000 && block_weight > (BLOCK_WEIGHT_UNITS - BLOCK_RESERVED_WEIGHT);
let queue_is_empty = mempool_array.len() == 0 && modified.is_empty(); let queue_is_empty = mempool_array.len() == 0 && modified.is_empty();
if (exceeded_package_tries || queue_is_empty) && blocks.len() < 7 { if (exceeded_package_tries || queue_is_empty) && blocks.len() < (MAX_BLOCKS - 1) {
// finalize this block // finalize this block
if transactions.len() > 0 { if transactions.len() > 0 {
blocks.push(transactions); blocks.push(transactions);
@ -178,7 +196,13 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
for overflowed in &overflow { for overflowed in &overflow {
if let Some(overflowed_tx) = audit_pool.get(overflowed) { if let Some(overflowed_tx) = audit_pool.get(overflowed) {
if overflowed_tx.modified { if overflowed_tx.modified {
modified.push(*overflowed, TxPriority{ uid: *overflowed, score: overflowed_tx.score}); modified.push(
*overflowed,
TxPriority {
uid: *overflowed,
score: overflowed_tx.score,
},
);
} else { } else {
mempool_array.push_front(*overflowed); mempool_array.push_front(*overflowed);
} }
@ -200,13 +224,13 @@ fn make_block_templates(mempool: &mut HashMap<u32,ThreadTransaction>) -> (Vec<Ve
} }
} }
return (blocks, rates, cluster_array); Some((blocks, rates, cluster_array))
} }
fn set_relatives(txid: u32, audit_pool: &mut HashMap<u32, AuditTransaction>) { fn set_relatives(txid: u32, audit_pool: &mut HashMap<u32, AuditTransaction>) {
let mut parents: HashSet<u32> = HashSet::new(); let mut parents: HashSet<u32> = HashSet::new();
if let Some(tx) = audit_pool.get(&txid) { if let Some(tx) = audit_pool.get(&txid) {
if tx.is_relatives_set { if tx.relatives_set_flag {
return; return;
} }
for input in &tx.inputs { for input in &tx.inputs {
@ -220,8 +244,7 @@ fn set_relatives(txid: u32, audit_pool: &mut HashMap<u32, AuditTransaction>) {
for parent_id in &parents { for parent_id in &parents {
set_relatives(*parent_id, audit_pool); set_relatives(*parent_id, audit_pool);
let parent_entry: Option<&mut AuditTransaction> = audit_pool.get_mut(&parent_id); match audit_pool.get_mut(&parent_id) {
match parent_entry {
Some(parent) => { Some(parent) => {
ancestors.insert(*parent_id); ancestors.insert(*parent_id);
parent.children.insert(txid); parent.children.insert(txid);
@ -250,13 +273,23 @@ fn set_relatives(txid: u32, audit_pool: &mut HashMap<u32, AuditTransaction>) {
tx.ancestor_fee = tx.fee + total_fee; tx.ancestor_fee = tx.fee + total_fee;
tx.ancestor_weight = tx.weight + total_weight; tx.ancestor_weight = tx.weight + total_weight;
tx.ancestor_sigops = tx.sigops + total_sigops; tx.ancestor_sigops = tx.sigops + total_sigops;
tx.score = (tx.ancestor_fee as f64) / (if tx.ancestor_weight != 0 {tx.ancestor_weight as f64 / 4.0} else { 1.0 }); tx.score = (tx.ancestor_fee as f64)
tx.is_relatives_set = true; / (if tx.ancestor_weight != 0 {
tx.ancestor_weight as f64 / 4.0
} else {
1.0
});
tx.relatives_set_flag = true;
} }
} }
// iterate over remaining descendants, removing the root as a valid ancestor & updating the ancestor score // iterate over remaining descendants, removing the root as a valid ancestor & updating the ancestor score
fn update_descendants(root_txid: u32, audit_pool: &mut HashMap<u32, AuditTransaction>, modified: &mut PriorityQueue<u32, TxPriority>, cluster_rate: f64) { fn update_descendants(
root_txid: u32,
audit_pool: &mut HashMap<u32, AuditTransaction>,
modified: &mut PriorityQueue<u32, TxPriority>,
cluster_rate: f64,
) {
let mut visited: HashSet<u32> = HashSet::new(); let mut visited: HashSet<u32> = HashSet::new();
let mut descendant_stack: Vec<u32> = Vec::new(); let mut descendant_stack: Vec<u32> = Vec::new();
let root_fee: u64; let root_fee: u64;
@ -284,14 +317,31 @@ fn update_descendants(root_txid: u32, audit_pool: &mut HashMap<u32, AuditTransac
descendant.ancestor_weight -= root_weight; descendant.ancestor_weight -= root_weight;
descendant.ancestor_sigops -= root_sigops; descendant.ancestor_sigops -= root_sigops;
let current_score = descendant.score; let current_score = descendant.score;
descendant.score = (descendant.ancestor_fee as f64) / (if descendant.ancestor_weight != 0 {descendant.ancestor_weight as f64 / 4.0} else { 1.0 }); descendant.score = (descendant.ancestor_fee as f64)
/ (if descendant.ancestor_weight != 0 {
descendant.ancestor_weight as f64 / 4.0
} else {
1.0
});
descendant.dependency_rate = descendant.dependency_rate.min(cluster_rate); descendant.dependency_rate = descendant.dependency_rate.min(cluster_rate);
descendant.modified = true; descendant.modified = true;
// update modified priority if score has changed // update modified priority if score has changed
if !descendant.modified || descendant.score < current_score { if !descendant.modified || descendant.score < current_score {
modified.push_decrease(descendant.uid, TxPriority { uid: descendant.uid, score: descendant.score}); modified.push_decrease(
descendant.uid,
TxPriority {
uid: descendant.uid,
score: descendant.score,
},
);
} else if descendant.score > current_score { } else if descendant.score > current_score {
modified.push_increase(descendant.uid, TxPriority { uid: descendant.uid, score: descendant.score}); modified.push_increase(
descendant.uid,
TxPriority {
uid: descendant.uid,
score: descendant.score,
},
);
} }
// add this node's children to the stack // add this node's children to the stack

46
backend/rust-gbt/src/lib.rs
View File

@ -1,33 +1,35 @@
use neon::{prelude::*, types::buffer::TypedArray}; use neon::{prelude::*, types::buffer::TypedArray};
use once_cell::sync::Lazy;
use std::collections::HashMap; use std::collections::HashMap;
use std::ops::DerefMut; use std::ops::DerefMut;
use std::sync::Mutex; use std::sync::Mutex;
use once_cell::sync::Lazy;
mod audit_transaction;
mod gbt; mod gbt;
mod thread_transaction; mod thread_transaction;
mod audit_transaction;
mod utils; mod utils;
use thread_transaction::ThreadTransaction; use thread_transaction::ThreadTransaction;
static THREAD_TRANSACTIONS: Lazy<Mutex<HashMap<u32, ThreadTransaction>>> = Lazy::new(|| { static THREAD_TRANSACTIONS: Lazy<Mutex<HashMap<u32, ThreadTransaction>>> =
Mutex::new(HashMap::new()) Lazy::new(|| Mutex::new(HashMap::new()));
});
fn make(mut cx: FunctionContext) -> JsResult<JsUndefined> { fn make(mut cx: FunctionContext) -> JsResult<JsUndefined> {
let mempool_arg = cx.argument::<JsArrayBuffer>(0)?.root(&mut cx).into_inner(&mut cx); let mempool_arg = cx
.argument::<JsArrayBuffer>(0)?
.root(&mut cx)
.into_inner(&mut cx);
let callback = cx.argument::<JsFunction>(1)?.root(&mut cx); let callback = cx.argument::<JsFunction>(1)?.root(&mut cx);
let channel = cx.channel(); let channel = cx.channel();
let buffer = mempool_arg.as_slice(&mut cx); let buffer = mempool_arg.as_slice(&mut cx);
let thread_transactions = ThreadTransaction::batch_from_buffer(buffer);
let mut map = THREAD_TRANSACTIONS.lock().unwrap(); let mut map = HashMap::new();
map.clear(); for tx in ThreadTransaction::batch_from_buffer(buffer) {
for tx in thread_transactions {
map.insert(tx.uid, tx); map.insert(tx.uid, tx);
} }
drop(map);
let mut global_map = THREAD_TRANSACTIONS.lock().unwrap();
*global_map = map;
run_in_thread(channel, callback); run_in_thread(channel, callback);
@ -35,21 +37,25 @@ fn make(mut cx: FunctionContext) -> JsResult<JsUndefined> {
} }
fn update(mut cx: FunctionContext) -> JsResult<JsUndefined> { fn update(mut cx: FunctionContext) -> JsResult<JsUndefined> {
let new_txs_arg = cx.argument::<JsArrayBuffer>(0)?.root(&mut cx).into_inner(&mut cx); let new_txs_arg = cx
let remove_txs_arg = cx.argument::<JsArrayBuffer>(1)?.root(&mut cx).into_inner(&mut cx); .argument::<JsArrayBuffer>(0)?
.root(&mut cx)
.into_inner(&mut cx);
let remove_txs_arg = cx
.argument::<JsArrayBuffer>(1)?
.root(&mut cx)
.into_inner(&mut cx);
let callback = cx.argument::<JsFunction>(2)?.root(&mut cx); let callback = cx.argument::<JsFunction>(2)?.root(&mut cx);
let channel = cx.channel(); let channel = cx.channel();
let mut map = THREAD_TRANSACTIONS.lock().unwrap(); let mut map = THREAD_TRANSACTIONS.lock().unwrap();
let new_tx_buffer = new_txs_arg.as_slice(&mut cx); let new_tx_buffer = new_txs_arg.as_slice(&mut cx);
let thread_transactions = ThreadTransaction::batch_from_buffer(new_tx_buffer); for tx in ThreadTransaction::batch_from_buffer(new_tx_buffer) {
for tx in thread_transactions {
map.insert(tx.uid, tx); map.insert(tx.uid, tx);
} }
let remove_tx_buffer = remove_txs_arg.as_slice(&mut cx); let remove_tx_buffer = remove_txs_arg.as_slice(&mut cx);
let remove_ids = utils::txids_from_buffer(remove_tx_buffer); for txid in &utils::txids_from_buffer(remove_tx_buffer) {
for txid in &remove_ids {
map.remove(txid); map.remove(txid);
} }
drop(map); drop(map);
@ -62,7 +68,7 @@ fn update(mut cx: FunctionContext) -> JsResult<JsUndefined> {
fn run_in_thread(channel: Channel, callback: Root<JsFunction>) { fn run_in_thread(channel: Channel, callback: Root<JsFunction>) {
std::thread::spawn(move || { std::thread::spawn(move || {
let mut map = THREAD_TRANSACTIONS.lock().unwrap(); let mut map = THREAD_TRANSACTIONS.lock().unwrap();
let (blocks, rates, clusters) = gbt::gbt(map.deref_mut()); let (blocks, rates, clusters) = gbt::gbt(map.deref_mut()).unwrap();
drop(map); drop(map);
channel.send(move |mut cx| { channel.send(move |mut cx| {
@ -104,9 +110,7 @@ fn run_in_thread(channel: Channel, callback: Root<JsFunction>) {
let callback = callback.into_inner(&mut cx); let callback = callback.into_inner(&mut cx);
let this = cx.undefined(); let this = cx.undefined();
let args = vec![ let args = vec![result.upcast()];
result.upcast()
];
callback.call(&mut cx, this, args)?; callback.call(&mut cx, this, args)?;

8
backend/rust-gbt/src/thread_transaction.rs
View File

@ -1,7 +1,5 @@
// use neon::{types::{JsObject, JsNumber, JsArray, JsValue, JsBoolean, JsArrayBuffer, buffer::TypedArray}, prelude::{Object, FunctionContext, Handle}};
extern crate bytes;
use std::io::Cursor;
use bytes::buf::Buf; use bytes::buf::Buf;
use std::io::Cursor;
pub struct ThreadTransaction { pub struct ThreadTransaction {
pub uid: u32, pub uid: u32,
@ -20,7 +18,7 @@ impl ThreadTransaction {
let size = cursor.get_u32(); let size = cursor.get_u32();
for _ in 0..size { for _ in 0..size {
let uid = cursor.get_u32(); let uid = cursor.get_u32();
let fee = cursor.get_f64() as u64; let fee = cursor.get_f64().round() as u64;
let weight = cursor.get_u32(); let weight = cursor.get_u32();
let sigops = cursor.get_u32(); let sigops = cursor.get_u32();
let fee_per_vsize = cursor.get_f64(); let fee_per_vsize = cursor.get_f64();
@ -41,6 +39,6 @@ impl ThreadTransaction {
}) })
} }
return transactions; transactions
} }
} }

9
backend/rust-gbt/src/utils.rs
View File

@ -1,14 +1,13 @@
extern crate bytes;
use std::io::Cursor;
use bytes::buf::Buf; use bytes::buf::Buf;
use std::io::Cursor;
pub fn txids_from_buffer(buffer: &[u8]) -> Vec<u32> { pub fn txids_from_buffer(buffer: &[u8]) -> Vec<u32> {
let mut txids: Vec<u32> = Vec::new(); let mut txids: Vec<u32> = Vec::new();
let mut cursor = Cursor::new(buffer); let mut cursor: Cursor<&[u8]> = Cursor::new(buffer);
let size = cursor.get_u32(); let size: u32 = cursor.get_u32();
for _ in 0..size { for _ in 0..size {
txids.push(cursor.get_u32()); txids.push(cursor.get_u32());
} }
return txids; txids
} }