Merge pull request #1570 from ethcore/dynamic-gas-price

Miner's gas price gets updated dynamically

ethcore/Cargo.toml

@@ -28,7 +28,7 @@ lazy_static = "0.2"
 ethcore-devtools = { path = "../devtools" }
 ethjson = { path = "../json" }
 bloomchain = "0.1"
-"ethcore-ipc" = { path = "../ipc/rpc" }
+ethcore-ipc = { path = "../ipc/rpc" }
 rayon = "0.3.1"
 ethstore = { path = "../ethstore" }
 semver = "0.2"

ethcore/src/miner/miner.rs

@@ -32,6 +32,7 @@ use engine::Engine;
 use miner::{MinerService, MinerStatus, TransactionQueue, AccountDetails, TransactionOrigin};
 use miner::work_notify::WorkPoster;
 use client::TransactionImportResult;
+use miner::price_info::PriceInfo;
 
 
 /// Different possible definitions for pending transaction set.
@@ -88,10 +89,78 @@ impl Default for MinerOptions {
 	}
 }
 
+/// Options for the dynamic gas price recalibrator.
+pub struct GasPriceCalibratorOptions {
+	/// Base transaction price to match against.
+	pub usd_per_tx: f32,
+	/// How frequently we should recalibrate. 
+	pub recalibration_period: Duration,
+}
+
+/// The gas price validator variant for a GasPricer.
+pub struct GasPriceCalibrator {
+	options: GasPriceCalibratorOptions,
+
+	next_calibration: Instant,
+}
+
+impl GasPriceCalibrator {
+	fn recalibrate<F: Fn(U256) + Sync + Send + 'static>(&mut self, set_price: F) {
+		trace!(target: "miner", "Recalibrating {:?} versus {:?}", Instant::now(), self.next_calibration);
+		if Instant::now() >= self.next_calibration {
+			let usd_per_tx = self.options.usd_per_tx;
+			trace!(target: "miner", "Getting price info");
+			if let Ok(_) = PriceInfo::get(move |price: PriceInfo| {
+				trace!(target: "miner", "Price info arrived: {:?}", price);
+				let usd_per_eth = price.ethusd;
+				let wei_per_usd: f32 = 1.0e18 / usd_per_eth;
+				let gas_per_tx: f32 = 21000.0;
+				let wei_per_gas: f32 = wei_per_usd * usd_per_tx / gas_per_tx;
+				info!(target: "miner", "Updated conversion rate to Ξ1 = {} ({} wei/gas)", format!("US${}", usd_per_eth).apply(Colour::White.bold()), format!("{}", wei_per_gas).apply(Colour::Yellow.bold()));
+				set_price(U256::from_dec_str(&format!("{:.0}", wei_per_gas)).unwrap());
+			}) {
+				self.next_calibration = Instant::now() + self.options.recalibration_period;
+			} else {
+				warn!(target: "miner", "Unable to update Ether price.");
+			}
+		}
+	}
+}
+
+/// Struct to look after updating the acceptable gas price of a miner.
+pub enum GasPricer {
+	/// A fixed gas price in terms of Wei - always the argument given.
+	Fixed(U256),
+	/// Gas price is calibrated according to a fixed amount of USD.
+	Calibrated(GasPriceCalibrator),
+}
+
+impl GasPricer {
+	/// Create a new Calibrated `GasPricer`.
+	pub fn new_calibrated(options: GasPriceCalibratorOptions) -> GasPricer {
+		GasPricer::Calibrated(GasPriceCalibrator {
+			options: options,
+			next_calibration: Instant::now(),
+		})
+	}
+
+	/// Create a new Fixed `GasPricer`.
+	pub fn new_fixed(gas_price: U256) -> GasPricer {
+		GasPricer::Fixed(gas_price)
+	}
+
+	fn recalibrate<F: Fn(U256) + Sync + Send + 'static>(&mut self, set_price: F) {
+		match *self {
+			GasPricer::Fixed(ref max) => set_price(max.clone()),
+			GasPricer::Calibrated(ref mut cal) => cal.recalibrate(set_price),
+		}
+	}
+}
+
 /// Keeps track of transactions using priority queue and holds currently mined block.
 pub struct Miner {
 	// NOTE [ToDr]  When locking always lock in this order!
-	transaction_queue: Mutex<TransactionQueue>,
+	transaction_queue: Arc<Mutex<TransactionQueue>>,
 	sealing_work: Mutex<UsingQueue<ClosedBlock>>,
 
 	// for sealing...
@@ -106,13 +175,14 @@ pub struct Miner {
 
 	accounts: Option<Arc<AccountProvider>>,
 	work_poster: Option<WorkPoster>,
+	gas_pricer: Mutex<GasPricer>,
 }
 
 impl Miner {
 	/// Creates new instance of miner without accounts, but with given spec.
 	pub fn with_spec(spec: Spec) -> Miner {
 		Miner {
-			transaction_queue: Mutex::new(TransactionQueue::new()),
+			transaction_queue: Arc::new(Mutex::new(TransactionQueue::new())),
 			options: Default::default(),
 			sealing_enabled: AtomicBool::new(false),
 			next_allowed_reseal: Mutex::new(Instant::now()),
@@ -124,14 +194,16 @@ impl Miner {
 			accounts: None,
 			spec: spec,
 			work_poster: None,
+			gas_pricer: Mutex::new(GasPricer::new_fixed(20_000_000_000u64.into())),
 		}
 	}
 
 	/// Creates new instance of miner
-	pub fn new(options: MinerOptions, spec: Spec, accounts: Option<Arc<AccountProvider>>) -> Arc<Miner> {
+	pub fn new(options: MinerOptions, gas_pricer: GasPricer, spec: Spec, accounts: Option<Arc<AccountProvider>>) -> Arc<Miner> {
 		let work_poster = if !options.new_work_notify.is_empty() { Some(WorkPoster::new(&options.new_work_notify)) } else { None };
+		let txq = Arc::new(Mutex::new(TransactionQueue::with_limits(options.tx_queue_size, options.tx_gas_limit)));
 		Arc::new(Miner {
-			transaction_queue: Mutex::new(TransactionQueue::with_limits(options.tx_queue_size, options.tx_gas_limit)),
+			transaction_queue: txq,
 			sealing_enabled: AtomicBool::new(options.force_sealing || !options.new_work_notify.is_empty()),
 			next_allowed_reseal: Mutex::new(Instant::now()),
 			sealing_block_last_request: Mutex::new(0),
@@ -143,6 +215,7 @@ impl Miner {
 			accounts: accounts,
 			spec: spec,
 			work_poster: work_poster,
+			gas_pricer: Mutex::new(gas_pricer),
 		})
 	}
 
@@ -160,6 +233,16 @@ impl Miner {
 	fn prepare_sealing(&self, chain: &MiningBlockChainClient) {
 		trace!(target: "miner", "prepare_sealing: entering");
 
+		{
+			trace!(target: "miner", "recalibrating...");
+			let txq = self.transaction_queue.clone();
+			self.gas_pricer.lock().unwrap().recalibrate(move |price| {
+				trace!(target: "miner", "Got gas price! {}", price);
+				txq.lock().unwrap().set_minimal_gas_price(price);
+			});
+			trace!(target: "miner", "done recalibration.");
+		}
+
 		let (transactions, mut open_block, original_work_hash) = {
 			let transactions = {self.transaction_queue.locked().top_transactions()};
 			let mut sealing_work = self.sealing_work.locked();
@@ -498,8 +581,11 @@ impl MinerService for Miner {
 		self.gas_range_target.unwrapped_read().1
 	}
 
-	fn import_external_transactions(&self, chain: &MiningBlockChainClient, transactions: Vec<SignedTransaction>) ->
+	fn import_external_transactions(
-		Vec<Result<TransactionImportResult, Error>> {
+		&self,
+		chain: &MiningBlockChainClient,
+		transactions: Vec<SignedTransaction>
+	) -> Vec<Result<TransactionImportResult, Error>> {
 
 		let results = {
 			let mut transaction_queue = self.transaction_queue.locked();

ethcore/src/miner/mod.rs

@@ -46,9 +46,10 @@ mod miner;
 mod external;
 mod transaction_queue;
 mod work_notify;
+mod price_info;
 
 pub use self::transaction_queue::{TransactionQueue, AccountDetails, TransactionOrigin};
-pub use self::miner::{Miner, MinerOptions, PendingSet};
+pub use self::miner::{Miner, MinerOptions, PendingSet, GasPricer, GasPriceCalibratorOptions};
 pub use self::external::{ExternalMiner, ExternalMinerService};
 pub use client::TransactionImportResult;
 

ethcore/src/miner/price_info.rs

@@ -0,0 +1,93 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+use rustc_serialize::json::Json;
+use std::thread;
+use std::io::Read;
+use std::time::Duration;
+use std::str::FromStr;
+use std::sync::mpsc;
+use hyper::client::{Handler, Request, Response, Client};
+use hyper::{Next, Encoder, Decoder};
+use hyper::net::HttpStream;
+
+#[derive(Debug)]
+pub struct PriceInfo {
+	pub ethusd: f32,
+}
+
+pub struct SetPriceHandler<F: Fn(PriceInfo) + Sync + Send + 'static> {
+	set_price: F,
+	channel: mpsc::Sender<()>,
+}
+
+impl<F: Fn(PriceInfo) + Sync + Send + 'static> Drop for SetPriceHandler<F> {
+	fn drop(&mut self) {
+		let _ = self.channel.send(());
+	}
+}
+
+impl<F: Fn(PriceInfo) + Sync + Send + 'static> Handler<HttpStream> for SetPriceHandler<F> {
+	fn on_request(&mut self, _: &mut Request) -> Next { Next::read().timeout(Duration::from_secs(3)) }
+	fn on_request_writable(&mut self, _: &mut Encoder<HttpStream>) -> Next { Next::read().timeout(Duration::from_secs(3)) }
+	fn on_response(&mut self, _: Response) -> Next { Next::read().timeout(Duration::from_secs(3)) }
+
+	fn on_response_readable(&mut self, r: &mut Decoder<HttpStream>) -> Next {
+		let mut body = String::new();
+		let _ = r.read_to_string(&mut body).ok()
+			.and_then(|_| Json::from_str(&body).ok())
+			.and_then(|json| json.find_path(&["result", "ethusd"])
+				.and_then(|obj| match *obj {
+					Json::String(ref s) => Some((self.set_price)(PriceInfo {
+						ethusd: FromStr::from_str(s).unwrap()
+					})),
+					_ => None,
+				}));
+		Next::end()
+	}
+
+}
+
+impl PriceInfo {
+	pub fn get<F: Fn(PriceInfo) + Sync + Send + 'static>(set_price: F) -> Result<(), ()> {
+		// TODO: Handle each error type properly
+		let client = try!(Client::new().map_err(|_| ()));
+		thread::spawn(move || {
+			let (tx, rx) = mpsc::channel();
+			let _ = client.request(FromStr::from_str("http://api.etherscan.io/api?module=stats&action=ethprice").unwrap(), SetPriceHandler {
+				set_price: set_price,
+				channel: tx,
+			}).ok().and_then(|_| rx.recv().ok());
+			client.close();
+		});
+		Ok(())
+	}
+}
+
+//#[ignore]
+#[test]
+fn should_get_price_info() {
+	use std::sync::{Condvar, Mutex, Arc};
+	use std::time::Duration;
+	use util::log::init_log;
+	init_log();
+	let done = Arc::new((Mutex::new(PriceInfo { ethusd: 0f32 }), Condvar::new()));
+	let rdone = done.clone();
+	PriceInfo::get(move |price| { let mut p = rdone.0.lock().unwrap(); *p = price; rdone.1.notify_one(); }).unwrap();
+	let p = done.1.wait_timeout(done.0.lock().unwrap(), Duration::from_millis(10000)).unwrap();
+	assert!(!p.1.timed_out());
+	assert!(p.0.ethusd != 0f32);
+}

parity/cli.rs

@@ -170,6 +170,10 @@ Sealing/Mining Options:
                            amount in USD, a web service or 'auto' to use each
                            web service in turn and fallback on the last known
                            good value [default: auto].
+  --price-update-period T  T will be allowed to pass between each gas price
+                           update. T may be daily, hourly, a number of seconds,
+                           or a time string of the form "2 days", "30 minutes"
+                           etc. [default: hourly].
   --gas-floor-target GAS   Amount of gas per block to target when sealing a new
                            block [default: 4700000].
   --gas-cap GAS            A cap on how large we will raise the gas limit per
@@ -335,6 +339,7 @@ pub struct Args {
 	pub flag_author: Option<String>,
 	pub flag_usd_per_tx: String,
 	pub flag_usd_per_eth: String,
+  pub flag_price_update_period: String,
 	pub flag_gas_floor_target: String,
 	pub flag_gas_cap: String,
 	pub flag_extra_data: Option<String>,

parity/configuration.rs

@@ -29,11 +29,10 @@ use util::log::Colour::*;
 use ethcore::account_provider::AccountProvider;
 use util::network_settings::NetworkSettings;
 use ethcore::client::{append_path, get_db_path, Mode, ClientConfig, DatabaseCompactionProfile, Switch, VMType};
-use ethcore::miner::{MinerOptions, PendingSet};
+use ethcore::miner::{MinerOptions, PendingSet, GasPricer, GasPriceCalibratorOptions};
 use ethcore::ethereum;
 use ethcore::spec::Spec;
 use ethsync::SyncConfig;
-use price_info::PriceInfo;
 use rpc::IpcConfiguration;
 
 pub struct Configuration {
@@ -154,35 +153,54 @@ impl Configuration {
 		})
 	}
 
-	pub fn gas_price(&self) -> U256 {
+	fn to_duration(s: &str) -> Duration {
+		let bad = |_| {
+			die!("{}: Invalid duration given. See parity --help for more information.", s)
+		};
+		Duration::from_secs(match s {
+			"daily" => 24 * 60 * 60,
+			"twice-daily" => 12 * 60 * 60,
+			"hourly" => 60 * 60,
+			"half-hourly" => 30 * 60,
+			"1second" | "1 second" | "second" => 1,
+			"1minute" | "1 minute" | "minute" => 60,
+			"1hour" | "1 hour" | "hour" => 60 * 60,
+			"1day" | "1 day" | "day" => 24 * 60 * 60,
+			x if x.ends_with("seconds") => FromStr::from_str(&x[0..x.len() - 7]).unwrap_or_else(bad),
+			x if x.ends_with("minutes") => FromStr::from_str(&x[0..x.len() - 7]).unwrap_or_else(bad) * 60,
+			x if x.ends_with("hours") => FromStr::from_str(&x[0..x.len() - 5]).unwrap_or_else(bad) * 60 * 60,
+			x if x.ends_with("days") => FromStr::from_str(&x[0..x.len() - 4]).unwrap_or_else(bad) * 24 * 60 * 60,
+ 			x => FromStr::from_str(x).unwrap_or_else(bad),
+		})
+	} 
+
+	pub fn gas_pricer(&self) -> GasPricer {
 		match self.args.flag_gasprice.as_ref() {
 			Some(d) => {
-				U256::from_dec_str(d).unwrap_or_else(|_| {
+				GasPricer::Fixed(U256::from_dec_str(d).unwrap_or_else(|_| {
 					die!("{}: Invalid gas price given. Must be a decimal unsigned 256-bit number.", d)
-				})
+				}))
 			}
 			_ => {
 				let usd_per_tx: f32 = FromStr::from_str(&self.args.flag_usd_per_tx).unwrap_or_else(|_| {
 					die!("{}: Invalid basic transaction price given in USD. Must be a decimal number.", self.args.flag_usd_per_tx)
 				});
-				let usd_per_eth = match self.args.flag_usd_per_eth.as_str() {
+				match self.args.flag_usd_per_eth.as_str() {
-					"auto" => PriceInfo::get().map_or_else(|| {
+					"auto" => {
-						let last_known_good = 9.69696;
+						GasPricer::new_calibrated(GasPriceCalibratorOptions {
-						// TODO: use #1083 to read last known good value.
+							usd_per_tx: usd_per_tx,
-						last_known_good
+							recalibration_period: Self::to_duration(self.args.flag_price_update_period.as_str()),
-					}, |x| x.ethusd),
+						})
-					"etherscan" => PriceInfo::get().map_or_else(|| {
+					},
-						die!("Unable to retrieve USD value of ETH from etherscan. Rerun with a different value for --usd-per-eth.")
+					x => {
-					}, |x| x.ethusd),
+						let usd_per_eth: f32 = FromStr::from_str(x).unwrap_or_else(|_| die!("{}: Invalid ether price given in USD. Must be a decimal number.", x));
-					x => FromStr::from_str(x).unwrap_or_else(|_| die!("{}: Invalid ether price given in USD. Must be a decimal number.", x))
+						let wei_per_usd: f32 = 1.0e18 / usd_per_eth;
-				};
+						let gas_per_tx: f32 = 21000.0;
-				// TODO: use #1083 to write last known good value as use_per_eth.
+						let wei_per_gas: f32 = wei_per_usd * usd_per_tx / gas_per_tx;
-
+						info!("Using a fixed conversion rate of Ξ1 = {} ({} wei/gas)", format!("US${}", usd_per_eth).apply(White.bold()), format!("{}", wei_per_gas).apply(Yellow.bold()));
-				let wei_per_usd: f32 = 1.0e18 / usd_per_eth;
+						GasPricer::Fixed(U256::from_dec_str(&format!("{:.0}", wei_per_gas)).unwrap())
-				let gas_per_tx: f32 = 21000.0;
+					}
-				let wei_per_gas: f32 = wei_per_usd * usd_per_tx / gas_per_tx;
+				}
-				info!("Using a conversion rate of Ξ1 = {} ({} wei/gas)", format!("US${}", usd_per_eth).apply(White.bold()), format!("{}", wei_per_gas).apply(Yellow.bold()));
-				U256::from_dec_str(&format!("{:.0}", wei_per_gas)).unwrap()
 			}
 		}
 	}

parity/io_handler.rs

@@ -49,14 +49,12 @@ impl IoHandler<NetSyncMessage> for ClientIoHandler {
 	fn message(&self, _io: &IoContext<NetSyncMessage>, message: &NetSyncMessage) {
 		match *message {
 			NetworkIoMessage::User(SyncMessage::StartNetwork) => {
-				info!("Starting network");
 				if let Some(network) = self.network.upgrade() {
 					network.start().unwrap_or_else(|e| warn!("Error starting network: {:?}", e));
 					EthSync::register(&*network, self.sync.clone()).unwrap_or_else(|e| warn!("Error registering eth protocol handler: {}", e));
 				}
 			},
 			NetworkIoMessage::User(SyncMessage::StopNetwork) => {
-				info!("Stopping network");
 				if let Some(network) = self.network.upgrade() {
 					network.stop().unwrap_or_else(|e| warn!("Error stopping network: {:?}", e));
 				}

parity/main.rs

@@ -57,7 +57,6 @@ extern crate ethcore_signer;
 
 #[macro_use]
 mod die;
-mod price_info;
 mod upgrade;
 mod setup_log;
 mod rpc;
@@ -223,12 +222,11 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
 	let account_service = Arc::new(conf.account_service());
 
 	// Miner
-	let miner = Miner::new(conf.miner_options(), conf.spec(), Some(account_service.clone()));
+	let miner = Miner::new(conf.miner_options(), conf.gas_pricer(), conf.spec(), Some(account_service.clone()));
 	miner.set_author(conf.author().unwrap_or_default());
 	miner.set_gas_floor_target(conf.gas_floor_target());
 	miner.set_gas_ceil_target(conf.gas_ceil_target());
 	miner.set_extra_data(conf.extra_data());
-	miner.set_minimal_gas_price(conf.gas_price());
 	miner.set_transactions_limit(conf.args.flag_tx_queue_size);
 
 	// Build client

parity/price_info.rs

@@ -1,47 +0,0 @@
-// Copyright 2015, 2016 Ethcore (UK) Ltd.
-// This file is part of Parity.
-
-// Parity is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-
-// Parity is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU General Public License for more details.
-
-// You should have received a copy of the GNU General Public License
-// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
-
-use rustc_serialize::json::Json;
-use std::io::Read;
-use hyper::Client;
-use hyper::header::Connection;
-use std::str::FromStr;
-
-pub struct PriceInfo {
-	pub ethusd: f32,
-}
-
-impl PriceInfo {
-	pub fn get() -> Option<PriceInfo> {
-		let mut body = String::new();
-		// TODO: Handle each error type properly
-		let mut client = Client::new();
-		client.set_read_timeout(Some(::std::time::Duration::from_secs(3)));
-		client.get("http://api.etherscan.io/api?module=stats&action=ethprice")
-			.header(Connection::close())
-			.send()
-			.ok()
-			.and_then(|mut s| s.read_to_string(&mut body).ok())
-			.and_then(|_| Json::from_str(&body).ok())
-			.and_then(|json| json.find_path(&["result", "ethusd"])
-				.and_then(|obj| match *obj {
-					Json::String(ref s) => Some(PriceInfo {
-						ethusd: FromStr::from_str(s).unwrap()
-					}),
-					_ => None
-				}))
-	}
-}

rpc/src/v1/tests/eth.rs

@@ -25,7 +25,7 @@ use ethcore::spec::{Genesis, Spec};
 use ethcore::block::Block;
 use ethcore::views::BlockView;
 use ethcore::ethereum;
-use ethcore::miner::{MinerOptions, MinerService, ExternalMiner, Miner, PendingSet};
+use ethcore::miner::{MinerOptions, GasPricer, MinerService, ExternalMiner, Miner, PendingSet};
 use ethcore::account_provider::AccountProvider;
 use devtools::RandomTempPath;
 use util::Hashable;
@@ -64,6 +64,7 @@ fn miner_service(spec: Spec, accounts: Arc<AccountProvider>) -> Arc<Miner> {
 			work_queue_size: 50,
 			enable_resubmission: true,
 		},
+		GasPricer::new_fixed(20_000_000_000u64.into()),
 		spec,
 		Some(accounts)
 	)

sync/src/lib.rs

@@ -38,13 +38,18 @@
 //! use ethcore::client::{Client, ClientConfig};
 //! use ethsync::{EthSync, SyncConfig};
 //! use ethcore::ethereum;
-//! use ethcore::miner::Miner;
+//! use ethcore::miner::{GasPricer, Miner};
 //!
 //! fn main() {
 //! 	let mut service = NetworkService::new(NetworkConfiguration::new()).unwrap();
 //! 	service.start().unwrap();
 //! 	let dir = env::temp_dir();
-//! 	let miner = Miner::new(Default::default(), ethereum::new_frontier(), None);
+//! 	let miner = Miner::new(
+//! 		Default::default(),
+//! 		GasPricer::new_fixed(20_000_000_000u64.into()),
+//! 		ethereum::new_frontier(),
+//! 		None
+//! 	);
 //! 	let client = Client::new(
 //!			ClientConfig::default(),
 //!			ethereum::new_frontier(),

util/src/log.rs

@@ -55,7 +55,7 @@ lazy_static! {
 			builder.parse(&log);
 		}
 
-		if let Ok(_) = builder.init() {
+		if builder.init().is_ok() {
 			println!("logger initialized");
 		}
 		true