Stratum up (#4233)

* flush work * flush work * flush work * flush work * generalized notifiers * general setup with modules * general setup with modules * all binded * catch up with master * all dependencies injected * stratum another up * tcp update * submitwork routine * finalize & fix warnings * merge bugs, review fixes * merge bugs, review fixes * new cli mess cleanup * usage.txt swap * flush work * cli adopt * compilation with new cli sorted * subid space in json * serialization issues * grumbles addressed * more grumbles * remove last_work note for now * fix compilation * fix tests * merge bugs * no obliged ipc * moving notifiers * no optional feature now * refactored again * working on tests * refactor to new tcp/ip * stratum lib ok * ethcore crate ok * wip on tests * final test working * fix warnings, \n-terminated response * new compatibility * re-pushing work once anybody submitted * various review and general fixes * reviewe fixes * remove redundant notifier * one symbol -> huge bug * ensure write lock isn't held when calling handlers * extern declarations moved * options to stratum mod, SocketAddr strongly-typed instantiation * Minor style fix. * Whitespace and phrasing * Whitespace
2017-01-25 13:03:36 +03:00
parent 67284cc1a2
commit 1acc8031ce
22 changed files with 830 additions and 258 deletions
--- a/parity/cli/mod.rs
+++ b/parity/cli/mod.rs
@@ -233,6 +233,15 @@ usage! {
 		flag_refuse_service_transactions: bool = false,
 			or |c: &Config| otry!(c.mining).refuse_service_transactions.clone(),

+		flag_stratum: bool = false,
+			or |c: &Config| Some(c.stratum.is_some()),
+		flag_stratum_interface: String = "local",
+			or |c: &Config| otry!(c.stratum).interface.clone(),
+		flag_stratum_port: u16 = 8008u16,
+			or |c: &Config| otry!(c.stratum).port.clone(),
+		flag_stratum_secret: Option<String> = None,
+			or |c: &Config| otry!(c.stratum).secret.clone().map(Some),
+
 		// -- Footprint Options
 		flag_tracing: String = "auto",
 			or |c: &Config| otry!(c.footprint).tracing.clone(),
@@ -313,6 +322,7 @@ struct Config {
 	snapshots: Option<Snapshots>,
 	vm: Option<VM>,
 	misc: Option<Misc>,
+	stratum: Option<Stratum>,
 }

 #[derive(Default, Debug, PartialEq, RustcDecodable)]
@@ -421,6 +431,13 @@ struct Mining {
 	refuse_service_transactions: Option<bool>,
 }

+#[derive(Default, Debug, PartialEq, RustcDecodable)]
+struct Stratum {
+	interface: Option<String>,
+	port: Option<u16>,
+	secret: Option<String>,
+}
+
 #[derive(Default, Debug, PartialEq, RustcDecodable)]
 struct Footprint {
 	tracing: Option<String>,
@@ -638,6 +655,11 @@ mod tests {
 			flag_notify_work: Some("http://localhost:3001".into()),
 			flag_refuse_service_transactions: false,

+			flag_stratum: false,
+			flag_stratum_interface: "local".to_owned(),
+			flag_stratum_port: 8008u16,
+			flag_stratum_secret: None,
+
 			// -- Footprint Options
 			flag_tracing: "auto".into(),
 			flag_pruning: "auto".into(),
@@ -843,7 +865,8 @@ mod tests {
 				logging: Some("own_tx=trace".into()),
 				log_file: Some("/var/log/parity.log".into()),
 				color: Some(true),
-			})
+			}),
+			stratum: None,
 		});
 	}
 }
--- a/parity/cli/usage.txt
+++ b/parity/cli/usage.txt
@@ -260,6 +260,11 @@ Sealing/Mining Options:
                                 (default: {flag_notify_work:?})
  --refuse-service-transactions  Always refuse service transactions.
                                 (default: {flag_refuse_service_transactions}).
+  --stratum                      Run Stratum server for miner push notification. (default: {flag_stratum})
+  --stratum-interface IP         Interface address for Stratum server. (default: {flag_stratum_interface})
+  --stratum-port PORT            Port for Stratum server to listen on. (default: {flag_stratum_port})
+  --stratum-secret STRING        Secret for authorizing Stratum server for peers.
+	                               (default: {flag_stratum_secret:?})

 Footprint Options:
  --tracing BOOL                 Indicates if full transaction tracing should be
--- a/parity/configuration.rs
+++ b/parity/configuration.rs
@@ -20,12 +20,12 @@ use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use std::cmp::max;
 use cli::{Args, ArgsError};
