GRIN/grin-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

slog-rs logging (#171)

Browse Source 
* added global slog instance, changed all logging macro formats to include logger instance
* adding configuration to logging, allowing for multiple log outputs
* updates to test, changes to build docs
* rustfmt
* moving logging functions into util crate
This commit is contained in:
Yeastplume
2017-10-12 17:56:44 +01:00
committed by Ignotus Peverell
parent b85006ebe5
commit 8e382a7593
62 changed files with 973 additions and 1006 deletions
Download Patch File Download Diff File Expand all files Collapse all files
grin/src/adapters.rs
+53 -41
View File
@@ -27,6 +27,7 @@ use secp::pedersen::Commitment;
use util::OneTime;
use store;
use sync;
use util::LOGGER;
use core::global::{MiningParameterMode, MINING_PARAMETER_MODE};
/// Implementation of the NetAdapter for the blockchain. Gets notified when new
@@ -51,19 +52,23 @@ impl NetAdapter for NetToChainAdapter {
identifier: "?.?.?.?".to_string(),
};
if let Err(e) = self.tx_pool.write().unwrap().add_to_memory_pool(source, tx) {
error!("Transaction rejected: {:?}", e);
error!(LOGGER, "Transaction rejected: {:?}", e);
}
}
fn block_received(&self, b: core::Block) {
let bhash = b.hash();
debug!("Received block {} from network, going to process.", bhash);
debug!(
LOGGER,
"Received block {} from network, going to process.",
bhash
);
// pushing the new block through the chain pipeline
let res = self.chain.process_block(b, self.chain_opts());
if let Err(e) = res {
debug!("Block {} refused by chain: {:?}", bhash, e);
debug!(LOGGER, "Block {} refused by chain: {:?}", bhash, e);
}
if self.syncer.borrow().syncing() {
@@ -82,6 +87,7 @@ impl NetAdapter for NetToChainAdapter {
}
Err(chain::Error::Unfit(s)) => {
info!(
LOGGER,
"Received unfit block header {} at {}: {}.",
bh.hash(),
bh.height,
@@ -89,16 +95,25 @@ impl NetAdapter for NetToChainAdapter {
);
}
Err(chain::Error::StoreErr(e)) => {
error!("Store error processing block header {}: {:?}", bh.hash(), e);
error!(
LOGGER,
"Store error processing block header {}: {:?}",
bh.hash(),
e
);
return;
}
Err(e) => {
info!("Invalid block header {}: {:?}.", bh.hash(), e);
info!(LOGGER, "Invalid block header {}: {:?}.", bh.hash(), e);
// TODO penalize peer somehow
}
}
}
info!("Added {} headers to the header chain.", added_hs.len());
info!(
LOGGER,
"Added {} headers to the header chain.",
added_hs.len()
);
if self.syncer.borrow().syncing() {
self.syncer.borrow().headers_received(added_hs);
@@ -118,7 +133,7 @@ impl NetAdapter for NetToChainAdapter {
return self.locate_headers(locator[1..].to_vec());
}
Err(e) => {
error!("Could not build header locator: {:?}", e);
error!(LOGGER, "Could not build header locator: {:?}", e);
return vec![];
}
};
@@ -132,7 +147,7 @@ impl NetAdapter for NetToChainAdapter {
Ok(head) => headers.push(head),
Err(chain::Error::StoreErr(store::Error::NotFoundErr)) => break,
Err(e) => {
error!("Could not build header locator: {:?}", e);
error!(LOGGER, "Could not build header locator: {:?}", e);
return vec![];
}
}
@@ -152,18 +167,15 @@ impl NetAdapter for NetToChainAdapter {
/// Find good peers we know with the provided capability and return their
/// addresses.
fn find_peer_addrs(&self, capab: p2p::Capabilities) -> Vec<SocketAddr> {
let peers = self.peer_store.find_peers(
State::Healthy,
capab,
p2p::MAX_PEER_ADDRS as usize,
);
debug!("Got {} peer addrs to send.", peers.len());
let peers = self.peer_store
.find_peers(State::Healthy, capab, p2p::MAX_PEER_ADDRS as usize);
debug!(LOGGER, "Got {} peer addrs to send.", peers.len());
map_vec!(peers, |p| p.addr)
}
/// A list of peers has been received from one of our peers.
fn peer_addrs_received(&self, peer_addrs: Vec<SocketAddr>) {
debug!("Received {} peer addrs, saving.", peer_addrs.len());
debug!(LOGGER, "Received {} peer addrs, saving.", peer_addrs.len());
for pa in peer_addrs {
if let Ok(e) = self.peer_store.exists_peer(pa) {
if e {
@@ -177,14 +189,14 @@ impl NetAdapter for NetToChainAdapter {
flags: State::Healthy,
};
if let Err(e) = self.peer_store.save_peer(&peer) {
error!("Could not save received peer address: {:?}", e);
error!(LOGGER, "Could not save received peer address: {:?}", e);
}
}
}
/// Network successfully connected to a peer.
fn peer_connected(&self, pi: &p2p::PeerInfo) {
debug!("Saving newly connected peer {}.", pi.addr);
debug!(LOGGER, "Saving newly connected peer {}.", pi.addr);
let peer = PeerData {
addr: pi.addr,
capabilities: pi.capabilities,
@@ -192,17 +204,16 @@ impl NetAdapter for NetToChainAdapter {
flags: State::Healthy,
};
if let Err(e) = self.peer_store.save_peer(&peer) {
error!("Could not save connected peer: {:?}", e);
error!(LOGGER, "Could not save connected peer: {:?}", e);
}
}
}
impl NetToChainAdapter {
pub fn new(
chain_ref: Arc<chain::Chain>,
tx_pool: Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>,
peer_store: Arc<PeerStore>,
) -> NetToChainAdapter {
pub fn new(chain_ref: Arc<chain::Chain>,
tx_pool: Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>,
peer_store: Arc<PeerStore>)
-> NetToChainAdapter {
NetToChainAdapter {
chain: chain_ref,
peer_store: peer_store,
@@ -216,15 +227,15 @@ impl NetToChainAdapter {
pub fn start_sync(&self, sync: sync::Syncer) {
let arc_sync = Arc::new(sync);
self.syncer.init(arc_sync.clone());
let spawn_result = thread::Builder::new().name("syncer".to_string()).spawn(
move || {
let spawn_result = thread::Builder::new()
.name("syncer".to_string())
.spawn(move || {
let sync_run_result = arc_sync.run();
match sync_run_result {
Ok(_) => {}
Err(_) => {}
}
},
);
});
match spawn_result {
Ok(_) => {}
Err(_) => {}
@@ -261,6 +272,7 @@ impl ChainAdapter for ChainToPoolAndNetAdapter {
{
if let Err(e) = self.tx_pool.write().unwrap().reconcile_block(b) {
error!(
LOGGER,
"Pool could not update itself at block {}: {:?}",
b.hash(),
e
@@ -272,9 +284,8 @@ impl ChainAdapter for ChainToPoolAndNetAdapter {
}
impl ChainToPoolAndNetAdapter {
pub fn new(
tx_pool: Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>,
) -> ChainToPoolAndNetAdapter {
pub fn new(tx_pool: Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>)
-> ChainToPoolAndNetAdapter {
ChainToPoolAndNetAdapter {
tx_pool: tx_pool,
p2p: OneTime::new(),
@@ -306,19 +317,19 @@ impl PoolToChainAdapter {
impl pool::BlockChain for PoolToChainAdapter {
fn get_unspent(&self, output_ref: &Commitment) -> Result<Output, pool::PoolError> {
self.chain.borrow().get_unspent(output_ref).map_err(
|e| match e {
self.chain
.borrow()
.get_unspent(output_ref)
.map_err(|e| match e {
chain::types::Error::OutputNotFound => pool::PoolError::OutputNotFound,
chain::types::Error::OutputSpent => pool::PoolError::OutputSpent,
_ => pool::PoolError::GenericPoolError,
},
)
})
}
fn get_block_header_by_output_commit(
&self,
commit: &Commitment,
) -> Result<BlockHeader, pool::PoolError> {
fn get_block_header_by_output_commit(&self,
commit: &Commitment)
-> Result<BlockHeader, pool::PoolError> {
self.chain
.borrow()
.get_block_header_by_output_commit(commit)
@@ -326,8 +337,9 @@ impl pool::BlockChain for PoolToChainAdapter {
}
fn head_header(&self) -> Result<BlockHeader, pool::PoolError> {
self.chain.borrow().head_header().map_err(|_| {
pool::PoolError::GenericPoolError
})
self.chain
.borrow()
.head_header()
.map_err(|_| pool::PoolError::GenericPoolError)
}
}
grin/src/lib.rs
+1 -2
View File
@@ -22,8 +22,7 @@
#![warn(missing_docs)]
#[macro_use]
extern crate log;
extern crate env_logger;
extern crate slog;
extern crate futures;
extern crate futures_cpupool as cpupool;
extern crate hyper;
grin/src/miner.rs
+80 -76
View File
@@ -35,6 +35,7 @@ use pow::MiningWorker;
use pow::types::MinerConfig;
use core::ser;
use core::ser::AsFixedBytes;
use util::LOGGER;
// use core::genesis;
@@ -81,10 +82,8 @@ impl Default for HeaderPartWriter {
impl HeaderPartWriter {
pub fn parts_as_hex_strings(&self) -> (String, String) {
(
String::from(format!("{:02x}", self.pre_nonce.iter().format(""))),
String::from(format!("{:02x}", self.post_nonce.iter().format(""))),
)
(String::from(format!("{:02x}", self.pre_nonce.iter().format(""))),
String::from(format!("{:02x}", self.post_nonce.iter().format(""))))
}
}
@@ -129,11 +128,10 @@ pub struct Miner {
impl Miner {
/// Creates a new Miner. Needs references to the chain state and its
/// storage.
pub fn new(
config: MinerConfig,
chain_ref: Arc<chain::Chain>,
tx_pool: Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>,
) -> Miner {
pub fn new(config: MinerConfig,
chain_ref: Arc<chain::Chain>,
tx_pool: Arc<RwLock<pool::TransactionPool<PoolToChainAdapter>>>)
-> Miner {
Miner {
config: config,
chain: chain_ref,
@@ -149,18 +147,18 @@ impl Miner {
}
/// Inner part of the mining loop for cuckoo-miner async mode
pub fn inner_loop_async(
&self,
plugin_miner: &mut PluginMiner,
difficulty: Difficulty,
b: &mut Block,
cuckoo_size: u32,
head: &BlockHeader,
latest_hash: &Hash,
attempt_time_per_block: u32,
) -> Option<Proof> {
pub fn inner_loop_async(&self,
plugin_miner: &mut PluginMiner,
difficulty: Difficulty,
b: &mut Block,
cuckoo_size: u32,
head: &BlockHeader,
latest_hash: &Hash,
attempt_time_per_block: u32)
-> Option<Proof> {
debug!(
LOGGER,
"(Server ID: {}) Mining at Cuckoo{} for at most {} secs at height {} and difficulty {}.",
self.debug_output_id,
cuckoo_size,
@@ -201,7 +199,7 @@ impl Miner {
if let Some(s) = job_handle.get_solution() {
sol = Some(Proof::new(s.solution_nonces.to_vec()));
b.header.nonce = s.get_nonce_as_u64();
// debug!("Nonce: {}", b.header.nonce);
// debug!(LOGGER, "Nonce: {}", b.header.nonce);
break;
}
if time::get_time().sec > next_stat_output {
@@ -213,6 +211,7 @@ impl Miner {
let last_solution_time_secs = s.last_solution_time as f64 / 1000.0;
let last_hashes_per_sec = 1.0 / last_solution_time_secs;
debug!(
LOGGER,
"Mining: Plugin {} - Device {} ({}): Last Solution time: {}s; \
Solutions per second: {:.*} - Total Attempts: {}",
i,
@@ -228,7 +227,7 @@ impl Miner {
}
}
}
debug!("Total solutions per second: {}", sps_total);
debug!(LOGGER, "Total solutions per second: {}", sps_total);
next_stat_output = time::get_time().sec + stat_output_interval;
}
}
@@ -238,6 +237,7 @@ impl Miner {
}
if sol == None {
debug!(
LOGGER,
"(Server ID: {}) No solution found after {} seconds, continuing...",
self.debug_output_id,
attempt_time_per_block
@@ -250,22 +250,23 @@ impl Miner {
}
/// The inner part of mining loop for cuckoo miner sync mode
pub fn inner_loop_sync_plugin(
&self,
plugin_miner: &mut PluginMiner,
b: &mut Block,
cuckoo_size: u32,
head: &BlockHeader,
attempt_time_per_block: u32,
latest_hash: &mut Hash,
) -> Option<Proof> {
// look for a pow for at most 2 sec on the same block (to give a chance to new
pub fn inner_loop_sync_plugin(&self,
plugin_miner: &mut PluginMiner,
b: &mut Block,
cuckoo_size: u32,
head: &BlockHeader,
attempt_time_per_block: u32,
latest_hash: &mut Hash)
-> Option<Proof> {
// look for a pow for at most attempt_time_per_block sec on the same block (to
// give a chance to new
// transactions) and as long as the head hasn't changed
let deadline = time::get_time().sec + attempt_time_per_block as i64;
let stat_check_interval = 3;
let mut next_stat_check = time::get_time().sec + stat_check_interval;
debug!(
LOGGER,
"(Server ID: {}) Mining at Cuckoo{} for {} secs (will wait for last solution) \
on block {} at difficulty {}.",
self.debug_output_id,
@@ -278,6 +279,7 @@ impl Miner {
if self.config.slow_down_in_millis != None && self.config.slow_down_in_millis.unwrap() > 0 {
debug!(
LOGGER,
"(Server ID: {}) Artificially slowing down loop by {}ms per iteration.",
self.debug_output_id,
self.config.slow_down_in_millis.unwrap()
@@ -301,7 +303,8 @@ impl Miner {
for s in stats_vec.into_iter() {
let last_solution_time_secs = s.last_solution_time as f64 / 1000.0;
let last_hashes_per_sec = 1.0 / last_solution_time_secs;
println!(
debug!(
LOGGER,
"Plugin 0 - Device {} ({}) - Last Solution time: {}; Solutions per second: {:.*}",
s.device_id,
s.device_name,
@@ -319,8 +322,7 @@ impl Miner {
// Artificial slow down
if self.config.slow_down_in_millis != None &&
self.config.slow_down_in_millis.unwrap() > 0
{
self.config.slow_down_in_millis.unwrap() > 0 {
thread::sleep(std::time::Duration::from_millis(
self.config.slow_down_in_millis.unwrap(),
));
@@ -329,6 +331,7 @@ impl Miner {
if sol == None {
debug!(
LOGGER,
"(Server ID: {}) No solution found after {} iterations, continuing...",
self.debug_output_id,
iter_count
@@ -339,20 +342,20 @@ impl Miner {
}
/// The inner part of mining loop for the internal miner
pub fn inner_loop_sync_internal<T: MiningWorker>(
&self,
miner: &mut T,
b: &mut Block,
cuckoo_size: u32,
head: &BlockHeader,
attempt_time_per_block: u32,
latest_hash: &mut Hash,
) -> Option<Proof> {
pub fn inner_loop_sync_internal<T: MiningWorker>(&self,
miner: &mut T,
b: &mut Block,
cuckoo_size: u32,
head: &BlockHeader,
attempt_time_per_block: u32,
latest_hash: &mut Hash)
-> Option<Proof> {
// look for a pow for at most 2 sec on the same block (to give a chance to new
// transactions) and as long as the head hasn't changed
let deadline = time::get_time().sec + attempt_time_per_block as i64;
debug!(
LOGGER,
"(Server ID: {}) Mining at Cuckoo{} for at most {} secs on block {} at difficulty {}.",
self.debug_output_id,
cuckoo_size,
@@ -364,6 +367,7 @@ impl Miner {
if self.config.slow_down_in_millis != None && self.config.slow_down_in_millis.unwrap() > 0 {
debug!(
LOGGER,
"(Server ID: {}) Artificially slowing down loop by {}ms per iteration.",
self.debug_output_id,
self.config.slow_down_in_millis.unwrap()
@@ -388,8 +392,7 @@ impl Miner {
// Artificial slow down
if self.config.slow_down_in_millis != None &&
self.config.slow_down_in_millis.unwrap() > 0
{
self.config.slow_down_in_millis.unwrap() > 0 {
thread::sleep(std::time::Duration::from_millis(
self.config.slow_down_in_millis.unwrap(),
));
@@ -398,6 +401,7 @@ impl Miner {
if sol == None {
debug!(
LOGGER,
"(Server ID: {}) No solution found after {} iterations, continuing...",
self.debug_output_id,
iter_count
@@ -410,22 +414,18 @@ impl Miner {
/// chain anytime required and looking for PoW solution.
pub fn run_loop(&self, miner_config: MinerConfig, cuckoo_size: u32, proof_size: usize) {
info!("(Server ID: {}) Starting miner loop.", self.debug_output_id);
info!(
LOGGER,
"(Server ID: {}) Starting miner loop.",
self.debug_output_id
);
let mut plugin_miner = None;
let mut miner = None;
if miner_config.use_cuckoo_miner {
plugin_miner = Some(PluginMiner::new(
consensus::EASINESS,
cuckoo_size,
proof_size,
));
plugin_miner = Some(PluginMiner::new(consensus::EASINESS, cuckoo_size, proof_size));
plugin_miner.as_mut().unwrap().init(miner_config.clone());
} else {
miner = Some(cuckoo::Miner::new(
consensus::EASINESS,
cuckoo_size,
proof_size,
));
miner = Some(cuckoo::Miner::new(consensus::EASINESS, cuckoo_size, proof_size));
}
// to prevent the wallet from generating a new HD key derivation for each
@@ -434,7 +434,7 @@ impl Miner {
let mut pubkey = None;
loop {
debug!("in miner loop...");
debug!(LOGGER, "in miner loop...");
// get the latest chain state and build a block on top of it
let head = self.chain.head_header().unwrap();
@@ -484,6 +484,7 @@ impl Miner {
// if we found a solution, push our block out
if let Some(proof) = sol {
info!(
LOGGER,
"(Server ID: {}) Found valid proof of work, adding block {}.",
self.debug_output_id,
b.hash()
@@ -497,15 +498,20 @@ impl Miner {
let res = self.chain.process_block(b, opts);
if let Err(e) = res {
error!(
LOGGER,
"(Server ID: {}) Error validating mined block: {:?}",
self.debug_output_id,
e
);
}
debug!("resetting pubkey in miner to None");
debug!(LOGGER, "resetting pubkey in miner to None");
pubkey = None;
} else {
debug!("setting pubkey in miner to pubkey from block_fees - {:?}", block_fees);
debug!(
LOGGER,
"setting pubkey in miner to pubkey from block_fees - {:?}",
block_fees
);
pubkey = block_fees.pubkey();
}
}
@@ -513,11 +519,10 @@ impl Miner {
/// Builds a new block with the chain head as previous and eligible
/// transactions from the pool.
fn build_block(
&self,
head: &core::BlockHeader,
pubkey: Option<Identifier>,
) -> (core::Block, BlockFees) {
fn build_block(&self,
head: &core::BlockHeader,
pubkey: Option<Identifier>)
-> (core::Block, BlockFees) {
// prepare the block header timestamp
let mut now_sec = time::get_time().sec;
let head_sec = head.timestamp.to_timespec().sec;
@@ -530,9 +535,10 @@ impl Miner {
let difficulty = consensus::next_difficulty(diff_iter).unwrap();
// extract current transaction from the pool
let txs_box = self.tx_pool.read().unwrap().prepare_mineable_transactions(
MAX_TX,
);
let txs_box = self.tx_pool
.read()
.unwrap()
.prepare_mineable_transactions(MAX_TX);
let txs: Vec<&Transaction> = txs_box.iter().map(|tx| tx.as_ref()).collect();
// build the coinbase and the block itself
@@ -543,6 +549,7 @@ impl Miner {
let (output, kernel, block_fees) = self.get_coinbase(block_fees);
let mut b = core::Block::with_reward(head, txs, output, kernel).unwrap();
debug!(
LOGGER,
"(Server ID: {}) Built new block with {} inputs and {} outputs, difficulty: {}",
self.debug_output_id,
b.inputs.len(),
@@ -558,16 +565,13 @@ impl Miner {
b.header.nonce = rng.gen();
b.header.difficulty = difficulty;
b.header.timestamp = time::at_utc(time::Timespec::new(now_sec, 0));
self.chain.set_sumtree_roots(&mut b).expect(
"Error setting sum tree roots",
);
self.chain
.set_sumtree_roots(&mut b)
.expect("Error setting sum tree roots");
(b, block_fees)
}
fn get_coinbase(
&self,
block_fees: BlockFees,
) -> (core::Output, core::TxKernel, BlockFees) {
fn get_coinbase(&self, block_fees: BlockFees) -> (core::Output, core::TxKernel, BlockFees) {
if self.config.burn_reward {
let keychain = Keychain::from_random_seed().unwrap();
let pubkey = keychain.derive_pubkey(1).unwrap();
@@ -598,10 +602,10 @@ impl Miner {
let pubkey = ser::deserialize(&mut &pubkey_bin[..]).unwrap();
let block_fees = BlockFees {
pubkey: Some(pubkey),
.. block_fees
..block_fees
};
debug!("block_fees here: {:?}", block_fees);
debug!(LOGGER, "block_fees here: {:?}", block_fees);
(output, kernel, block_fees)
}
grin/src/seed.rs
+47 -50
View File
@@ -31,6 +31,7 @@ use tokio_core::reactor;
use tokio_timer::Timer;
use p2p;
use util::LOGGER;
const PEER_MAX_COUNT: u32 = 25;
const PEER_PREFERRED_COUNT: u32 = 8;
@@ -44,11 +45,10 @@ pub struct Seeder {
}
impl Seeder {
pub fn new(
capabilities: p2p::Capabilities,
peer_store: Arc<p2p::PeerStore>,
p2p: Arc<p2p::Server>,
) -> Seeder {
pub fn new(capabilities: p2p::Capabilities,
peer_store: Arc<p2p::PeerStore>,
p2p: Arc<p2p::Server>)
-> Seeder {
Seeder {
peer_store: peer_store,
p2p: p2p,
@@ -56,31 +56,27 @@ impl Seeder {
}
}
pub fn connect_and_monitor(
&self,
h: reactor::Handle,
seed_list: Box<Future<Item = Vec<SocketAddr>, Error = String>>,
) {
pub fn connect_and_monitor(&self,
h: reactor::Handle,
seed_list: Box<Future<Item = Vec<SocketAddr>, Error = String>>) {
// open a channel with a listener that connects every peer address sent below
// max peer count
let (tx, rx) = futures::sync::mpsc::unbounded();
h.spawn(self.listen_for_addrs(h.clone(), rx));
// check seeds and start monitoring connections
let seeder = self.connect_to_seeds(tx.clone(), seed_list).join(
self.monitor_peers(tx.clone()),
);
let seeder = self.connect_to_seeds(tx.clone(), seed_list)
.join(self.monitor_peers(tx.clone()));
h.spawn(seeder.map(|_| ()).map_err(|e| {
error!("Seeding or peer monitoring error: {}", e);
error!(LOGGER, "Seeding or peer monitoring error: {}", e);
()
}));
}
fn monitor_peers(
&self,
tx: mpsc::UnboundedSender<SocketAddr>,
) -> Box<Future<Item = (), Error = String>> {
fn monitor_peers(&self,
tx: mpsc::UnboundedSender<SocketAddr>)
-> Box<Future<Item = (), Error = String>> {
let peer_store = self.peer_store.clone();
let p2p_server = self.p2p.clone();
@@ -95,9 +91,9 @@ impl Seeder {
let disconnected = p2p_server.clean_peers();
for p in disconnected {
if p.is_banned() {
debug!("Marking peer {} as banned.", p.info.addr);
let update_result =
peer_store.update_state(p.info.addr, p2p::State::Banned);
debug!(LOGGER, "Marking peer {} as banned.", p.info.addr);
let update_result = peer_store
.update_state(p.info.addr, p2p::State::Banned);
match update_result {
Ok(()) => {}
Err(_) => {}
@@ -114,7 +110,11 @@ impl Seeder {
);
peers.retain(|p| !p2p_server.is_known(p.addr));
if peers.len() > 0 {
debug!("Got {} more peers from db, trying to connect.", peers.len());
debug!(
LOGGER,
"Got {} more peers from db, trying to connect.",
peers.len()
);
thread_rng().shuffle(&mut peers[..]);
let sz = min(PEER_PREFERRED_COUNT as usize, peers.len());
for p in &peers[0..sz] {
@@ -131,11 +131,10 @@ impl Seeder {
// Check if we have any pre-existing peer in db. If so, start with those,
// otherwise use the seeds provided.
fn connect_to_seeds(
&self,
tx: mpsc::UnboundedSender<SocketAddr>,
seed_list: Box<Future<Item = Vec<SocketAddr>, Error = String>>,
) -> Box<Future<Item = (), Error = String>> {
fn connect_to_seeds(&self,
tx: mpsc::UnboundedSender<SocketAddr>,
seed_list: Box<Future<Item = Vec<SocketAddr>, Error = String>>)
-> Box<Future<Item = (), Error = String>> {
let peer_store = self.peer_store.clone();
// a thread pool is required so we don't block the event loop with a
@@ -164,11 +163,11 @@ impl Seeder {
// connect to this first set of addresses
let sz = min(PEER_PREFERRED_COUNT as usize, peer_addrs.len());
for addr in &peer_addrs[0..sz] {
debug!("Connecting to seed: {}.", addr);
debug!(LOGGER, "Connecting to seed: {}.", addr);
tx.unbounded_send(*addr).unwrap();
}
if peer_addrs.len() == 0 {
warn!("No seeds were retrieved.");
warn!(LOGGER, "No seeds were retrieved.");
}
Ok(())
});
@@ -179,17 +178,16 @@ impl Seeder {
/// addresses to and initiate a connection if the max peer count isn't
/// exceeded. A request for more peers is also automatically sent after
/// connection.
fn listen_for_addrs(
&self,
h: reactor::Handle,
rx: mpsc::UnboundedReceiver<SocketAddr>,
) -> Box<Future<Item = (), Error = ()>> {
fn listen_for_addrs(&self,
h: reactor::Handle,
rx: mpsc::UnboundedReceiver<SocketAddr>)
-> Box<Future<Item = (), Error = ()>> {
let capab = self.capabilities;
let p2p_store = self.peer_store.clone();
let p2p_server = self.p2p.clone();
let listener = rx.for_each(move |peer_addr| {
debug!("New peer address to connect to: {}.", peer_addr);
debug!(LOGGER, "New peer address to connect to: {}.", peer_addr);
let inner_h = h.clone();
if p2p_server.peer_count() < PEER_MAX_COUNT {
connect_and_req(
@@ -226,8 +224,10 @@ pub fn web_seeds(h: reactor::Handle) -> Box<Future<Item = Vec<SocketAddr>, Error
})
.and_then(|res| {
// collect all chunks and split around whitespace to get a list of SocketAddr
res.body().collect().map_err(|e| e.to_string()).and_then(
|chunks| {
res.body()
.collect()
.map_err(|e| e.to_string())
.and_then(|chunks| {
let res = chunks.iter().fold("".to_string(), |acc, ref chunk| {
acc + str::from_utf8(&chunk[..]).unwrap()
});
@@ -235,8 +235,7 @@ pub fn web_seeds(h: reactor::Handle) -> Box<Future<Item = Vec<SocketAddr>, Error
.map(|s| s.parse().unwrap())
.collect::<Vec<_>>();
Ok(addrs)
},
)
})
})
});
Box::new(seeds)
@@ -244,9 +243,8 @@ pub fn web_seeds(h: reactor::Handle) -> Box<Future<Item = Vec<SocketAddr>, Error
/// Convenience function when the seed list is immediately known. Mostly used
/// for tests.
pub fn predefined_seeds(
addrs_str: Vec<String>,
) -> Box<Future<Item = Vec<SocketAddr>, Error = String>> {
pub fn predefined_seeds(addrs_str: Vec<String>)
-> Box<Future<Item = Vec<SocketAddr>, Error = String>> {
let seeds = future::ok(()).and_then(move |_| {
Ok(
addrs_str
@@ -258,13 +256,12 @@ pub fn predefined_seeds(
Box::new(seeds)
}
fn connect_and_req(
capab: p2p::Capabilities,
peer_store: Arc<p2p::PeerStore>,
p2p: Arc<p2p::Server>,
h: reactor::Handle,
addr: SocketAddr,
) -> Box<Future<Item = (), Error = ()>> {
fn connect_and_req(capab: p2p::Capabilities,
peer_store: Arc<p2p::PeerStore>,
p2p: Arc<p2p::Server>,
h: reactor::Handle,
addr: SocketAddr)
-> Box<Future<Item = (), Error = ()>> {
let fut = p2p.connect_peer(addr, h).then(move |p| {
match p {
Ok(Some(p)) => {
@@ -275,7 +272,7 @@ fn connect_and_req(
}
}
Err(e) => {
error!("Peer request error: {:?}", e);
error!(LOGGER, "Peer request error: {:?}", e);
let update_result = peer_store.update_state(addr, p2p::State::Defunct);
match update_result {
Ok(()) => {}
grin/src/server.rs
+9 -9
View File
@@ -35,6 +35,7 @@ use seed;
use sync;
use types::*;
use pow;
use util::LOGGER;
use core::global;
@@ -66,7 +67,7 @@ impl Server {
let forever = Timer::default()
.interval(time::Duration::from_secs(60))
.for_each(move |_| {
debug!("event loop running");
debug!(LOGGER, "event loop running");
Ok(())
})
.map_err(|_| ());
@@ -79,9 +80,10 @@ impl Server {
pub fn future(mut config: ServerConfig, evt_handle: &reactor::Handle) -> Result<Server, Error> {
let pool_adapter = Arc::new(PoolToChainAdapter::new());
let tx_pool = Arc::new(RwLock::new(
pool::TransactionPool::new(config.pool_config.clone(), pool_adapter.clone()),
));
let tx_pool = Arc::new(RwLock::new(pool::TransactionPool::new(
config.pool_config.clone(),
pool_adapter.clone(),
)));
let chain_adapter = Arc::new(ChainToPoolAndNetAdapter::new(tx_pool.clone()));
@@ -131,7 +133,7 @@ impl Server {
evt_handle.spawn(p2p_server.start(evt_handle.clone()).map_err(|_| ()));
info!("Starting rest apis at: {}", &config.api_http_addr);
info!(LOGGER, "Starting rest apis at: {}", &config.api_http_addr);
api::start_rest_apis(
config.api_http_addr.clone(),
@@ -139,7 +141,7 @@ impl Server {
tx_pool.clone(),
);
warn!("Grin server started.");
warn!(LOGGER, "Grin server started.");
Ok(Server {
config: config,
evt_handle: evt_handle.clone(),
@@ -174,9 +176,7 @@ impl Server {
let mut miner = miner::Miner::new(config.clone(), self.chain.clone(), self.tx_pool.clone());
miner.set_debug_output_id(format!("Port {}", self.config.p2p_config.unwrap().port));
thread::spawn(move || {
miner.run_loop(config.clone(), cuckoo_size as u32, proof_size);
});
thread::spawn(move || { miner.run_loop(config.clone(), cuckoo_size as u32, proof_size); });
}
/// The chain head
grin/src/sync.rs
+8 -4
View File
@@ -28,6 +28,7 @@ use core::core::hash::{Hash, Hashed};
use chain;
use p2p;
use types::Error;
use util::LOGGER;
pub struct Syncer {
chain: Arc<chain::Chain>,
@@ -58,7 +59,7 @@ impl Syncer {
/// Checks the local chain state, comparing it with our peers and triggers
/// syncing if required.
pub fn run(&self) -> Result<(), Error> {
debug!("Starting syncer.");
debug!(LOGGER, "Starting syncer.");
let start = Instant::now();
loop {
let pc = self.p2p.peer_count();
@@ -76,7 +77,7 @@ impl Syncer {
// main syncing loop, requests more headers and bodies periodically as long
// as a peer with higher difficulty exists and we're not fully caught up
info!("Starting sync loop.");
info!(LOGGER, "Starting sync loop.");
loop {
let tip = self.chain.get_header_head()?;
// TODO do something better (like trying to get more) if we lose peers
@@ -107,7 +108,7 @@ impl Syncer {
thread::sleep(Duration::from_secs(2));
}
info!("Sync done.");
info!(LOGGER, "Sync done.");
Ok(())
}
@@ -130,6 +131,7 @@ impl Syncer {
}
debug!(
LOGGER,
"Added {} full block hashes to download.",
blocks_to_download.len()
);
@@ -162,6 +164,7 @@ impl Syncer {
blocks_downloading.push((h, Instant::now()));
}
debug!(
LOGGER,
"Requesting more full block hashes to download, total: {}.",
blocks_to_download.len()
);
@@ -187,6 +190,7 @@ impl Syncer {
let locator = self.get_locator(&tip)?;
if let Some(p) = peer {
debug!(
LOGGER,
"Asking peer {} for more block headers starting from {} at {}.",
p.info.addr,
tip.last_block_h,
@@ -194,7 +198,7 @@ impl Syncer {
);
p.send_header_request(locator)?;
} else {
warn!("Could not get most worked peer to request headers.");
warn!(LOGGER, "Could not get most worked peer to request headers.");
}
Ok(())
}