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

Merge branch 'master' into webapps-mio

Browse Source 
Conflicts:
	Cargo.lock
	parity/main.rs
This commit is contained in:
Tomasz Drwięga
2016-04-15 17:11:11 +02:00
parent 0b589915b9 59f98a2a20
commit 13c25c5d49
41 changed files with 1466 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

158
parity/hypervisor/mod.rs Normal file
View File

@@ -0,0 +1,158 @@
// 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 <http://www.gnu.org/licenses/>.
//! Parity interprocess hypervisor module
// while not included in binary
#![allow(dead_code)]
pub mod service;
/// Default value for hypervisor ipc listener
pub const HYPERVISOR_IPC_URL: &'static str = "ipc:///tmp/parity-internal-hyper-status.ipc";
use nanoipc;
use std::sync::{Arc,RwLock};
use hypervisor::service::*;
use std::process::{Command,Child};
use std::collections::HashMap;
type BinaryId = &'static str;
const BLOCKCHAIN_DB_BINARY: BinaryId = "blockchain";
pub struct Hypervisor {
ipc_addr: String,
service: Arc<HypervisorService>,
ipc_worker: RwLock<nanoipc::Worker<HypervisorService>>,
processes: RwLock<HashMap<BinaryId, Child>>,
}
impl Hypervisor {
/// initializes the Hypervisor service with the open ipc socket for incoming clients
pub fn new() -> Hypervisor {
Hypervisor::with_url(HYPERVISOR_IPC_URL)
}
/// Starts on the specified address for ipc listener
fn with_url(addr: &str) -> Hypervisor{
Hypervisor::with_url_and_service(addr, HypervisorService::new())
}
/// Starts with the specified address for the ipc listener and
/// the specified list of modules in form of created service
fn with_url_and_service(addr: &str, service: Arc<HypervisorService>) -> Hypervisor {
let worker = nanoipc::Worker::new(&service);
Hypervisor{
ipc_addr: addr.to_owned(),
service: service,
ipc_worker: RwLock::new(worker),
processes: RwLock::new(HashMap::new()),
}
}
/// Since one binary can host multiple modules
/// we match binaries
fn match_module(module_id: &IpcModuleId) -> Option<BinaryId> {
match *module_id {
BLOCKCHAIN_MODULE_ID => Some(BLOCKCHAIN_DB_BINARY),
// none means the module is inside the main binary
_ => None
}
}
/// Creates IPC listener and starts all binaries
fn start(&self) {
let mut worker = self.ipc_worker.write().unwrap();
worker.add_reqrep(&self.ipc_addr).unwrap_or_else(|e| panic!("Hypervisor ipc worker can not start - critical! ({:?})", e));
for module_id in self.service.module_ids() {
self.start_module(module_id);
}
}
/// Start binary for the specified module
/// Does nothing when it is already started on module is inside the
/// main binary
fn start_module(&self, module_id: IpcModuleId) {
Self::match_module(&module_id).map(|binary_id| {
let mut processes = self.processes.write().unwrap();
{
if processes.get(binary_id).is_some() {
// already started for another module
return;
}
}
let child = Command::new(binary_id).spawn().unwrap_or_else(
|e| panic!("Hypervisor cannot start binary: {}", e));
processes.insert(binary_id, child);
});
}
/// Reports if all modules are checked in
pub fn modules_ready(&self) -> bool {
self.service.unchecked_count() == 0
}
/// Waits for every required module to check in
pub fn wait_for_startup(&self) {
let mut worker = self.ipc_worker.write().unwrap();
while !self.modules_ready() {
worker.poll()
}
}
}
mod tests {
use super::*;
use std::sync::atomic::{AtomicBool,Ordering};
use std::sync::Arc;
use super::service::*;
use nanoipc;
#[test]
fn can_init() {
let url = "ipc:///tmp/test-parity-hypervisor-10.ipc";
let test_module_id = 8080u64;
let hypervisor = Hypervisor::with_url_and_service(url, HypervisorService::with_modules(vec![test_module_id]));
assert_eq!(false, hypervisor.modules_ready());
}
#[test]
fn can_wait_for_startup() {
let url = "ipc:///tmp/test-parity-hypervisor-20.ipc";
let test_module_id = 8080u64;
let hypervisor_ready = Arc::new(AtomicBool::new(false));
let hypervisor_ready_local = hypervisor_ready.clone();
::std::thread::spawn(move || {
while !hypervisor_ready.load(Ordering::Relaxed) { }
let client = nanoipc::init_client::<HypervisorServiceClient<_>>(url).unwrap();
client.handshake().unwrap();
client.module_ready(test_module_id);
});
let hypervisor = Hypervisor::with_url_and_service(url, HypervisorService::with_modules(vec![test_module_id]));
hypervisor.start();
hypervisor_ready_local.store(true, Ordering::Relaxed);
hypervisor.wait_for_startup();
assert_eq!(true, hypervisor.modules_ready());
}
}

19
parity/hypervisor/service.rs Normal file
View File

@@ -0,0 +1,19 @@
// 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 <http://www.gnu.org/licenses/>.
//! Parity interprocess hypervisor IPC service
include!(concat!(env!("OUT_DIR"), "/hypervisor_service_cg.rs"));

