diff --git a/Cargo.lock b/Cargo.lock index 88796dae22..40a8436792 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1020,7 +1020,6 @@ dependencies = [ "flate2", "futures", "hex", - "hotwatch", "humanize-rs", "ic-agent", "ic-types", @@ -1318,25 +1317,6 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05f8140122fa0d5dcb9fc8627cfce2b37cc1500f752636d46ea28bc26785c2f9" -[[package]] -name = "fsevent" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ab7d1bd1bd33cc98b0889831b72da23c0aa4df9cec7e0702f46ecea04b35db6" -dependencies = [ - "bitflags", - "fsevent-sys", -] - -[[package]] -name = "fsevent-sys" -version = "2.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f41b048a94555da0f42f1d632e2e19510084fb8e303b0daa2816e733fb3644a0" -dependencies = [ - "libc", -] - [[package]] name = "fuchsia-cprng" version = "0.1.1" @@ -1565,16 +1545,6 @@ dependencies = [ "winapi 0.3.9", ] -[[package]] -name = "hotwatch" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba0add391e9cd7d19c29024617a44df79c867ab003bce7f3224c1636595ec740" -dependencies = [ - "log", - "notify", -] - [[package]] name = "http" version = "0.2.1" @@ -1762,26 +1732,6 @@ dependencies = [ "regex", ] -[[package]] -name = "inotify" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4816c66d2c8ae673df83366c18341538f234a26d65a9ecea5c348b453ac1d02f" -dependencies = [ - "bitflags", - "inotify-sys", - "libc", -] - -[[package]] -name = "inotify-sys" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e74a1aa87c59aeff6ef2cc2fa62d41bc43f54952f55652656b18a02fd5e356c0" -dependencies = [ - "libc", -] - [[package]] name = "instant" version = "0.1.7" @@ -1901,12 +1851,6 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" -[[package]] -name = "lazycell" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" - [[package]] name = "leb128" version = "0.2.4" @@ -2066,18 +2010,6 @@ dependencies = [ "winapi 0.2.8", ] -[[package]] -name = "mio-extras" -version = "2.0.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52403fe290012ce777c4626790c8951324a2b9e3316b3143779c72b029742f19" -dependencies = [ - "lazycell", - "log", - "mio", - "slab", -] - [[package]] name = "mio-uds" version = "0.6.8" @@ -2210,24 +2142,6 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61807f77802ff30975e01f4f071c8ba10c022052f98b3294119f3e615d13e5be" -[[package]] -name = "notify" -version = "4.0.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "80ae4a7688d1fab81c5bf19c64fc8db920be8d519ce6336ed4e7efe024724dbd" -dependencies = [ - "bitflags", - "filetime", - "fsevent", - "fsevent-sys", - "inotify", - "libc", - "mio", - "mio-extras", - "walkdir", - "winapi 0.3.9", -] - [[package]] name = "num-bigint" version = "0.3.0" diff --git a/e2e/bats/start.bash b/e2e/bats/start.bash new file mode 100644 index 0000000000..b51cfb7cfa --- /dev/null +++ b/e2e/bats/start.bash @@ -0,0 +1,48 @@ +#!/usr/bin/env bats + +load utils/_ + +setup() { + # We want to work from a temporary directory, different for every test. + cd $(mktemp -d -t dfx-e2e-XXXXXXXX) + + dfx_new hello +} + +teardown() { + dfx_stop +} + +@test "dfx restarts the replica" { + [ "$USE_IC_REF" ] && skip "skip for ic-ref" + + dfx_start + + install_asset greet + assert_command dfx deploy + assert_command dfx canister call hello greet '("Alpha")' + assert_eq '("Hello, Alpha!")' + + # ps "isn't a robust solution": https://dfinity.atlassian.net/browse/OPS-166 + # + # I guess we could make dfx write the replica pid somewhere. + # + # Anyway, if this test ends up sometimes killing a replica other than + # the one created by dfx for this test, then some other test might fail. + # + # Also, this does not work on linux: + # ps -o "ppid, pid, comm" + REPLICA_PID=$(ps x | grep [/[:space:]]replica | awk '{print $1}') + + echo "replica pid is $REPLICA_PID" + + kill -KILL $REPLICA_PID + assert_process_exits $REPLICA_PID 15s + + timeout 15s sh -c \ + 'until dfx ping; do echo waiting for replica to restart; sleep 1; done' \ + || (echo "replica did not restart" && ps aux && exit 1) + + assert_command dfx canister call hello greet '("Omega")' + assert_eq '("Hello, Omega!")' +} diff --git a/e2e/bats/utils/assertions.bash b/e2e/bats/utils/assertions.bash index 71878579de..0d726ea911 100644 --- a/e2e/bats/utils/assertions.bash +++ b/e2e/bats/utils/assertions.bash @@ -148,8 +148,8 @@ assert_process_exits() { # Asserts that `dfx start` and `replica` are no longer running assert_no_dfx_start_or_replica_processes() { - ! ( ps | grep "[/\s]dfx start" ) - ! ( ps | grep "[/\s]replica" ) + ! ( ps | grep "[/[:space:]]dfx start" ) + ! ( ps | grep "[/[:space:]]replica" ) } assert_file_eventually_exists() { diff --git a/src/dfx/Cargo.toml b/src/dfx/Cargo.toml index 3c9c263c64..7504943926 100644 --- a/src/dfx/Cargo.toml +++ b/src/dfx/Cargo.toml @@ -24,9 +24,9 @@ base64 = "0.11.0" candid = "0.6.7" chrono = "0.4.9" clap = "2.33.0" -ctrlc = { version = "3.1.6", features = [ "termination" ] } console = "0.7.7" crossbeam = "0.7.3" +ctrlc = { version = "3.1.6", features = [ "termination" ] } delay = "0.3.0" dialoguer = "0.6.2" erased-serde = "0.3.10" @@ -37,7 +37,6 @@ indicatif = "0.13.0" lazy-init = "0.3.0" lazy_static = "1.4.0" libflate = "0.1.27" -hotwatch = "0.4.3" humanize-rs = "0.1.5" mime = "0.3.16" mockall = "0.6.0" diff --git a/src/dfx/src/actors.rs b/src/dfx/src/actors.rs index 319d813b54..b748f45c03 100644 --- a/src/dfx/src/actors.rs +++ b/src/dfx/src/actors.rs @@ -1,2 +1,3 @@ pub mod replica; +pub mod replica_webserver_coordinator; pub mod shutdown_controller; diff --git a/src/dfx/src/actors/replica.rs b/src/dfx/src/actors/replica.rs index 7aeae93dbc..c79e0e2375 100644 --- a/src/dfx/src/actors/replica.rs +++ b/src/dfx/src/actors/replica.rs @@ -226,15 +226,22 @@ fn wait_for_child_or_receiver( receiver: &Receiver<()>, ) -> ChildOrReceiver { loop { - // Ping-pong between waiting 100 msec for a receiver signal, and check if - // the child quit. - if let Ok(()) = receiver.recv_timeout(std::time::Duration::from_millis(100)) { - return ChildOrReceiver::Receiver; - } - - if let Ok(Some(_)) = child.try_wait() { - return ChildOrReceiver::Child; - } + // Check if either the child exited or a shutdown has been requested. + // These can happen in either order in response to Ctrl-C, so increase the chance + // to notice a shutdown request even if the replica exited quickly. + let child_try_wait = child.try_wait(); + let receiver_signalled = receiver.recv_timeout(std::time::Duration::from_millis(100)); + + match (receiver_signalled, child_try_wait) { + (Ok(()), _) => { + // Prefer to indicate the shutdown request + return ChildOrReceiver::Receiver; + } + (Err(_), Ok(Some(_))) => { + return ChildOrReceiver::Child; + } + _ => {} + }; } } diff --git a/src/dfx/src/actors/replica_webserver_coordinator.rs b/src/dfx/src/actors/replica_webserver_coordinator.rs new file mode 100644 index 0000000000..9b5e23a587 --- /dev/null +++ b/src/dfx/src/actors/replica_webserver_coordinator.rs @@ -0,0 +1,130 @@ +use crate::actors::replica::signals::outbound::ReplicaReadySignal; +use crate::actors::replica::signals::PortReadySubscribe; +use crate::actors::replica::Replica; +use crate::actors::shutdown_controller::signals::outbound::Shutdown; +use crate::actors::shutdown_controller::signals::ShutdownSubscribe; +use crate::actors::shutdown_controller::ShutdownController; +use crate::lib::error::DfxResult; +use crate::lib::network::network_descriptor::NetworkDescriptor; +use crate::lib::webserver::run_webserver; +use actix::clock::{delay_for, Duration}; +use actix::fut::wrap_future; +use actix::{Actor, Addr, AsyncContext, Context, Handler, ResponseFuture}; +use actix_server::Server; +use futures::future; +use futures::future::FutureExt; +use slog::{debug, error, info, Logger}; +use std::net::SocketAddr; +use std::path::PathBuf; + +pub struct Config { + pub logger: Option, + pub replica_addr: Addr, + pub shutdown_controller: Addr, + pub bind: SocketAddr, + pub serve_dir: PathBuf, + pub providers: Vec, + pub build_output_root: PathBuf, + pub network_descriptor: NetworkDescriptor, +} + +/// +/// The ReplicaWebserverCoordinator runs a webserver for the replica. +/// +/// If the replica restarts, it will start a new webserver for the new replica. +pub struct ReplicaWebserverCoordinator { + logger: Logger, + config: Config, + server: Option, +} + +impl ReplicaWebserverCoordinator { + pub fn new(config: Config) -> Self { + let logger = + (config.logger.clone()).unwrap_or_else(|| Logger::root(slog::Discard, slog::o!())); + ReplicaWebserverCoordinator { + config, + logger, + server: None, + } + } + + fn start_server(&self, port: u16) -> DfxResult { + let mut providers = self.config.providers.clone(); + + let ic_replica_bind_addr = "http://localhost:".to_owned() + port.to_string().as_str(); + let ic_replica_bind_addr = ic_replica_bind_addr.as_str(); + let replica_api_uri = + url::Url::parse(ic_replica_bind_addr).expect("Failed to parse replica ingress url."); + providers.push(replica_api_uri); + info!( + self.logger, + "Starting webserver on port {} for replica at {:?}", port, ic_replica_bind_addr + ); + + run_webserver( + self.logger.clone(), + self.config.build_output_root.clone(), + self.config.network_descriptor.clone(), + self.config.bind, + providers, + self.config.serve_dir.clone(), + ) + } +} + +impl Actor for ReplicaWebserverCoordinator { + type Context = Context; + + fn started(&mut self, ctx: &mut Self::Context) { + self.config + .replica_addr + .do_send(PortReadySubscribe(ctx.address().recipient())); + self.config + .shutdown_controller + .do_send(ShutdownSubscribe(ctx.address().recipient::())); + } +} + +impl Handler for ReplicaWebserverCoordinator { + type Result = (); + + fn handle(&mut self, msg: ReplicaReadySignal, ctx: &mut Self::Context) { + debug!(self.logger, "replica ready on {}", msg.port); + + if let Some(server) = &self.server { + ctx.wait(wrap_future(server.stop(true))); + self.server = None; + ctx.address().do_send(msg); + } else { + match self.start_server(msg.port) { + Ok(server) => { + self.server = Some(server); + } + Err(e) => { + error!( + self.logger, + "Unable to start webserver on port {}: {}", msg.port, e + ); + ctx.wait(wrap_future(delay_for(Duration::from_secs(2)))); + ctx.address().do_send(msg); + } + } + } + } +} + +impl Handler for ReplicaWebserverCoordinator { + type Result = ResponseFuture>; + + fn handle(&mut self, _msg: Shutdown, _ctx: &mut Self::Context) -> Self::Result { + if let Some(server) = self.server.take() { + // We stop the webserver before shutting down because + // if we don't, the process will segfault + // while dropping actix stuff after main() returns. + Box::pin(server.stop(true).map(Ok)) + } else { + Box::pin(future::ok(())) + } + } +} diff --git a/src/dfx/src/commands/replica.rs b/src/dfx/src/commands/replica.rs index f73b57ea37..909a5d7d0a 100644 --- a/src/dfx/src/commands/replica.rs +++ b/src/dfx/src/commands/replica.rs @@ -62,7 +62,7 @@ fn get_config(env: &dyn Environment, args: &ArgMatches<'_>) -> DfxResult DfxRe ping_and_wait(&frontend_url_mod) } -// TODO(eftychis)/In progress: Rename to replica. /// Start the Internet Computer locally. Spawns a proxy to forward and /// manage browser requests. Responsible for running the network (one /// replica at the moment) and the proxy. @@ -118,202 +117,132 @@ pub fn exec(env: &dyn Environment, args: &ArgMatches<'_>) -> DfxResult { .get_config() .ok_or(DfxError::CommandMustBeRunInAProject)?; - let temp_dir = env.get_temp_dir(); + let network_descriptor = get_network_descriptor(env, args)?; - let (frontend_url, address_and_port) = frontend_address(args, &config)?; + let temp_dir = env.get_temp_dir(); + let build_output_root = temp_dir.join(&network_descriptor.name).join("canisters"); + let pid_file_path = temp_dir.join("pid"); let webserver_port_path = temp_dir.join("webserver-port"); - std::fs::write(&webserver_port_path, "")?; - - // don't write to file since this arg means we send_background() - if !args.is_present("background") { - std::fs::write(&webserver_port_path, address_and_port.port().to_string())?; - } - - let client_pathbuf = env.get_cache().get_binary_command_path("replica")?; - let ic_starter_pathbuf = env.get_cache().get_binary_command_path("ic-starter")?; - let state_root = env.get_state_dir(); - let pid_file_path = temp_dir.join("pid"); check_previous_process_running(&pid_file_path)?; + // As we know no start process is running in this project, we can // clean up the state if it is necessary. if args.is_present("clean") { - // Clean the contents of the provided directory including the - // directory itself. N.B. This does NOT follow symbolic links -- and I - // hope we do not need to. - if state_root.is_dir() { - fs::remove_dir_all(state_root.clone()).map_err(DfxError::CleanState)?; - } - let local_dir = temp_dir.join("local"); - if local_dir.is_dir() { - fs::remove_dir_all(local_dir).map_err(DfxError::CleanState)?; - } + clean_state(temp_dir, &state_root)?; } - let client_configuration_dir = temp_dir.join("client-configuration"); - fs::create_dir_all(&client_configuration_dir)?; - let state_dir = temp_dir.join("state/replicated_state"); - fs::create_dir_all(&state_dir)?; - let client_port_path = client_configuration_dir.join("client-1.port"); + std::fs::write(&pid_file_path, "")?; // make sure we can write to this file + std::fs::write(&webserver_port_path, "")?; - // Touch the client port file. This ensures it is empty prior to - // handing it over to the replica. If we read the file and it has - // contents we shall assume it is due to our spawned client - // process. - std::fs::write(&client_port_path, "")?; - // We are doing this here to make sure we can write to the temp - // pid file. - std::fs::write(&pid_file_path, "")?; + let (frontend_url, address_and_port) = frontend_address(args, &config)?; if args.is_present("background") { send_background()?; return fg_ping_and_wait(webserver_port_path, frontend_url); } - // Start the client. - let b = ProgressBar::new_spinner(); - b.set_draw_target(ProgressDrawTarget::stderr()); + write_pid(&pid_file_path); + std::fs::write(&webserver_port_path, address_and_port.port().to_string())?; - b.set_message("Starting up the replica..."); - b.enable_steady_tick(80); + let system = actix::System::new("dfx-start"); - // Must be unbounded, as a killed child should not deadlock. + let shutdown_controller = start_shutdown_controller(env)?; - let (request_stop, rcv_wait) = unbounded(); - let (broadcast_stop, is_killed_client) = unbounded(); - let (give_actix, actix_handler) = unbounded(); + let replica = start_replica(env, &state_root, shutdown_controller.clone())?; - let request_stop_echo = request_stop.clone(); - let rcv_wait_fwatcher = rcv_wait.clone(); - b.set_message("Generating IC local replica configuration."); - let replica_config = ReplicaConfig::new(state_root).with_random_port(&client_port_path); + let _webserver_coordinator = start_webserver_coordinator( + env, + network_descriptor, + address_and_port, + build_output_root, + replica, + shutdown_controller, + )?; - // TODO(eftychis): we need a proper manager type when we start - // spawning multiple client processes and registry. - let client_watchdog = std::thread::Builder::new().name("replica".into()).spawn({ - let is_killed_client = is_killed_client.clone(); - let b = b.clone(); - - move || { - start_client( - &client_pathbuf, - &ic_starter_pathbuf, - &pid_file_path, - is_killed_client, - request_stop, - replica_config, - b, - ) - } - })?; + system.run()?; + + Ok(()) +} + +fn clean_state(temp_dir: &Path, state_root: &Path) -> DfxResult { + // Clean the contents of the provided directory including the + // directory itself. N.B. This does NOT follow symbolic links -- and I + // hope we do not need to. + if state_root.is_dir() { + fs::remove_dir_all(state_root) + .map_err(|e| DfxError::CleanState(e, PathBuf::from(state_root)))?; + } + let local_dir = temp_dir.join("local"); + if local_dir.is_dir() { + fs::remove_dir_all(&local_dir).map_err(|e| DfxError::CleanState(e, local_dir))?; + } + Ok(()) +} + +fn start_shutdown_controller(env: &dyn Environment) -> DfxResult> { + let actor_config = shutdown_controller::Config { + logger: Some(env.get_logger().clone()), + }; + Ok(ShutdownController::new(actor_config).start()) +} - let bootstrap_dir = env.get_cache().get_binary_command_path("bootstrap")?; +fn start_replica( + env: &dyn Environment, + state_root: &Path, + shutdown_controller: Addr, +) -> DfxResult> { + let replica_path = env.get_cache().get_binary_command_path("replica")?; + let ic_starter_path = env.get_cache().get_binary_command_path("ic-starter")?; - // We have a long-lived replica process and a proxy. We use - // currently a messaging pattern to supervise. This is going to - // be tidied up over a more formal actor framework. - let is_killed = is_killed_client; + let temp_dir = env.get_temp_dir(); + let client_configuration_dir = temp_dir.join("client-configuration"); + fs::create_dir_all(&client_configuration_dir)?; + let state_dir = temp_dir.join("state/replicated_state"); + fs::create_dir_all(&state_dir)?; + let client_port_path = client_configuration_dir.join("client-1.port"); + // Touch the client port file. This ensures it is empty prior to + // handing it over to the replica. If we read the file and it has + // contents we shall assume it is due to our spawned client + // process. + std::fs::write(&client_port_path, "")?; + + let replica_config = ReplicaConfig::new(state_root).with_random_port(&client_port_path); + let actor_config = actors::replica::Config { + ic_starter_path, + replica_config, + replica_path, + shutdown_controller, + logger: Some(env.get_logger().clone()), + }; + Ok(actors::replica::Replica::new(actor_config).start()) +} + +fn start_webserver_coordinator( + env: &dyn Environment, + network_descriptor: NetworkDescriptor, + bind: SocketAddr, + build_output_root: PathBuf, + replica_addr: Addr, + shutdown_controller: Addr, +) -> DfxResult> { + let serve_dir = env.get_cache().get_binary_command_path("bootstrap")?; // By default we reach to no external IC nodes. let providers = Vec::new(); - let network_descriptor = get_network_descriptor(env, args)?; - let build_output_root = config.get_temp_path().join(network_descriptor.name.clone()); - let build_output_root = build_output_root.join("canisters"); - - let proxy_config = ProxyConfig { - logger: env.get_logger().clone(), - client_api_port: address_and_port.port(), - bind: address_and_port, - serve_dir: bootstrap_dir, + let actor_config = actors::replica_webserver_coordinator::Config { + logger: Some(env.get_logger().clone()), + replica_addr, + shutdown_controller, + bind, + serve_dir, providers, build_output_root, network_descriptor, }; - - let supervisor_actor_handle = CoordinateProxy { - inform_parent: give_actix, - server_receiver: actix_handler.clone(), - rcv_wait_fwatcher, - request_stop_echo, - is_killed, - }; - - let frontend_watchdog = spawn_and_update_proxy( - proxy_config, - client_port_path, - supervisor_actor_handle, - b.clone(), - )?; - - b.set_message("Pinging the Internet Computer replica..."); - ping_and_wait(&frontend_url)?; - b.finish_with_message("Internet Computer replica started..."); - - // TODO/In Progress(eftychis): Here we should define a Supervisor - // actor to keep track and spawn these two processes - // independently. - - // We have two side processes involving multiple threads running at - // this point. We first wait for a signal that one of the processes - // terminated. N.B. We do not handle the case where the proxy - // terminates abruptly and we have to terminate the client as that - // complicates the situation right now, and we need a watcher that - // terminates all sibling processes if a process returns an error, - // which we lack. We consider this a fine trade-off for now. - - rcv_wait.recv().map_err(|e| { - DfxError::RuntimeError(Error::new( - ErrorKind::Other, - format!("Failed while waiting for the manager -- {:?}", e), - )) - })?; - - // Signal the client to stop. Right now we have little control - // over the client and nodemanager as it provides little - // handling. This is mostly done for completeness. In the future - // we should also force kill, if it ends up being necessary. - let b = ProgressBar::new_spinner(); - b.set_draw_target(ProgressDrawTarget::stderr()); - b.set_message("Terminating..."); - b.enable_steady_tick(80); - broadcast_stop.send(()).expect("Failed to signal children"); - // We can now start terminating our proxy server, we block to - // ensure termination is done properly. At this point the client - // is down though. - - // Signal the actix server to stop. This will - // block. - - b.set_message("Terminating proxy..."); - block_on( - actix_handler - .recv() - .expect("Failed to receive server") - .stop(true), - ); - - b.set_message("Gathering proxy thread..."); - // Join and handle errors for the frontend watchdog thread. - frontend_watchdog.join().map_err(|e| { - DfxError::RuntimeError(Error::new( - ErrorKind::Other, - format!("Failed while running frontend proxy thead -- {:?}", e), - )) - })?; - - b.set_message("Gathering replica thread..."); - // Join and handle errors for the client watchdog thread. Here we - // check the result of client_watchdog and start_client. - client_watchdog.join().map_err(|e| { - DfxError::RuntimeError(Error::new( - ErrorKind::Other, - format!("Failed while running replica thread -- {:?}", e), - )) - })??; - b.finish_with_message("Terminated successfully... Have a great day!!!"); - Ok(()) + Ok(ReplicaWebserverCoordinator::new(actor_config).start()) } fn send_background() -> DfxResult<()> { @@ -356,7 +285,7 @@ fn frontend_address(args: &ArgMatches<'_>, config: &Config) -> DfxResult<(String Ok((frontend_url, address_and_port)) } -fn check_previous_process_running(dfx_pid_path: &PathBuf) -> DfxResult<()> { +fn check_previous_process_running(dfx_pid_path: &Path) -> DfxResult<()> { if dfx_pid_path.exists() { // Read and verify it's not running. If it is just return. if let Ok(s) = std::fs::read_to_string(&dfx_pid_path) { @@ -372,108 +301,8 @@ fn check_previous_process_running(dfx_pid_path: &PathBuf) -> DfxResult<()> { Ok(()) } -/// Starts the client. It is supposed to be used in a thread, thus -/// this function will panic when an error occurs that implies -/// termination of the replica and need the attention of the parent -/// thread. -/// -/// # Panics -/// We panic here to transmit an error to the parent thread. -fn start_client( - client_pathbuf: &PathBuf, - ic_starter_pathbuf: &PathBuf, - pid_file_path: &PathBuf, - is_killed_client: Receiver<()>, - request_stop: Sender<()>, - config: ReplicaConfig, - b: ProgressBar, -) -> DfxResult<()> { - b.set_message("Generating IC local replica configuration."); - - let ic_starter = ic_starter_pathbuf.as_path().as_os_str(); - let mut cmd = std::process::Command::new(ic_starter); - // if None is returned, then an empty String will be provided to replica-path - // TODO: figure out the right solution - cmd.args(&[ - "--replica-path", - client_pathbuf.to_str().unwrap_or_default(), - "--http-port-file", - config - .http_handler - .write_port_to - .unwrap_or_default() - .to_str() - .unwrap_or_default(), - "--state-dir", - config.state_manager.state_root.to_str().unwrap_or_default(), - "--create-funds-whitelist", - "*", - ]); - cmd.stdout(std::process::Stdio::inherit()); - cmd.stderr(std::process::Stdio::inherit()); - - // If the replica itself fails, we are probably into deeper trouble than - // we can solve at this point and the user is better rerunning the server. - let mut child = cmd.spawn().unwrap_or_else(|e| { - request_stop - .try_send(()) - .expect("Replica thread couldn't signal parent to stop"); - // We still want to send an error message. - panic!("Couldn't spawn ic-starter with command {:?}: {}", cmd, e); - }); - - // Update the pid file. +fn write_pid(pid_file_path: &Path) { if let Ok(pid) = sysinfo::get_current_pid() { let _ = std::fs::write(&pid_file_path, pid.to_string()); } - - // N.B. The logic below fixes errors from replica causing - // restarts. We do not want to respawn the replica on a failure. - // This should be substituted with a supervisor. - - // Did we receive a kill signal? - while is_killed_client.is_empty() { - // We have to wait for the child to exit here. We *should* - // always wait(). Read related documentation. - - // We check every 1s on the replica. This logic should be - // transferred / substituted by a supervisor object. - std::thread::sleep(Duration::from_millis(1000)); - - match child.try_wait() { - Ok(Some(status)) => { - // An error occurred: exit the loop. - b.set_message(format!("local replica exited with: {}", status).as_str()); - break; - } - Ok(None) => { - // No change in exit status. - continue; - } - Err(e) => { - request_stop - .send(()) - .expect("Could not signal parent thread from replica runner"); - panic!("Failed to check the status of the replica: {}", e) - } - } - } - // Terminate the replica; wait() then signal stop. Ignore errors - // -- we might get InvalidInput: that is fine -- process might - // have terminated already. - Process::new(child.id() as Pid, None, 0).kill(Signal::Term); - match child.wait() { - Ok(status) => b.set_message(format!("Replica exited with {}", status).as_str()), - Err(e) => b.set_message( - format!("Failed to properly wait for the replica to terminate {}", e).as_str(), - ), - } - - // We DO want to panic here, if we can not signal our - // parent. This is interpreted as an error via join by the - // parent thread. - request_stop - .send(()) - .expect("Could not signal parent thread from replica runner"); - Ok(()) } diff --git a/src/dfx/src/lib/error/mod.rs b/src/dfx/src/lib/error/mod.rs index d93da00d69..578d78ad46 100644 --- a/src/dfx/src/lib/error/mod.rs +++ b/src/dfx/src/lib/error/mod.rs @@ -79,7 +79,7 @@ pub enum DfxError { RuntimeError(std::io::Error), /// Failed to clean up state. - CleanState(std::io::Error), + CleanState(std::io::Error, PathBuf), /// The ide server shouldn't be started from a terminal. LanguageServerFromATerminal, @@ -93,9 +93,6 @@ pub enum DfxError { /// An error during parsing of a version string. VersionCouldNotBeParsed(semver::SemVerError), - /// String provided is not a port - CouldNotParsePort(std::num::ParseIntError), - /// A replica did not start successfully. ReplicaCouldNotBeStarted(), diff --git a/src/dfx/src/lib/mod.rs b/src/dfx/src/lib/mod.rs index 801bea92be..6c663a9ea6 100644 --- a/src/dfx/src/lib/mod.rs +++ b/src/dfx/src/lib/mod.rs @@ -14,8 +14,6 @@ pub mod operations; pub mod package_arguments; pub mod progress_bar; pub mod provider; -pub mod proxy; -pub mod proxy_process; pub mod replica_config; pub mod waiter; pub mod webserver; diff --git a/src/dfx/src/lib/proxy.rs b/src/dfx/src/lib/proxy.rs deleted file mode 100644 index 4ec2ced0f9..0000000000 --- a/src/dfx/src/lib/proxy.rs +++ /dev/null @@ -1,140 +0,0 @@ -use crate::lib::network::network_descriptor::NetworkDescriptor; -use crate::lib::webserver::run_webserver; -use actix_server::Server; -use crossbeam::channel::{Receiver, Sender}; -use std::io::Result; -use std::io::{Error, ErrorKind}; -use std::net::SocketAddr; -use std::path::PathBuf; - -/// A proxy that forwards requests from the browser to the network. -#[derive(Clone, Debug)] -pub struct Proxy { - config: ProxyConfig, - server_handle: ProxyServer, -} - -/// Provide basic information to the proxy about the API port, the -/// address and the serve directory. -#[derive(Clone, Debug)] -pub struct ProxyConfig { - pub client_api_port: u16, - pub bind: SocketAddr, - pub serve_dir: PathBuf, - pub providers: Vec, - pub logger: slog::Logger, - pub build_output_root: PathBuf, - pub network_descriptor: NetworkDescriptor, -} - -#[derive(Clone, Debug)] -enum ProxyServer { - Down, - Up(ServerHandle), -} - -#[derive(Clone, Debug)] -struct ServerHandle { - sender: Sender, - receiver: Receiver, -} - -impl Proxy { - pub fn new(config: ProxyConfig) -> Self { - Self { - config, - server_handle: ProxyServer::Down, - } - } - - // Shutdown and start are private (for now). - async fn shutdown(self) -> Result { - match self.server_handle { - // In case the server is down we recall new() as in the - // future we might add more bookkeeping logic, which will - // end up in bugs. This makes this more readable to on - // what we want. The compiler can optimize this away. - ProxyServer::Down => Ok(Proxy::new(self.config)), - ProxyServer::Up(handler) => { - handler - .receiver - .try_recv() - .map_err(|e| { - Error::new( - ErrorKind::Other, - format!("Failed to shutdown proxy -- {:?}", e), - ) - })? - .stop(true) - .await; - Ok(Self { - config: self.config, - server_handle: ProxyServer::Down, - }) - } - } - } - - /// Start a proxy with the provided configuration. Returns a proxy - /// handle. Can fail to return a new proxy. - /// # Panics - /// Currently, we panic if the underlying webserver does not start. - pub fn start(self, sender: Sender, receiver: Receiver) -> Result { - let mut providers = self.config.providers.clone(); - - let ic_client_bind_addr = "http://localhost:".to_owned() + self.port().to_string().as_str(); - let ic_client_bind_addr = ic_client_bind_addr.as_str(); - let client_api_uri = - url::Url::parse(ic_client_bind_addr).expect("Failed to parse replica ingress url."); - // Add the localhost as an option. - providers.push(client_api_uri); - eprintln!("replica address: {:?}", ic_client_bind_addr); - - run_webserver( - self.config.logger.clone(), - self.config.build_output_root.clone(), - self.config.network_descriptor.clone(), - self.config.bind, - providers, - self.config.serve_dir.clone(), - sender.clone(), - )?; - - let mut new_server = Proxy::new(self.config); - let handle = ServerHandle { sender, receiver }; - new_server.server_handle = ProxyServer::Up(handle); - Ok(new_server) - } - - /// Set the api port used by the replica. Returns a new proxy - /// object, but does not restart the proxy. - pub fn set_client_api_port(self, client_api_port: u16) -> Self { - let mut handle = self; - handle.config.client_api_port = client_api_port; - handle - } - - /// Restart a proxy with a new configuration. - pub async fn restart(self, sender: Sender, receiver: Receiver) -> Result { - let config = self.config.clone(); - let mut handle = self.shutdown().await?; - handle.config = config; - handle.start(sender, receiver) - } - - /// Return proxy client api port. - fn port(&self) -> u16 { - self.config.client_api_port - } -} - -/// Supervise a Proxy. -// This should be used to refactor and simplify handling of both proxy -// and replica. -pub struct CoordinateProxy { - pub inform_parent: Sender, - pub server_receiver: Receiver, - pub rcv_wait_fwatcher: Receiver<()>, - pub request_stop_echo: Sender<()>, - pub is_killed: Receiver<()>, -} diff --git a/src/dfx/src/lib/proxy_process.rs b/src/dfx/src/lib/proxy_process.rs deleted file mode 100644 index dee4d059ae..0000000000 --- a/src/dfx/src/lib/proxy_process.rs +++ /dev/null @@ -1,111 +0,0 @@ -use crate::lib::error::DfxError; -use crate::lib::proxy::{CoordinateProxy, Proxy, ProxyConfig}; - -use crossbeam::unbounded; -use futures::executor::block_on; -use hotwatch::{Event, Hotwatch}; -use indicatif::ProgressBar; -use std::fs; -use std::path::PathBuf; -use std::time::Duration; - -pub fn spawn_and_update_proxy( - proxy_config: ProxyConfig, - client_port_path: PathBuf, - proxy_supervisor: CoordinateProxy, - b: ProgressBar, -) -> std::io::Result> { - std::thread::Builder::new() - .name("Proxy".into()) - .spawn(move || { - let proxy = Proxy::new(proxy_config); - // Start the proxy first. Below, we panic to propagate the error - // to the parent thread as an error via join(). - - // Check the port and then start the proxy. Below, we panic to propagate the error - // to the parent thread as an error via join(). - b.set_message("Checking replica!"); - - let (send_port, rcv_port) = unbounded(); - - let mut hotwatch = Hotwatch::new_with_custom_delay(std::time::Duration::from_secs(1)) - .expect("hotwatch failed to initialize!"); - let is_killed = proxy_supervisor.is_killed.clone(); - // We start a hotwatch watcher. It will run on the - // background, with "watch life" equal to the lifetime of - // the value. We attempt on each sensible event to read - // the port. - hotwatch - .watch(&client_port_path, { - let client_port_path = client_port_path.clone(); - move |event: Event| { - if !is_killed.is_empty() { - // We are in a weird state where the replica exited with an error, - // but we are still waiting for the pid file to change. As this change - // is never going to occur we need to exit our wait and stop tracking - // the file. We need to re-send the error to properly handle it later - // on. Worst case we will panic at this point. - // - // Disconnect the sender. This should unblock the port receiver. - let _ = send_port; - } - match event { - // We pretty much want to unblock for any events - // except a rescan. A move, create etc event should - // lead to a failure. - Event::Rescan => {} - _ => { - let port = fs::read_to_string(&client_port_path) - .map_err(DfxError::RuntimeError) - .expect("failed to read port file") - .parse::(); - if let Ok(port) = port { - send_port.send(port).expect("Failed to send port"); - } - } - } - } - }) - .expect("failed to watch replica port file!"); - - let port_after_enter = - fs::read_to_string(&client_port_path).unwrap_or_else(|_| "".to_owned()); - - let port = if port_after_enter != "" { - port_after_enter - .parse::() - .map_err(DfxError::CouldNotParsePort) - } else { - // Fail if sender is disconnected. - Ok(rcv_port.recv().expect("Failed to receive port")) - } - .unwrap_or_else(|e| { - proxy_supervisor - .request_stop_echo - .try_send(()) - .expect("Replica thread couldn't signal parent to stop"); - - panic!("Failed to watch port configuration file {:?}", e); - }); - // Stop watching. - let _ = hotwatch; - let proxy = proxy.set_client_api_port(port); - b.set_message(format!("Replica bound at {}", port).as_str()); - block_on(proxy.restart( - proxy_supervisor.inform_parent.clone(), - proxy_supervisor.server_receiver.clone(), - )) - .unwrap_or_else(|e| { - proxy_supervisor - .request_stop_echo - .try_send(()) - .expect("Replica thread couldn't signal parent to stop"); - panic!("Failed to restart the proxy {:?}", e); - }); - - while proxy_supervisor.is_killed.is_empty() { - std::thread::sleep(Duration::from_millis(1000)); - //wait! - } - }) -} diff --git a/src/dfx/src/lib/replica_config.rs b/src/dfx/src/lib/replica_config.rs index 11798bf6bb..9420082675 100644 --- a/src/dfx/src/lib/replica_config.rs +++ b/src/dfx/src/lib/replica_config.rs @@ -58,7 +58,7 @@ pub struct ReplicaConfig { } impl ReplicaConfig { - pub fn new(state_root: PathBuf) -> Self { + pub fn new(state_root: &Path) -> Self { ReplicaConfig { http_handler: HttpHandlerConfig { write_port_to: None, diff --git a/src/dfx/src/lib/webserver.rs b/src/dfx/src/lib/webserver.rs index bc038cf757..2f827c5658 100644 --- a/src/dfx/src/lib/webserver.rs +++ b/src/dfx/src/lib/webserver.rs @@ -172,8 +172,7 @@ pub fn run_webserver( bind: SocketAddr, providers: Vec, serve_dir: PathBuf, - inform_parent: Sender, -) -> Result<(), std::io::Error> { +) -> DfxResult { info!(logger, "binding to: {:?}", bind); const SHUTDOWN_WAIT_TIME: u64 = 60; @@ -189,7 +188,6 @@ pub fn run_webserver( .join(", ") ); - let _sys = System::new("dfx-frontend-http-server"); let forward_data = Arc::new(Mutex::new(ForwardActixData { providers, logger: logger.clone(), @@ -222,15 +220,9 @@ pub fn run_webserver( .bind(bind)? // N.B. This is an arbitrary timeout for now. .shutdown_timeout(SHUTDOWN_WAIT_TIME) - .system_exit() .run(); - // Warning: Note that HttpServer provides its own signal - // handler. That means if we provide signal handling beyond basic - // we need to either as normal "re-signal" or disable_signals(). - let _ = inform_parent.send(handler); - - Ok(()) + Ok(handler) } pub fn webserver( @@ -262,16 +254,21 @@ pub fn webserver( .spawn({ let serve_dir = serve_dir.to_path_buf(); move || { - run_webserver( + let _sys = System::new("dfx-frontend-http-server"); + let server = run_webserver( logger, build_output_root, network_descriptor, bind, clients_api_uri, serve_dir, - inform_parent, ) - .unwrap() + .unwrap(); + + // Warning: Note that HttpServer provides its own signal + // handler. That means if we provide signal handling beyond basic + // we need to either as normal "re-signal" or disable_signals(). + let _ = inform_parent.send(server); } }) .map_err(DfxError::from)