Add batch run and streammed batch run

Cargo.toml

@@ -14,6 +14,7 @@ A code runner library for online judge system
 bon = "3.8.1"
 byte-unit = "5.2.0"
 cgroups-rs = "0.5.0"
+futures-lite = "2.6.1"
 serde = { version = "1.0.228", features = ["derive"], optional = true }
 state-shift = "2.1.1"
 tokio = { version = "1.48.0", features = ["full"] }

src/judge.rs

@@ -1,14 +1,16 @@
 use std::{env, io, marker::PhantomData, path::PathBuf, process::Stdio, time::Duration};
 
 use bon::bon;
+use byte_unit::Byte;
+use futures_lite::{Stream, StreamExt};
 use state_shift::{impl_state, type_state};
 use tokio::{
     fs,
     io::{AsyncReadExt, AsyncWriteExt},
 };
 use uuid::Uuid;
 
-use crate::{Language, Metrics, Resource, Sandbox, Verdict};
+use crate::{AggregatedMetrics, Language, Metrics, Resource, Sandbox, Verdict};
 
 const MAIN: &str = "main";
 const CHECKER: &str = "checker";
@@ -203,4 +205,70 @@ impl Judge {
             memory_usage,
         })
     }
+
+    #[require(Compiled)]
+    pub async fn batch_run(
+        &self,
+        inputs: impl Iterator<Item = &[u8]>,
+    ) -> io::Result<AggregatedMetrics> {
+        let mut verdict = Verdict::Accepted;
+        let mut total_run_time = Duration::ZERO;
+        let mut total_memory_usage = Byte::default();
+        let mut count = 0;
+
+        // running sequentially to enable early exit, saving resources
+        for input in inputs {
+            let metrics = self.run(input).await?;
+            total_run_time += metrics.run_time;
+            total_memory_usage = total_memory_usage
+                .add(metrics.memory_usage)
+                .expect("memory usage should not overflow u32");
+            count += 1;
+            if metrics.verdict != Verdict::Accepted {
+                verdict = metrics.verdict;
+                break;
+            }
+        }
+
+        Ok(AggregatedMetrics {
+            verdict,
+            average_run_time: total_run_time / count,
+            average_memory_usage: total_memory_usage
+                .divide(count as usize)
+                .expect("count must be greater than 0"),
+        })
+    }
+
+    #[require(Compiled)]
+    pub async fn streamed_batch_run(
+        &self,
+        mut inputs: impl Stream<Item = &[u8]> + std::marker::Unpin,
+    ) -> io::Result<AggregatedMetrics> {
+        let mut verdict = Verdict::Accepted;
+        let mut total_run_time = Duration::ZERO;
+        let mut total_memory_usage = Byte::default();
+        let mut count = 0;
+
+        // running sequentially to enable early exit, saving resources
+        while let Some(input) = inputs.next().await {
+            let metrics = self.run(input).await?;
+            total_run_time += metrics.run_time;
+            total_memory_usage = total_memory_usage
+                .add(metrics.memory_usage)
+                .expect("memory usage should not overflow u32");
+            count += 1;
+            if metrics.verdict != Verdict::Accepted {
+                verdict = metrics.verdict;
+                break;
+            }
+        }
+
+        Ok(AggregatedMetrics {
+            verdict,
+            average_run_time: total_run_time / count,
+            average_memory_usage: total_memory_usage
+                .divide(count as usize)
+                .expect("count must be greater than 0"),
+        })
+    }
 }

src/metrics.rs

@@ -15,11 +15,18 @@ pub enum Verdict {
 }
 
 #[derive(Debug, Clone)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize))]
 pub struct Metrics {
     pub verdict: Verdict,
     pub run_time: Duration,
     pub memory_usage: Byte,
     pub stdout: Vec<u8>,
     pub stderr: Vec<u8>,
 }
+
+#[derive(Debug, Clone)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize))]
+pub struct AggregatedMetrics {
+    pub verdict: Verdict,
+    pub average_run_time: Duration,
+    pub average_memory_usage: Byte,
+}