Sort pool tx by fee over weight for mining (#1346)

* Sort pool tx by fee over weight for mining. Fixes #1105
* Bucketing dependent transactions before weighing. Minor tx weight fixes.
* Limit length of tx chain, cleanup and test fixes
* Cleanup all mining references to a hardcoded tx count
* Small test improvement, cleanup

config/tests/config.rs

@@ -6,7 +6,6 @@ use config::GlobalConfig;
 
 #[test]
 fn file_config_equal_to_defaults() {
-	let x = true;
 	let global_config_without_file = GlobalConfig::default();
 
 	let global_config_with_file = GlobalConfig::new(Some("../grin.toml")).unwrap_or_else(|e| {

core/src/core/transaction.rs

@@ -385,18 +385,30 @@ impl TransactionBody {
 
 	/// Calculate transaction weight
 	pub fn body_weight(&self) -> u32 {
-		TransactionBody::weight(self.inputs.len(), self.outputs.len())
+		TransactionBody::weight(self.inputs.len(), self.outputs.len(), self.kernels.len())
+	}
+
+	/// Calculate weight of transaction using block weighing
+	pub fn body_weight_as_block(&self) -> u32 {
+		TransactionBody::weight_as_block(self.inputs.len(), self.outputs.len(), self.kernels.len())
 	}
 
 	/// Calculate transaction weight from transaction details
-	pub fn weight(input_len: usize, output_len: usize) -> u32 {
+	pub fn weight(input_len: usize, output_len: usize, kernel_len: usize) -> u32 {
-		let mut body_weight = -1 * (input_len as i32) + (4 * output_len as i32) + 1;
+		let mut body_weight = -1 * (input_len as i32) + (4 * output_len as i32) + kernel_len as i32;
 		if body_weight < 1 {
 			body_weight = 1;
 		}
 		body_weight as u32
 	}
 
+	/// Calculate transaction weight using block weighing from transaction details
+	pub fn weight_as_block(input_len: usize, output_len: usize, kernel_len: usize) -> u32 {
+		(input_len * consensus::BLOCK_INPUT_WEIGHT
+			+ output_len * consensus::BLOCK_OUTPUT_WEIGHT
+			+ kernel_len * consensus::BLOCK_KERNEL_WEIGHT) as u32
+	}
+
 	/// Lock height of a body is the max lock height of the kernels.
 	pub fn lock_height(&self) -> u64 {
 		self.kernels
@@ -427,9 +439,11 @@ impl TransactionBody {
 		// if as_block check the body as if it was a block, with an additional output and
 		// kernel for reward
 		let reserve = if with_reward { 0 } else { 1 };
-		let tx_block_weight = self.inputs.len() * consensus::BLOCK_INPUT_WEIGHT
+		let tx_block_weight = TransactionBody::weight_as_block(
-			+ (self.outputs.len() + reserve) * consensus::BLOCK_OUTPUT_WEIGHT
+			self.inputs.len(),
-			+ (self.kernels.len() + reserve) * consensus::BLOCK_KERNEL_WEIGHT;
+			self.outputs.len() + reserve,
+			self.kernels.len() + reserve,
+		) as usize;
 
 		if tx_block_weight > consensus::MAX_BLOCK_WEIGHT {
 			return Err(Error::TooHeavy);
@@ -683,9 +697,14 @@ impl Transaction {
 		self.body.body_weight()
 	}
 
+	/// Calculate transaction weight as a block
+	pub fn tx_weight_as_block(&self) -> u32 {
+		self.body.body_weight_as_block()
+	}
+
 	/// Calculate transaction weight from transaction details
-	pub fn weight(input_len: usize, output_len: usize) -> u32 {
+	pub fn weight(input_len: usize, output_len: usize, kernel_len: usize) -> u32 {
-		TransactionBody::weight(input_len, output_len)
+		TransactionBody::weight(input_len, output_len, kernel_len)
 	}
 }
 
@@ -720,9 +739,14 @@ pub fn cut_through(inputs: &mut Vec<Input>, outputs: &mut Vec<Output>) -> Result
 /// cut_through. Optionally allows passing a reward output and kernel for
 /// block building.
 pub fn aggregate(
-	transactions: Vec<Transaction>,
+	mut transactions: Vec<Transaction>,
 	reward: Option<(Output, TxKernel)>,
 ) -> Result<Transaction, Error> {
+	// convenience short-circuiting
+	if reward.is_none() && transactions.len() == 1 {
+		return Ok(transactions.pop().unwrap());
+	}
+
 	let mut inputs: Vec<Input> = vec![];
 	let mut outputs: Vec<Output> = vec![];
 	let mut kernels: Vec<TxKernel> = vec![];

pool/src/pool.rs

@@ -15,9 +15,10 @@
 //! Transaction pool implementation.
 //! Used for both the txpool and stempool layers in the pool.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
+use core::consensus;
 use core::core::hash::Hashed;
 use core::core::id::ShortIdentifiable;
 use core::core::transaction;
@@ -25,6 +26,13 @@ use core::core::{Block, CompactBlock, Transaction, TxKernel};
 use types::{BlockChain, PoolEntry, PoolEntryState, PoolError};
 use util::LOGGER;
 
+// max weight leaving minimum space for a coinbase
+const MAX_MINEABLE_WEIGHT: usize =
+	consensus::MAX_BLOCK_WEIGHT - consensus::BLOCK_OUTPUT_WEIGHT - consensus::BLOCK_KERNEL_WEIGHT;
+
+// longest chain of dependent transactions that can be included in a block
+const MAX_TX_CHAIN: usize = 20;
+
 pub struct Pool<T> {
 	/// Entries in the pool (tx + info + timer) in simple insertion order.
 	pub entries: Vec<PoolEntry>,
@@ -64,17 +72,41 @@ where
 				}
 			}
 		}
-
 		txs
 	}
 
-	/// Take the first num_to_fetch txs based on insertion order.
+	/// Take pool transactions, filtering and ordering them in a way that's
-	pub fn prepare_mineable_transactions(&self, num_to_fetch: u32) -> Vec<Transaction> {
+	/// appropriate to put in a mined block. Aggregates chains of dependent
-		self.entries
+	/// transactions, orders by fee over weight and ensures to total weight
-			.iter()
+	/// doesn't exceed block limits.
-			.take(num_to_fetch as usize)
+	pub fn prepare_mineable_transactions(&self) -> Vec<Transaction> {
-			.map(|x| x.tx.clone())
+		let tx_buckets = self.bucket_transactions();
-			.collect()
+
+		// flatten buckets using aggregate (with cut-through)
+		let mut flat_txs: Vec<Transaction> = tx_buckets
+			.into_iter()
+			.filter_map(|mut bucket| {
+				bucket.truncate(MAX_TX_CHAIN);
+				transaction::aggregate(bucket, None).ok()
+			})
+			.collect();
+
+		// sort by fees over weight, multiplying by 1000 to keep some precision
+		// don't think we'll ever see a >max_u64/1000 fee transaction
+		flat_txs.sort_unstable_by_key(|tx| tx.fee() * 1000 / tx.tx_weight() as u64);
+
+		// accumulate as long as we're not above the block weight
+		let mut weight = 0;
+		flat_txs.retain(|tx| {
+			weight += tx.tx_weight_as_block() as usize;
+			weight < MAX_MINEABLE_WEIGHT
+		});
+
+		// make sure those txs are all valid together, no Error is expected
+		// when passing None
+		self.blockchain
+			.validate_raw_txs(flat_txs, None)
+			.expect("should never happen")
 	}
 
 	pub fn all_transactions(&self) -> Vec<Transaction> {
@@ -186,6 +218,41 @@ where
 		Ok(())
 	}
 
+	// Group dependent transactions in buckets (vectors), each bucket
+	// is therefore independent from the others. Relies on the entries
+	// Vec having parent transactions first (should always be the case)
+	fn bucket_transactions(&self) -> Vec<Vec<Transaction>> {
+		let mut tx_buckets = vec![];
+		let mut output_commits = HashMap::new();
+
+		for entry in &self.entries {
+			// check the commits index to find parents and their position
+			// picking the last one for bucket (so all parents come first)
+			let mut insert_pos: i32 = -1;
+			for input in entry.tx.inputs() {
+				if let Some(pos) = output_commits.get(&input.commitment()) {
+					if *pos > insert_pos {
+						insert_pos = *pos;
+					}
+				}
+			}
+			if insert_pos == -1 {
+				// no parent, just add to the end in its own bucket
+				insert_pos = tx_buckets.len() as i32;
+				tx_buckets.push(vec![entry.tx.clone()]);
+			} else {
+				// parent found, add to its bucket
+				tx_buckets[insert_pos as usize].push(entry.tx.clone());
+			}
+
+			// update the commits index
+			for out in entry.tx.outputs() {
+				output_commits.insert(out.commitment(), insert_pos);
+			}
+		}
+		tx_buckets
+	}
+
 	// Filter txs in the pool based on the latest block.
 	// Reject any txs where we see a matching tx kernel in the block.
 	// Also reject any txs where we see a conflicting tx,

pool/src/transaction_pool.rs

@@ -174,7 +174,7 @@ where
 
 	/// Returns a vector of transactions from the txpool so we can build a
 	/// block from them.
-	pub fn prepare_mineable_transactions(&self, num_to_fetch: u32) -> Vec<Transaction> {
+	pub fn prepare_mineable_transactions(&self) -> Vec<Transaction> {
-		self.txpool.prepare_mineable_transactions(num_to_fetch)
+		self.txpool.prepare_mineable_transactions()
 	}
 }

pool/tests/block_building.rs

@@ -20,8 +20,8 @@ extern crate grin_pool as pool;
 extern crate grin_util as util;
 extern crate grin_wallet as wallet;
 
-extern crate rand;
 extern crate chrono;
+extern crate rand;
 
 pub mod common;
 
@@ -29,8 +29,8 @@ use std::sync::{Arc, RwLock};
 
 use core::core::{Block, BlockHeader};
 
+use chain::txhashset;
 use chain::types::Tip;
-use chain::{txhashset, ChainStore};
 use core::core::target::Difficulty;
 
 use keychain::{ExtKeychain, Keychain};
@@ -40,6 +40,7 @@ use common::*;
 
 #[test]
 fn test_transaction_pool_block_building() {
+	util::init_test_logger();
 	let keychain: ExtKeychain = Keychain::from_random_seed().unwrap();
 
 	let db_root = ".grin_block_building".to_string();
@@ -48,11 +49,10 @@ fn test_transaction_pool_block_building() {
 
 	// Initialize the chain/txhashset with an initial block
 	// so we have a non-empty UTXO set.
-	let header = {
+	let add_block = |height, txs| {
-		let height = 1;
 		let key_id = keychain.derive_key_id(height as u32).unwrap();
 		let reward = libtx::reward::output(&keychain, &key_id, 0, height).unwrap();
-		let block = Block::new(&BlockHeader::default(), vec![], Difficulty::one(), reward).unwrap();
+		let block = Block::new(&BlockHeader::default(), txs, Difficulty::one(), reward).unwrap();
 
 		let mut txhashset = chain.txhashset.write().unwrap();
 		let mut batch = chain.store.batch().unwrap();
@@ -67,6 +67,7 @@ fn test_transaction_pool_block_building() {
 
 		block.header
 	};
+	let header = add_block(1, vec![]);
 
 	// Initialize a new pool with our chain adapter.
 	let pool = RwLock::new(test_setup(&Arc::new(chain.clone())));
@@ -75,14 +76,8 @@ fn test_transaction_pool_block_building() {
 	// Provides us with some useful outputs to test with.
 	let initial_tx = test_transaction_spending_coinbase(&keychain, &header, vec![10, 20, 30, 40]);
 
-	// Add this tx to the pool (stem=false, direct to txpool).
+	// Mine that initial tx so we can spend it with multiple txs
-	{
+	let header = add_block(2, vec![initial_tx]);
-		let mut write_pool = pool.write().unwrap();
-		write_pool
-			.add_to_pool(test_source(), initial_tx, false)
-			.unwrap();
-		assert_eq!(write_pool.total_size(), 1);
-	}
 
 	let root_tx_1 = test_transaction(&keychain, vec![10, 20], vec![24]);
 	let root_tx_2 = test_transaction(&keychain, vec![30], vec![28]);
@@ -113,14 +108,15 @@ fn test_transaction_pool_block_building() {
 			.add_to_pool(test_source(), child_tx_2.clone(), false)
 			.unwrap();
 
-		assert_eq!(write_pool.total_size(), 6);
+		assert_eq!(write_pool.total_size(), 5);
 	}
 
 	let txs = {
 		let read_pool = pool.read().unwrap();
-		read_pool.prepare_mineable_transactions(4)
+		read_pool.prepare_mineable_transactions()
 	};
-	assert_eq!(txs.len(), 4);
+	// children should have been aggregated into parents
+	assert_eq!(txs.len(), 3);
 
 	let block = {
 		let key_id = keychain.derive_key_id(2).unwrap();
@@ -145,8 +141,6 @@ fn test_transaction_pool_block_building() {
 		let mut write_pool = pool.write().unwrap();
 		write_pool.reconcile_block(&block).unwrap();
 
-		assert_eq!(write_pool.total_size(), 2);
+		assert_eq!(write_pool.total_size(), 0);
-		assert_eq!(write_pool.txpool.entries[0].tx, child_tx_1);
-		assert_eq!(write_pool.txpool.entries[1].tx, child_tx_2);
 	}
 }

servers/src/mining/mine_block.rs

@@ -78,18 +78,11 @@ pub fn get_block(
 	chain: &Arc<chain::Chain>,
 	tx_pool: &Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>,
 	key_id: Option<Identifier>,
-	max_tx: u32,
 	wallet_listener_url: Option<String>,
 ) -> (core::Block, BlockFees) {
 	let wallet_retry_interval = 5;
 	// get the latest chain state and build a block on top of it
-	let mut result = build_block(
+	let mut result = build_block(chain, tx_pool, key_id.clone(), wallet_listener_url.clone());
-		chain,
-		tx_pool,
-		key_id.clone(),
-		max_tx,
-		wallet_listener_url.clone(),
-	);
 	while let Err(e) = result {
 		match e {
 			self::Error::Chain(c) => match c.kind() {
@@ -116,13 +109,7 @@ pub fn get_block(
 			}
 		}
 		thread::sleep(Duration::from_millis(100));
-		result = build_block(
+		result = build_block(chain, tx_pool, key_id.clone(), wallet_listener_url.clone());
-			chain,
-			tx_pool,
-			key_id.clone(),
-			max_tx,
-			wallet_listener_url.clone(),
-		);
 	}
 	return result.unwrap();
 }
@@ -133,7 +120,6 @@ fn build_block(
 	chain: &Arc<chain::Chain>,
 	tx_pool: &Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>,
 	key_id: Option<Identifier>,
-	max_tx: u32,
 	wallet_listener_url: Option<String>,
 ) -> Result<(core::Block, BlockFees), Error> {
 	// prepare the block header timestamp
@@ -150,10 +136,7 @@ fn build_block(
 	let difficulty = consensus::next_difficulty(diff_iter).unwrap();
 
 	// extract current transaction from the pool
-	let txs = tx_pool
+	let txs = tx_pool.read().unwrap().prepare_mineable_transactions();
-		.read()
-		.unwrap()
-		.prepare_mineable_transactions(max_tx);
 
 	// build the coinbase and the block itself
 	let fees = txs.iter().map(|tx| tx.fee()).sum();

servers/src/mining/stratumserver.rs

@@ -14,6 +14,7 @@
 
 //! Mining Stratum Server
 use bufstream::BufStream;
+use chrono::prelude::Utc;
 use serde;
 use serde_json;
 use serde_json::Value;
@@ -23,22 +24,18 @@ use std::net::{TcpListener, TcpStream};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, SystemTime};
 use std::{cmp, thread};
-use chrono::prelude::{Utc};
 
 use chain;
 use common::adapters::PoolToChainAdapter;
 use common::stats::{StratumStats, WorkerStats};
 use common::types::{StratumServerConfig, SyncState};
 use core::core::Block;
-use core::{pow, global};
+use core::{global, pow};
 use keychain;
 use mining::mine_block;
 use pool;
 use util::LOGGER;
 
-// Max number of transactions this miner will assemble in a block
-const MAX_TX: u32 = 5000;
-
 // ----------------------------------------
 // http://www.jsonrpc.org/specification
 // RPC Methods
@@ -438,10 +435,9 @@ impl StratumServer {
 		worker: &mut Worker,
 		worker_stats: &mut WorkerStats,
 	) -> Result<(Value, bool), Value> {
-
 		// Validate parameters
 		let params: SubmitParams = parse_params(params)?;
-		
+
 		let share_difficulty: u64;
 		let mut share_is_block = false;
 		if params.height != self.current_block_versions.last().unwrap().header.height {
@@ -567,7 +563,10 @@ impl StratumServer {
 		} else {
 			submit_response = "ok".to_string();
 		}
-		return Ok((serde_json::to_value(submit_response).unwrap(), share_is_block));
+		return Ok((
+			serde_json::to_value(submit_response).unwrap(),
+			share_is_block,
+		));
 	} // handle submit a solution
 
 	// Purge dead/sick workers - remove all workers marked in error state
@@ -711,7 +710,8 @@ impl StratumServer {
 			// or We are rebuilding the current one to include new transactions
 			// and we're not synching
 			// and there is at least one worker connected
-			if (current_hash != latest_hash || Utc::now().timestamp() >= deadline) && !mining_stopped
+			if (current_hash != latest_hash || Utc::now().timestamp() >= deadline)
+				&& !mining_stopped
 				&& num_workers > 0
 			{
 				let mut wallet_listener_url: Option<String> = None;
@@ -727,7 +727,6 @@ impl StratumServer {
 					&self.chain,
 					&self.tx_pool,
 					self.current_key_id.clone(),
-					MAX_TX.clone(),
 					wallet_listener_url,
 				);
 				self.current_difficulty = (new_block.header.total_difficulty.clone()
@@ -763,12 +762,11 @@ impl StratumServer {
 	} // fn run_loop()
 } // StratumServer
 
-
 // Utility function to parse a JSON RPC parameter object, returning a proper
 // error if things go wrong.
 fn parse_params<T>(params: Option<Value>) -> Result<T, Value>
 where
-	for<'de> T: serde::Deserialize<'de>
+	for<'de> T: serde::Deserialize<'de>,
 {
 	params
 		.and_then(|v| serde_json::from_value(v).ok())
@@ -780,4 +778,3 @@ where
 			serde_json::to_value(e).unwrap()
 		})
 }
-

servers/src/mining/test_miner.rs

@@ -1,4 +1,3 @@
-
 // Copyright 2018 The Grin Developers
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -18,9 +17,9 @@
 //! header with its proof-of-work.  Any valid mined blocks are submitted to the
 //! network.
 
+use chrono::prelude::Utc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, RwLock};
-use chrono::prelude::{Utc};
 
 use chain;
 use common::adapters::PoolToChainAdapter;
@@ -33,9 +32,6 @@ use mining::mine_block;
 use pool;
 use util::LOGGER;
 
-// Max number of transactions this miner will assemble in a block
-const MAX_TX: u32 = 5000;
-
 pub struct Miner {
 	config: StratumServerConfig,
 	chain: Arc<chain::Chain>,
@@ -153,7 +149,6 @@ impl Miner {
 				&self.chain,
 				&self.tx_pool,
 				key_id.clone(),
-				MAX_TX.clone(),
 				wallet_listener_url.clone(),
 			);
 

wallet/src/libtx/mod.rs

@@ -36,11 +36,16 @@ pub use libtx::error::{Error, ErrorKind};
 const DEFAULT_BASE_FEE: u64 = consensus::MILLI_GRIN;
 
 /// Transaction fee calculation
-pub fn tx_fee(input_len: usize, output_len: usize, base_fee: Option<u64>) -> u64 {
+pub fn tx_fee(
+	input_len: usize,
+	output_len: usize,
+	kernel_len: usize,
+	base_fee: Option<u64>,
+) -> u64 {
 	let use_base_fee = match base_fee {
 		Some(bf) => bf,
 		None => DEFAULT_BASE_FEE,
 	};
 
-	(Transaction::weight(input_len, output_len) as u64) * use_base_fee
+	(Transaction::weight(input_len, output_len, kernel_len) as u64) * use_base_fee
 }

wallet/src/libtx/proof.rs

@@ -83,7 +83,8 @@ where
 	K: Keychain,
 {
 	let nonce = create_nonce(k, &commit)?;
-	let proof_message = k.secp()
+	let proof_message = k
+		.secp()
 		.rewind_bullet_proof(commit, nonce, extra_data, proof);
 	let proof_info = match proof_message {
 		Ok(p) => p,

wallet/src/libtx/slate.rs

@@ -267,7 +267,12 @@ impl Slate {
 		// double check the fee amount included in the partial tx
 		// we don't necessarily want to just trust the sender
 		// we could just overwrite the fee here (but we won't) due to the sig
-		let fee = tx_fee(self.tx.inputs().len(), self.tx.outputs().len(), None);
+		let fee = tx_fee(
+			self.tx.inputs().len(),
+			self.tx.outputs().len(),
+			self.tx.kernels().len(),
+			None,
+		);
 		if fee > self.tx.fee() {
 			return Err(ErrorKind::Fee(
 				format!("Fee Dispute Error: {}, {}", self.tx.fee(), fee,).to_string(),

wallet/src/libwallet/internal/selection.rs

@@ -254,7 +254,7 @@ where
 	// TODO - Does this not potentially reveal the senders private key?
 	//
 	// First attempt to spend without change
-	let mut fee = tx_fee(coins.len(), 1, None);
+	let mut fee = tx_fee(coins.len(), 1, 1, None);
 	let mut total: u64 = coins.iter().map(|c| c.value).sum();
 	let mut amount_with_fee = amount + fee;
 
@@ -277,7 +277,7 @@ where
 
 	// We need to add a change address or amount with fee is more than total
 	if total != amount_with_fee {
-		fee = tx_fee(coins.len(), num_outputs, None);
+		fee = tx_fee(coins.len(), num_outputs, 1, None);
 		amount_with_fee = amount + fee;
 
 		// Here check if we have enough outputs for the amount including fee otherwise
@@ -300,7 +300,7 @@ where
 				max_outputs,
 				selection_strategy_is_use_all,
 			);
-			fee = tx_fee(coins.len(), num_outputs, None);
+			fee = tx_fee(coins.len(), num_outputs, 1, None);
 			total = coins.iter().map(|c| c.value).sum();
 			amount_with_fee = amount + fee;
 		}

wallet/src/libwallet/internal/tx.rs

@@ -191,7 +191,7 @@ where
 
 	debug!(LOGGER, "selected some coins - {}", coins.len());
 
-	let fee = tx_fee(coins.len(), 2, None);
+	let fee = tx_fee(coins.len(), 2, 1, None);
 	let num_change_outputs = 1;
 	let (mut parts, _) =
 		selection::inputs_and_change(&coins, wallet, amount, fee, num_change_outputs)?;

wallet/tests/transaction.rs

@@ -137,6 +137,7 @@ fn basic_transaction_api(
 		let fee = wallet::libtx::tx_fee(
 			wallet1_info.last_confirmed_height as usize - cm as usize,
 			2,
+			1,
 			None,
 		);
 		// we should have a transaction entry for this slate
@@ -184,6 +185,7 @@ fn basic_transaction_api(
 		let fee = wallet::libtx::tx_fee(
 			wallet1_info.last_confirmed_height as usize - 1 - cm as usize,
 			2,
+			1,
 			None,
 		);
 		assert!(wallet1_refreshed);