Integrate client and pull from minio (#139)

* Integrate client to judger server

* Fix typo

* Prepare for problem package sync

* Sync package for client

* Perf package

* Perf minio config file name

judger/Cargo.toml

@@ -52,8 +52,3 @@ path ="src/cli/main.rs"
 [[bin]]
 name ="judger-server"
 path ="src/server/main.rs"
-
-[[bin]]
-name ="judger-client"
-path ="src/client/main.rs"
-

judger/README.md

@@ -31,7 +31,7 @@ cargo run --bin judger-cli -- batch-judge \
 
 ### How to run
 
-`cargo run --bin judger-server -- --env-path ./judge-server/src/environment/.env.development`
+`cargo run --bin judger-server -- --env-path ./judger/src/server/environment/.env.development`
 
 ### How to visit OpenAPI
 

judger/src/lib.rs

@@ -1,2 +1 @@
-#[allow(dead_code)]
 pub mod service;

judger/src/client/main.rs → judger/src/server/client/mod.rs

@@ -1,9 +1,7 @@
-mod client;
-mod environment;
-mod error;
-
-use client::HttpClient;
-use error::ClientError;
+mod httpclient;
+use crate::error::ClientError;
+use crate::service::state;
+use httpclient::HttpClient;
 use judge_core::judge;
 use judge_core::{
     compiler::Language,
@@ -12,6 +10,7 @@ use judge_core::{
     judge::JudgeConfig,
     package::PackageType,
 };
+use judger::service::package_manager::package;
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
 use std::{fs, path::PathBuf};
@@ -47,13 +46,9 @@ struct JudgeTask {
     redis_stream_id: String,
 }
 
-#[tokio::main]
-async fn main() {
-    let opt = environment::load_option();
-    environment::setup_logger();
-    let mut interval = interval(Duration::from_secs(10));
-    let base_url = opt.base_url;
-    let client = client::HttpClient::new(base_url);
+pub async fn run_client(base_url: String, interval_sec: u64) {
+    let mut interval = interval(Duration::from_secs(interval_sec));
+    let client = HttpClient::new(base_url);
 
     loop {
         interval.tick().await;
@@ -88,7 +83,9 @@ async fn pick_task(client: &HttpClient) -> Result<JudgeTask, ClientError> {
 
     match response.status() {
         reqwest::StatusCode::OK => Ok(response.json::<PickResponse>().await?.task),
-        _ => Err(ClientError::PickFail(anyhow::anyhow!("Queue is empty"))),
+        _ => Err(ClientError::InternalError(anyhow::anyhow!(
+            "Queue is empty"
+        ))),
     }
 }
 
@@ -117,12 +114,18 @@ async fn report_task(
             );
             Ok(())
         }
-        _ => Err(ClientError::ReportFail(anyhow::anyhow!("Report Failed"))),
+        _ => Err(ClientError::InternalError(anyhow::anyhow!("Report Failed"))),
     }
 }
 
 fn run_judge(task: JudgeTask) -> Result<Vec<JudgeResultInfo>, ClientError> {
-    let problem_package_dir = PathBuf::from("data/dev-problem-package");
+    if let Err(sync_err) = package::sync_package(&PathBuf::from("data"), "oj-lab-problem-package") {
+        return Err(ClientError::PackageError(sync_err));
+    };
+    if let Err(set_err) = state::set_busy() {
+        return Err(ClientError::InternalError(set_err));
+    }
+    let problem_package_dir = PathBuf::from(format!("data/{}", package::PACKAGE_SAVE_DIRNAME));
     let problem_slug = task.problem_slug;
     let uuid = uuid::Uuid::new_v4();
     let runtime_path = PathBuf::from("/tmp").join(uuid.to_string());
@@ -166,5 +169,6 @@ fn run_judge(task: JudgeTask) -> Result<Vec<JudgeResultInfo>, ClientError> {
     }
 
     log::debug!("Judge finished");
+    state::set_idle();
     Ok(results)
 }

judger/src/server/environment/.env.development

@@ -1,2 +1,4 @@
 PORT=8081
+BASE_URL="http://localhost:8080/api/v1/judge"
+INTERVAL=10
 RUST_LOG=DEBUG

judger/src/server/environment/mod.rs

@@ -14,6 +14,12 @@ pub struct JudgeServerOpt {
 
     #[structopt(long, default_value = "data/dev-problem-package")]
     pub problem_package_dir: PathBuf,
+
+    #[structopt(env = "BASE_URL", default_value = "http://localhost:8080/api/v1/judge")]
+    pub base_url: String,
+
+    #[structopt(env = "INTERVAL", default_value = "10")]
+    pub interval: i32,
 }
 
 pub fn load_option() -> JudgeServerOpt {

judger/src/server/error.rs

@@ -2,6 +2,7 @@
 
 use actix_web::{HttpResponse, ResponseError};
 use judge_core::error::JudgeCoreError;
+use judger::service::error::JudgeServiceError;
 
 #[derive(Debug, thiserror::Error)]
 pub enum ServiceError {
@@ -19,6 +20,12 @@ pub enum ServiceError {
     UnauthorizedWithMsg(anyhow::Error, String),
 }
 
+#[derive(Debug)]
+pub enum ClientError {
+    InternalError(anyhow::Error),
+    PackageError(JudgeServiceError),
+}
+
 #[derive(Serialize)]
 struct ServiceErrorBody {
     msg: Option<String>,
@@ -67,3 +74,15 @@ impl From<JudgeCoreError> for ServiceError {
         Self::InternalError(anyhow::anyhow!("{:?}", value))
     }
 }
+
+impl From<reqwest::Error> for ClientError {
+    fn from(value: reqwest::Error) -> Self {
+        Self::InternalError(anyhow::anyhow!("{:?}", value))
+    }
+}
+
+impl From<JudgeCoreError> for ClientError {
+    fn from(value: JudgeCoreError) -> Self {
+        Self::InternalError(anyhow::anyhow!("{:?}", value))
+    }
+}

judger/src/server/main.rs

@@ -1,3 +1,4 @@
+mod client;
 mod environment;
 mod error;
 mod service;
@@ -22,6 +23,10 @@ async fn main() -> std::io::Result<()> {
     //     }
     // });
 
+    tokio::spawn(async move {
+        client::run_client(opt.base_url, opt.interval as u64).await;
+    });
+
     let port = opt.port;
 
     HttpServer::new(move || {

judger/src/server/service/mod.rs

@@ -1,6 +1,6 @@
 mod greet;
 mod judge;
-mod state;
+pub mod state;
 
 use actix_web::web;
 use utoipa::OpenApi;

judger/src/service/error.rs

@@ -6,6 +6,7 @@ use judge_core::error::JudgeCoreError;
 pub enum JudgeServiceError {
     JudgeCoreError(JudgeCoreError),
     IOError(io::Error),
+    RcloneError(anyhow::Error),
     AnyhowError(anyhow::Error),
 }
 

judger/src/service/package_manager/description.rs

@@ -0,0 +1,100 @@
+use std::{
+    collections::HashMap,
+    fs,
+    path::{Path, PathBuf},
+};
+
+use judge_core::{error::JudgeCoreError, package::PackageType};
+use serde_derive::{Deserialize, Serialize};
+
+use crate::service::error::JudgeServiceError;
+
+pub const PACKAGES_DESCRIPTION_FILE_NAME: &str = "judge-pd.json";
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct PackageDescription {
+    pub name: String,
+    pub revision: u32,
+    pub package_type: PackageType,
+}
+
+impl PackageDescription {
+    pub fn new(name: String, package_type: PackageType) -> Result<Self, JudgeServiceError> {
+        Ok(Self {
+            name,
+            revision: 0,
+            package_type,
+        })
+    }
+}
+
+pub struct StoragedPackageDescriptionMap {
+    pub folder_path: PathBuf,
+    pub package_description_map: HashMap<String, PackageDescription>,
+}
+
+impl StoragedPackageDescriptionMap {
+    pub fn init(folder_path: PathBuf) -> Result<Self, JudgeServiceError> {
+        init_package_description_file(&folder_path)?;
+        let package_description_map = HashMap::new();
+        Ok(Self {
+            folder_path,
+            package_description_map,
+        })
+    }
+
+    pub fn load(folder_path: PathBuf) -> Result<Self, JudgeServiceError> {
+        let package_description_map = load_package_description_map(&folder_path)?;
+        Ok(Self {
+            folder_path,
+            package_description_map,
+        })
+    }
+
+    pub fn insert(
+        &mut self,
+        package_description: PackageDescription,
+    ) -> Result<(), JudgeCoreError> {
+        self.package_description_map
+            .insert(package_description.name.clone(), package_description);
+        update_package_description_file(&self.folder_path, &self.package_description_map)?;
+        Ok(())
+    }
+
+    pub fn get(&self, package_name: &str) -> Option<&PackageDescription> {
+        self.package_description_map.get(package_name)
+    }
+}
+
+fn load_package_description_map(
+    folder: &Path,
+) -> Result<HashMap<String, PackageDescription>, JudgeCoreError> {
+    let description_file_content = fs::read_to_string(folder.join(PACKAGES_DESCRIPTION_FILE_NAME))?;
+    let package_description_map: HashMap<String, PackageDescription> =
+        serde_json::from_str(&description_file_content)?;
+    Ok(package_description_map)
+}
+
+fn init_package_description_file(folder: &Path) -> Result<(), JudgeCoreError> {
+    let package_description_map = HashMap::<String, PackageDescription>::new();
+    let package_description_file_content = serde_json::to_string_pretty(&package_description_map)?;
+
+    fs::write(
+        folder.join(PACKAGES_DESCRIPTION_FILE_NAME),
+        package_description_file_content,
+    )?;
+    Ok(())
+}
+
+fn update_package_description_file(
+    folder: &Path,
+    package_description_map: &HashMap<String, PackageDescription>,
+) -> Result<(), JudgeCoreError> {
+    let package_description_file_content = serde_json::to_string_pretty(package_description_map)?;
+
+    fs::write(
+        folder.join(PACKAGES_DESCRIPTION_FILE_NAME),
+        package_description_file_content,
+    )?;
+    Ok(())
+}