69
parity/hypervisor/service.rs.in Normal file
View File

@@ -0,0 +1,69 @@
// 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 <http://www.gnu.org/licenses/>.
use std::sync::{RwLock,Arc};
use std::ops::*;
use ipc::IpcConfig;
use std::collections::HashMap;
pub type IpcModuleId = u64;
/// Blockhain database module id
pub const BLOCKCHAIN_MODULE_ID: IpcModuleId = 2000;
/// IPC service that handles module management
pub struct HypervisorService {
check_list: RwLock<HashMap<IpcModuleId, bool>>,
}
#[derive(Ipc)]
impl HypervisorService {
fn module_ready(&self, module_id: u64) -> bool {
let mut check_list = self.check_list.write().unwrap();
check_list.get_mut(&module_id).map(|mut status| *status = true);
check_list.iter().any(|(_, status)| !status)
}
}
impl HypervisorService {
/// New service with the default list of modules
pub fn new() -> Arc<HypervisorService> {
HypervisorService::with_modules(vec![])
}
/// New service with list of modules that will report for being ready
pub fn with_modules(module_ids: Vec<IpcModuleId>) -> Arc<HypervisorService> {
let mut check_list = HashMap::new();
for module_id in module_ids {
check_list.insert(module_id, false);
}
Arc::new(HypervisorService {
check_list: RwLock::new(check_list),
})
}
/// Number of modules still being waited for check-in
pub fn unchecked_count(&self) -> usize {
self.check_list.read().unwrap().iter().filter(|&(_, status)| !status).count()
}
/// List of all modules within this service
pub fn module_ids(&self) -> Vec<IpcModuleId> {
self.check_list.read().unwrap().iter().map(|(module_id, _)| module_id).cloned().collect()
}
}
impl ::ipc::IpcConfig for HypervisorService {}

47
parity/main.rs
View File

