grassrootseconomics/openethereum
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

finished removing ipc

Browse Source
This commit is contained in:
debris
2017-10-16 18:18:43 +02:00
parent fa019bd03e
commit 16d84f8ecf
54 changed files with 6 additions and 5600 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
parity/boot.rs
View File

@@ -1,119 +0,0 @@
// Copyright 2015-2017 Parity Technologies (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 <http://www.gnu.org/licenses/>.
//! Parity micro-service helpers
use nanoipc;
use ipc;
use std;
use std::sync::Arc;
use hypervisor::HypervisorServiceClient;
use hypervisor::service::IpcModuleId;
use ctrlc::CtrlC;
use std::sync::atomic::{AtomicBool, Ordering};
use nanoipc::{IpcInterface, GuardedSocket, NanoSocket};
use ipc::WithSocket;
use ethcore_logger::{Config as LogConfig, setup_log};
use docopt::Docopt;
#[derive(Debug)]
pub enum BootError {
ReadArgs(std::io::Error),
DecodeArgs(ipc::binary::BinaryError),
DependencyConnect(nanoipc::SocketError),
}
pub fn host_service<T: ?Sized + Send + Sync + 'static>(addr: &str, stop_guard: Arc<AtomicBool>, service: Arc<T>) where T: IpcInterface {
let socket_url = addr.to_owned();
std::thread::spawn(move || {
let mut worker = nanoipc::Worker::<T>::new(&service);
worker.add_reqrep(&socket_url).unwrap();
while !stop_guard.load(Ordering::SeqCst) {
worker.poll();
}
});
}
pub fn payload<B: ipc::BinaryConvertable>() -> Result<B, BootError> {
use std::io;
use std::io::Read;
let mut buffer = Vec::new();
io::stdin().read_to_end(&mut buffer).map_err(BootError::ReadArgs)?;
ipc::binary::deserialize::<B>(&buffer).map_err(BootError::DecodeArgs)
}
pub fn register(hv_url: &str, control_url: &str, module_id: IpcModuleId) -> GuardedSocket<HypervisorServiceClient<NanoSocket>>{
let hypervisor_client = nanoipc::fast_client::<HypervisorServiceClient<_>>(hv_url).unwrap();
hypervisor_client.handshake().unwrap();
hypervisor_client.module_ready(module_id, control_url.to_owned());
hypervisor_client
}
pub fn dependency<C: WithSocket<NanoSocket>>(url: &str)
-> Result<GuardedSocket<C>, BootError>
{
nanoipc::generic_client::<C>(url).map_err(BootError::DependencyConnect)
}
pub fn main_thread() -> Arc<AtomicBool> {
let stop = Arc::new(AtomicBool::new(false));
let ctrc_stop = stop.clone();
CtrlC::set_handler(move || {
ctrc_stop.store(true, Ordering::Relaxed);
});
stop
}
pub fn setup_cli_logger(svc_name: &str) {
let usage = format!("
Ethcore {} service
Usage:
parity {} [options]
Options:
-l --logging LOGGING Specify the logging level. Must conform to the same
format as RUST_LOG.
--log-file FILENAME Specify a filename into which logging should be
directed.
--no-color Don't use terminal color codes in output.
", svc_name, svc_name);
#[derive(Debug, RustcDecodable)]
struct Args {
flag_logging: Option<String>,
flag_log_file: Option<String>,
flag_no_color: bool,
}
impl Args {
pub fn log_settings(&self) -> LogConfig {
LogConfig {
color: self.flag_no_color || cfg!(windows),
mode: self.flag_logging.clone(),
file: self.flag_log_file.clone(),
}
}
}
let args: Args = Docopt::new(usage)
.and_then(|d| d.decode())
.unwrap_or_else(|e| e.exit());
setup_log(&args.log_settings()).expect("Log initialization failure");
}

14
parity/main.rs
View File

@@ -52,9 +52,6 @@ extern crate toml;
extern crate ethcore;
extern crate ethcore_devtools as devtools;
extern crate ethcore_io as io;
extern crate ethcore_ipc as ipc;
extern crate ethcore_ipc_hypervisor as hypervisor;
extern crate ethcore_ipc_nano as nanoipc;
extern crate ethcore_light as light;
extern crate ethcore_logger;
extern crate ethcore_util as util;
@@ -125,11 +122,6 @@ mod url;
mod user_defaults;
mod whisper;
#[cfg(feature="ipc")]
mod boot;
#[cfg(feature="ipc")]
mod sync;
#[cfg(feature="stratum")]
mod stratum;
@@ -203,14 +195,8 @@ fn stratum_main(alt_mains: &mut HashMap<String, fn()>) {
alt_mains.insert("stratum".to_owned(), stratum::main);
}
#[cfg(not(feature="ipc"))]
fn sync_main(_: &mut HashMap<String, fn()>) {}
#[cfg(feature="ipc")]
fn sync_main(alt_mains: &mut HashMap<String, fn()>) {
alt_mains.insert("sync".to_owned(), sync::main);
}
fn updates_path(name: &str) -> PathBuf {
let mut dest = PathBuf::from(default_hypervisor_path());
dest.push(name);

152
parity/modules.rs
View File

@@ -15,167 +15,19 @@
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::sync::Arc;
use std::path::Path;
use ethcore::client::BlockChainClient;
use hypervisor::Hypervisor;
use ethsync::{AttachedProtocol, SyncConfig, NetworkConfiguration, NetworkError, Params, ConnectionFilter};
use ethcore::snapshot::SnapshotService;
use light::Provider;
#[cfg(not(feature="ipc"))]
use self::no_ipc_deps::*;
#[cfg(not(feature="ipc"))]
pub use ethsync::{EthSync, SyncProvider, ManageNetwork};
pub use ethcore::client::ChainNotify;
use ethcore_logger::Config as LogConfig;
#[cfg(feature="ipc")]
use self::ipc_deps::*;
#[cfg(feature="ipc")]
pub mod service_urls {
use std::path::PathBuf;
pub const CLIENT: &'static str = "parity-chain.ipc";
pub const SNAPSHOT: &'static str = "parity-snapshot.ipc";
pub const SYNC: &'static str = "parity-sync.ipc";
pub const SYNC_NOTIFY: &'static str = "parity-sync-notify.ipc";
pub const NETWORK_MANAGER: &'static str = "parity-manage-net.ipc";
pub const SYNC_CONTROL: &'static str = "parity-sync-control.ipc";
pub const LIGHT_PROVIDER: &'static str = "parity-light-provider.ipc";
#[cfg(feature="stratum")]
pub const STRATUM_CONTROL: &'static str = "parity-stratum-control.ipc";
pub fn with_base(data_dir: &str, service_path: &str) -> String {
let mut path = PathBuf::from(data_dir);
path.push(service_path);
format!("ipc://{}", path.to_str().unwrap())
}
}
#[cfg(not(feature="ipc"))]
mod no_ipc_deps {
pub use ethsync::{EthSync, SyncProvider, ManageNetwork};
pub use ethcore::client::ChainNotify;
}
#[cfg(feature="ipc")]
pub type SyncModules = (
GuardedSocket<SyncClient<NanoSocket>>,
GuardedSocket<NetworkManagerClient<NanoSocket>>,
GuardedSocket<ChainNotifyClient<NanoSocket>>
);
#[cfg(not(feature="ipc"))]
pub type SyncModules = (Arc<SyncProvider>, Arc<ManageNetwork>, Arc<ChainNotify>);
#[cfg(feature="ipc")]
mod ipc_deps {
pub use ethsync::remote::{SyncClient, NetworkManagerClient};
pub use ethsync::ServiceConfiguration;
pub use ethcore::client::remote::ChainNotifyClient;
pub use hypervisor::{SYNC_MODULE_ID, BootArgs, HYPERVISOR_IPC_URL};
pub use nanoipc::{GuardedSocket, NanoSocket, generic_client, fast_client};
pub use ipc::IpcSocket;
pub use ipc::binary::serialize;
pub use light::remote::LightProviderClient;
}
#[cfg(feature="ipc")]
pub fn hypervisor(base_path: &Path) -> Option<Hypervisor> {
Some(Hypervisor
::with_url(&service_urls::with_base(base_path.to_str().unwrap(), HYPERVISOR_IPC_URL))
.io_path(base_path.to_str().unwrap()))
}
#[cfg(not(feature="ipc"))]
pub fn hypervisor(_: &Path) -> Option<Hypervisor> {
None
}
#[cfg(feature="ipc")]
fn sync_arguments(io_path: &str, sync_cfg: SyncConfig, net_cfg: NetworkConfiguration, log_settings: &LogConfig) -> BootArgs {
let service_config = ServiceConfiguration {
sync: sync_cfg,
net: net_cfg,
io_path: io_path.to_owned(),
};
// initialisation payload is passed via stdin
let service_payload = serialize(&service_config).expect("Any binary-derived struct is serializable by definition");
// client service url and logging settings are passed in command line
let mut cli_args = Vec::new();
cli_args.push("sync".to_owned());
if !log_settings.color { cli_args.push("--no-color".to_owned()); }
if let Some(ref mode) = log_settings.mode {
cli_args.push("-l".to_owned());
cli_args.push(mode.to_owned());
}
if let Some(ref file) = log_settings.file {
cli_args.push("--log-file".to_owned());
cli_args.push(file.to_owned());
}
BootArgs::new().stdin(service_payload).cli(cli_args)
}
#[cfg(feature="ipc")]
pub fn stratum(
hypervisor_ref: &mut Option<Hypervisor>,
config: &::ethcore::miner::StratumOptions
) {
use ethcore_stratum;
let mut hypervisor = hypervisor_ref.take().expect("There should be hypervisor for ipc configuration");
let args = BootArgs::new().stdin(
serialize(&ethcore_stratum::ServiceConfiguration {
io_path: hypervisor.io_path.to_owned(),
port: config.port,
listen_addr: config.listen_addr.to_owned(),
secret: config.secret,
}).expect("Any binary-derived struct is serializable by definition")
).cli(vec!["stratum".to_owned()]);
hypervisor = hypervisor.module(super::stratum::MODULE_ID, args);
*hypervisor_ref = Some(hypervisor);
}
#[cfg(feature="ipc")]
pub fn sync(
hypervisor_ref: &mut Option<Hypervisor>,
sync_cfg: SyncConfig,
net_cfg: NetworkConfiguration,
_client: Arc<BlockChainClient>,
_snapshot_service: Arc<SnapshotService>,
_provider: Arc<Provider>,
log_settings: &LogConfig,
_attached_protos: Vec<AttachedProtocol>,
) -> Result<SyncModules, NetworkError> {
let mut hypervisor = hypervisor_ref.take().expect("There should be hypervisor for ipc configuration");
let args = sync_arguments(&hypervisor.io_path, sync_cfg, net_cfg, log_settings);
hypervisor = hypervisor.module(SYNC_MODULE_ID, args);
hypervisor.start();
hypervisor.wait_for_startup();
let sync_client = generic_client::<SyncClient<_>>(
&service_urls::with_base(&hypervisor.io_path, service_urls::SYNC)).unwrap();
let notify_client = generic_client::<ChainNotifyClient<_>>(
&service_urls::with_base(&hypervisor.io_path, service_urls::SYNC_NOTIFY)).unwrap();
let manage_client = generic_client::<NetworkManagerClient<_>>(
&service_urls::with_base(&hypervisor.io_path, service_urls::NETWORK_MANAGER)).unwrap();
let provider_client = generic_client::<LightProviderClient<_>>(
&service_urls::with_base(&hypervisor.io_path, service_urls::LIGHT_PROVIDER)).unwrap();
*hypervisor_ref = Some(hypervisor);
Ok((sync_client, manage_client, notify_client))
}
#[cfg(not(feature="ipc"))]
pub fn sync(
_hypervisor: &mut Option<Hypervisor>,
sync_cfg: SyncConfig,
net_cfg: NetworkConfiguration,
client: Arc<BlockChainClient>,

8
parity/run.rs
View File

@@ -579,9 +579,6 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
// set network path.
net_conf.net_config_path = Some(db_dirs.network_path().to_string_lossy().into_owned());
// create supervisor
let mut hypervisor = modules::hypervisor(&cmd.dirs.ipc_path());
// create client service.
let service = ClientService::start(
client_config,
@@ -661,7 +658,6 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
// create sync object
let (sync_provider, manage_network, chain_notify) = modules::sync(
&mut hypervisor,
sync_config,
net_conf.clone().into(),
client.clone(),
@@ -868,10 +864,6 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
// just Arc is dropping here, to allow other reference release in its default time
drop(informant);
// hypervisor should be shutdown first while everything still works and can be
// terminated gracefully
drop(hypervisor);
Ok(restart)
}

103
parity/sync.rs
View File

@@ -1,103 +0,0 @@
// Copyright 2015-2017 Parity Technologies (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 <http://www.gnu.org/licenses/>.
//! Parity sync service
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use hypervisor::{SYNC_MODULE_ID, HYPERVISOR_IPC_URL, ControlService};
use ethcore::client::ChainNotify;
use ethcore::client::remote::RemoteClient;
use ethcore::snapshot::remote::RemoteSnapshotService;
use light::remote::LightProviderClient;
use ethsync::{SyncProvider, EthSync, ManageNetwork, ServiceConfiguration};
use modules::service_urls;
use boot;
use nanoipc;
#[derive(Default)]
struct SyncControlService {
pub stop: Arc<AtomicBool>,
}
impl ControlService for SyncControlService {
fn shutdown(&self) -> bool {
trace!(target: "hypervisor", "Received shutdown from control service");
self.stop.store(true, ::std::sync::atomic::Ordering::SeqCst);
true
}
}
pub fn main() {
boot::setup_cli_logger("sync");
let service_config: ServiceConfiguration = boot::payload()
.unwrap_or_else(|e| panic!("Fatal: error reading boot arguments ({:?})", e));
let remote_client = dependency!(RemoteClient, &service_urls::with_base(&service_config.io_path, service_urls::CLIENT));
let remote_snapshot = dependency!(RemoteSnapshotService, &service_urls::with_base(&service_config.io_path, service_urls::SNAPSHOT));
let remote_provider = dependency!(LightProviderClient, &service_urls::with_base(&service_config.io_path, service_urls::LIGHT_PROVIDER));
let sync = EthSync::new(Params {
config: service_config.sync,
chain: remote_client.service().clone(),
snapshot_service: remote_snapshot.service().clone(),
provider: remote_provider.service().clone(),
network_config: service_config.net
attached_protos: Vec::new(),
}).unwrap();
let _ = boot::main_thread();
let service_stop = Arc::new(AtomicBool::new(false));
let hypervisor = boot::register(
&service_urls::with_base(&service_config.io_path, HYPERVISOR_IPC_URL),
&service_urls::with_base(&service_config.io_path, service_urls::SYNC_CONTROL),
SYNC_MODULE_ID
);
boot::host_service(
&service_urls::with_base(&service_config.io_path, service_urls::SYNC),
service_stop.clone(),
sync.clone() as Arc<SyncProvider>
);
boot::host_service(
&service_urls::with_base(&service_config.io_path, service_urls::NETWORK_MANAGER),
service_stop.clone(),
sync.clone() as Arc<ManageNetwork>
);
boot::host_service(
&service_urls::with_base(&service_config.io_path, service_urls::SYNC_NOTIFY),
service_stop.clone(),
sync.clone() as Arc<ChainNotify>
);
let control_service = Arc::new(SyncControlService::default());
let as_control = control_service.clone() as Arc<ControlService>;
let mut worker = nanoipc::Worker::<ControlService>::new(&as_control);
let thread_stop = control_service.stop.clone();
worker.add_reqrep(
&service_urls::with_base(&service_config.io_path, service_urls::SYNC_CONTROL)
).unwrap();
while !thread_stop.load(::std::sync::atomic::Ordering::SeqCst) {
worker.poll();
}
service_stop.store(true, ::std::sync::atomic::Ordering::SeqCst);
hypervisor.module_shutdown(SYNC_MODULE_ID);
trace!(target: "hypervisor", "Sync process terminated gracefully");
}