// Copyright 2015, 2016 Ethcore (UK) Ltd. // This file is part of Parity. // Parity is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // Parity is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU General Public License for more details. // You should have received a copy of the GNU General Public License // along with Parity. If not, see . //! Creates and registers client and network services. use util::*; use io::*; use spec::Spec; use error::*; use client::{Client, ClientConfig, ChainNotify}; use miner::Miner; use snapshot::ManifestData; use snapshot::service::Service as SnapshotService; use std::sync::atomic::AtomicBool; #[cfg(feature="ipc")] use nanoipc; #[cfg(feature="ipc")] use client::BlockChainClient; /// Message type for external and internal events #[derive(Clone)] pub enum ClientIoMessage { /// Best Block Hash in chain has been changed NewChainHead, /// A block is ready BlockVerified, /// New transaction RLPs are ready to be imported NewTransactions(Vec), /// Begin snapshot restoration BeginRestoration(ManifestData), /// Feed a state chunk to the snapshot service FeedStateChunk(H256, Bytes), /// Feed a block chunk to the snapshot service FeedBlockChunk(H256, Bytes), /// Take a snapshot for the block with given number. TakeSnapshot(u64), } /// Client service setup. Creates and registers client and network services with the IO subsystem. pub struct ClientService { io_service: Arc>, client: Arc, snapshot: Arc, panic_handler: Arc, _stop_guard: ::devtools::StopGuard, } impl ClientService { /// Start the service in a separate thread. pub fn start( config: ClientConfig, spec: &Spec, db_path: &Path, ipc_path: &Path, miner: Arc, ) -> Result { let panic_handler = PanicHandler::new_in_arc(); let io_service = try!(IoService::::start()); panic_handler.forward_from(&io_service); info!("Configured for {} using {} engine", Colour::White.bold().paint(spec.name.clone()), Colour::Yellow.bold().paint(spec.engine.name())); if spec.fork_name.is_some() { warn!("Your chain is an alternative fork. {}", Colour::Red.bold().paint("TRANSACTIONS MAY BE REPLAYED ON THE MAINNET!")); } let pruning = config.pruning; let client = try!(Client::new(config, &spec, db_path, miner, io_service.channel())); let snapshot = try!(SnapshotService::new(spec, pruning, db_path.into(), io_service.channel(), client.clone())); let snapshot = Arc::new(snapshot); panic_handler.forward_from(&*client); let client_io = Arc::new(ClientIoHandler { client: client.clone(), snapshot: snapshot.clone(), }); try!(io_service.register_handler(client_io)); let stop_guard = ::devtools::StopGuard::new(); run_ipc(ipc_path, client.clone(), snapshot.clone(), stop_guard.share()); Ok(ClientService { io_service: Arc::new(io_service), client: client, snapshot: snapshot, panic_handler: panic_handler, _stop_guard: stop_guard, }) } /// Add a node to network pub fn add_node(&mut self, _enode: &str) { unimplemented!(); } /// Get general IO interface pub fn register_io_handler(&self, handler: Arc + Send>) -> Result<(), IoError> { self.io_service.register_handler(handler) } /// Get client interface pub fn client(&self) -> Arc { self.client.clone() } /// Get snapshot interface. pub fn snapshot_service(&self) -> Arc { self.snapshot.clone() } /// Get network service component pub fn io(&self) -> Arc> { self.io_service.clone() } /// Set the actor to be notified on certain chain events pub fn add_notify(&self, notify: Arc) { self.client.add_notify(notify); } } impl MayPanic for ClientService { fn on_panic(&self, closure: F) where F: OnPanicListener { self.panic_handler.on_panic(closure); } } /// IO interface for the Client handler struct ClientIoHandler { client: Arc, snapshot: Arc, } const CLIENT_TICK_TIMER: TimerToken = 0; const SNAPSHOT_TICK_TIMER: TimerToken = 1; const CLIENT_TICK_MS: u64 = 5000; const SNAPSHOT_TICK_MS: u64 = 10000; impl IoHandler for ClientIoHandler { fn initialize(&self, io: &IoContext) { io.register_timer(CLIENT_TICK_TIMER, CLIENT_TICK_MS).expect("Error registering client timer"); io.register_timer(SNAPSHOT_TICK_TIMER, SNAPSHOT_TICK_MS).expect("Error registering snapshot timer"); } fn timeout(&self, _io: &IoContext, timer: TimerToken) { match timer { CLIENT_TICK_TIMER => self.client.tick(), SNAPSHOT_TICK_TIMER => self.snapshot.tick(), _ => warn!("IO service triggered unregistered timer '{}'", timer), } } #[cfg_attr(feature="dev", allow(single_match))] fn message(&self, _io: &IoContext, net_message: &ClientIoMessage) { match *net_message { ClientIoMessage::BlockVerified => { self.client.import_verified_blocks(); } ClientIoMessage::NewTransactions(ref transactions) => { self.client.import_queued_transactions(transactions); } ClientIoMessage::BeginRestoration(ref manifest) => { if let Err(e) = self.snapshot.init_restore(manifest.clone()) { warn!("Failed to initialize snapshot restoration: {}", e); } } ClientIoMessage::FeedStateChunk(ref hash, ref chunk) => self.snapshot.feed_state_chunk(*hash, chunk), ClientIoMessage::FeedBlockChunk(ref hash, ref chunk) => self.snapshot.feed_block_chunk(*hash, chunk), ClientIoMessage::TakeSnapshot(num) => { if let Err(e) = self.snapshot.take_snapshot(&*self.client, num) { warn!("Failed to take snapshot at block #{}: {}", num, e); } } _ => {} // ignore other messages } } } #[cfg(feature="ipc")] fn run_ipc(base_path: &Path, client: Arc, snapshot_service: Arc, stop: Arc) { let mut path = base_path.to_owned(); path.push("parity-chain.ipc"); let socket_addr = format!("ipc://{}", path.to_string_lossy()); let s = stop.clone(); ::std::thread::spawn(move || { let mut worker = nanoipc::Worker::new(&(client as Arc)); worker.add_reqrep(&socket_addr).expect("Ipc expected to initialize with no issues"); while !s.load(::std::sync::atomic::Ordering::Relaxed) { worker.poll(); } }); let mut path = base_path.to_owned(); path.push("parity-snapshot.ipc"); let socket_addr = format!("ipc://{}", path.to_string_lossy()); ::std::thread::spawn(move || { let mut worker = nanoipc::Worker::new(&(snapshot_service as Arc<::snapshot::SnapshotService>)); worker.add_reqrep(&socket_addr).expect("Ipc expected to initialize with no issues"); while !stop.load(::std::sync::atomic::Ordering::Relaxed) { worker.poll(); } }); } #[cfg(not(feature="ipc"))] fn run_ipc(_base_path: &Path, _client: Arc, _snapshot_service: Arc, _stop: Arc) { } #[cfg(test)] mod tests { use super::*; use tests::helpers::*; use devtools::*; use client::ClientConfig; use std::sync::Arc; use miner::Miner; #[test] fn it_can_be_started() { let temp_path = RandomTempPath::new(); let mut path = temp_path.as_path().to_owned(); path.push("pruning"); path.push("db"); let spec = get_test_spec(); let service = ClientService::start( ClientConfig::default(), &spec, &path, &path, Arc::new(Miner::with_spec(&spec)), ); assert!(service.is_ok()); drop(service.unwrap()); ::std::thread::park_timeout(::std::time::Duration::from_millis(100)); } }