minerd: programmatic control daemon

bin/minerd/src/lib.rs

@@ -0,0 +1,223 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2024 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::{collections::HashSet, sync::Arc};
+
+use log::{error, info};
+use smol::{
+    channel::{Receiver, Sender},
+    lock::Mutex,
+};
+use url::Url;
+
+use darkfi::{
+    rpc::server::{listen_and_serve, RequestHandler},
+    system::{ExecutorPtr, StoppableTask, StoppableTaskPtr},
+    Error, Result,
+};
+
+/// Daemon error codes
+mod error;
+
+/// JSON-RPC server methods
+mod rpc;
+
+/// Atomic pointer to the DarkFi mining node
+pub type MinerNodePtr = Arc<MinerNode>;
+
+/// Structure representing a DarkFi mining node
+pub struct MinerNode {
+    /// PoW miner number of threads to use
+    threads: usize,
+    /// Sender to stop miner threads
+    sender: Sender<()>,
+    /// Receiver to stop miner threads
+    stop_signal: Receiver<()>,
+    /// JSON-RPC connection tracker
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+}
+
+impl MinerNode {
+    pub fn new(threads: usize, sender: Sender<()>, stop_signal: Receiver<()>) -> MinerNodePtr {
+        Arc::new(Self { threads, sender, stop_signal, rpc_connections: Mutex::new(HashSet::new()) })
+    }
+}
+
+/// Atomic pointer to the DarkFi mining daemon
+pub type MinerdPtr = Arc<Minerd>;
+
+/// Structure representing a DarkFi mining daemon
+pub struct Minerd {
+    /// Miner node instance conducting the mining operations
+    node: MinerNodePtr,
+    /// JSON-RPC background task
+    rpc_task: StoppableTaskPtr,
+}
+
+impl Minerd {
+    /// Initialize a DarkFi mining daemon.
+    ///
+    /// Corresponding communication channels are setup to generate a new `MinerNode`,
+    /// and a new task is generated to handle the JSON-RPC API.
+    pub fn init(threads: usize) -> MinerdPtr {
+        info!(target: "minerd::Minerd::init", "Initializing a new mining daemon...");
+
+        // Initialize the smol channels to send signal between the threads
+        let (sender, stop_signal) = smol::channel::bounded(1);
+
+        // Generate the node
+        let node = MinerNode::new(threads, sender, stop_signal);
+
+        // Generate the JSON-RPC task
+        let rpc_task = StoppableTask::new();
+
+        info!(target: "minerd::Minerd::init", "Mining daemon initialized successfully!");
+
+        Arc::new(Self { node, rpc_task })
+    }
+
+    /// Start the DarkFi mining daemon in the given executor, using the provided JSON-RPC listen url.
+    pub fn start(&self, executor: &ExecutorPtr, rpc_listen: &Url) {
+        info!(target: "minerd::Minerd::start", "Starting mining daemon...");
+
+        // Start the JSON-RPC task
+        let node_ = self.node.clone();
+        self.rpc_task.clone().start(
+            listen_and_serve(rpc_listen.clone(), self.node.clone(), None, executor.clone()),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::RpcServerStopped) => node_.stop_connections().await,
+                    Err(e) => error!(target: "minerd::Minerd::start", "Failed stopping JSON-RPC server: {}", e),
+                }
+            },
+            Error::RpcServerStopped,
+            executor.clone(),
+        );
+
+        info!(target: "minerd::Minerd::start", "Mining daemon started successfully!");
+    }
+
+    /// Stop the DarkFi mining daemon.
+    pub async fn stop(&self) -> Result<()> {
+        info!(target: "minerd::Minerd::stop", "Terminating mining daemon...");
+
+        // Stop the mining node
+        info!(target: "minerd::Minerd::stop", "Stopping miner threads...");
+        self.node.sender.send(()).await?;
+
+        // Stop the JSON-RPC task
+        info!(target: "minerd::Minerd::stop", "Stopping JSON-RPC server...");
+        self.rpc_task.stop().await;
+
+        // Consume channel item so its empty again
+        if self.node.stop_signal.is_full() {
+            self.node.stop_signal.recv().await?;
+        }
+
+        info!(target: "minerd::Minerd::stop", "Mining daemon terminated successfully!");
+        Ok(())
+    }
+}
+
+#[test]
+/// Test the programmatic control of `Minerd`.
+///
+/// First we initialize a daemon, start it and then perform
+/// couple of restarts to verify everything works as expected.
+fn minerd_programmatic_control() -> Result<()> {
+    // Initialize logger
+    let mut cfg = simplelog::ConfigBuilder::new();
+
+    // We check this error so we can execute same file tests in parallel,
+    // otherwise second one fails to init logger here.
+    if simplelog::TermLogger::init(
+        simplelog::LevelFilter::Info,
+        //simplelog::LevelFilter::Debug,
+        //simplelog::LevelFilter::Trace,
+        cfg.build(),
+        simplelog::TerminalMode::Mixed,
+        simplelog::ColorChoice::Auto,
+    )
+    .is_err()
+    {
+        log::debug!(target: "minerd_programmatic_control", "Logger initialized");
+    }
+
+    // Daemon configuration
+    let threads = 4;
+    let rpc_listen = Url::parse("tcp://127.0.0.1:28467")?;
+
+    // Create an executor and communication signals
+    let ex = Arc::new(smol::Executor::new());
+    let (signal, shutdown) = smol::channel::unbounded::<()>();
+
+    // Generate a dummy mining job
+    let target = darkfi::rpc::util::JsonValue::String(
+        num_bigint::BigUint::from_bytes_be(&[0xFF; 32]).to_string(),
+    );
+    let block = darkfi::rpc::util::JsonValue::String(darkfi::util::encoding::base64::encode(
+        &darkfi_serial::serialize(&darkfi::blockchain::BlockInfo::default()),
+    ));
+    let mining_job = darkfi::rpc::jsonrpc::JsonRequest::new(
+        "mine",
+        darkfi::rpc::util::JsonValue::Array(vec![target, block]),
+    );
+
+    easy_parallel::Parallel::new()
+        .each(0..threads, |_| smol::block_on(ex.run(shutdown.recv())))
+        .finish(|| {
+            smol::block_on(async {
+                // Initialize a daemon
+                let daemon = Minerd::init(threads);
+
+                // Start it
+                daemon.start(&ex, &rpc_listen);
+
+                // Generate a JSON-RPC client to send mining jobs
+                let rpc_client =
+                    darkfi::rpc::client::RpcClient::new(rpc_listen.clone(), ex.clone())
+                        .await
+                        .unwrap();
+
+                // Send a mining job but stop the daemon after it starts mining
+                smol::future::or(
+                    async {
+                        rpc_client.request(mining_job).await.unwrap();
+                    },
+                    async {
+                        // Wait node to start mining
+                        darkfi::system::sleep(2).await;
+                        daemon.stop().await.unwrap();
+                    },
+                )
+                .await;
+                rpc_client.stop().await;
+
+                // Start it again
+                daemon.start(&ex, &rpc_listen);
+
+                // Stop it
+                daemon.stop().await.unwrap();
+
+                // Shutdown entirely
+                drop(signal);
+            })
+        });
+
+    Ok(())
+}

