Cleanup stratum a bit (#11161)

* Cleanup stratum a bit

Salvage some code from https://github.com/paritytech/parity-ethereum/pull/10884 + some cleanup and typos.

* HashSet::new does not allocate before first insert

* Remove unused method push_work()
This commit is contained in:
David 2019-10-11 11:52:09 +02:00 committed by GitHub
parent 4f25d43516
commit f3015ce0c6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 33 additions and 68 deletions

View File

@ -225,8 +225,6 @@ impl NotifyWork for Stratum {
self.service.push_work_all( self.service.push_work_all(
self.dispatcher.payload(pow_hash, difficulty, number) self.dispatcher.payload(pow_hash, difficulty, number)
).unwrap_or_else(
|e| warn!(target: "stratum", "Error while pushing work: {:?}", e)
); );
} }
} }
@ -239,16 +237,13 @@ impl Stratum {
let dispatcher = Arc::new(StratumJobDispatcher::new(miner, client)); let dispatcher = Arc::new(StratumJobDispatcher::new(miner, client));
let stratum_svc = StratumService::start( let service = StratumService::start(
&SocketAddr::new(options.listen_addr.parse::<IpAddr>()?, options.port), &SocketAddr::new(options.listen_addr.parse::<IpAddr>()?, options.port),
dispatcher.clone(), dispatcher.clone(),
options.secret.clone(), options.secret.clone(),
)?; )?;
Ok(Stratum { Ok(Stratum { dispatcher, service })
dispatcher: dispatcher,
service: stratum_svc,
})
} }
/// Start STRATUM job dispatcher and register it in the miner /// Start STRATUM job dispatcher and register it in the miner

View File

