From 96042ddec40b55f5d6629291221f276a1a727c29 Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Mon, 28 Oct 2024 14:05:13 +0800 Subject: [PATCH 1/2] support lfs chunk download --- gemini/Cargo.toml | 2 + gemini/src/cache/mod.rs | 156 ++++++++++++++++++++++++++++++++-------- 2 files changed, 130 insertions(+), 28 deletions(-) diff --git a/gemini/Cargo.toml b/gemini/Cargo.toml index a9d51c040..9a81f3cf0 100644 --- a/gemini/Cargo.toml +++ b/gemini/Cargo.toml @@ -26,3 +26,5 @@ tracing = { workspace = true } tokio = { workspace = true, features = ["net"] } chrono = { workspace = true } secp256k1 = { workspace = true , features = ["serde", "rand","hashes"] } +ring = "0.17.8" +hex = { workspace = true } \ No newline at end of file diff --git a/gemini/src/cache/mod.rs b/gemini/src/cache/mod.rs index 2e8c0c936..a6a611337 100755 --- a/gemini/src/cache/mod.rs +++ b/gemini/src/cache/mod.rs @@ -1,6 +1,6 @@ use std::{ fs::{self, create_dir_all, File}, - io::{Read, Write}, + io::Read, path::PathBuf, time::Duration, }; @@ -13,10 +13,12 @@ use crate::{ LFSInfo, RepoInfo, }; use callisto::ztm_path_mapping; +use ceres::lfs::lfs_structs::FetchchunkResponse; use common::utils::generate_id; use jupiter::context::Context; use reqwest::{get, Client}; -use tokio::process::Command; +use ring::digest::SHA256; +use tokio::{fs::OpenOptions, io::AsyncWriteExt, process::Command}; use vault::get_peerid; pub async fn cache_public_repo_and_lfs( @@ -303,36 +305,13 @@ async fn download_and_upload_lfs( return; } }; - let url = format!("http://localhost:{}/objects/{}", local_port, lfs.file_hash); - let response = match get(url.clone()).await { - Ok(response) => response, - Err(_) => { - tracing::error!("Download lfs failed {}", url); - return; - } - }; - if !response.status().is_success() { - tracing::error!("Download lfs failed {}", url); - return; - } - // create temp file - let base_dir = - PathBuf::from(std::env::var("MEGA_BASE_DIR").unwrap_or_else(|_| "/tmp/.mega".to_string())); - let base_dir = base_dir.join("tmp"); - if !base_dir.exists() { - create_dir_all(base_dir.clone()).unwrap(); - } - let target_path = base_dir.join(file_hash.clone()); + download_lfs_by_chunk(local_port, lfs.clone()).await; - if target_path.exists() { - fs::remove_file(&target_path).unwrap(); + if !checksum_lfs(lfs.clone()) { + return; } - let mut file = File::create(target_path.clone()).unwrap(); - let bytes = response.bytes().await.unwrap(); - file.write_all(&bytes).unwrap(); - //upload to local mega lfs let client = Client::new(); let json_str = format!( @@ -365,6 +344,13 @@ async fn download_and_upload_lfs( return; } + let target_path = match get_lfs_tmp_file(lfs.clone()) { + Some(p) => p, + None => { + return; + } + }; + let mut file = File::open(target_path.clone()).unwrap(); let mut file_content = Vec::new(); file.read_to_end(&mut file_content).unwrap(); @@ -394,6 +380,120 @@ async fn download_and_upload_lfs( .await; } +async fn download_lfs_by_chunk(local_port: u16, lfs: LFSInfo) { + tracing::info!("Prepare to download LFS {} by chunks", lfs.file_hash); + // fetch chunks info http://localhost:{localport}/objects/{object_id}/chunks + let url = format!( + "http://localhost:{}/objects/{}/chunks", + local_port, lfs.file_hash + ); + let chunk_info = match get(url.clone()).await { + Ok(response) => { + if !response.status().is_success() { + tracing::error!("Get lfs chuncks info failed {}", url); + return; + } + response.json::().await.unwrap() + } + Err(_) => { + tracing::error!("Get lfs chuncks info failed {}", url); + return; + } + }; + let mut chunks = chunk_info.chunks; + chunks.sort_unstable_by(|a, b| a.offset.cmp(&b.offset)); + + // create temp file + let base_dir = + PathBuf::from(std::env::var("MEGA_BASE_DIR").unwrap_or_else(|_| "/tmp/.mega".to_string())); + let base_dir = base_dir.join("tmp"); + if !base_dir.exists() { + create_dir_all(base_dir.clone()).unwrap(); + } + let target_path = base_dir.join(lfs.file_hash.clone()); + if target_path.exists() { + fs::remove_file(&target_path).unwrap(); + } + + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(target_path) + .await + .unwrap(); + + tracing::info!( + "Start to download LFS chunks,size: {}B, chunk_num: {}", + chunk_info.size, + chunks.len() + ); + for chunk in chunks { + // http://localhost:{localport}/objects/{object_id}/chunks + let url = format!("http://localhost:{}/objects/{}", local_port, chunk.sub_oid); + let data = match get(url.clone()).await { + Ok(response) => { + if !response.status().is_success() { + tracing::error!("Get lfs chuncks info failed {}", url); + return; + } + response.bytes().await.unwrap() + } + Err(_) => { + tracing::error!("Get lfs chuncks info failed {}", url); + return; + } + }; + file.write_all(&data).await.unwrap(); + tracing::info!("Chunk offset[{}] download successfully", chunk.offset); + } + tracing::info!("Download LFS {} by chunks successfully", lfs.file_hash); +} + +fn checksum_lfs(lfs: LFSInfo) -> bool { + tracing::info!("Prepare to check LFS SHA256"); + let target_path = match get_lfs_tmp_file(lfs.clone()) { + Some(p) => p, + None => { + return false; + } + }; + let mut file = File::open(target_path).unwrap(); + + let mut context = ring::digest::Context::new(&SHA256); + let mut buffer = [0u8; 1024]; + + loop { + let count = file.read(&mut buffer).unwrap(); + if count == 0 { + break; + } + context.update(&buffer[..count]); + } + + let checksum = hex::encode(context.finish().as_ref()); + let result = checksum == lfs.file_hash; + if result { + tracing::info!("Check LFS SHA256({}) successfully", lfs.file_hash); + } else { + tracing::error!("Check LFS SHA256 failed"); + } + result +} + +fn get_lfs_tmp_file(lfs: LFSInfo) -> Option { + let base_dir = + PathBuf::from(std::env::var("MEGA_BASE_DIR").unwrap_or_else(|_| "/tmp/.mega".to_string())); + let base_dir = base_dir.join("tmp"); + if !base_dir.exists() { + create_dir_all(base_dir.clone()).unwrap(); + } + let target_path = base_dir.join(lfs.file_hash.clone()); + if !target_path.exists() { + return None; + } + Some(target_path) +} + #[cfg(test)] mod tests { // use std::{fs::File, io::Read}; From 51bcc6920eae5601e95172672d9096d771d21c06 Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Mon, 28 Oct 2024 16:14:19 +0800 Subject: [PATCH 2/2] add gemini lfs doc --- gemini/src/cache/mod.rs | 8 ++- gemini/src/lfs/mod.rs | 155 ++++++++++++++++++++++++++++++++++++++-- 2 files changed, 154 insertions(+), 9 deletions(-) diff --git a/gemini/src/cache/mod.rs b/gemini/src/cache/mod.rs index a6a611337..5c96d6675 100755 --- a/gemini/src/cache/mod.rs +++ b/gemini/src/cache/mod.rs @@ -393,7 +393,9 @@ async fn download_lfs_by_chunk(local_port: u16, lfs: LFSInfo) { tracing::error!("Get lfs chuncks info failed {}", url); return; } - response.json::().await.unwrap() + let body = response.text().await.unwrap(); + let chunck_info: FetchchunkResponse = serde_json::from_str(&body).unwrap(); + chunck_info } Err(_) => { tracing::error!("Get lfs chuncks info failed {}", url); @@ -427,7 +429,7 @@ async fn download_lfs_by_chunk(local_port: u16, lfs: LFSInfo) { chunk_info.size, chunks.len() ); - for chunk in chunks { + for (index, chunk) in chunks.iter().enumerate() { // http://localhost:{localport}/objects/{object_id}/chunks let url = format!("http://localhost:{}/objects/{}", local_port, chunk.sub_oid); let data = match get(url.clone()).await { @@ -444,7 +446,7 @@ async fn download_lfs_by_chunk(local_port: u16, lfs: LFSInfo) { } }; file.write_all(&data).await.unwrap(); - tracing::info!("Chunk offset[{}] download successfully", chunk.offset); + tracing::info!("Chunk[{}] download from {} successfully", index, url); } tracing::info!("Download LFS {} by chunks successfully", lfs.file_hash); } diff --git a/gemini/src/lfs/mod.rs b/gemini/src/lfs/mod.rs index 056dafc7b..3028a8620 100644 --- a/gemini/src/lfs/mod.rs +++ b/gemini/src/lfs/mod.rs @@ -6,6 +6,27 @@ use crate::{ util::handle_response, ztm::get_or_create_remote_mega_tunnel, LFSInfo, LFSInfoPostBody, }; +/// share lfs +/// +/// ## paras +/// - `bootstrap_node`: bootstrap_node +/// - `file_hash`: file_hash +/// - `hash_type`: hash_type +/// - `file_size`: file_size +/// - `origin`: origin +/// +/// for example +/// ``` +/// { +/// "bootstrap_node":"https://gitmono.org/relay", +/// "file_hash":"52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6", +/// "hash_type":"sha256", +/// "file_size":199246498, +/// "origin":"p2p://t14id7uQxwneJ2PnPtaA3GSUwxTx6HTaq1UkayQVWSPT/sha256/52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6" +///} +/// ``` +/// This method will send a Post request to the relay to share lfs +/// pub async fn share_lfs( bootstrap_node: String, file_hash: String, @@ -41,11 +62,30 @@ pub async fn share_lfs( } } +/// createa lfs download local ports +/// +/// ## Paras +/// - `bootstrap_node`: bootstrap_node +/// - `ztm_agent_port`: ztm_agent_port +/// - `file_uri`: file_uri +/// +/// for example +/// ``` +/// { +/// "bootstrap_node":"https://gitmono.org/relay", +/// "ztm_agent_port":777, +/// "file_uri":"p2p://t14id7uQxwneJ2PnPtaA3GSUwxTx6HTaq1UkayQVWSPT/sha256/52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6" +///} +/// ``` +/// ## Return +/// local_port1, local_port2,... +/// +/// Each port is for a remote peer pub async fn create_lfs_download_tunnel( bootstrap_node: String, ztm_agent_port: u16, file_uri: String, -) -> Result, String> { +) -> Result, String> { let file_hash = match get_file_hash_from_origin(file_uri) { Ok(file_hash) => file_hash, Err(_) => { @@ -76,11 +116,11 @@ pub async fn create_lfs_download_tunnel( .collect(); tracing::info!("Search lfs[{}] download peer:{:?}", file_hash, peer_list); - let mut tunnel_list: Vec = vec![]; + let mut tunnel_list: Vec = vec![]; for peer_id in peer_list { match get_or_create_remote_mega_tunnel(ztm_agent_port, peer_id).await { Ok(port) => { - tunnel_list.push(format!("http://localhost:{}", port)); + tunnel_list.push(port); } Err(s) => { tracing::error!("{}", s); @@ -101,14 +141,117 @@ pub fn get_file_hash_from_origin(origin: String) -> Result { #[cfg(test)] mod tests { - // use crate::lfs::create_lfs_download_tunnel; + // use reqwest::get; + // use std::path::PathBuf; + // use tokio::fs::OpenOptions; + // use crate::lfs::create_lfs_download_tunnel; + // use ceres::lfs::lfs_structs::FetchchunkResponse; + // use ring::digest::SHA256; + // use std::fs::{self, create_dir_all}; + // use std::{fs::File, io::Read}; + // use tokio::io::AsyncWriteExt; // #[tokio::test] // async fn create_lfs_download_tunnel_test() { - // let result = create_lfs_download_tunnel("http://222.20.126.106:8001".to_string() + // let local_port_list = create_lfs_download_tunnel("http://222.20.126.106:8001".to_string() // , 7777, // "p2p://t14id7uQxwneJ2PnPtaA3GSUwxTx6HTaq1UkayQVWSPT/sha256/52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6".to_string() // ).await.unwrap(); - // println!("{:?}", result); + // println!("{:?}", local_port_list); + // if local_port_list.len() == 0 { + // return; + // } + + // let file_hash = "52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6"; + // let local_port = local_port_list[0]; + // println!("Prepare to download LFS {} by chunks", file_hash); + // // fetch chunks info http://localhost:{localport}/objects/{object_id}/chunks + // let url = format!( + // "http://localhost:{}/objects/{}/chunks", + // local_port, file_hash + // ); + // let chunk_info: FetchchunkResponse = match get(url.clone()).await { + // Ok(response) => { + // if !response.status().is_success() { + // println!("Get lfs chuncks info failed {}", url); + // return; + // } + // let body = response.text().await.unwrap(); + // let chunck_info: FetchchunkResponse = serde_json::from_str(&body).unwrap(); + // chunck_info + // } + // Err(_) => { + // println!("Get lfs chuncks info failed {}", url); + // return; + // } + // }; + // let mut chunks = chunk_info.chunks; + // chunks.sort_unstable_by(|a, b| a.offset.cmp(&b.offset)); + + // // create temp file + // let base_dir = PathBuf::from("tmp"); + // if !base_dir.exists() { + // create_dir_all(base_dir.clone()).unwrap(); + // } + // let target_path = base_dir.join(file_hash); + // if target_path.exists() { + // fs::remove_file(&target_path).unwrap(); + // } + + // let mut file = OpenOptions::new() + // .create(true) + // .append(true) + // .open(target_path.clone()) + // .await + // .unwrap(); + + // println!( + // "Start to download LFS chunks,size: {}B, chunk_num: {}", + // chunk_info.size, + // chunks.len() + // ); + // for (index, chunk) in chunks.iter().enumerate() { + // // http://localhost:{localport}/objects/{object_id}/chunks + // let local_port = local_port_list[index % local_port_list.len()]; + // let url = format!("http://localhost:{}/objects/{}", local_port, chunk.sub_oid); + // let data = match get(url.clone()).await { + // Ok(response) => { + // if !response.status().is_success() { + // println!("Get lfs chuncks failed {}", url); + // return; + // } + // response.bytes().await.unwrap() + // } + // Err(_) => { + // println!("Get lfs chuncks failed {}", url); + // return; + // } + // }; + // file.write_all(&data).await.unwrap(); + // println!("Chunk[{}] download from {} successfully", index, url); + // } + + // println!("Download LFS {} by chunks successfully", file_hash); + + // let mut file = File::open(target_path).unwrap(); + + // let mut context = ring::digest::Context::new(&SHA256); + // let mut buffer = [0u8; 1024]; + + // loop { + // let count = file.read(&mut buffer).unwrap(); + // if count == 0 { + // break; + // } + // context.update(&buffer[..count]); + // } + + // let checksum = hex::encode(context.finish().as_ref()); + // let result = checksum == file_hash; + // if result { + // println!("Check LFS SHA256({}) successfully", file_hash); + // } else { + // println!("Check LFS SHA256 failed"); + // } // } }