bin/minerd/src/main.rs

@@ -16,34 +16,20 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashSet, sync::Arc};
+use std::sync::Arc;
 
-use log::{error, info};
-use smol::{
-    channel::{Receiver, Sender},
-    lock::Mutex,
-    stream::StreamExt,
-    Executor,
-};
+use log::info;
+use smol::{stream::StreamExt, Executor};
 use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
 use url::Url;
 
-use darkfi::{
-    async_daemonize, cli_desc,
-    rpc::server::{listen_and_serve, RequestHandler},
-    system::{StoppableTask, StoppableTaskPtr},
-    Error, Result,
-};
+use darkfi::{async_daemonize, cli_desc, Result};
+
+use minerd::Minerd;
 
 const CONFIG_FILE: &str = "minerd.toml";
 const CONFIG_FILE_CONTENTS: &str = include_str!("../minerd.toml");
 
-/// Daemon error codes
-mod error;
-
-/// JSON-RPC server methods
-mod rpc;
-
 #[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
 #[serde(default)]
 #[structopt(name = "minerd", about = cli_desc!())]
@@ -69,55 +55,18 @@ struct Args {
     verbose: u8,
 }
 
-/// Daemon structure
-pub struct Minerd {
-    /// PoW miner number of threads to use
-    threads: usize,
-    // Sender to stop miner threads
-    sender: Sender<()>,
-    // Receiver to stop miner threads
-    stop_signal: Receiver<()>,
-    /// JSON-RPC connection tracker
-    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
-}
-
-impl Minerd {
-    pub fn new(threads: usize, sender: Sender<()>, stop_signal: Receiver<()>) -> Self {
-        Self { threads, sender, stop_signal, rpc_connections: Mutex::new(HashSet::new()) }
-    }
-}
-
 async_daemonize!(realmain);
 async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "minerd", "Starting DarkFi Mining Daemon...");
-    let (sender, recvr) = smol::channel::bounded(1);
-    let minerd = Arc::new(Minerd::new(args.threads, sender.clone(), recvr));
-
-    info!(target: "minerd", "Starting JSON-RPC server on {}", args.rpc_listen);
-    let minerd_ = Arc::clone(&minerd);
-    let rpc_task = StoppableTask::new();
-    rpc_task.clone().start(
-        listen_and_serve(args.rpc_listen, minerd.clone(), None, ex.clone()),
-        |res| async move {
-            match res {
-                Ok(()) | Err(Error::RpcServerStopped) => minerd_.stop_connections().await,
-                Err(e) => error!(target: "minerd", "Failed stopping JSON-RPC server: {}", e),
-            }
-        },
-        Error::RpcServerStopped,
-        ex.clone(),
-    );
+    let daemon = Minerd::init(args.threads);
+    daemon.start(&ex, &args.rpc_listen);
 
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new(ex)?;
     signals_handler.wait_termination(signals_task).await?;
     info!(target: "minerd", "Caught termination signal, cleaning up and exiting");
 
-    info!(target: "minerd", "Stopping miner threads...");
-    sender.send(()).await?;
-
-    info!(target: "minerd", "Stopping JSON-RPC server...");
-    rpc_task.stop().await;
+    daemon.stop().await?;
 
     info!(target: "minerd", "Shut down successfully");
     Ok(())

bin/minerd/src/rpc.rs

@@ -38,11 +38,11 @@ use darkfi_serial::{async_trait, deserialize_async};
 
 use crate::{
     error::{server_error, RpcError},
-    Minerd,
+    MinerNode,
 };
 
 #[async_trait]
-impl RequestHandler for Minerd {
+impl RequestHandler for MinerNode {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         debug!(target: "minerd::rpc", "--> {}", req.stringify().unwrap());
 
@@ -59,7 +59,7 @@ impl RequestHandler for Minerd {
     }
 }
 
-impl Minerd {
+impl MinerNode {
     // RPCAPI:
     // Signals miner daemon to abort mining pending request.
     // Returns `true` on success.