From 815bc126a738123d67fe40a207f9c105580b801c Mon Sep 17 00:00:00 2001
From: weisd <weishidavip@163.com>
Date: Tue, 23 Jul 2024 11:25:20 +0800
Subject: [PATCH] todo:ec decode

---
 ecstore/src/disk.rs     | 21 +++++++++++++++++++--
 ecstore/src/disk_api.rs |  1 +
 ecstore/src/erasure.rs  | 32 ++++++++++++++++++++++++++++----
 3 files changed, 48 insertions(+), 6 deletions(-)

diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs
index 4feeb57ca..96b7fc996 100644
--- a/ecstore/src/disk.rs
+++ b/ecstore/src/disk.rs
@@ -1,5 +1,6 @@
 use std::{
     fs::Metadata,
+    io::SeekFrom,
     path::{Path, PathBuf},
     sync::Arc,
 };
@@ -9,8 +10,11 @@ use bytes::Bytes;
 use futures::future::join_all;
 use path_absolutize::Absolutize;
 use time::OffsetDateTime;
-use tokio::fs::{self, File};
-use tokio::io::ErrorKind;
+use tokio::io::{AsyncReadExt, ErrorKind};
+use tokio::{
+    fs::{self, File},
+    io::AsyncSeekExt,
+};
 use tracing::{debug, warn};
 use uuid::Uuid;
 
@@ -501,7 +505,20 @@ impl DiskAPI for LocalDisk {
 
         // Ok(())
     }
+    async fn read_file(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<(Vec<u8>, usize)> {
+        let p = self.get_object_path(&volume, &path)?;
+        let mut file = File::options().read(true).open(&p).await?;
 
+        file.seek(SeekFrom::Start(offset as u64)).await?;
+
+        let mut buffer = vec![0; length];
+
+        let bytes_read = file.read(&mut buffer).await?;
+
+        buffer.truncate(bytes_read);
+
+        Ok((buffer, bytes_read))
+    }
     async fn rename_data(
         &self,
         src_volume: &str,
diff --git a/ecstore/src/disk_api.rs b/ecstore/src/disk_api.rs
index 0f438f3a5..69b55498e 100644
--- a/ecstore/src/disk_api.rs
+++ b/ecstore/src/disk_api.rs
@@ -19,6 +19,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>;
     async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result<FileWriter>;
     async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter>;
+    async fn read_file(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<(Vec<u8>, usize)>;
     async fn rename_data(
         &self,
         src_volume: &str,
diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs
index 904b9c433..fb82cc8b1 100644
--- a/ecstore/src/erasure.rs
+++ b/ecstore/src/erasure.rs
@@ -15,14 +15,16 @@ pub struct Erasure {
     // data_shards: usize,
     // parity_shards: usize,
     encoder: ReedSolomon,
+    block_size: usize,
     id: Uuid,
 }
 
 impl Erasure {
-    pub fn new(data_shards: usize, parity_shards: usize) -> Self {
+    pub fn new(data_shards: usize, parity_shards: usize, block_size: usize) -> Self {
         Erasure {
             // data_shards,
             // parity_shards,
+            block_size,
             encoder: ReedSolomon::new(data_shards, parity_shards).unwrap(),
             id: Uuid::new_v4(),
         }
@@ -32,7 +34,7 @@ impl Erasure {
         &self,
         body: S,
         writers: &mut Vec<W>,
-        block_size: usize,
+        // block_size: usize,
         data_size: usize,
         _write_quorum: usize,
     ) -> Result<usize>
@@ -40,7 +42,7 @@ impl Erasure {
         S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'static,
         W: AsyncWrite + Unpin,
     {
-        let mut stream = ChunkedStream::new(body, data_size, block_size, true);
+        let mut stream = ChunkedStream::new(body, data_size, self.block_size, true);
         let mut total: usize = 0;
         let mut idx = 0;
         while let Some(result) = stream.next().await {
@@ -93,6 +95,14 @@ impl Erasure {
         // }
     }
 
+    pub async fn decode<R, W>(&self, writer: W, readers: Vec<R>, offset: usize, length: usize, total_length: usize) -> Result<()>
+    where
+        R: ReadAt,
+        W: AsyncWrite + Unpin,
+    {
+        unimplemented!()
+    }
+
     pub fn encode_data(&self, data: &[u8]) -> Result<Vec<Vec<u8>>> {
         let (shard_size, total_size) = self.need_size(data.len());
 
@@ -137,6 +147,20 @@ impl Erasure {
     }
 }
 
+pub trait ReadAt {
+    async fn read_at(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<(Vec<u8>, usize)>;
+}
+
+pub struct ShardReader<R> {
+    readers: Vec<R>,
+}
+
+impl<R: ReadAt> ShardReader<R> {
+    pub fn new(readers: Vec<R>, offset: usize, total_length: usize) -> Self {
+        Self { readers }
+    }
+}
+
 // fn shards_to_option_shards<T: Clone>(shards: &[Vec<T>]) -> Vec<Option<Vec<T>>> {
 //     let mut result = Vec::with_capacity(shards.len());
 
@@ -157,7 +181,7 @@ mod test {
         let data_shards = 3;
         let parity_shards = 2;
         let data: &[u8] = &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11];
-        let ec = Erasure::new(data_shards, parity_shards);
+        let ec = Erasure::new(data_shards, parity_shards, 1);
         let shards = ec.encode_data(data).unwrap();
         println!("shards:{:?}", shards);