Refactor into system service, add exe redirect.

This commit is contained in:
Gav Wood
2016-12-10 23:58:39 +01:00
parent 316c25d9f8
commit fa30dfd4b9
18 changed files with 392 additions and 383 deletions

View File

@@ -32,7 +32,6 @@ use ethcore::verification::queue::VerifierSettings;
use cache::CacheConfig;
use informant::{Informant, MillisecondDuration};
use params::{SpecType, Pruning, Switch, tracing_switch_to_bool, fatdb_switch_to_bool};
use io_handler::ImportIoHandler;
use helpers::{to_client_config, execute_upgrades};
use dir::Directories;
use user_defaults::UserDefaults;
@@ -231,11 +230,8 @@ fn execute_import(cmd: ImportBlockchain) -> Result<String, String> {
}
};
let informant = Informant::new(client.clone(), None, None, None, cmd.with_color);
try!(service.register_io_handler(Arc::new(ImportIoHandler {
info: Arc::new(informant),
})).map_err(|_| "Unable to register informant handler".to_owned()));
let informant = Arc::new(Informant::new(client.clone(), None, None, None, cmd.with_color));
service.register_io_handler(informant).map_err(|_| "Unable to register informant handler".to_owned())?;
let do_import = |bytes| {
while client.queue_info().is_full() { sleep(Duration::from_secs(1)); }
@@ -251,7 +247,6 @@ fn execute_import(cmd: ImportBlockchain) -> Result<String, String> {
Ok(())
};
match format {
DataFormat::Binary => {
loop {

View File

@@ -42,6 +42,8 @@ Operating Options:
--no-consensus Force the binary to run even if there are known
issues regarding consensus. Not recommended.
(default: {flag_no_consensus}).
--force-direct Run the originally installed version of Parity,
ignoring any updates that have since been installed.
--chain CHAIN Specify the blockchain type. CHAIN may be either a
JSON chain specification file or olympic, frontier,
homestead, mainnet, morden, ropsten, classic, expanse,

View File

@@ -209,7 +209,6 @@ pub fn default_network_config() -> ::ethsync::NetworkConfiguration {
#[cfg_attr(feature = "dev", allow(too_many_arguments))]
pub fn to_client_config(
cache_config: &CacheConfig,
update_policy: UpdatePolicy,
mode: Mode,
tracing: bool,
fat_db: bool,
@@ -243,7 +242,6 @@ pub fn to_client_config(
// in bytes
client_config.jump_table_size = cache_config.jump_tables() as usize * mb;
client_config.update_policy = update_policy;
client_config.mode = mode;
client_config.tracing.enabled = tracing;
client_config.fat_db = fat_db;

View File

@@ -222,3 +222,14 @@ impl ChainNotify for Informant {
}
}
impl IoHandler<ClientIoMessage> for Informant {
fn initialize(&self, io: &IoContext<ClientIoMessage>) {
io.register_timer(INFO_TIMER, 5000).expect("Error registering timer");
}
fn timeout(&self, _io: &IoContext<ClientIoMessage>, timer: TimerToken) {
if timer == INFO_TIMER && !self.shutdown.load(Ordering::SeqCst) {
self.info.tick();
}
}
}

View File

@@ -1,64 +0,0 @@
// 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::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use ethcore::client::Client;
use ethcore::service::ClientIoMessage;
use ethsync::{SyncProvider, ManageNetwork};
use ethcore::account_provider::AccountProvider;
use io::{TimerToken, IoHandler, IoContext};
use informant::Informant;
const INFO_TIMER: TimerToken = 0;
pub struct ClientIoHandler {
pub client: Arc<Client>,
pub sync: Arc<SyncProvider>,
pub net: Arc<ManageNetwork>,
pub accounts: Arc<AccountProvider>,
pub info: Arc<Informant>,
pub shutdown: Arc<AtomicBool>
}
impl IoHandler<ClientIoMessage> for ClientIoHandler {
fn initialize(&self, io: &IoContext<ClientIoMessage>) {
io.register_timer(INFO_TIMER, 5000).expect("Error registering timer");
}
fn timeout(&self, _io: &IoContext<ClientIoMessage>, timer: TimerToken) {
if timer == INFO_TIMER && !self.shutdown.load(Ordering::SeqCst) {
self.info.tick();
}
}
}
pub struct ImportIoHandler {
pub info: Arc<Informant>,
}
impl IoHandler<ClientIoMessage> for ImportIoHandler {
fn initialize(&self, io: &IoContext<ClientIoMessage>) {
io.register_timer(INFO_TIMER, 5000).expect("Error registering timer");
}
fn timeout(&self, _io: &IoContext<ClientIoMessage>, timer: TimerToken) {
if let INFO_TIMER = timer {
self.info.tick()
}
}
}

View File

@@ -109,13 +109,18 @@ mod sync;
#[cfg(feature="ipc")]
mod boot;
mod user_defaults;
mod updater;
mod operations;
mod fetch;
#[cfg(feature="stratum")]
mod stratum;
use std::{process, env};
use std::collections::HashMap;
use std::io::{self as stdio, BufReader, Write};
use std::fs::File;
use std::path::PathBuf;
use util::sha3::sha3;
use cli::Args;
use configuration::{Cmd, Execute, Configuration};
@@ -132,25 +137,31 @@ fn print_hash_of(maybe_file: Option<String>) -> Result<String, String> {
}
}
fn execute(command: Execute) -> Result<String, String> {
enum PostExecutionAction {
Print(String),
Restart,
Quit,
}
fn execute(command: Execute) -> Result<PostExecutionAction, String> {
let logger = setup_log(&command.logger).expect("Logger is initialized only once; qed");
match command.cmd {
Cmd::Run(run_cmd) => {
try!(run::execute(run_cmd, logger));
Ok("".into())
let restart = run::execute(run_cmd, logger)?;
Ok(if restart { PostExecutionAction::Restart } else { PostExecutionAction::Quit })
},
Cmd::Version => Ok(Args::print_version()),
Cmd::Hash(maybe_file) => print_hash_of(maybe_file),
Cmd::Account(account_cmd) => account::execute(account_cmd),
Cmd::ImportPresaleWallet(presale_cmd) => presale::execute(presale_cmd),
Cmd::Blockchain(blockchain_cmd) => blockchain::execute(blockchain_cmd),
Cmd::SignerToken(signer_cmd) => signer::execute(signer_cmd),
Cmd::Snapshot(snapshot_cmd) => snapshot::execute(snapshot_cmd),
Cmd::Version => Ok(PostExecutionAction::Print(Args::print_version())),
Cmd::Hash(maybe_file) => print_hash_of(maybe_file).map(|s| PostExecutionAction::Print(s)),
Cmd::Account(account_cmd) => account::execute(account_cmd).map(|s| PostExecutionAction::Print(s)),
Cmd::ImportPresaleWallet(presale_cmd) => presale::execute(presale_cmd).map(|s| PostExecutionAction::Print(s)),
Cmd::Blockchain(blockchain_cmd) => blockchain::execute(blockchain_cmd).map(|s| PostExecutionAction::Print(s)),
Cmd::SignerToken(signer_cmd) => signer::execute(signer_cmd).map(|s| PostExecutionAction::Print(s)),
Cmd::Snapshot(snapshot_cmd) => snapshot::execute(snapshot_cmd).map(|s| PostExecutionAction::Print(s)),
}
}
fn start() -> Result<String, String> {
fn start() -> Result<PostExecutionAction, String> {
let args: Vec<String> = env::args().collect();
let conf = Configuration::parse(&args).unwrap_or_else(|e| e.exit());
@@ -163,58 +174,108 @@ fn start() -> Result<String, String> {
execute(cmd)
}
#[cfg(feature="stratum")]
mod stratum_optional {
pub fn probably_run() -> bool {
// just redirect to the stratum::main()
if ::std::env::args().nth(1).map_or(false, |arg| arg == "stratum") {
super::stratum::main();
true
}
else { false }
}
}
#[cfg(not(feature="stratum"))]
mod stratum_optional {
pub fn probably_run() -> bool {
false
}
fn stratum_main(_: &mut HashMap<String, fn()>) {}
#[cfg(feature="stratum")]
fn stratum_main(alt_mains: &mut HashMap<String, fn()>) {
alt_mains.insert("stratum".to_owned(), stratum::main);
}
#[cfg(not(feature="ipc"))]
fn sync_main() -> bool {
false
}
fn sync_main(_: &mut HashMap<String, fn()>) {}
#[cfg(feature="ipc")]
fn sync_main() -> bool {
// just redirect to the sync::main()
if std::env::args().nth(1).map_or(false, |arg| arg == "sync") {
sync::main();
true
fn sync_main(alt_mains: &mut HashMap<String, fn()>) {
alt_mains.insert("sync".to_owned(), sync::main);
}
// TODO: merge with version in Updater.
fn updates_latest() -> PathBuf {
let mut dest = PathBuf::from(env::home_dir().unwrap().to_str().expect("env filesystem paths really should be valid; qed"));
dest.push(".parity-updates");
dest.push("parity");
dest
}
// Starts ~/.parity-updates/parity and returns the code it exits with.
fn run_parity() -> Option<i32> {
let exe = updates_latest();
process::Command::new(exe)
.args(&env::args_os().collect::<Vec<_>>())
.status()
.map(|es| es.code().unwrap_or(128))
.ok()
}
const PLEASE_RESTART_EXIT_CODE: i32 = 69;
// Run our version of parity.
// Returns the exit error code.
fn main_direct() -> i32 {
let mut alt_mains = HashMap::new();
sync_main(&mut alt_mains);
stratum_main(&mut alt_mains);
if let Some(f) = std::env::args().nth(1).and_then(|arg| alt_mains.get(&arg.to_string())) {
f();
0
} else {
false
match start() {
Ok(result) => match result {
PostExecutionAction::Print(s) => { info!("{}", s); 0 },
PostExecutionAction::Restart => PLEASE_RESTART_EXIT_CODE,
PostExecutionAction::Quit => 0,
},
Err(err) => {
writeln!(&mut stdio::stderr(), "{}", err).expect("StdErr available; qed");
1
},
}
}
}
fn println_trace_main(s: String) {
if env::var("RUST_LOG").ok().and_then(|s| s.find("main=trace")).is_some() {
println!("{}", s);
}
}
#[macro_export]
macro_rules! trace_main {
($arg:expr) => (println_trace_main($arg.into()));
($($arg:tt)*) => (println_trace_main(format!("{}", format_args!($($arg)*))));
}
fn main() {
// Always print backtrace on panic.
::std::env::set_var("RUST_BACKTRACE", "1");
env::set_var("RUST_BACKTRACE", "1");
if sync_main() {
return;
}
if stratum_optional::probably_run() { return; }
match start() {
Ok(result) => {
info!("{}", result);
},
Err(err) => {
writeln!(&mut stdio::stderr(), "{}", err).expect("StdErr available; qed");
process::exit(1);
// assuming the user is not running with `--force-direct`, then:
// if argv[0] == "parity" and this executable != ~/.parity-updates/parity, run that instead.
let force_direct = std::env::args().any(|arg| arg == "--force-direct");
let exe = std::env::current_exe().ok();
let development = exe.and_then(|p| p.parent().and_then(|p| p.parent()).and_then(|p| p.file_name()).map(|n| n == "target")).unwrap_or(false);
let same_name = exe.and_then(|p| p.file_stem().map_or(false, |s| s == "parity"));
let have_update = updates_latest().exists();
let is_non_updated_current = exe.map_or(false, p.canonicalize() != updates_latest().canonicalize());
trace_main!("Starting up {} (force-direct: {}, development: {}, have-update: {}, non-updated-current: {})", std::env::current_exe().map(|x| format!("{}", x.display())).unwrap_or("<unknown>".to_owned()), force_direct, development, have_update, is_non_updated_current);
if !force_direct && ! development && have_update && is_non_updated_current {
// looks like we're not running ~/.parity-updates/parity when the user is expecting otherwise.
// Everything run inside a loop, so we'll be able to restart from the child into a new version seamlessly.
loop {
// If we fail to run the updated parity then fallback to local version.
trace_main!("Attempting to run latest update...");
let exit_code = run_parity().unwrap_or_else(|| { trace_main!("Falling back to local..."); main_direct() });
trace_main!("Latest exited with {}", exit_code);
if exit_code != PLEASE_RESTART_EXIT_CODE {
trace_main!("Quitting...");
process::exit(exit_code);
}
trace_main!("Rerunning...");
}
} else {
trace_main!("Running direct");
// Otherwise, we're presumably running the version we want. Just run and fall-through.
process::exit(main_direct());
}
}

359
parity/operations.rs Normal file

File diff suppressed because one or more lines are too long

View File

@@ -23,7 +23,7 @@ use ethsync::NetworkConfiguration;
use util::{Colour, version, RotatingLogger};
use io::{MayPanic, ForwardPanic, PanicHandler};
use ethcore_logger::{Config as LogConfig};
use ethcore::client::{Mode, UpdatePolicy, DatabaseCompactionProfile, VMType, ChainNotify, BlockChainClient};
use ethcore::client::{Mode, UpdatePolicy, Updater, DatabaseCompactionProfile, VMType, ChainNotify, BlockChainClient};
use ethcore::service::ClientService;
use ethcore::account_provider::AccountProvider;
use ethcore::miner::{Miner, MinerService, ExternalMiner, MinerOptions};
@@ -31,11 +31,11 @@ use ethcore::snapshot;
use ethcore::verification::queue::VerifierSettings;
use ethsync::SyncConfig;
use informant::Informant;
use updater::Updater;
use rpc::{HttpServer, IpcServer, HttpConfiguration, IpcConfiguration};
use signer::SignerServer;
use dapps::WebappServer;
use io_handler::ClientIoHandler;
use params::{
SpecType, Pruning, AccountsConfig, GasPricerConfig, MinerExtras, Switch,
tracing_switch_to_bool, fatdb_switch_to_bool, mode_switch_to_bool
@@ -116,12 +116,12 @@ pub fn open_ui(dapps_conf: &dapps::Configuration, signer_conf: &signer::Configur
Ok(())
}
pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<bool, String> {
if cmd.ui && cmd.dapps_conf.enabled {
// Check if Parity is already running
let addr = format!("{}:{}", cmd.dapps_conf.interface, cmd.dapps_conf.port);
if !TcpListener::bind(&addr as &str).is_ok() {
return open_ui(&cmd.dapps_conf, &cmd.signer_conf);
return open_ui(&cmd.dapps_conf, &cmd.signer_conf).map(|_| false);
}
}
@@ -244,7 +244,6 @@ pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
// create client config
let mut client_config = to_client_config(
&cmd.cache_config,
update_policy,
mode.clone(),
tracing,
fat_db,
@@ -312,6 +311,12 @@ pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
chain_notify.start();
}
// the updater service
let updater = Updater::new(service.client(), update_policy);
if let Some(ref u) = updater {
service.add_notify(u.clone());
}
// set up dependencies for rpc servers
let signer_path = cmd.signer_conf.signer_path.clone();
let deps_for_rpc_apis = Arc::new(rpc_apis::Dependencies {
@@ -348,24 +353,23 @@ pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
let http_server = try!(rpc::new_http(cmd.http_conf, &dependencies));
let ipc_server = try!(rpc::new_ipc(cmd.ipc_conf, &dependencies));
// the dapps server
let dapps_deps = dapps::Dependencies {
panic_handler: panic_handler.clone(),
apis: deps_for_rpc_apis.clone(),
client: client.clone(),
sync: sync_provider.clone(),
};
// start dapps server
let dapps_server = try!(dapps::new(cmd.dapps_conf.clone(), dapps_deps));
// the signer server
let signer_deps = signer::Dependencies {
panic_handler: panic_handler.clone(),
apis: deps_for_rpc_apis.clone(),
};
// start signer server
let signer_server = try!(signer::start(cmd.signer_conf.clone(), signer_deps));
// the informant
let informant = Arc::new(Informant::new(
service.client(),
Some(sync_provider.clone()),
@@ -373,17 +377,8 @@ pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
Some(snapshot_service.clone()),
cmd.logger_config.color
));
let info_notify: Arc<ChainNotify> = informant.clone();
service.add_notify(info_notify);
let io_handler = Arc::new(ClientIoHandler {
client: service.client(),
info: informant,
sync: sync_provider.clone(),
net: manage_network.clone(),
accounts: account_provider.clone(),
shutdown: Default::default(),
});
service.register_io_handler(io_handler.clone()).expect("Error registering IO handler");
service.add_notify(informant.clone());
service.register_io_handler(informant.clone()).map_err(|_| "Unable to register informant handler".to_owned())?;
// save user defaults
user_defaults.pruning = algorithm;
@@ -392,13 +387,11 @@ pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
user_defaults.mode = mode;
try!(user_defaults.save(&user_defaults_path));
let on_mode_change = move |mode: &Mode| {
// tell client how to save the default mode if it gets changed.
client.on_mode_change(move |mode: &Mode| {
user_defaults.mode = mode.clone();
let _ = user_defaults.save(&user_defaults_path); // discard failures - there's nothing we can do
};
// tell client how to save the default mode if it gets changed.
client.on_mode_change(on_mode_change);
});
// the watcher must be kept alive.
let _watcher = match cmd.no_periodic_snapshot {
@@ -424,7 +417,9 @@ pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
}
// Handle exit
wait_for_exit(panic_handler, http_server, ipc_server, dapps_server, signer_server);
wait_for_exit(panic_handler, http_server, ipc_server, dapps_server, signer_server, updater);
info!("Finishing work, please wait...");
// to make sure timer does not spawn requests while shutdown is in progress
io_handler.shutdown.store(true, ::std::sync::atomic::Ordering::SeqCst);
@@ -435,7 +430,7 @@ pub fn execute(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<(), String> {
// terminated gracefully
drop(hypervisor);
Ok(())
Ok(false)
}
#[cfg(not(windows))]
@@ -443,11 +438,11 @@ fn daemonize(pid_file: String) -> Result<(), String> {
extern crate daemonize;
daemonize::Daemonize::new()
.pid_file(pid_file)
.chown_pid_file(true)
.start()
.map(|_| ())
.map_err(|e| format!("Couldn't daemonize; {}", e))
.pid_file(pid_file)
.chown_pid_file(true)
.start()
.map(|_| ())
.map_err(|e| format!("Couldn't daemonize; {}", e))
}
#[cfg(windows)]
@@ -478,20 +473,26 @@ fn wait_for_exit(
_http_server: Option<HttpServer>,
_ipc_server: Option<IpcServer>,
_dapps_server: Option<WebappServer>,
_signer_server: Option<SignerServer>
) {
let exit = Arc::new(Condvar::new());
_signer_server: Option<SignerServer>,
updater: Option<Arc<Updater>>
) -> bool {
let exit = Arc::new((Mutex::new(false), Condvar::new()));
// Handle possible exits
let e = exit.clone();
CtrlC::set_handler(move || { e.notify_all(); });
CtrlC::set_handler(move || { e.1.notify_all(); });
// Handle panics
let e = exit.clone();
panic_handler.on_panic(move |_reason| { e.notify_all(); });
panic_handler.on_panic(move |_reason| { e.1.notify_all(); });
// Handle updater wanting to restart us
if let Some(ref u) = updater {
let e = exit.clone();
u.set_exit_handler(move || { e.0.lock() = true; e.1.notify_all(); });
}
// Wait for signal
let mutex = Mutex::new(());
let _ = exit.wait(mutex.lock().unwrap());
info!("Finishing work, please wait...");
let _ = exit.1.wait(exit.0.lock().unwrap());
*exit.0.lock()
}

367
parity/updater.rs Normal file
View File

@@ -0,0 +1,367 @@
// 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::{Weak};
use std::{io, os, fs, env};
use std::path::{Path, PathBuf};
use util::misc::{VersionInfo, ReleaseTrack/*, platform*/};
use util::{Address, H160, H256, FixedHash, Mutex};
use super::operations::Operations;
use ethcore::client::{Client, BlockId};
use fetch::HashFetch;
use fetch;
/// Filter for releases.
#[derive(Debug, Eq, PartialEq, Clone)]
pub enum UpdateFilter {
/// All releases following the same track.
All,
/// As with `All`, but only those which are known to be critical.
Critical,
/// None.
None,
}
/// The policy for auto-updating.
#[derive(Debug, Eq, PartialEq, Clone)]
pub struct UpdatePolicy {
/// Download potential updates.
pub enable_downloading: bool,
/// Which of those downloaded should be automatically installed.
pub filter: UpdateFilter,
}
impl Default for UpdatePolicy {
fn default() -> Self {
UpdatePolicy {
enable_downloading: false,
filter: UpdateFilter::None,
}
}
}
/// Information regarding a particular release of Parity
#[derive(Debug, Clone, PartialEq)]
pub struct ReleaseInfo {
/// Information on the version.
pub version: VersionInfo,
/// Does this release contain critical security updates?
pub is_critical: bool,
/// The latest fork that this release can handle.
pub fork: u64,
/// Our platform's binary, if known.
pub binary: Option<H256>,
}
/// Information on our operations environment.
#[derive(Debug, Clone, PartialEq)]
pub struct OperationsInfo {
/// Our blockchain's latest fork.
pub fork: u64,
/// Last fork our client supports, if known.
pub this_fork: Option<u64>,
/// Information on our track's latest release.
pub track: ReleaseInfo,
/// Information on our minor version's latest release.
pub minor: Option<ReleaseInfo>,
}
#[derive(Debug, Default)]
struct UpdaterState {
latest: Option<OperationsInfo>,
fetching: Option<ReleaseInfo>,
ready: Option<ReleaseInfo>,
installed: Option<ReleaseInfo>,
}
/// Service for checking for updates and determining whether we can achieve consensus.
pub struct Updater {
// Useful environmental stuff.
update_policy: UpdatePolicy,
weak_self: Weak<Updater>,
client: Weak<Client>,
fetcher: Option<fetch::Client>,
operations: Mutex<Option<Operations>>,
exit_handler: Mutex<Option<Fn()>>,
// Our version info (static)
this: VersionInfo,
// All the other info - this changes so leave it behind a Mutex.
state: Mutex<UpdaterState>,
}
const CLIENT_ID: &'static str = "parity";
// TODO!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! REMOVE!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
fn platform() -> String {
"test".to_owned()
}
#[cfg(windows)]
fn symlink<P: AsRef<Path>, Q: AsRef<Path>>(src: P, dst: Q) -> io::Result<()> {
os::windows::fs::symlink_file(src, dst)
}
#[cfg(not(windows))]
fn symlink<P: AsRef<Path>, Q: AsRef<Path>>(src: P, dst: Q) -> io::Result<()> {
os::unix::fs::symlink(src, dst)
}
impl Updater {
pub fn new(client: Weak<BlockChainClient>, update_policy: UpdatePolicy) -> Arc<Self> {
let mut u = Updater {
update_policy: update_policy,
weak_self: Default::default(),
client: client.clone(),
fetcher: None,
operations: Mutex::new(None),
exit_handler: Mutex::new(None),
this: VersionInfo::this(),
state: Mutex::new(Default::default()),
};
// TODO!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! REMOVE!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
if u.this.track == ReleaseTrack::Unknown {
u.this.track = ReleaseTrack::Nightly;
}
let r = Arc::new(u);
r.as_mut().weak_self = Arc::downgrade(&r);
r.as_mut().fetcher = Some(fetch::Client::new(r));
r
}
/// Is the currently running client capable of supporting the current chain?
/// `Some` answer or `None` if information on the running client is not available.
pub fn is_capable(&self) -> Option<bool> {
self.state.lock().latest.as_ref().and_then(|latest| {
latest.this_fork.map(|this_fork| {
let current_number = self.client.upgrade().map_or(0, |c| c.block_number(BlockId::Latest).unwrap_or(0));
this_fork >= latest.fork || current_number < latest.fork
})
})
}
/// The release which is ready to be upgraded to, if any. If this returns `Some`, then
/// `execute_upgrade` may be called.
pub fn upgrade_ready(&self) -> Option<ReleaseInfo> {
self.state.lock().ready.clone()
}
/// Actually upgrades the client. Assumes that the binary has been downloaded.
/// @returns `true` on success.
pub fn execute_upgrade(&mut self) -> bool {
(|| -> Result<bool, String> {
let s = state.lock();
if let Some(r) = s.ready.take() {
let p = Self::update_file_path(&r.version);
let n = Self::updates_latest();
let _ = fs::remove_file(&n);
match symlink(p, n) {
Ok(_) => {
info!("Completed upgrade to {}", &r.version);
s.installed = Some(r);
if let Some(ref h) = self.exit_handler().lock() {
(*h)();
}
Ok(true)
}
Err(e) => {
s.ready = Some(r);
Err(format!("Unable to create soft-link for update {:?}", e))
}
}
} else {
warn!("Execute upgrade called when no upgrade ready.");
Ok(false)
}
})().unwrap_or_else(|e| { warn!("{}", e); false })
}
/// Returns true iff the current version is capable of forming consensus.
pub fn is_consensus_capable(&self) -> bool {
/* if let Some(ref latest) = self.latest {
*/ unimplemented!()
}
/// Our version info.
pub fn version_info(&self) -> &VersionInfo { &self.this }
/// Information gathered concerning the release.
pub fn info(&self) -> Option<OperationsInfo> { self.state.lock().latest.clone() }
/// Set a closure to call when we want to restart the client
pub fn set_exit_handler(&self, f: Fn()) {
*self.exit_handler.lock() = f;
}
fn collect_release_info(&self, release_id: &H256) -> Result<ReleaseInfo, String> {
let (fork, track, semver, is_critical) = self.operations.release(CLIENT_ID, release_id)?;
let latest_binary = self.operations.checksum(CLIENT_ID, release_id, &platform())?;
Ok(ReleaseInfo {
version: VersionInfo::from_raw(semver, track, release_id.clone().into()),
is_critical: is_critical,
fork: fork as u64,
binary: if latest_binary.is_zero() { None } else { Some(latest_binary) },
})
}
fn collect_latest(&self) -> Result<OperationsInfo, String> {
let this_fork = u.operations.release(CLIENT_ID, &u.this.hash.into()).ok()
.and_then(|(fork, track, _, _)| if track > 0 {Some(fork as u64)} else {None});
if self.this.track == ReleaseTrack::Unknown {
return Err(format!("Current executable ({}) is unreleased.", H160::from(self.this.hash)));
}
let latest_in_track = self.operations.latest_in_track(CLIENT_ID, self.this.track.into())?;
let in_track = self.collect_release_info(&latest_in_track)?;
let mut in_minor = Some(in_track.clone());
const PROOF: &'static str = "in_minor initialised and assigned with Some; loop breaks if None assigned; qed";
while in_minor.as_ref().expect(PROOF).version.track != self.this.track {
let track = match in_minor.as_ref().expect(PROOF).version.track {
ReleaseTrack::Beta => ReleaseTrack::Stable,
ReleaseTrack::Nightly => ReleaseTrack::Beta,
_ => { in_minor = None; break; }
};
in_minor = Some(self.collect_release_info(&self.operations.latest_in_track(CLIENT_ID, track.into())?)?);
}
Ok(OperationsInfo {
fork: self.operations.latest_fork()? as u64,
this_fork: this_fork,
track: in_track,
minor: in_minor,
})
}
fn update_file_path(v: &VersionInfo) -> PathBuf {
let mut dest = PathBuf::from(env::home_dir().unwrap().to_str().expect("env filesystem paths really should be valid; qed"));
dest.push(".parity-updates");
dest.push(format!("parity-{}.{}.{}-{:?}", v.version.major, v.version.minor, v.version.patch, v.hash));
dest
}
fn updates_latest() -> PathBuf {
let mut dest = PathBuf::from(env::home_dir().unwrap().to_str().expect("env filesystem paths really should be valid; qed"));
dest.push(".parity-updates");
dest.push("parity");
dest
}
fn fetch_done(&mut self, result: Result<PathBuf, fetch::Error>) {
(|| -> Result<(), String> {
let auto = {
let mut s = state.lock();
let fetched = s.fetching.take().unwrap();
let b = result.map_err(|e| format!("Unable to fetch update ({}): {:?}", fetched.version, e))?;
info!("Fetched latest version ({}) OK to {}", fetched.version, b.display());
let dest = Self::update_file_path(&fetched.version);
fs::create_dir_all(dest.parent().expect("at least one thing pushed; qed")).map_err(|e| format!("Unable to create updates path: {:?}", e))?;
fs::copy(&b, &dest).map_err(|e| format!("Unable to copy update: {:?}", e))?;
info!("Copied file to {}", dest.display());
let auto = match self.update_policy.filter {
UpdateFilter::All => true,
UpdateFilter::Critical if fetched.is_critical /* TODO: or is on a bad fork */ => true,
_ => false,
};
s.ready = Some(fetched);
auto
};
if auto {
self.execute_upgrade();
}
Ok(())
})().unwrap_or_else(|e| warn!("{}", e));
}
fn poll(&mut self) {
info!(target: "updater", "Current release is {}", self.this);
if *self.operations.lock().is_none() {
if let Some(ops_addr) = client.upgrade().registry_address("operations") {
trace!(target: "client", "Found operations at {}", ops_addr);
let client = self.client.clone();
*self.operations.lock() = Some(Operations::new(ops_addr, move |a, d| client.upgrade().ok_or("No client!".into()).and_then(|c| c.call_contract(a, d))));
} else {
// No Operations contract - bail.
return;
}
}
u.latest = u.collect_latest().ok();
let current_number = self.client.upgrade().map_or(0, |c| c.block_number(BlockId::Latest).unwrap_or(0));
let latest = self.collect_latest().ok();
if let Some(ref latest) = latest {
info!(target: "updater", "Latest release in our track is v{} it is {}critical ({} binary is {})",
latest.track.version,
if latest.track.is_critical {""} else {"non-"},
platform(),
if let Some(ref b) = latest.track.binary {
format!("{}", b)
} else {
"unreleased".into()
}
);
let mut s = self.state.lock();
let running_latest = latest.track.version.hash == self.version_info().hash;
let already_have_latest = s.installed.as_ref().or(s.ready.as_ref()).map_or(false, |t| *t == latest.track);
if self.update_policy.enable_downloading && !running_latest && !already_have_latest {
if let Some(b) = latest.track.binary {
if fetching.is_none() {
info!("Attempting to get parity binary {}", b);
s.fetching = Some(latest.track.clone());
let weak_self = self.weak_self.clone();
let f = move |r: Result<PathBuf, fetch::Error>| if let Some(this) = weak_self.upgrade().as_mut() { this.fetch_done(r) }};
fetcher.fetch(b, Box::new(f)).ok();
}
}
}
info!(target: "updater", "Fork: this/current/latest/latest-known: {}/#{}/#{}/#{}", match s.latest.this_fork { Some(f) => format!("#{}", f), None => "unknown".into(), }, current_number, s.latest.track.fork, s.latest.fork);
}
(*self.state.lock()).latest = latest;
}
}
impl ChainNotify for Updater {
fn new_blocks(&self, _imported: Vec<H256>, _invalid: Vec<H256>, _enacted: Vec<H256>, _retracted: Vec<H256>, _sealed: Vec<H256>, duration: u64) {
// TODO: something like this
// if !self.client.upgrade().map_or(true, |c| c.is_major_syncing()) {
self.poll();
// }
}
}
impl fetch::urlhint::ContractClient for Updater {
fn registrar(&self) -> Result<Address, String> {
self.client.upgrade().ok_or_else(|| "Client not available".to_owned())?
.registrar_address()
.ok_or_else(|| "Registrar not available".into())
}
fn call(&self, address: Address, data: Bytes) -> Result<Bytes, String> {
self.client.upgrade().ok_or_else(|| "Client not available".to_owned())?
.call_contract(address, data)
}
}