made raft more testable

moved main function to tokio instead axtic

Author: Klaven

cli/Cargo.toml

@@ -10,4 +10,5 @@ reqwest = { version = "0.11", features = ["json"] }
 tokio = { version = "1", features = ["full"] }
 clap = { version = "3.0", features = ["derive"] }
 uuid = { version = "0.8.2", features = ["serde", "v4"] }
-rand = "0.8.5"
+rand = "0.8.5"
+futures = "0.3.30"

cli/src/main.rs

@@ -1,9 +1,11 @@
 use clap::{Parser, Subcommand};
-use rand::distributions::{Alphanumeric};
+use futures::stream::{self, StreamExt};
+use rand::distributions::Alphanumeric;
 use rand::{thread_rng, Rng};
 use std::convert::TryInto;
 use std::{thread, time};
-
+use tokio::io::AsyncRead;
+use tokio::sync::SemaphorePermit;
 mod null_client;
 
 #[derive(Parser)]
@@ -55,6 +57,13 @@ enum Commands {
         host: String,
     },
 
+    BenchDisk {
+        #[clap(long, default_value = "localhost")]
+        host: String,
+        #[clap(long, default_value = "8001")]
+        port: String,
+    },
+
     Compact {
         #[clap(long, default_value = "localhost")]
         host: String,
@@ -79,9 +88,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args = Cli::parse();
 
     match &args.command {
-        Commands::Put { key, value, host, port } => {
+        Commands::Put {
+            key,
+            value,
+            host,
+            port,
+        } => {
             println!("putting data {}", value);
-            println!("connecting at: {}",format!("http://{}:{}/{}/{}", host, port, "v1/data", key));
+            println!(
+                "connecting at: {}",
+                format!("http://{}:{}/{}/{}", host, port, "v1/data", key)
+            );
             let client = reqwest::Client::new();
             let data = value.clone();
             let resp = client
@@ -111,7 +128,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         Commands::Get { key, host, port } => {
             let then = time::Instant::now();
             println!("getting data for key {}", key);
-            let resp = reqwest::get(format!("http://{}:{}/{}/{}\n", host,port, "v1/data", key))
+            let resp = reqwest::get(format!("http://{}:{}/{}/{}\n", host, port, "v1/data", key))
                 .await?
                 .text()
                 .await?;
@@ -142,11 +159,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             benchmark(*records, *duration, host.to_string()).await;
         }
 
+        Commands::BenchDisk { host, port } => {
+            println!("benchmarking disk");
+            load_disks(1_000, 32, host.to_string(), port.to_string()).await;
+        }
+
         Commands::Compact { host, port } => {
             println!("Making a compation request!");
             let _resp = reqwest::get(format!(
                 "http://{}:{}/{}\n",
-                host,port, "v1/management/compact"
+                host, port, "v1/management/compact"
             ))
             .await?
             .text()
@@ -202,6 +224,23 @@ async fn benchmark(records: i32, duration: i32, host: String) -> Option<()> {
     return Some(());
 }
 
+async fn load_disks(records: u32, threads: u32, host: String, port: String) {
+    let client = null_client::NullClient::new(format!("http://{host}:{port}/v1/data").to_string());
+
+    let _ = futures::stream::iter(0..records)
+        .map(|_| {
+            let c = client.clone();
+            tokio::spawn(async move {
+                let _ = c.post(
+                "testdata".to_string(),
+                "This is the value of our record, it's awesome, not super short but kind of short."
+                    .to_string()).await;
+            })
+        })
+        .buffered(threads as usize)
+        .count()
+        .await;
+}
 fn get_random_string(length: usize) -> String {
     let chars: Vec<u8> = rand::thread_rng()
         .sample_iter(&Alphanumeric)

cli/src/null_client.rs

@@ -1,5 +1,6 @@
 use reqwest::{Error, Response};
 
+#[derive(Debug, Clone)]
 pub struct NullClient {
     url: String,
     client: reqwest::Client,

database/Cargo.toml

@@ -22,6 +22,7 @@ serde = { version = "1.0.130", features = ["derive"] }
 serde_json = "1.0.70"
 tempfile = "3.8.0"
 tokio = { version = "1.26.0", features = ["full"] }
+tokio-util = "0.7.11"
 tonic = { version = "0.10.2" }
 
 

database/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.74.1 as builder
+FROM rust:1.79.0 as builder
 #! /bin/bash
 RUN apt update
 RUN apt install -y protobuf-compiler

database/compose.yaml

@@ -1,45 +1,34 @@
 services:
   node1:
     build: .
-    command: 'null-db --roster=node2:3001,node3:3002 --id=3000 --encoding=json'
+    command: "null-db --roster=node1:3001,node2:3002,node3:3003 --id=node1:3001 --encoding=json"
     environment:
       - RUST_LOG=info
     ports:
-      - 3000:3000
+      - 3001:3001
       - 30001:8080
     volumes:
       - ./node1:/db
-    develop:
-      watch:
-        - action: rebuild
-          path: . 
 
   node2:
     build: .
-    command: 'null-db --roster=node1:3000,node3:3002 --id=3001 --encoding=json'
+    command: "null-db --roster=node1:3001,node2:3002,node3:3003 --id=node2:3002 --encoding=json"
     environment:
       - RUST_LOG=info
     ports:
-      - 3001:3001
-      - 30011:8080
+      - 3002:3002
+      - 30002:8080
     volumes:
       - ./node2:/db
-    develop:
-      watch:
-        - action: rebuild
-          path: . 
 
   node3:
     build: .
-    command: 'null-db --roster=node2:3001,node1:3000 --id=3002 --encoding=json'
+    command: "null-db --roster=node1:3001,node2:3002,node3:3003 --id=node3:3003 --encoding=json"
     environment:
       - RUST_LOG=info
     ports:
-      - 3002:3002
-      - 30021:8080
+      - 3003:3003
+      - 30003:8080
     volumes:
       - ./node3:/db
     develop:
-      watch:
-        - action: rebuild
-          path: . 

database/hack/reset.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+# Reset the database
+# Usage: ./reset.sh
+# Author: Arun Prakash Jana
+# Date: 26th May 2020
+# Version: 1.0
+# Description: This script will reset the database
+
+# remove all files from the node directories
+rm -rf node1/*
+rm -rf node2/*
+rm -rf node3/*

database/hack/run_3.sh

@@ -0,0 +1,13 @@
+#! /bin/bash
+# Run 3 Null DB's with 3 different seeds
+
+export RUST_LOG=info
+
+#node 1
+cargo run -- --roster=localhost:3002,localhost:3003 --id=3001 --port=8001 --encoding=json --dir=node1/ &
+#node 2
+cargo run -- --roster=localhost:3001,localhost:3003 --id=3002 --port=8002 --encoding=json --dir=node2/ &
+#node 3
+cargo run -- --roster=localhost:3001,localhost:3002 --id=3003 --port=8003 --encoding=json --dir=node3/ &
+
+wait

database/src/file_compactor.rs

@@ -1,7 +1,7 @@
 use super::file::record;
 use super::utils;
 use crate::index::generate_index_for_segment;
-use crate::nulldb::NullDB;
+use crate::nulldb::{DatabaseLog, NullDB};
 use actix_web::web::Data;
 use anyhow::anyhow;
 use std::collections::HashMap;

database/src/main.rs

@@ -7,10 +7,14 @@ use actix_web::{
 use clap::Parser;
 use errors::NullDbReadError;
 use file_reader::EasyReader;
-use nulldb::{Config, NullDB};
+use nulldb::{Config, DatabaseLog, NullDB};
 use raft::grpcserver::RaftEvent;
-use std::path::PathBuf;
-use tokio::sync::mpsc::Sender;
+use std::{path::PathBuf, time::Duration};
+use tokio::{
+    signal::unix::{signal, SignalKind},
+    sync::mpsc::Sender,
+};
+use tokio_util::sync::CancellationToken;
 
 mod errors;
 mod file;
@@ -25,20 +29,30 @@ mod utils;
 #[clap(author, version, about, long_about = None)]
 struct Args {
     #[clap(short, long)]
+    #[arg(default_value = "false")]
     compaction: bool,
+
     #[clap(short, long)]
     #[arg(default_value=get_work_dir().into_os_string())]
     dir: PathBuf,
+
     #[clap(short, long)]
-    roster: String,
+    roster: Option<String>,
+
     #[clap(short, long)]
+    #[arg(default_value = "localhost:3000")]
     id: String,
+
     #[clap(short, long)]
     #[arg(default_value = "html")]
     encoding: String,
+
+    #[clap(short, long)]
+    #[arg(default_value = "8080")]
+    port: u16,
 }
 
-#[actix_web::main]
+#[tokio::main]
 async fn main() -> Result<(), std::io::Error> {
     env_logger::init();
     let Args {
@@ -47,24 +61,58 @@ async fn main() -> Result<(), std::io::Error> {
         roster,
         id,
         encoding,
+        port,
     } = Args::parse();
 
-    let nodes = roster.split(",").map(Into::into).collect::<Vec<String>>();
+    let nodes = parse_roster(roster);
 
     let (sender, receiver) = tokio::sync::mpsc::channel(1000);
     let config = Config::new(dir, compaction, encoding);
     let raft_config = raft::config::RaftConfig {
         roster: nodes,
         candidate_id: id,
     };
+
+    let mut sigterm = signal(SignalKind::terminate()).unwrap();
+    let mut sigquit = signal(SignalKind::quit()).unwrap();
+    let mut sigint = signal(SignalKind::interrupt()).unwrap();
+    let cancel = CancellationToken::new();
+    let cancel_clone = cancel.clone();
+
+    let tx = sender.clone();
+    let sender_ark = Data::new(sender);
+
     let db_mutex = create_db(config).expect("could not start db");
     let mut raft = raft::RaftNode::new(raft_config, receiver, db_mutex);
-    let tx = sender.clone();
+    let db_thread = tokio::spawn(async move {
+        let _ = raft.run(tx, cancel_clone).await;
+    });
+    println!("listening for signals");
     tokio::spawn(async move {
-        let _ = raft.run(tx).await;
+        tokio::select! {
+            _ = sigterm.recv() => {
+                println!("Received SIGTERM");
+                cancel.cancel();
+                db_thread.await.unwrap();
+            }
+            _ = sigint.recv() => {
+                println!("Received SIGINT");
+                cancel.cancel();
+                db_thread.await.unwrap();
+            }
+            _ = sigquit.recv() => {
+                println!("Received SIGQUIT");
+                cancel.cancel();
+                db_thread.await.unwrap();
+            }
+            _ = tokio::signal::ctrl_c() => {
+                println!("Received ctrl_c");
+                cancel.cancel();
+                db_thread.await.unwrap();
+            }
+        }
     });
 
-    let sender_ark = Data::new(sender);
     println!("starting web server");
     HttpServer::new(move || {
         App::new()
@@ -75,9 +123,17 @@ async fn main() -> Result<(), std::io::Error> {
             .service(compact_data)
             .service(get_index)
     })
-    .bind("0.0.0.0:8080")?
+    .bind(format!("0.0.0.0:{port}"))?
     .run()
-    .await
+    .await;
+    Ok(())
+}
+
+fn parse_roster(roster: Option<String>) -> Option<Vec<String>> {
+    match roster {
+        Some(r) => Some(r.split(",").map(Into::into).collect::<Vec<String>>()),
+        None => None,
+    }
 }
 
 fn get_work_dir() -> PathBuf {
@@ -116,7 +172,6 @@ pub async fn put_value_for_key(
     key: web::Path<String>,
     req_body: String,
 ) -> impl Responder {
-    println!("putting data {req_body}");
     let (tx, receiver) = tokio::sync::oneshot::channel();
     let event = RaftEvent::NewEntry {
         key: key.into_inner(),
@@ -185,7 +240,7 @@ mod tests {
             let tmp_dir = TempDir::new().expect("could not get temp dir");
             let _workdir = setup_base_data(tmp_dir.path(), cargo_path);
 
-            let config = Config::new(tmp_dir.into_path(), false);
+            let config = Config::new(tmp_dir.into_path(), false, "html".to_string());
             let db = create_db(config).expect("could not start database");
 
             let result = db.get_value_for_key("name").expect("should retrive value");