@@ -37,6 +37,10 @@ extern crate time;
extern crate number_prefix;
extern crate rpassword;
extern crate semver;
extern crate ethcore_ipc as ipc;
extern crate ethcore_ipc_nano as nanoipc;
extern crate serde;
extern crate bincode;
// for price_info.rs
#[macro_use] extern crate hyper;
@@ -73,6 +77,7 @@ use webapp::Server as WebappServer;
mod price_info;
mod upgrade;
mod hypervisor;
fn die_with_message(msg: &str) -> ! {
println!("ERROR: {}", msg);
@@ -133,8 +138,7 @@ API and Console Options:
--jsonrpc-interface IP Specify the hostname portion of the JSONRPC API
server, IP should be an interface's IP address, or
all (all interfaces) or local [default: local].
--jsonrpc-cors URL Specify CORS header for JSON-RPC API responses
[default: null].
--jsonrpc-cors URL Specify CORS header for JSON-RPC API responses.
--jsonrpc-apis APIS Specify the APIs available through the JSONRPC
interface. APIS is a comma-delimited list of API
name. Possible name are web3, eth and net.
@@ -242,7 +246,7 @@ struct Args {
flag_jsonrpc: bool,
flag_jsonrpc_interface: String,
flag_jsonrpc_port: u16,
flag_jsonrpc_cors: String,
flag_jsonrpc_cors: Option<String>,
flag_jsonrpc_apis: String,
flag_webapp: bool,
flag_webapp_port: u16,
@@ -307,7 +311,7 @@ fn setup_rpc_server(
secret_store: Arc<AccountService>,
miner: Arc<Miner>,
url: &SocketAddr,
cors_domain: &str,
cors_domain: Option<String>,
apis: Vec<&str>,
) -> RpcServer {
use rpc::v1::*;
@@ -380,7 +384,7 @@ fn setup_rpc_server(
_secret_store: Arc<AccountService>,
_miner: Arc<Miner>,
_url: &SocketAddr,
_cors_domain: &str,
_cors_domain: Option<String>,
_apis: Vec<&str>,
) -> ! {
die!("Your Parity version has been compiled without JSON-RPC support.")
@@ -550,6 +554,7 @@ impl Configuration {
let jdb_types = [journaldb::Algorithm::Archive, journaldb::Algorithm::EarlyMerge, journaldb::Algorithm::OverlayRecent, journaldb::Algorithm::RefCounted];
for i in jdb_types.into_iter() {
let db = journaldb::new(&append_path(&get_db_path(&Path::new(&self.path()), *i, spec.genesis_header().hash()), "state"), *i);
trace!(target: "parity", "Looking for best DB: {} at {:?}", i, db.latest_era());
match (latest_era, db.latest_era()) {
(Some(best), Some(this)) if best >= this => {}
(_, None) => {}
@@ -582,7 +587,7 @@ impl Configuration {
"auto" => self.find_best_db(spec).unwrap_or(journaldb::Algorithm::OverlayRecent),
_ => { die!("Invalid pruning method given."); }
};
info!("Using state DB of {}", client_config.pruning);
trace!(target: "parity", "Using pruning strategy of {}", client_config.pruning);
client_config.name = self.args.flag_identity.clone();
client_config.queue.max_mem_use = self.args.flag_queue_max_size;
client_config
@@ -601,6 +606,18 @@ impl Configuration {
print_version();
return;
}
match ::upgrade::upgrade(Some(&self.path())) {
Ok(upgrades_applied) => {
if upgrades_applied > 0 {
println!("Executed {} upgrade scripts - ok", upgrades_applied);
}
},
Err(e) => {
die!("Error upgrading parity data: {:?}", e);
}
}
if self.args.cmd_daemon {
Daemonize::new()
.pid_file(self.args.arg_pid_file.clone())
@@ -621,9 +638,9 @@ impl Configuration {
let mut secret_store = SecretStore::new_in(Path::new(&self.keys_path()));
if self.args.cmd_new {
println!("Please note that password is NOT RECOVERABLE.");
println!("Type password: ");
print!("Type password: ");
let password = read_password().unwrap();
println!("Repeat password: ");
print!("Repeat password: ");
let password_repeat = read_password().unwrap();
if password != password_repeat {
println!("Passwords do not match!");
@@ -712,7 +729,7 @@ impl Configuration {
self.args.flag_rpcport.unwrap_or(self.args.flag_jsonrpc_port)
);
let addr = SocketAddr::from_str(&url).unwrap_or_else(|_| die!("{}: Invalid JSONRPC listen host/port given.", url));
let cors_domain = self.args.flag_rpccorsdomain.as_ref().unwrap_or(&self.args.flag_jsonrpc_cors);
let cors_domain = self.args.flag_jsonrpc_cors.clone().or(self.args.flag_rpccorsdomain.clone());
Some(setup_rpc_server(
service.client(),
@@ -720,7 +737,7 @@ impl Configuration {
account_service.clone(),
miner.clone(),
&addr,
&cors_domain,
cors_domain,
apis.split(',').collect()
))
} else {
@@ -815,16 +832,6 @@ fn die_with_io_error(e: std::io::Error) -> ! {
}
fn main() {
match ::upgrade::upgrade() {
Ok(upgrades_applied) => {
if upgrades_applied > 0 {
println!("Executed {} upgrade scripts - ok", upgrades_applied);
}
},
Err(e) => {
die!("Error upgrading parity data: {:?}", e);
}
}
Configuration::parse().execute();
}

25
parity/upgrade.rs
View File

@@ -18,14 +18,15 @@
use semver::Version;
use std::collections::*;
use std::fs::File;
use std::fs::{File, create_dir_all};
use std::env;
use std::io::{Read, Write};
#[cfg_attr(feature="dev", allow(enum_variant_names))]
#[derive(Debug)]
pub enum Error {
CannotLockVersionFile,
CannotCreateConfigPath,
CannotWriteVersionFile,
CannotUpdateVersionFile,
}
@@ -66,7 +67,7 @@ fn dummy_upgrade() -> Result<(), Error> {
Ok(())
}
fn push_updrades(upgrades: &mut UpgradeList)
fn push_upgrades(upgrades: &mut UpgradeList)
{
// dummy upgrade (remove when the first one is in)
upgrades.insert(
@@ -76,7 +77,7 @@ fn push_updrades(upgrades: &mut UpgradeList)
fn upgrade_from_version(previous_version: &Version) -> Result<usize, Error> {
let mut upgrades = HashMap::new();
push_updrades(&mut upgrades);
push_upgrades(&mut upgrades);
let current_version = Version::parse(CURRENT_VERSION).unwrap();
@@ -91,11 +92,15 @@ fn upgrade_from_version(previous_version: &Version) -> Result<usize, Error> {
Ok(count)
}
fn with_locked_version<F>(script: F) -> Result<usize, Error>
fn with_locked_version<F>(db_path: Option<&str>, script: F) -> Result<usize, Error>
where F: Fn(&Version) -> Result<usize, Error>
{
let mut path = env::home_dir().expect("Applications should have a home dir");
path.push(".parity");
let mut path = db_path.map_or({
let mut path = env::home_dir().expect("Applications should have a home dir");
path.push(".parity");
path
}, |s| ::std::path::PathBuf::from(s));
try!(create_dir_all(&path).map_err(|_| Error::CannotCreateConfigPath));
path.push("ver.lock");
let version =
@@ -108,7 +113,7 @@ fn with_locked_version<F>(script: F) -> Result<usize, Error>
})
.unwrap_or_else(|| Version::parse("0.9.0").unwrap());
let mut lock = try!(File::create(&path).map_err(|_| Error::CannotLockVersionFile));
let mut lock = try!(File::create(&path).map_err(|_| Error::CannotWriteVersionFile));
let result = script(&version);
let written_version = Version::parse(CURRENT_VERSION).unwrap();
@@ -116,8 +121,8 @@ fn with_locked_version<F>(script: F) -> Result<usize, Error>
result
}
pub fn upgrade() -> Result<usize, Error> {
with_locked_version(|ver| {
pub fn upgrade(db_path: Option<&str>) -> Result<usize, Error> {
with_locked_version(db_path, |ver| {
upgrade_from_version(ver)
})
}