Skip to content

fix(overlaybd): probe p2p before remote runtime initialization #172

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
WindSongShaoGe wants to merge 1 commit into
base: main
Choose a base branch
from
+132 −12
Open

fix(overlaybd): probe p2p before remote runtime initialization #172

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 45 additions & 1 deletion storage/overlaybd/src/image/image_file.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -602,7 +602,7 @@ impl ImageFile {
.context("lower layer local file is missing and p2p uuid facade is not configured")?;
let url = format!("{}/{}", p2p_uuid_address.trim_end_matches('/'), uuid);
let remote_file = image_service
.open_source_blob_with_size(&url, (layer.size != 0).then_some(layer.size))
.open_unaccelerated_source_blob_with_size(&url, (layer.size != 0).then_some(layer.size))
.await?;
let tar_file = new_tar_file_adaptor(remote_file).await?;
let switch_file = new_switch_file(tar_file, false, Some(&url)).await?;
Expand Down Expand Up @@ -1168,10 +1168,54 @@ mod tests {
miss: bool,
}

async fn proxy_p2p_origin_request(
headers: HttpHeaderMap,
origin_url: String,
) -> Response<Body> {
let mut origin_request = reqwest::Client::new().get(origin_url);
if let Some(range) = headers.get(reqwest::header::RANGE) {
origin_request = origin_request.header(reqwest::header::RANGE, range.clone());
}
let origin_response = match origin_request.send().await {
Ok(response) => response,
Err(_) => {
return Response::builder()
.status(HttpStatusCode::BAD_GATEWAY)
.body(Body::empty())
.expect("502 response");
}
};
let status = origin_response.status();
let headers = origin_response.headers().clone();
let body = match origin_response.bytes().await {
Ok(body) => body,
Err(_) => {
return Response::builder()
.status(HttpStatusCode::BAD_GATEWAY)
.body(Body::empty())
.expect("502 response");
}
};
let mut response = Response::builder()
.status(status)
.body(Body::from(body))
.expect("origin proxy response");
*response.headers_mut() = headers;
response
}

async fn handle_p2p_uuid_request(
State(state): State<P2pUuidLayerState>,
request: Request,
) -> Response<Body> {
let origin_url = request
.uri()
.path()
.strip_prefix("/p2p-http/")
.map(str::to_owned);
if let Some(origin_url) = origin_url {
return proxy_p2p_origin_request(request.headers().clone(), origin_url).await;
}
let path = request.uri().path();
if !path.starts_with("/p2p-uuid/") {
return Response::builder()
Expand Down
98 changes: 87 additions & 11 deletions storage/overlaybd/src/image/image_service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ struct ImageServiceInner {

struct RemoteRuntime {
underlay_registryfs: RegistryFsV2,
direct_registryfs: RegistryFsV2,
oss_backend: Option<OssBackend>,
file_cache: Option<FileCacheBackend>,
}
Expand Down Expand Up @@ -193,6 +194,12 @@ impl ImageService {

async fn build_remote_runtime(&self) -> Result<RemoteRuntime> {
let underlay_registryfs = RegistryFsV2::from_global_config(&self.inner.global_config)?;
let direct_registryfs = RegistryFsV2::from_global_config(&self.inner.global_config)?;
direct_registryfs.set_accelerate_address("");
// Probe before the OnceCell captures this runtime. Otherwise a caller
// that reaches `remote_runtime` before `create_image_file` freezes an
// empty acceleration address into the first burst of remote reads.
self.enable_acceleration();

@github-actions github-actions Bot Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

performance · medium
enable_acceleration() performs synchronous DNS resolution and TcpStream::connect_timeout calls (up to one second per resolved address). Calling it directly while initializing this async OnceCell can block a Tokio worker whenever any remote-open path first initializes the runtime, delaying unrelated tasks. Run the probe via tokio::task::spawn_blocking (or replace it with an async, bounded probe) before setting the registry address.

Suggestion:

Suggested change
self.enable_acceleration();
let service = self.clone();
tokio::task::spawn_blocking(move || service.enable_acceleration())
.await
.context("join acceleration probe task")?;

underlay_registryfs.set_accelerate_address(self.current_accelerate_address());
let oss_backend = if self.inner.global_config.oss_config.enable {
Some(OssBackend::new(&self.inner.global_config.oss_config)?)
Expand All @@ -203,6 +210,7 @@ impl ImageService {

Ok(RemoteRuntime {
underlay_registryfs,
direct_registryfs,
oss_backend,
file_cache,
})
Expand Down Expand Up @@ -444,13 +452,42 @@ impl ImageService {
self.open_backend_source_with_size(url, source_size).await
}

pub(crate) async fn open_unaccelerated_source_blob_with_size(
&self,
url: &str,
source_size: Option<u64>,
) -> Result<Arc<dyn VirtualFile>> {
let remote_runtime = self.remote_runtime().await?;
if Self::is_oss_url(url) {
let oss = remote_runtime
.oss_backend
.as_ref()
.ok_or_else(|| anyhow::anyhow!("OSS backend not enabled in config"))?;
return oss.open_with_size_hint(url, source_size);
}

Self::open_registry_source_with_size(&remote_runtime.direct_registryfs, url, source_size)
.await
}

fn is_oss_url(url: &str) -> bool {
match reqwest::Url::parse(url) {
Ok(parsed) => matches!(parsed.scheme(), "s3" | "oss"),
Err(_) => false,
}
}

async fn open_registry_source_with_size(
registryfs: &RegistryFsV2,
url: &str,
source_size: Option<u64>,
) -> Result<Arc<dyn VirtualFile>> {
match source_size {
Some(size) => Ok(registryfs.open_with_size_hint(url.to_string(), Some(size))),
None => registryfs.open(url.to_string()).await,
}
}

async fn open_backend_source_with_size(
&self,
url: &str,
Expand All @@ -465,17 +502,8 @@ impl ImageService {
return oss.open_with_size_hint(url, source_size);
}

match source_size {
Some(size) => Ok(remote_runtime
.underlay_registryfs
.open_with_size_hint(url.to_string(), Some(size))),
None => {
remote_runtime
.underlay_registryfs
.open(url.to_string())
.await
}
}
Self::open_registry_source_with_size(&remote_runtime.underlay_registryfs, url, source_size)
.await
}

pub async fn export_upper_as_oss_sealed(
Expand Down Expand Up @@ -958,6 +986,54 @@ mod tests {
);
}

#[tokio::test]
async fn test_remote_runtime_initialization_probes_p2p_first() {
let tmp = TempDir::new().expect("tempdir");
let global_path = tmp.path().join("overlaybd.json");
let (facade_base, server_handle) = spawn_server(Router::new()).await;
let p2p_address = format!("{facade_base}/p2p-http");

write_json(
&global_path,
&serde_json::json!({
"registryFsVersion": "v2",
"ioEngine": 0,
"cacheConfig": {
"cacheType": "file",
"cacheDir": tmp.path().join("cache"),
"cacheSizeGB": 1,
"refillSize": 262144,
"blockSize": 65536
},
"p2pConfig": {
"enable": true,
"address": p2p_address
}
}),
);

let service = ImageService::from_config_path(&global_path)
.await
.expect("service");
assert!(service.inner.remote_runtime.get().is_none());
assert_eq!(
*service.inner.remote_mode.read(),
RemoteOpenMode::Cached,
"P2P must not be marked reachable before the first probe"
);

service
.remote_runtime()
.await
.expect("initialize remote runtime before create_image_file");

assert!(service.inner.remote_runtime.get().is_some());
assert_eq!(*service.inner.remote_mode.read(), RemoteOpenMode::Direct);
assert_eq!(service.current_accelerate_address(), p2p_address);

server_handle.abort();
}

#[tokio::test]
async fn test_open_remote_blob_with_aliyun_endpoint_uses_unsigned_payload_sigv4() {
let tmp = TempDir::new().expect("tempdir");
Expand Down
Loading