Add effects and reactor implementation.

Cargo.toml

@@ -10,7 +10,14 @@ license-file = "LICENSE"
 
 [dependencies]
 anyhow = "1.0.28"
+async-trait = "0.1.31"
+either = "1.5.3"
+enum-iterator = "0.6.0"
+futures = "0.3.5"
 serde = { version = "1.0.110", features = ["derive"] }
+smallvec = "1.4.0"
 structopt = "0.3.14"
-tokio = { version = "0.2.20", features = ["macros", "rt-threaded"] }
+tokio = { version = "0.2.20", features = ["macros", "rt-threaded", "sync"] }
 toml = "0.5.6"
+tracing = "0.1.14"
+tracing-subscriber = "0.2.5"

README.md

@@ -6,10 +6,28 @@ The is the core application for the CasperLabs blockchain.
 
 To compile this application, simply run `cargo build` on a recent stable Rust (`>= 1.43.1`) version.
 
+## Running a validator node
+
+Launching a validator node with the default configuration is done by simply launching the application:
+
+```
+casper-node validator
+```
+
+It is very likely that the configuration requires editing though, so typically one will want to generate a configuration file first, edit it and then launch:
+
+```
+casper-node generate-config > mynode.toml
+# ... edit mynode.toml
+casper-node validator -c mynode.toml
+```
+
 ## Development
 
 A good starting point is to build the documentation and read it in your browser:
 
 ```
 cargo doc --no-deps --open
-```
+```
+
+When generating a configuration file, it is usually helpful to set the log-level to `DEBUG` during development.

src/cli.rs

@@ -5,7 +5,7 @@
 use std::{io, io::Write, path};
 use structopt::StructOpt;
 
-use crate::config;
+use crate::{config, reactor};
 
 // Note: The docstring on `Cli` is the help shown when calling the binary with `--help`.
 #[derive(Debug, StructOpt)]
@@ -14,7 +14,10 @@ pub enum Cli {
     /// Generate a configuration file from defaults and dump it to stdout.
     GenerateConfig {},
     /// Run the validator node.
-    RunValidator {
+    ///
+    /// Loads the configuration values from the given configuration file or uses defaults if not
+    /// given, then launches the reactor.
+    Validator {
         #[structopt(short, long, env)]
         /// Path to configuration file.
         config: Option<path::PathBuf>,
@@ -28,18 +31,20 @@ impl Cli {
             Cli::GenerateConfig {} => {
                 let cfg_str = config::to_string(&Default::default())?;
                 io::stdout().write_all(cfg_str.as_bytes())?;
+
+                Ok(())
             }
-            Cli::RunValidator { config } => {
+            Cli::Validator { config } => {
                 // We load the specified config, if any, otherwise use defaults.
                 let cfg = config
                     .map(config::load_from_file)
                     .transpose()?
                     .unwrap_or_default();
 
-                println!("{:?}", cfg);
+                cfg.log.setup_logging()?;
+
+                reactor::launch::<reactor::validator::Reactor>(cfg).await
             }
         }
-
-        Ok(())
     }
 }

src/config.rs

@@ -4,14 +4,61 @@
 //! sensible defaults.
 //!
 //! The `cli` offers an option to generate a configuration from defaults for editing.
+//!
+//! # Adding a configuration section
+//!
+//! When adding a section to the configuration, ensure that
+//!
+//! * it has an entry in the root configuration `Config`,
+//! * `Default` is implemented (derived or manually) with sensible defaults, and
+//! * it is completely documented.
 
 use anyhow::Context;
 use serde::{Deserialize, Serialize};
-use std::{fs, path};
+use std::{fs, io, path};
+use tracing::debug;
 
-/// Root configuration
+/// Root configuration.
 #[derive(Debug, Default, Deserialize, Serialize)]
-pub struct Config {}
+pub struct Config {
+    /// Log configuration.
+    pub log: Log,
+}
+
+/// Log configuration.
+#[derive(Debug, Deserialize, Serialize)]
+pub struct Log {
+    /// Log level.
+    #[serde(with = "log_level")]
+    pub level: tracing::Level,
+}
+
+impl Default for Log {
+    fn default() -> Self {
+        Log {
+            level: tracing::Level::INFO,
+        }
+    }
+}
+
+impl Log {
+    /// Initialize logging system based on settings in configuration.
+    ///
+    /// Will setup logging as described in this configuration for the whole application. This
+    /// function should only be called once during the lifetime of the application.
+    pub fn setup_logging(&self) -> anyhow::Result<()> {
+        // Setup a new tracing-subscriber writing to `stderr` for logging.
+        tracing::subscriber::set_global_default(
+            tracing_subscriber::fmt()
+                .with_writer(io::stderr)
+                .with_max_level(self.level.clone())
+                .finish(),
+        )?;
+        debug!("debug output enabled");
+
+        Ok(())
+    }
+}
 
 /// Loads a TOML-formatted configuration from a given file.
 pub fn load_from_file<P: AsRef<path::Path>>(config_path: P) -> anyhow::Result<Config> {
@@ -27,3 +74,25 @@ pub fn load_from_file<P: AsRef<path::Path>>(config_path: P) -> anyhow::Result<Co
 pub fn to_string(cfg: &Config) -> anyhow::Result<String> {
     toml::to_string_pretty(cfg).with_context(|| "Failed to serialize default configuration")
 }
+
+/// Serialization/deserialization
+mod log_level {
+    use serde::{self, Deserialize};
+    use std::str::FromStr;
+    use tracing::Level;
+
+    pub fn serialize<S>(value: &Level, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.serialize_str(value.to_string().as_str())
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Level, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        Level::from_str(s.as_str()).map_err(serde::de::Error::custom)
+    }
+}

src/effect.rs

@@ -0,0 +1,125 @@
+//! Effects subsystem.
+//!
+//! Effects describe things that the creator of the effect intends to happen,
+//! producing a value upon completion. They are, in fact, futures.
+//!
+//! A boxed, pinned future returning an event is called an effect and typed as an `Effect<Ev>`,
+//! where `Ev` is the event's type.
+//!
+//! ## Using effects
+//!
+//! To create an effect, an events factory is used that implements one or more of the factory
+//! traits of this module. For example, given an events factory `eff`, we can create a
+//! `set_timeout` future and turn it into an effect:
+//!
+//! ```
+//! # use std::time;
+//! use crate::effect::EffectExt;
+//!
+//! enum Event {
+//!     ThreeSecondsElapsed(time::Duration)
+//! }
+//!
+//! eff.set_timeout(time::Duration::from_secs(3))
+//!    .event(Event::ThreeSecondsElapsed)
+//! ```
+//!
+//! This example will produce an effect that, after three seconds, creates an
+//! `Event::ThreeSecondsElapsed`. Note that effects do nothing on their own, they need to be passed
+//! to the `Reactor` (see `reactor` module) to be executed.
+//!
+//! ## Chaining futures and effects
+//!
+//! Effects are built from futures, which can be combined before being finalized
+//! into an effect. However, only one effect can be created as the end result
+//! of such a chain.
+//!
+//! It is possible to create an effect from multiple effects being run in parallel using `.also`:
+//!
+//! ```
+//! # use std::time;
+//! use crate::effect::{EffectExt, EffectAlso};
+//!
+//! enum Event {
+//!     ThreeSecondsElapsed(time::Duration),
+//!     FiveSecondsElapsed(time::Duration),
+//! }
+//!
+//! // This effect produces a single event after five seconds:
+//! eff.set_timeout(time::Duration::from_secs(3))
+//!    .then(|_| eff.set_timeout(time::Duration::from_secs(2))
+//!    .event(Event::FiveSecondsElapsed);
+//!
+//! // Here, two effects are run in parallel, resulting in two events:
+//! eff.set_timeout(time::Duration::from_secs(3))
+//!    .event(Event::ThreeSecondsElapsed)
+//!    .also(eff.set_timeout(time::Duration::from_secs(5))
+//!             .event(Event::FiveSecondsElapsed));
+//! ```
+//!
+//! ## Arbitrary effects
+//!
+//! While it is technically possible to turn any future into an effect, it is advisable to only use
+//! the effects explicitly listed in this module through traits to create them. Post-processing on
+//! effects to turn them into events should also be kept brief.
+
+use crate::util::Multiple;
+use futures::future::BoxFuture;
+use futures::FutureExt;
+use smallvec::smallvec;
+use std::future::Future;
+use std::time;
+
+/// Effect type.
+///
+/// Effects are just boxed futures that produce one or more events.
+pub type Effect<Ev> = BoxFuture<'static, Multiple<Ev>>;
+
+/// Effect extension for futures.
+///
+/// Used to convert futures into actual effects.
+pub trait EffectExt: Future + Send {
+    /// Finalize a future into an effect that returns an event.
+    ///
+    /// The passed in function `f` is used to translate the resulting value from an effect into
+    fn event<U, F>(self, f: F) -> Multiple<Effect<U>>
+    where
+        F: FnOnce(Self::Output) -> U + 'static + Send,
+        U: 'static,
+        Self: Sized;
+
+    /// Finalize a future into an effect that runs but drops the result.
+    fn ignore<Ev>(self) -> Multiple<Effect<Ev>>;
+}
+
+impl<T: ?Sized> EffectExt for T
+where
+    T: Future + Send + 'static + Sized,
+{
+    fn event<U, F>(self, f: F) -> Multiple<Effect<U>>
+    where
+        F: FnOnce(Self::Output) -> U + 'static + Send,
+        U: 'static,
+    {
+        smallvec![self.map(f).map(|item| smallvec![item]).boxed()]
+    }
+
+    fn ignore<Ev>(self) -> Multiple<Effect<Ev>> {
+        smallvec![self.map(|_| Multiple::new()).boxed()]
+    }
+}
+
+/// Core effects.
+pub trait Core {
+    /// Do not do anything.
+    ///
+    /// Immediately completes, can be used to trigger an event.
+    fn immediately(self) -> BoxFuture<'static, ()>;
+
+    /// Set a timeout.
+    ///
+    /// Once the timeout fires, it will return the actual elapsed time since the execution (not
+    /// creation!) of this effect. Event loops typically execute effects right after a called event
+    /// handling function completes.
+    fn set_timeout(self, timeout: time::Duration) -> BoxFuture<'static, time::Duration>;
+}

src/main.rs

@@ -1,20 +1,23 @@
 //! # CasperLabs blockchain node
 //!
-//! This crate contain the core application for the CasperLabs blockchain. Run with `--help` to
-//! see available command-line arguments.
+//! This crate contain the core application for the CasperLabs blockchain. Run with `--help` to see
+//! available command-line arguments.
 //!
 //! ## Application structure
 //!
-//! While the `main` function is the central entrypoint for the node application, its core event
-//! loop is found inside the reactor. To get a tour of the sourcecode, be sure to run
-//! `cargo doc --open`.
+//! While the [`main`](fn.main.html) function is the central entrypoint for the node application,
+//! its core event loop is found inside the [reactor](reactor/index.html). To get a tour of the
+//! sourcecode, be sure to run `cargo doc --open`.
 
 mod cli;
 mod config;
+mod effect;
+mod reactor;
+mod util;
 
 use structopt::StructOpt;
 
-/// Parse command-line arguments and run application.
+/// Parse [command-line arguments](cli/index.html) and run application.
 #[tokio::main]
 pub async fn main() -> anyhow::Result<()> {
     // Parse CLI args and run selected subcommand.