v2.5.10 stable (#11239)
* ropsten #6631425 foundation #8798209 (#11201) * [stable] builtin, istanbul and mordor testnet backports (#11234) * ethcore-builtin (#10850) * [builtin]: support `multiple prices and activations` in chain spec (#11039) * [chain specs]: activate `Istanbul` on mainnet (#11228) * ethcore/res: add mordor testnet configuration (#11200) * Update list of bootnodes for xDai chain (#11236) * ethcore: remove `test-helper feat` from build (#11047) * Secret store: fix Instant::now() related race in net_keep_alive (#11155) (#11159) * [stable]: backport #10691 and #10683 (#11143) * Fix compiler warning (that will become an error) (#10683) * Refactor Clique stepping (#10691) * Add Constantinople eips to the dev (instant_seal) config (#10809) * Add cargo-remote dir to .gitignore (?) * Insert explicit warning into the panic hook (#11225) * Fix docker centos build (#11226) * Update MIX bootnodes. (#11203) * Use provided usd-per-eth value if an endpoint is specified (#11209) * Add new line after writing block to hex file. (#10984) * Type annotation for next_key() matching of json filter options (#11192) (but no `FilterOption` in 2.5 so…) * Upgrade jsonrpc to latest (#11206) * [CI] check evmbin build (#11096) * Correct EIP-712 encoding (#11092) * [client]: Fix for incorrectly dropped consensus messages (#11086) * Fix block detail updating (#11015) * Switching sccache from local to Redis (#10971) * Made ecrecover implementation trait public (#11188) * [dependencies]: jsonrpc `14.0.1` (#11183) * [receipt]: add `sender` & `receiver` to `RichReceipts` (#11179) * [ethcore/builtin]: do not panic in blake2pricer on short input (#11180) * util Host: fix a double Read Lock bug in fn Host::session_readable() (#11175) * ethcore client: fix a double Read Lock bug in fn Client::logs() (#11172) * Change how RPCs eth_call and eth_estimateGas handle "Pending" (#11127) * Cleanup stratum a bit (#11161) * Upgrade to jsonrpc v14 (#11151) * SecretStore: expose restore_key_public in HTTP API (#10241)
This commit is contained in:
Talha Cross
s3krit
@@ -76,7 +76,7 @@ impl Stratum {
|
||||
|
||||
let implementation = Arc::new(StratumImpl {
|
||||
subscribers: RwLock::default(),
|
||||
job_que: RwLock::default(),
|
||||
job_queue: RwLock::default(),
|
||||
dispatcher,
|
||||
workers: Arc::new(RwLock::default()),
|
||||
secret,
|
||||
@@ -106,13 +106,9 @@ impl 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)
|
||||
}
|
||||
|
||||
fn push_work(&self, payloads: Vec<String>) -> Result<(), Error> {
|
||||
self.implementation.push_work(payloads, &self.tcp_dispatcher)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Stratum {
|
||||
@@ -126,14 +122,14 @@ struct StratumImpl {
|
||||
/// Subscribed clients
|
||||
subscribers: RwLock<Vec<SocketAddr>>,
|
||||
/// List of workers supposed to receive job update
|
||||
job_que: RwLock<HashSet<SocketAddr>>,
|
||||
job_queue: RwLock<HashSet<SocketAddr>>,
|
||||
/// Payload manager
|
||||
dispatcher: Arc<JobDispatcher>,
|
||||
/// Authorized workers (socket - worker_id)
|
||||
workers: Arc<RwLock<HashMap<SocketAddr, String>>>,
|
||||
/// Secret if any
|
||||
secret: Option<H256>,
|
||||
/// Dispatch notify couinter
|
||||
/// Dispatch notify counter
|
||||
notify_counter: RwLock<u32>,
|
||||
}
|
||||
|
||||
@@ -143,7 +139,7 @@ impl StratumImpl {
|
||||
use std::str::FromStr;
|
||||
|
||||
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());
|
||||
|
||||
Ok(match self.dispatcher.initial() {
|
||||
@@ -160,7 +156,7 @@ impl StratumImpl {
|
||||
|
||||
/// rpc method `mining.authorize`
|
||||
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 {
|
||||
let hash = keccak(secret);
|
||||
if hash != valid_secret {
|
||||
@@ -184,15 +180,15 @@ impl StratumImpl {
|
||||
_ => None
|
||||
})
|
||||
.collect::<Vec<String>>()) {
|
||||
Ok(()) => {
|
||||
self.update_peers(&meta.tcp_dispatcher.expect("tcp_dispatcher is always initialized; qed"));
|
||||
to_value(true)
|
||||
},
|
||||
Err(submit_err) => {
|
||||
warn!("Error while submitting share: {:?}", submit_err);
|
||||
to_value(false)
|
||||
}
|
||||
Ok(()) => {
|
||||
self.update_peers(&meta.tcp_dispatcher.expect("tcp_dispatcher is always initialized; qed"));
|
||||
to_value(true)
|
||||
},
|
||||
Err(submit_err) => {
|
||||
warn!("Error while submitting share: {:?}", submit_err);
|
||||
to_value(false)
|
||||
}
|
||||
}
|
||||
},
|
||||
_ => {
|
||||
trace!(target: "stratum", "Invalid submit work format {:?}", params);
|
||||
@@ -204,36 +200,37 @@ impl StratumImpl {
|
||||
/// Helper method
|
||||
fn update_peers(&self, tcp_dispatcher: &Dispatcher) {
|
||||
if let Some(job) = self.dispatcher.job() {
|
||||
if let Err(e) = self.push_work_all(job, tcp_dispatcher) {
|
||||
warn!("Failed to update some of the peers: {:?}", e);
|
||||
}
|
||||
self.push_work_all(job, tcp_dispatcher)
|
||||
}
|
||||
}
|
||||
|
||||
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 workers = self.workers.read();
|
||||
let next_request_id = {
|
||||
let mut counter = self.notify_counter.write();
|
||||
if *counter == ::std::u32::MAX { *counter = NOTIFY_COUNTER_INITIAL; }
|
||||
else { *counter = *counter + 1 }
|
||||
if *counter == ::std::u32::MAX {
|
||||
*counter = NOTIFY_COUNTER_INITIAL;
|
||||
} else {
|
||||
*counter = *counter + 1
|
||||
}
|
||||
*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);
|
||||
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);
|
||||
match tcp_dispatcher.push_message(addr, workers_msg.clone()) {
|
||||
Err(PushMessageError::NoSuchPeer) => {
|
||||
trace!(target: "stratum", "Worker no longer connected: {}", &addr);
|
||||
hup_peers.insert(*addr.clone());
|
||||
trace!(target: "stratum", "Worker no longer connected: {}", addr);
|
||||
hup_peers.insert(addr.clone());
|
||||
},
|
||||
Err(e) => {
|
||||
warn!(target: "stratum", "Unexpected transport error: {:?}", e);
|
||||
},
|
||||
Ok(_) => { },
|
||||
Ok(_) => {},
|
||||
}
|
||||
}
|
||||
hup_peers
|
||||
@@ -241,33 +238,10 @@ impl StratumImpl {
|
||||
|
||||
if !hup_peers.is_empty() {
|
||||
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))
|
||||
.and_then(move |stream| {
|
||||
trace!(target: "stratum", "Pusing work to peers");
|
||||
stratum.push_work_all(r#"{ "00040008", "100500" }"#.to_owned())
|
||||
.expect("Pushing work should produce no errors");
|
||||
stratum.push_work_all(r#"{ "00040008", "100500" }"#.to_owned());
|
||||
Timeout::new(future::ok(stream), ::std::time::Duration::from_millis(100))
|
||||
})
|
||||
.map_err(|err: timeout::Error<()>| panic!("Timeout: {:?}", err))
|
||||
@@ -497,4 +470,11 @@ mod tests {
|
||||
"{ \"id\": 17, \"method\": \"mining.notify\", \"params\": { \"00040008\", \"100500\" } }\n",
|
||||
response);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jsonprc_server_is_send_and_sync() {
|
||||
fn is_send_and_sync<T: Send + Sync>() {}
|
||||
|
||||
is_send_and_sync::<JsonRpcServer>();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,10 +55,7 @@ pub trait JobDispatcher: Send + Sync {
|
||||
/// Interface that can handle requests to push job for workers
|
||||
pub trait PushWorkHandler: Send + Sync {
|
||||
/// 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>;
|
||||
|
||||
/// push the work packages worker-wise (`payload`: json of pow-specific set of work specification)
|
||||
fn push_work(&self, payloads: Vec<String>) -> Result<(), Error>;
|
||||
fn push_work_all(&self, payload: String);
|
||||
}
|
||||
|
||||
pub struct ServiceConfiguration {
|
||||
|
||||
Reference in New Issue
Block a user