blockgen/miner/
stratum.rs

1// Copyright 2019-2020 Conflux Foundation. All rights reserved.
2// Conflux is free software and distributed under GNU General Public License.
3// See http://www.gnu.org/licenses/
4
5// Copyright 2015-2019 Parity Technologies (UK) Ltd.
6// This file is part of Parity Ethereum.
7
8// Parity Ethereum is free software: you can redistribute it and/or modify
9// it under the terms of the GNU General Public License as published by
10// the Free Software Foundation, either version 3 of the License, or
11// (at your option) any later version.
12
13// Parity Ethereum is distributed in the hope that it will be useful,
14// but WITHOUT ANY WARRANTY; without even the implied warranty of
15// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16// GNU General Public License for more details.
17
18// You should have received a copy of the GNU General Public License
19// along with Parity Ethereum.  If not, see <http://www.gnu.org/licenses/>.
20
21// Copyright 2019 Conflux Foundation. All rights reserved.
22// Conflux is free software and distributed under GNU General Public License.
23// See http://www.gnu.org/licenses/
24
25//! Client-side stratum job dispatcher and mining notifier handler
26
27use crate::{BlockGenerator, SolutionReceiver};
28
29use super::MineWorker;
30use cfx_stratum::{
31    Error as StratumServiceError, JobDispatcher, PushWorkHandler,
32    Stratum as StratumService,
33};
34use cfx_types::{H256, U256};
35use cfxcore::pow::{PowComputer, ProofOfWorkProblem, ProofOfWorkSolution};
36use log::{info, trace, warn};
37use parking_lot::Mutex;
38use std::{
39    collections::HashSet,
40    fmt,
41    net::{AddrParseError, SocketAddr},
42    sync::{mpsc, Arc},
43};
44
45/// Configures stratum server options.
46#[derive(Debug, PartialEq, Clone)]
47pub struct Options {
48    /// Network address
49    pub listen_addr: String,
50    /// Port
51    pub port: u16,
52    /// Secret for peers
53    pub secret: Option<H256>,
54}
55
56fn clean_0x(s: &str) -> &str { s.strip_prefix("0x").unwrap_or(s) }
57
58struct SubmitPayload {
59    worker_id: String,
60    nonce: U256,
61    pow_hash: H256,
62}
63
64impl SubmitPayload {
65    fn from_args(payload: Vec<String>) -> Result<Self, PayloadError> {
66        if payload.len() != 4 {
67            return Err(PayloadError::ArgumentsAmountUnexpected(payload.len()));
68        }
69
70        let worker_id = payload[0].clone();
71
72        let nonce = match clean_0x(&payload[2]).parse::<U256>() {
73            Ok(nonce) => nonce,
74            Err(e) => {
75                warn!(target: "stratum", "submit_work ({}): invalid nonce ({:?})", &payload[0], e);
76                return Err(PayloadError::InvalidNonce(payload[0].clone()));
77            }
78        };
79
80        let pow_hash = match clean_0x(&payload[3]).parse::<H256>() {
81            Ok(pow_hash) => pow_hash,
82            Err(e) => {
83                warn!(target: "stratum", "submit_work ({}): invalid hash ({:?})", &payload[1], e);
84                return Err(PayloadError::InvalidPowHash(payload[1].clone()));
85            }
86        };
87
88        Ok(SubmitPayload {
89            worker_id,
90            nonce,
91            pow_hash,
92        })
93    }
94}
95
96#[derive(Debug)]
97#[allow(dead_code)]
98pub enum PayloadError {
99    ArgumentsAmountUnexpected(usize),
100    InvalidNonce(String),
101    InvalidPowHash(String),
102}
103
104impl fmt::Display for PayloadError {
105    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
106        fmt::Debug::fmt(&self, f)
107    }
108}
109
110/// Job dispatcher for stratum service
111pub struct StratumJobDispatcher {
112    recent_problems: Mutex<Vec<(ProofOfWorkProblem, HashSet<U256>)>>,
113    solution_sender: Mutex<mpsc::Sender<ProofOfWorkSolution>>,
114    pow: Arc<PowComputer>,
115    window_size: usize,
116}
117
118impl JobDispatcher for StratumJobDispatcher {
119    fn submit(&self, payload: Vec<String>) -> Result<(), StratumServiceError> {
120        let payload = SubmitPayload::from_args(payload)
121            .map_err(|e| StratumServiceError::Dispatch(e.to_string()))?;
122
123        trace!(
124            target: "stratum",
125            "submit_work: Decoded: nonce={}, pow_hash={}, worker_id={}",
126            payload.nonce,
127            payload.pow_hash,
128            payload.worker_id,
129        );
130
131        let sol = ProofOfWorkSolution {
132            nonce: payload.nonce,
133        };
134        {
135            let mut probs = self.recent_problems.lock();
136            let mut found = false;
137            for (pow_prob, solved_nonce) in probs.iter_mut() {
138                if pow_prob.block_hash == payload.pow_hash {
139                    if solved_nonce.contains(&sol.nonce) {
140                        return Err(StratumServiceError::InvalidSolution(
141                            format!(
142                                "Problem already solved with nonce = {}! worker_id = {}",
143                                sol.nonce, payload.worker_id
144                            ),
145                        ));
146                    } else if self.pow.validate(pow_prob, &sol) {
147                        solved_nonce.insert(sol.nonce);
148                        info!(
149                            "Stratum worker {} mined a block!",
150                            payload.worker_id
151                        );
152                        found = true;
153                    } else {
154                        return Err(StratumServiceError::InvalidSolution(
155                            format!(
156                                "Incorrect Nonce! worker_id = {}!",
157                                payload.worker_id
158                            ),
159                        ));
160                    }
161                }
162            }
163            if !found {
164                return Err(StratumServiceError::InvalidSolution(format!(
165                    "Solution for a stale job! worker_id = {}",
166                    payload.worker_id
167                )));
168            }
169
170            match self.solution_sender.lock().send(sol) {
171                Ok(_) => {}
172                Err(e) => {
173                    warn!("{}", e);
174                }
175            }
176        }
177
178        Ok(())
179    }
180}
181
182impl StratumJobDispatcher {
183    /// New stratum job dispatcher given the miner and client
184    fn new(
185        solution_sender: mpsc::Sender<ProofOfWorkSolution>,
186        pow: Arc<PowComputer>, pow_window_size: usize,
187    ) -> StratumJobDispatcher {
188        StratumJobDispatcher {
189            recent_problems: Mutex::new(vec![]),
190            solution_sender: Mutex::new(solution_sender),
191            pow,
192            window_size: pow_window_size,
193        }
194    }
195
196    fn notify_new_problem(&self, current_problem: &ProofOfWorkProblem) {
197        let mut probs = self.recent_problems.lock();
198        if probs.len() == self.window_size {
199            probs.remove(0);
200        }
201        probs.push((*current_problem, HashSet::new()));
202    }
203
204    /// Serializes payload for stratum service
205    fn payload(
206        &self, block_height: u64, pow_hash: H256, boundary: U256,
207    ) -> String {
208        // Now we just fill the job_id as pow_hash. This will be more consistent
209        // with the convention.
210        format!(
211            r#"["0x{:x}", "{}", "0x{:x}","0x{:x}"]"#,
212            pow_hash, block_height, pow_hash, boundary
213        )
214    }
215}
216
217/// Wrapper for dedicated stratum service
218pub struct Stratum {
219    dispatcher: Arc<StratumJobDispatcher>,
220    service: Arc<StratumService>,
221}
222
223#[derive(Debug)]
224/// Stratum error
225pub enum Error {
226    #[allow(unused)]
227    /// IPC sockets error
228    Service(StratumServiceError),
229    #[allow(unused)]
230    /// Invalid network address
231    Address(AddrParseError),
232}
233
234impl From<StratumServiceError> for Error {
235    fn from(service_err: StratumServiceError) -> Error {
236        Error::Service(service_err)
237    }
238}
239
240impl From<AddrParseError> for Error {
241    fn from(err: AddrParseError) -> Error { Error::Address(err) }
242}
243
244impl MineWorker for Stratum {
245    fn receive_problem(&self, prob: ProofOfWorkProblem) {
246        trace!(target: "stratum", "Notify work");
247
248        self.dispatcher.notify_new_problem(&prob);
249        self.service.push_work_all(
250            self.dispatcher.payload(prob.block_height, prob.block_hash, prob.boundary)
251        ).unwrap_or_else(
252            |e| warn!(target: "stratum", "Error while pushing work: {:?}", e)
253        );
254    }
255}
256
257impl Stratum {
258    pub fn spawn(bg: &BlockGenerator) -> (Self, SolutionReceiver) {
259        let (solution_sender, solution_receiver) = mpsc::channel();
260        let cfg = Options {
261            listen_addr: bg.pow_config.stratum_listen_addr.clone(),
262            port: bg.pow_config.stratum_port,
263            secret: bg.pow_config.stratum_secret,
264        };
265        let stratum = Stratum::start(
266            &cfg,
267            bg.pow.clone(),
268            bg.pow_config.pow_problem_window_size,
269            solution_sender,
270        )
271        .expect("Failed to start Stratum service.");
272
273        (stratum, solution_receiver)
274    }
275
276    /// New stratum job dispatcher, given the miner, client and dedicated
277    /// stratum service
278    pub fn start(
279        options: &Options, pow: Arc<PowComputer>, pow_window_size: usize,
280        solution_sender: mpsc::Sender<ProofOfWorkSolution>,
281    ) -> Result<Stratum, Error> {
282        use std::net::IpAddr;
283
284        let dispatcher = Arc::new(StratumJobDispatcher::new(
285            solution_sender,
286            pow,
287            pow_window_size,
288        ));
289
290        let stratum_svc = StratumService::start(
291            &SocketAddr::new(
292                options.listen_addr.parse::<IpAddr>()?,
293                options.port,
294            ),
295            dispatcher.clone(),
296            options.secret,
297        )?;
298
299        Ok(Stratum {
300            dispatcher,
301            service: stratum_svc,
302        })
303    }
304}
blockgen/miner/stratum.rs

blockgen/miner/
stratum.rs