-use util::{Hashable, U256, Uint, Bytes, version_data, Address};
+use util::{Hashable, H256, U256, Uint, Bytes, version_data, Address};
 use util::log::Colour;
 use ethsync::{NetworkConfiguration, is_valid_node_url, AllowIP};
 use ethcore::ethstore::ethkey::Secret;
 use ethcore::client::{VMType};
-use ethcore::miner::{MinerOptions, Banning};
+use ethcore::miner::{MinerOptions, Banning, StratumOptions};
 use ethcore::verification::queue::VerifierSettings;

 use rpc::{IpcConfiguration, HttpConfiguration};
@@ -329,6 +329,7 @@ impl Configuration {
 				acc_conf: self.accounts_config()?,
 				gas_pricer: self.gas_pricer_config()?,
 				miner_extras: self.miner_extras()?,
+				stratum: self.stratum_options()?,
 				update_policy: update_policy,
 				mode: mode,
 				tracing: tracing,
@@ -465,6 +466,17 @@ impl Configuration {
 		Ok(cfg)
 	}

+	fn stratum_options(&self) -> Result<Option<StratumOptions>, String> {
+		if self.args.flag_stratum {
+			Ok(Some(StratumOptions {
+				io_path: self.directories().db,
+				listen_addr: self.stratum_interface(),
+				port: self.args.flag_stratum_port,
+				secret: self.args.flag_stratum_secret.as_ref().map(|s| s.parse::<H256>().unwrap_or_else(|_| s.sha3())),
+			}))
+		} else { Ok(None) }
+	}
+
 	fn miner_options(&self) -> Result<MinerOptions, String> {
 		let reseal = self.args.flag_reseal_on_txs.parse::<ResealPolicy>()?;

@@ -827,6 +839,14 @@ impl Configuration {
 		}.into()
 	}

+	fn stratum_interface(&self) -> String {
+		match self.args.flag_stratum_interface.as_str() {
+			"local" => "127.0.0.1",
+			"all" => "0.0.0.0",
+			x => x,
+		}.into()
+	}
+
 	fn dapps_enabled(&self) -> bool {
 		!self.args.flag_dapps_off && !self.args.flag_no_dapps && cfg!(feature = "dapps")
 	}
@@ -1086,6 +1106,7 @@ mod tests {
 			custom_bootnodes: false,
 			fat_db: Default::default(),
 			no_periodic_snapshot: false,
+			stratum: None,
 			check_seal: true,
 			download_old_blocks: true,
 			verifier_settings: Default::default(),
--- a/parity/modules.rs
+++ b/parity/modules.rs
@@ -45,10 +45,7 @@ pub mod service_urls {
 	pub const LIGHT_PROVIDER: &'static str = "parity-light-provider.ipc";

 	#[cfg(feature="stratum")]
-	pub const STRATUM: &'static str = "parity-stratum.ipc";
-	#[cfg(feature="stratum")]
-	pub const MINING_JOB_DISPATCHER: &'static str = "parity-mining-jobs.ipc";
-
+	pub const STRATUM_CONTROL: &'static str = "parity-stratum-control.ipc";

 	pub fn with_base(data_dir: &str, service_path: &str) -> String {
 		let mut path = PathBuf::from(data_dir);
@@ -126,18 +123,35 @@ fn sync_arguments(io_path: &str, sync_cfg: SyncConfig, net_cfg: NetworkConfigura
 }

 #[cfg(feature="ipc")]
-pub fn sync
-	(
-		hypervisor_ref: &mut Option<Hypervisor>,
-		sync_cfg: SyncConfig,
-		net_cfg: NetworkConfiguration,
-		_client: Arc<BlockChainClient>,
-		_snapshot_service: Arc<SnapshotService>,
-		_provider: Arc<Provider>,		
-		log_settings: &LogConfig,
-	)
-	-> Result<SyncModules, NetworkError>
-{
+pub fn stratum(
+	hypervisor_ref: &mut Option<Hypervisor>,
+	config: &::ethcore::miner::StratumOptions
+) {
+	use ethcore_stratum;
+
+	let mut hypervisor = hypervisor_ref.take().expect("There should be hypervisor for ipc configuration");
+	let args = BootArgs::new().stdin(
+			serialize(&ethcore_stratum::ServiceConfiguration {
+				io_path: hypervisor.io_path.to_owned(),
+				port: config.port,
+				listen_addr: config.listen_addr.to_owned(),
+				secret: config.secret,
+			}).expect("Any binary-derived struct is serializable by definition")
+		).cli(vec!["stratum".to_owned()]);
+	hypervisor = hypervisor.module(super::stratum::MODULE_ID, args);
+	*hypervisor_ref = Some(hypervisor);
+}
+
+#[cfg(feature="ipc")]
+pub fn sync(
+	hypervisor_ref: &mut Option<Hypervisor>,
+	sync_cfg: SyncConfig,
+	net_cfg: NetworkConfiguration,
+	_client: Arc<BlockChainClient>,
+	_snapshot_service: Arc<SnapshotService>,
+	_provider: Arc<Provider>,
+	log_settings: &LogConfig,
+) -> Result<SyncModules, NetworkError> {
 	let mut hypervisor = hypervisor_ref.take().expect("There should be hypervisor for ipc configuration");
 	let args = sync_arguments(&hypervisor.io_path, sync_cfg, net_cfg, log_settings);
 	hypervisor = hypervisor.module(SYNC_MODULE_ID, args);
@@ -153,29 +167,26 @@ pub fn sync
 		&service_urls::with_base(&hypervisor.io_path, service_urls::NETWORK_MANAGER)).unwrap();
 	let provider_client = generic_client::<LightProviderClient<_>>(
 		&service_urls::with_base(&hypervisor.io_path, service_urls::LIGHT_PROVIDER)).unwrap();
-		
+
 	*hypervisor_ref = Some(hypervisor);
 	Ok((sync_client, manage_client, notify_client))
 }

 #[cfg(not(feature="ipc"))]
-pub fn sync
-	(
-		_hypervisor: &mut Option<Hypervisor>,
-		sync_cfg: SyncConfig,
-		net_cfg: NetworkConfiguration,
-		client: Arc<BlockChainClient>,
-		snapshot_service: Arc<SnapshotService>,
-		provider: Arc<Provider>,		
-		_log_settings: &LogConfig,
-	)
-	-> Result<SyncModules, NetworkError>
-{
+pub fn sync(
+	_hypervisor: &mut Option<Hypervisor>,
+	sync_cfg: SyncConfig,
+	net_cfg: NetworkConfiguration,
+	client: Arc<BlockChainClient>,
+	snapshot_service: Arc<SnapshotService>,
+	provider: Arc<Provider>,
+	_log_settings: &LogConfig,
+) -> Result<SyncModules, NetworkError> {
 	let eth_sync = EthSync::new(Params {
-		config: sync_cfg, 
+		config: sync_cfg,
 		chain: client,
 		provider: provider,
-		snapshot_service: snapshot_service, 
+		snapshot_service: snapshot_service,
 		network_config: net_cfg,
 	})?;

--- a/parity/run.rs
+++ b/parity/run.rs
@@ -23,6 +23,7 @@ use ethsync::NetworkConfiguration;
 use util::{Colour, version, RotatingLogger, Mutex, Condvar};
 use io::{MayPanic, ForwardPanic, PanicHandler};
 use ethcore_logger::{Config as LogConfig};
+use ethcore::miner::{StratumOptions, Stratum};
 use ethcore::client::{Mode, DatabaseCompactionProfile, VMType, BlockChainClient};
 use ethcore::service::ClientService;
 use ethcore::account_provider::AccountProvider;
@@ -97,6 +98,7 @@ pub struct RunCmd {
 	pub ui: bool,
 	pub name: String,
 	pub custom_bootnodes: bool,
+	pub stratum: Option<StratumOptions>,
 	pub no_periodic_snapshot: bool,
 	pub check_seal: bool,
 	pub download_old_blocks: bool,
@@ -315,6 +317,12 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
 	// create external miner
 	let external_miner = Arc::new(ExternalMiner::default());

+	// start stratum
+	if let Some(ref stratum_config) = cmd.stratum {
+		Stratum::register(stratum_config, miner.clone(), Arc::downgrade(&client))
+			.map_err(|e| format!("Stratum start error: {:?}", e))?;
+	}
+
 	// create sync object
 	let (sync_provider, manage_network, chain_notify) = modules::sync(
 		&mut hypervisor,
--- a/parity/stratum.rs
+++ b/parity/stratum.rs
@@ -16,42 +16,104 @@

 //! Parity sync service

-use std;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use ethcore_stratum::{Stratum as StratumServer, PushWorkHandler, RemoteJobDispatcher, ServiceConfiguration};
-use std::thread;
 use modules::service_urls;
 use boot;
 use hypervisor::service::IpcModuleId;
-use std::net::SocketAddr;
+use hypervisor::{HYPERVISOR_IPC_URL, ControlService};
+use std::net::{SocketAddr, IpAddr};
 use std::str::FromStr;
+use nanoipc;
+use std::thread;
+use ethcore::miner::stratum::{STRATUM_SOCKET_NAME, JOB_DISPATCHER_SOCKET_NAME};

-const STRATUM_MODULE_ID: IpcModuleId = 8000;
+pub const MODULE_ID: IpcModuleId = 8000;
+
+#[derive(Default)]
+struct StratumControlService {
+	pub stop: Arc<AtomicBool>,
+}
+
+impl ControlService for StratumControlService {
+	fn shutdown(&self) -> bool {
+		trace!(target: "hypervisor", "Received shutdown from control service");
+		self.stop.store(true, ::std::sync::atomic::Ordering::Relaxed);
+		true
+	}
+}

 pub fn main() {
 	boot::setup_cli_logger("stratum");

 	let service_config: ServiceConfiguration = boot::payload()
-		.unwrap_or_else(|e| panic!("Fatal: error reading boot arguments ({:?})", e));
+		.unwrap_or_else(|e| {
+			println!("Fatal: error reading boot arguments ({:?})", e);
+			std::process::exit(1)
+		});

-	let job_dispatcher = dependency!(RemoteJobDispatcher, service_urls::MINING_JOB_DISPATCHER);
+	let job_dispatcher = dependency!(
+		RemoteJobDispatcher,
+		&service_urls::with_base(&service_config.io_path, JOB_DISPATCHER_SOCKET_NAME)
+	);
+
+	let _ = boot::main_thread();
+	let service_stop = Arc::new(AtomicBool::new(false));

-	let stop = boot::main_thread();
 	let server =
 		StratumServer::start(
-			&SocketAddr::from_str(&service_config.listen_addr)
-				.unwrap_or_else(|e| panic!("Fatal: invalid listen address ({:?})", e)),
+			&SocketAddr::new(
+				IpAddr::from_str(&service_config.listen_addr)
+					.unwrap_or_else(|e|
+						println!("Fatal: invalid listen address: '{}' ({:?})", &service_config.listen_addr, e);
+						std::process::exit(1)
+					),
+				service_config.port,
+			),
 			job_dispatcher.service().clone(),
 			service_config.secret
 		).unwrap_or_else(
-			|e| panic!("Fatal: cannot start stratum server({:?})", e)
+			|e| {
+				println!("Fatal: cannot start stratum server({:?})", e);
+				std::process::exit(1)
+			}
 		);

-	boot::host_service(service_urls::STRATUM, stop.clone(), server.clone() as Arc<PushWorkHandler>);
+	boot::host_service(
+		&service_urls::with_base(&service_config.io_path, STRATUM_SOCKET_NAME),
+		service_stop.clone(),
+		server.clone() as Arc<PushWorkHandler>
+	);

-	let _ = boot::register(STRATUM_MODULE_ID);
+	let hypervisor = boot::register(
+		&service_urls::with_base(&service_config.io_path, HYPERVISOR_IPC_URL),
+		&service_urls::with_base(&service_config.io_path, service_urls::STRATUM_CONTROL),
+		MODULE_ID
+	);

-	while !stop.load(::std::sync::atomic::Ordering::Relaxed) {
-		thread::park_timeout(std::time::Duration::from_millis(1000));
+	let timer_svc = server.clone();
+	let timer_stop = service_stop.clone();
+	thread::spawn(move || {
+		while !timer_stop.load(Ordering::SeqCst) {
+			thread::park_timeout(::std::time::Duration::from_millis(2000));
+			// It almost always not doing anything, only greets new peers with a job
+			timer_svc.maintain();
+		}
+	});
+
+	let control_service = Arc::new(StratumControlService::default());
+	let as_control = control_service.clone() as Arc<ControlService>;
+	let mut worker = nanoipc::Worker::<ControlService>::new(&as_control);
+	worker.add_reqrep(
+		&service_urls::with_base(&service_config.io_path, service_urls::STRATUM_CONTROL)
+	).unwrap();
+
+	while !control_service.stop.load(Ordering::SeqCst) {
+		worker.poll();
 	}
+	service_stop.store(true, Ordering::SeqCst);
+
+	hypervisor.module_shutdown(MODULE_ID);
+	trace!(target: "hypervisor", "Stratum process terminated gracefully");
 }