From 43bc4db87c3106fab09024f5d818770e06170694 Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 6 Dec 2024 13:50:42 +0800 Subject: [PATCH] metacache done --- ecstore/src/disk/mod.rs | 2 +- ecstore/src/io.rs | 175 +++++++++++++++++++++++++++++++- ecstore/src/metacache/writer.rs | 97 ++++++++++-------- 3 files changed, 228 insertions(+), 46 deletions(-) diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 56d975414..5fec3bc29 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -594,7 +594,7 @@ pub struct MetadataResolutionParams { pub candidates: Vec>, } -#[derive(Clone, Debug, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)] pub struct MetaCacheEntry { // name is the full name of the object including prefixes pub name: String, diff --git a/ecstore/src/io.rs b/ecstore/src/io.rs index ca2c3904e..cdaabaebc 100644 --- a/ecstore/src/io.rs +++ b/ecstore/src/io.rs @@ -1,3 +1,5 @@ +use std::io::Read; +use std::io::Write; use std::pin::Pin; use std::task::{Context, Poll}; use tokio::fs::File; @@ -62,7 +64,174 @@ impl AsyncWrite for Writer { } } -// #[tokio::test] -// async fn test_reader{ +pub struct AsyncToSync { + inner: R, +} -// } +impl AsyncToSync { + pub fn new_reader(inner: R) -> Self { + Self { inner } + } + fn read_async(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll> { + let mut read_buf = ReadBuf::new(buf); + // Poll the underlying AsyncRead to fill the ReadBuf + match Pin::new(&mut self.inner).poll_read(cx, &mut read_buf) { + Poll::Ready(Ok(())) => Poll::Ready(Ok(read_buf.filled().len())), + Poll::Ready(Err(e)) => Poll::Ready(Err(e)), + Poll::Pending => Poll::Pending, + } + } +} + +impl AsyncToSync { + pub fn new_writer(inner: R) -> Self { + Self { inner } + } + // This function will perform a write using AsyncWrite + fn write_async(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let result = Pin::new(&mut self.inner).poll_write(cx, buf); + match result { + Poll::Ready(Ok(n)) => Poll::Ready(Ok(n)), + Poll::Ready(Err(e)) => Poll::Ready(Err(e)), + Poll::Pending => Poll::Pending, + } + } + + // This function will perform a flush using AsyncWrite + fn flush_async(&mut self, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } +} + +impl Read for AsyncToSync { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref()); + loop { + match self.read_async(&mut cx, buf) { + Poll::Ready(Ok(n)) => return Ok(n), + Poll::Ready(Err(e)) => return Err(e), + Poll::Pending => { + // If Pending, we need to wait for the readiness. + // Here, we can use an arbitrary mechanism to yield control, + // this might be blocking until some readiness occurs can be complex. + // A full blocking implementation would require an async runtime to block on. + std::thread::sleep(std::time::Duration::from_millis(1)); // Replace with proper waiting if needed + } + } + } + } +} + +impl Write for AsyncToSync { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref()); + loop { + match self.write_async(&mut cx, buf) { + Poll::Ready(Ok(n)) => return Ok(n), + Poll::Ready(Err(e)) => return Err(e), + Poll::Pending => { + // Here we are blocking and waiting for the async operation to complete. + std::thread::sleep(std::time::Duration::from_millis(1)); // Not efficient, see notes. + } + } + } + } + + fn flush(&mut self) -> std::io::Result<()> { + let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref()); + loop { + match self.flush_async(&mut cx) { + Poll::Ready(Ok(())) => return Ok(()), + Poll::Ready(Err(e)) => return Err(e), + Poll::Pending => { + // Again, blocking to wait for flush. + std::thread::sleep(std::time::Duration::from_millis(1)); // Not efficient, see notes. + } + } + } + } +} + +pub struct VecAsyncWriter { + buffer: Vec, +} + +impl VecAsyncWriter { + /// Create a new VecAsyncWriter with an empty Vec. + pub fn new(buffer: Vec) -> Self { + VecAsyncWriter { buffer } + } + + /// Retrieve the underlying buffer. + pub fn get_buffer(&self) -> &[u8] { + &self.buffer + } +} + +// Implementing AsyncWrite trait for VecAsyncWriter +impl AsyncWrite for VecAsyncWriter { + fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let len = buf.len(); + + // Assume synchronous writing for simplicity + self.get_mut().buffer.extend_from_slice(buf); + + // Returning the length of written data + Poll::Ready(Ok(len)) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + // In this case, flushing is a no-op for a Vec + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + // Similar to flush, shutdown has no effect here + Poll::Ready(Ok(())) + } +} + +pub struct VecAsyncReader { + buffer: Vec, + position: usize, +} + +impl VecAsyncReader { + /// Create a new VecAsyncReader with the given Vec. + pub fn new(buffer: Vec) -> Self { + VecAsyncReader { buffer, position: 0 } + } + + /// Reset the reader position. + pub fn reset(&mut self) { + self.position = 0; + } +} + +// Implementing AsyncRead trait for VecAsyncReader +impl AsyncRead for VecAsyncReader { + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf) -> Poll> { + let this = self.get_mut(); + + // Check how many bytes are available to read + let len = this.buffer.len(); + let bytes_available = len - this.position; + + if bytes_available == 0 { + // If there's no more data to read, return ready with an Eof + return Poll::Ready(Ok(())); + } + + // Calculate how much we can read into the provided buffer + let to_read = std::cmp::min(bytes_available, buf.remaining()); + + // Write the data to the buf + buf.put_slice(&this.buffer[this.position..this.position + to_read]); + + // Update the position + this.position += to_read; + + // Indicate how many bytes were read + Poll::Ready(Ok(())) + } +} diff --git a/ecstore/src/metacache/writer.rs b/ecstore/src/metacache/writer.rs index 0be530eb8..57b0d189e 100644 --- a/ecstore/src/metacache/writer.rs +++ b/ecstore/src/metacache/writer.rs @@ -1,30 +1,23 @@ -use std::io::ErrorKind; -use std::io::Read; -use std::io::Write; -use std::str::from_utf8; - use crate::disk::MetaCacheEntry; use crate::error::Error; use crate::error::Result; +use std::io::Read; +use std::io::Write; +use std::str::from_utf8; const METACACHE_STREAM_VERSION: u8 = 2; pub struct MetacacheWriter { wr: W, - buf: Vec, created: bool, } -impl MetacacheWriter { - pub fn new(wr: W, block_size: usize) -> Self { - Self { - wr, - buf: Vec::with_capacity(block_size), - created: false, - } +impl MetacacheWriter { + pub fn new(wr: W) -> Self { + Self { wr, created: false } } - async fn write(&mut self, objs: &[MetaCacheEntry]) -> Result<()> { + pub fn write(&mut self, objs: &[MetaCacheEntry]) -> Result<()> { if objs.is_empty() { return Ok(()); } @@ -49,9 +42,8 @@ impl MetacacheWriter { Ok(()) } - async fn close(&mut self) -> Result<()> { + pub fn close(&mut self) -> Result<()> { rmp::encode::write_bool(&mut self.wr, false)?; - self.wr.flush()?; Ok(()) } @@ -74,8 +66,15 @@ impl MetacacheReader { } } - pub fn check_init(&mut self) { + fn check_init(&mut self) { if !self.init { + // let mut buf = match self.read_buf(1).await { + // Ok(res) => res, + // Err(err) => { + // self.err = Some(Error::msg(err.to_string())); + // return; + // } + // }; let ver = match rmp::decode::read_u8(&mut self.rd) { Ok(res) => res, Err(err) => { @@ -94,7 +93,7 @@ impl MetacacheReader { } } - pub fn peek(&mut self) -> Result { + pub fn peek(&mut self) -> Result> { self.check_init(); if let Some(err) = &self.err { @@ -104,8 +103,7 @@ impl MetacacheReader { match rmp::decode::read_bool(&mut self.rd) { Ok(res) => { if !res { - self.err = Some(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof))); - return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof))); + return Ok(None); } } Err(err) => { @@ -158,49 +156,64 @@ impl MetacacheReader { let metadata = self.buf.clone(); - Ok(MetaCacheEntry { + Ok(Some(MetaCacheEntry { name, metadata, cached: None, reusable: false, - }) + })) + } + + pub fn read_all(&mut self) -> Result> { + let mut ret = Vec::new(); + + loop { + if let Some(entry) = self.peek()? { + ret.push(entry); + continue; + } + + break; + } + + Ok(ret) } } #[tokio::test] async fn test_writer() { - use std::fs::File; - use std::fs::OpenOptions; + use crate::io::AsyncToSync; + use crate::io::VecAsyncReader; + use crate::io::VecAsyncWriter; - let file_path = "./test_writer.txt"; - let f = OpenOptions::new() - .create(true) - .read(true) - .write(true) - .truncate(true) - .open(file_path) - .unwrap(); + let mut f = VecAsyncWriter::new(Vec::new()); - // let wr = Writer::File(f); - - let mut w = MetacacheWriter::new(f, 1024); + let mut w = MetacacheWriter::new(AsyncToSync::new_writer(&mut f)); let mut objs = Vec::new(); for i in 0..10 { - objs.push(MetaCacheEntry { + let info = MetaCacheEntry { name: format!("item{}", i), metadata: vec![0u8, 10], cached: None, reusable: false, - }); + }; + println!("old {:?}", &info); + objs.push(info); } - w.write(&objs).await.unwrap(); - w.close().await.unwrap(); + w.write(&objs).unwrap(); - let nf = File::open(file_path).unwrap(); + w.close().unwrap(); - let meta = nf.metadata().unwrap(); + let nf = VecAsyncReader::new(f.get_buffer().to_vec()); - println!("{}", meta.len()); + let mut r = MetacacheReader::new(AsyncToSync::new_reader(nf)); + let nobjs = r.read_all().unwrap(); + + for info in nobjs.iter() { + println!("new {:?}", &info); + } + + assert_eq!(objs, nobjs) }