@ -76,7 +76,7 @@ impl Stratum {
let implementation = Arc::new(StratumImpl { let implementation = Arc::new(StratumImpl {
subscribers: RwLock::default(), subscribers: RwLock::default(),
job_que: RwLock::default(), job_queue: RwLock::default(),
dispatcher, dispatcher,
workers: Arc::new(RwLock::default()), workers: Arc::new(RwLock::default()),
secret, secret,
@ -106,13 +106,9 @@ impl Stratum {
} }
impl PushWorkHandler for Stratum { impl PushWorkHandler for Stratum {
fn push_work_all(&self, payload: String) -> Result<(), Error> { fn push_work_all(&self, payload: String) {
self.implementation.push_work_all(payload, &self.tcp_dispatcher) self.implementation.push_work_all(payload, &self.tcp_dispatcher)
} }
fn push_work(&self, payloads: Vec<String>) -> Result<(), Error> {
self.implementation.push_work(payloads, &self.tcp_dispatcher)
}
} }
impl Drop for Stratum { impl Drop for Stratum {
@ -126,14 +122,14 @@ struct StratumImpl {
/// Subscribed clients /// Subscribed clients
subscribers: RwLock<Vec<SocketAddr>>, subscribers: RwLock<Vec<SocketAddr>>,
/// List of workers supposed to receive job update /// List of workers supposed to receive job update
job_que: RwLock<HashSet<SocketAddr>>, job_queue: RwLock<HashSet<SocketAddr>>,
/// Payload manager /// Payload manager
dispatcher: Arc<dyn JobDispatcher>, dispatcher: Arc<dyn JobDispatcher>,
/// Authorized workers (socket - worker_id) /// Authorized workers (socket - worker_id)
workers: Arc<RwLock<HashMap<SocketAddr, String>>>, workers: Arc<RwLock<HashMap<SocketAddr, String>>>,
/// Secret if any /// Secret if any
secret: Option<H256>, secret: Option<H256>,
/// Dispatch notify couinter /// Dispatch notify counter
notify_counter: RwLock<u32>, notify_counter: RwLock<u32>,
} }
@ -143,7 +139,7 @@ impl StratumImpl {
use std::str::FromStr; use std::str::FromStr;
self.subscribers.write().push(meta.addr().clone()); self.subscribers.write().push(meta.addr().clone());
self.job_que.write().insert(meta.addr().clone()); self.job_queue.write().insert(meta.addr().clone());
trace!(target: "stratum", "Subscription request from {:?}", meta.addr()); trace!(target: "stratum", "Subscription request from {:?}", meta.addr());
Ok(match self.dispatcher.initial() { Ok(match self.dispatcher.initial() {
@ -160,7 +156,7 @@ impl StratumImpl {
/// rpc method `mining.authorize` /// rpc method `mining.authorize`
fn authorize(&self, params: Params, meta: SocketMetadata) -> RpcResult { fn authorize(&self, params: Params, meta: SocketMetadata) -> RpcResult {
params.parse::<(String, String)>().map(|(worker_id, secret)|{ params.parse::<(String, String)>().map(|(worker_id, secret)| {
if let Some(valid_secret) = self.secret { if let Some(valid_secret) = self.secret {
let hash = keccak(secret); let hash = keccak(secret);
if hash != valid_secret { if hash != valid_secret {
@ -184,15 +180,15 @@ impl StratumImpl {
_ => None _ => None
}) })
.collect::<Vec<String>>()) { .collect::<Vec<String>>()) {
Ok(()) => { Ok(()) => {
self.update_peers(&meta.tcp_dispatcher.expect("tcp_dispatcher is always initialized; qed")); self.update_peers(&meta.tcp_dispatcher.expect("tcp_dispatcher is always initialized; qed"));
to_value(true) to_value(true)
}, },
Err(submit_err) => { Err(submit_err) => {
warn!("Error while submitting share: {:?}", submit_err); warn!("Error while submitting share: {:?}", submit_err);
to_value(false) to_value(false)
}
} }
}
}, },
_ => { _ => {
trace!(target: "stratum", "Invalid submit work format {:?}", params); trace!(target: "stratum", "Invalid submit work format {:?}", params);
@ -204,36 +200,37 @@ impl StratumImpl {
/// Helper method /// Helper method
fn update_peers(&self, tcp_dispatcher: &Dispatcher) { fn update_peers(&self, tcp_dispatcher: &Dispatcher) {
if let Some(job) = self.dispatcher.job() { if let Some(job) = self.dispatcher.job() {
if let Err(e) = self.push_work_all(job, tcp_dispatcher) { self.push_work_all(job, tcp_dispatcher)
warn!("Failed to update some of the peers: {:?}", e);
}
} }
} }
fn push_work_all(&self, payload: String, tcp_dispatcher: &Dispatcher) -> Result<(), Error> { fn push_work_all(&self, payload: String, tcp_dispatcher: &Dispatcher) {
let hup_peers = { let hup_peers = {
let workers = self.workers.read(); let workers = self.workers.read();
let next_request_id = { let next_request_id = {
let mut counter = self.notify_counter.write(); let mut counter = self.notify_counter.write();
if *counter == ::std::u32::MAX { *counter = NOTIFY_COUNTER_INITIAL; } if *counter == ::std::u32::MAX {
else { *counter = *counter + 1 } *counter = NOTIFY_COUNTER_INITIAL;
} else {
*counter = *counter + 1
}
*counter *counter
}; };
let mut hup_peers = HashSet::with_capacity(0); // most of the cases won't be needed, hence avoid allocation let mut hup_peers = HashSet::new();
let workers_msg = format!("{{ \"id\": {}, \"method\": \"mining.notify\", \"params\": {} }}", next_request_id, payload); let workers_msg = format!("{{ \"id\": {}, \"method\": \"mining.notify\", \"params\": {} }}", next_request_id, payload);
trace!(target: "stratum", "pushing work for {} workers (payload: '{}')", workers.len(), &workers_msg); trace!(target: "stratum", "pushing work for {} workers (payload: '{}')", workers.len(), &workers_msg);
for (ref addr, _) in workers.iter() { for (addr, _) in workers.iter() {
trace!(target: "stratum", "pusing work to {}", addr); trace!(target: "stratum", "pusing work to {}", addr);
match tcp_dispatcher.push_message(addr, workers_msg.clone()) { match tcp_dispatcher.push_message(addr, workers_msg.clone()) {
Err(PushMessageError::NoSuchPeer) => { Err(PushMessageError::NoSuchPeer) => {
trace!(target: "stratum", "Worker no longer connected: {}", &addr); trace!(target: "stratum", "Worker no longer connected: {}", addr);
hup_peers.insert(*addr.clone()); hup_peers.insert(addr.clone());
}, },
Err(e) => { Err(e) => {
warn!(target: "stratum", "Unexpected transport error: {:?}", e); warn!(target: "stratum", "Unexpected transport error: {:?}", e);
}, },
Ok(_) => { }, Ok(_) => {},
} }
} }
hup_peers hup_peers
@ -241,33 +238,10 @@ impl StratumImpl {
if !hup_peers.is_empty() { if !hup_peers.is_empty() {
let mut workers = self.workers.write(); let mut workers = self.workers.write();
for hup_peer in hup_peers { workers.remove(&hup_peer); } for hup_peer in hup_peers {
workers.remove(&hup_peer);
}
} }
Ok(())
}
fn push_work(&self, payloads: Vec<String>, tcp_dispatcher: &Dispatcher) -> Result<(), Error> {
if !payloads.len() > 0 {
return Err(Error::NoWork);
}
let workers = self.workers.read();
let addrs = workers.keys().collect::<Vec<&SocketAddr>>();
if !workers.len() > 0 {
return Err(Error::NoWorkers);
}
let mut que = payloads;
let mut addr_index = 0;
while que.len() > 0 {
let next_worker = addrs[addr_index];
let mut next_payload = que.drain(0..1);
tcp_dispatcher.push_message(
next_worker,
next_payload.nth(0).expect("drained successfully of 0..1, so 0-th element should exist")
)?;
addr_index = addr_index + 1;
}
Ok(())
} }
} }
@ -475,8 +449,7 @@ mod tests {
.map_err(|err: timeout::Error<()>| panic!("Timeout: {:?}", err)) .map_err(|err: timeout::Error<()>| panic!("Timeout: {:?}", err))
.and_then(move |stream| { .and_then(move |stream| {
trace!(target: "stratum", "Pusing work to peers"); trace!(target: "stratum", "Pusing work to peers");
stratum.push_work_all(r#"{ "00040008", "100500" }"#.to_owned()) stratum.push_work_all(r#"{ "00040008", "100500" }"#.to_owned());
.expect("Pushing work should produce no errors");
Timeout::new(future::ok(stream), ::std::time::Duration::from_millis(100)) Timeout::new(future::ok(stream), ::std::time::Duration::from_millis(100))
}) })
.map_err(|err: timeout::Error<()>| panic!("Timeout: {:?}", err)) .map_err(|err: timeout::Error<()>| panic!("Timeout: {:?}", err))

View File

@ -55,10 +55,7 @@ pub trait JobDispatcher: Send + Sync {
/// Interface that can handle requests to push job for workers /// Interface that can handle requests to push job for workers
pub trait PushWorkHandler: Send + Sync { pub trait PushWorkHandler: Send + Sync {
/// push the same work package for all workers (`payload`: json of pow-specific set of work specification) /// push the same work package for all workers (`payload`: json of pow-specific set of work specification)
fn push_work_all(&self, payload: String) -> Result<(), Error>; fn push_work_all(&self, payload: String);
/// push the work packages worker-wise (`payload`: json of pow-specific set of work specification)
fn push_work(&self, payloads: Vec<String>) -> Result<(), Error>;
} }
pub struct ServiceConfiguration { pub struct ServiceConfiguration {