refactor: actor-based dfx start

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!")'
+}

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() {

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"

src/dfx/src/actors.rs

@@ -1,2 +1,3 @@
 pub mod replica;
+pub mod replica_webserver_coordinator;
 pub mod shutdown_controller;

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;
+            }
+            _ => {}
+        };
     }
 }
 

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<Logger>,
+    pub replica_addr: Addr<Replica>,
+    pub shutdown_controller: Addr<ShutdownController>,
+    pub bind: SocketAddr,
+    pub serve_dir: PathBuf,
+    pub providers: Vec<url::Url>,
+    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<Server>,
+}
+
+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<Server> {
+        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<Self>;
+
+    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::<Shutdown>()));
+    }
+}
+
+impl Handler<ReplicaReadySignal> 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<Shutdown> for ReplicaWebserverCoordinator {
+    type Result = ResponseFuture<Result<(), ()>>;
+
+    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(()))
+        }
+    }
+}

src/dfx/src/commands/replica.rs

@@ -62,7 +62,7 @@ fn get_config(env: &dyn Environment, args: &ArgMatches<'_>) -> DfxResult<Replica
     let state_dir = temp_dir.join("state/replicated_state");
     std::fs::create_dir_all(&state_dir)?;
 
-    let mut replica_config = ReplicaConfig::new(env.get_state_dir());
+    let mut replica_config = ReplicaConfig::new(&env.get_state_dir());
     replica_config.http_handler = http_handler;
     replica_config.scheduler = scheduler;
     Ok(replica_config)