mirrors/krata
1
0
Fork 0
mirror of https://github.com/edera-dev/krata.git synced 2025-08-03 05:10:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: concurrent image pulls

Browse Source
This commit is contained in:
Alex Zenla
2024-04-16 08:20:09 +00:00
parent 3d29a68645
commit a99541460d
13 changed files with 331 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -1452,6 +1452,7 @@ dependencies = [
"log",
"prost",
"redb",
"scopeguard",
"signal-hook",
"tokio",
"tokio-stream",

1
crates/daemon/Cargo.toml
View File

@ -23,6 +23,7 @@ krata-runtime = { path = "../runtime", version = "^0.0.9" }
log = { workspace = true }
prost = { workspace = true }
redb = { workspace = true }
scopeguard = { workspace = true }
signal-hook = { workspace = true }
tokio = { workspace = true }
tokio-stream = { workspace = true }

52
crates/daemon/src/control.rs
View File

@ -18,7 +18,7 @@ use krata::{
};
use krataoci::{
name::ImageName,
packer::{service::OciPackerService, OciImagePacked, OciPackedFormat},
packer::{service::OciPackerService, OciPackedFormat, OciPackedImage},
progress::{OciProgress, OciProgressContext},
};
use std::{pin::Pin, str::FromStr};
@ -90,8 +90,8 @@ enum ConsoleDataSelect {
}
enum PullImageSelect {
Progress(usize),
Completed(Result<Result<OciImagePacked, anyhow::Error>, JoinError>),
Progress(Option<OciProgress>),
Completed(Result<Result<OciPackedImage, anyhow::Error>, JoinError>),
}
#[tonic::async_trait]
@ -367,32 +367,46 @@ impl ControlService for DaemonControlService {
OciImageFormat::Erofs => OciPackedFormat::Erofs,
OciImageFormat::Tar => OciPackedFormat::Tar,
};
let (sender, mut receiver) = channel::<OciProgress>(100);
let context = OciProgressContext::new(sender);
let (context, mut receiver) = OciProgressContext::create();
let our_packer = self.packer.clone();
let output = try_stream! {
let mut task = tokio::task::spawn(async move {
our_packer.request(name, format, context).await
});
let abort_handle = task.abort_handle();
let _task_cancel_guard = scopeguard::guard(abort_handle, |handle| {
handle.abort();
});
loop {
let mut progresses = Vec::new();
let what = select! {
x = receiver.recv_many(&mut progresses, 10) => PullImageSelect::Progress(x),
x = receiver.recv() => PullImageSelect::Progress(x.ok()),
x = &mut task => PullImageSelect::Completed(x),
};
match what {
PullImageSelect::Progress(count) => {
if count > 0 {
let progress = progresses.remove(progresses.len() - 1);
let reply = PullImageReply {
progress: Some(convert_oci_progress(progress)),
digest: String::new(),
format: OciImageFormat::Unknown.into(),
};
yield reply;
PullImageSelect::Progress(Some(mut progress)) => {
let mut drain = 0;
loop {
if drain >= 10 {
break;
}
if let Ok(latest) = receiver.try_recv() {
progress = latest;
} else {
break;
}
drain += 1;
}
let reply = PullImageReply {
progress: Some(convert_oci_progress(progress)),
digest: String::new(),
format: OciImageFormat::Unknown.into(),
};
yield reply;
},
PullImageSelect::Completed(result) => {
@ -414,6 +428,10 @@ impl ControlService for DaemonControlService {
yield reply;
break;
},
_ => {
continue;
}
}
}
};

29
crates/oci/examples/squashify.rs
View File

@ -5,10 +5,10 @@ use env_logger::Env;
use krataoci::{
name::ImageName,
packer::{service::OciPackerService, OciPackedFormat},
progress::{OciProgress, OciProgressContext},
progress::OciProgressContext,
registry::OciPlatform,
};
use tokio::{fs, sync::mpsc::channel};
use tokio::fs;
#[tokio::main]
async fn main() -> Result<()> {
@ -22,14 +22,28 @@ async fn main() -> Result<()> {
fs::create_dir(&cache_dir).await?;
}
let (sender, mut receiver) = channel::<OciProgress>(100);
let (context, mut receiver) = OciProgressContext::create();
tokio::task::spawn(async move {
loop {
let mut progresses = Vec::new();
let _ = receiver.recv_many(&mut progresses, 100).await;
let Some(progress) = progresses.last() else {
continue;
let Ok(mut progress) = receiver.recv().await else {
return;
};
let mut drain = 0;
loop {
if drain >= 10 {
break;
}
if let Ok(latest) = receiver.try_recv() {
progress = latest;
} else {
break;
}
drain += 1;
}
println!("phase {:?}", progress.phase);
for (id, layer) in &progress.layers {
println!(
@ -39,7 +53,6 @@ async fn main() -> Result<()> {
}
}
});
let context = OciProgressContext::new(sender);
let service = OciPackerService::new(seed, &cache_dir, OciPlatform::current())?;
let packed = service
.request(image.clone(), OciPackedFormat::Squashfs, context)

47
crates/oci/src/assemble.rs
View File

@ -5,8 +5,10 @@ use crate::vfs::{VfsNode, VfsTree};
use anyhow::{anyhow, Result};
use log::{debug, trace, warn};
use oci_spec::image::{ImageConfiguration, ImageManifest};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::fs;
use tokio::io::AsyncRead;
@ -34,11 +36,12 @@ impl Drop for OciImageAssembled {
pub struct OciImageAssembler {
downloader: OciImageFetcher,
resolved: OciResolvedImage,
resolved: Option<OciResolvedImage>,
progress: OciBoundProgress,
work_dir: PathBuf,
disk_dir: PathBuf,
tmp_dir: Option<PathBuf>,
success: AtomicBool,
}
impl OciImageAssembler {
@ -82,11 +85,12 @@ impl OciImageAssembler {
Ok(OciImageAssembler {
downloader,
resolved,
resolved: Some(resolved),
progress,
work_dir,
disk_dir: target_dir,
tmp_dir,
success: AtomicBool::new(false),
})
}
@ -98,11 +102,11 @@ impl OciImageAssembler {
self.assemble_with(&layer_dir).await
}
async fn assemble_with(self, layer_dir: &Path) -> Result<OciImageAssembled> {
let local = self
.downloader
.download(self.resolved.clone(), layer_dir)
.await?;
async fn assemble_with(mut self, layer_dir: &Path) -> Result<OciImageAssembled> {
let Some(ref resolved) = self.resolved else {
return Err(anyhow!("resolved image was not available when expected"));
};
let local = self.downloader.download(resolved, layer_dir).await?;
let mut vfs = VfsTree::new();
for layer in &local.layers {
debug!(
@ -146,13 +150,20 @@ impl OciImageAssembler {
fs::remove_file(&layer.path).await?;
}
}
Ok(OciImageAssembled {
let Some(resolved) = self.resolved.take() else {
return Err(anyhow!("resolved image was not available when expected"));
};
let assembled = OciImageAssembled {
vfs: Arc::new(vfs),
digest: self.resolved.digest,
manifest: self.resolved.manifest,
digest: resolved.digest,
manifest: resolved.manifest,
config: local.config,
tmp_dir: self.tmp_dir,
})
tmp_dir: self.tmp_dir.clone(),
};
self.success.store(true, Ordering::Release);
Ok(assembled)
}
async fn process_whiteout_entry(
@ -223,6 +234,18 @@ impl OciImageAssembler {
}
}
impl Drop for OciImageAssembler {
fn drop(&mut self) {
if !self.success.load(Ordering::Acquire) {
if let Some(tmp_dir) = self.tmp_dir.clone() {
tokio::task::spawn(async move {
let _ = fs::remove_dir_all(tmp_dir).await;
});
}
}
}
}
async fn delete_disk_paths(node: &VfsNode) -> Result<()> {
let mut queue = vec![node];
while !queue.is_empty() {

4
crates/oci/src/fetch.rs
View File

@ -219,7 +219,7 @@ impl OciImageFetcher {
pub async fn download(
&self,
image: OciResolvedImage,
image: &OciResolvedImage,
layer_dir: &Path,
) -> Result<OciLocalImage> {
let config: OciSchema<ImageConfiguration>;
@ -270,7 +270,7 @@ impl OciImageFetcher {
.await;
}
Ok(OciLocalImage {
image,
image: image.clone(),
config,
layers,
})

27
crates/oci/src/packer/backend.rs
View File

@ -7,7 +7,12 @@ use crate::{
};
use anyhow::{anyhow, Result};
use log::warn;
use tokio::{fs::File, pin, process::Command, select};
use tokio::{
fs::File,
pin,
process::{Child, Command},
select,
};
#[derive(Debug, Clone, Copy)]
pub enum OciPackerBackendType {
@ -56,7 +61,7 @@ impl OciPackerBackend for OciPackerMkSquashfs {
})
.await;
let mut child = Command::new("mksquashfs")
let child = Command::new("mksquashfs")
.arg("-")
.arg(file)
.arg("-comp")
@ -66,7 +71,9 @@ impl OciPackerBackend for OciPackerMkSquashfs {
.stderr(Stdio::null())
.stdout(Stdio::null())
.spawn()?;
let mut child = ChildProcessKillGuard(child);
let stdin = child
.0
.stdin
.take()
.ok_or(anyhow!("unable to acquire stdin stream"))?;
@ -77,7 +84,7 @@ impl OciPackerBackend for OciPackerMkSquashfs {
}
Ok(())
}));
let wait = child.wait();
let wait = child.0.wait();
pin!(wait);
let status_result = loop {
if let Some(inner) = writer.as_mut() {
@ -138,7 +145,7 @@ impl OciPackerBackend for OciPackerMkfsErofs {
})
.await;
let mut child = Command::new("mkfs.erofs")
let child = Command::new("mkfs.erofs")
.arg("-L")
.arg("root")
.arg("--tar=-")
@ -147,14 +154,16 @@ impl OciPackerBackend for OciPackerMkfsErofs {
.stderr(Stdio::null())
.stdout(Stdio::null())
.spawn()?;
let mut child = ChildProcessKillGuard(child);
let stdin = child
.0
.stdin
.take()
.ok_or(anyhow!("unable to acquire stdin stream"))?;
let mut writer = Some(tokio::task::spawn(
async move { vfs.write_to_tar(stdin).await },
));
let wait = child.wait();
let wait = child.0.wait();
pin!(wait);
let status_result = loop {
if let Some(inner) = writer.as_mut() {
@ -229,3 +238,11 @@ impl OciPackerBackend for OciPackerTar {
Ok(())
}
}
struct ChildProcessKillGuard(Child);
impl Drop for ChildProcessKillGuard {
fn drop(&mut self) {
let _ = self.0.start_kill();
}
}

10
crates/oci/src/packer/cache.rs
View File

@ -1,5 +1,5 @@
use crate::{
packer::{OciImagePacked, OciPackedFormat},
packer::{OciPackedFormat, OciPackedImage},
schema::OciSchema,
};
@ -25,7 +25,7 @@ impl OciPackerCache {
&self,
digest: &str,
format: OciPackedFormat,
) -> Result<Option<OciImagePacked>> {
) -> Result<Option<OciPackedImage>> {
let mut fs_path = self.cache_dir.clone();
let mut config_path = self.cache_dir.clone();
let mut manifest_path = self.cache_dir.clone();
@ -46,7 +46,7 @@ impl OciPackerCache {
let config_bytes = fs::read(&config_path).await?;
let config: ImageConfiguration = serde_json::from_slice(&config_bytes)?;
debug!("cache hit digest={}", digest);
Some(OciImagePacked::new(
Some(OciPackedImage::new(
digest.to_string(),
fs_path.clone(),
format,
@ -63,7 +63,7 @@ impl OciPackerCache {
)
}
pub async fn store(&self, packed: OciImagePacked) -> Result<OciImagePacked> {
pub async fn store(&self, packed: OciPackedImage) -> Result<OciPackedImage> {
debug!("cache store digest={}", packed.digest);
let mut fs_path = self.cache_dir.clone();
let mut manifest_path = self.cache_dir.clone();
@ -74,7 +74,7 @@ impl OciPackerCache {
fs::rename(&packed.path, &fs_path).await?;
fs::write(&config_path, packed.config.raw()).await?;
fs::write(&manifest_path, packed.manifest.raw()).await?;
Ok(OciImagePacked::new(
Ok(OciPackedImage::new(
packed.digest,
fs_path.clone(),
packed.format,

10
crates/oci/src/packer/mod.rs
View File

@ -9,7 +9,7 @@ pub mod backend;
pub mod cache;
pub mod service;
#[derive(Debug, Default, Clone, Copy)]
#[derive(Debug, Default, Clone, Copy, Eq, PartialEq, Hash)]
pub enum OciPackedFormat {
#[default]
Squashfs,
@ -36,7 +36,7 @@ impl OciPackedFormat {
}
#[derive(Clone)]
pub struct OciImagePacked {
pub struct OciPackedImage {
pub digest: String,
pub path: PathBuf,
pub format: OciPackedFormat,
@ -44,15 +44,15 @@ pub struct OciImagePacked {
pub manifest: OciSchema<ImageManifest>,
}
impl OciImagePacked {
impl OciPackedImage {
pub fn new(
digest: String,
path: PathBuf,
format: OciPackedFormat,
config: OciSchema<ImageConfiguration>,
manifest: OciSchema<ImageManifest>,
) -> OciImagePacked {
OciImagePacked {
) -> OciPackedImage {
OciPackedImage {
digest,
path,
format,

172
crates/oci/src/packer/service.rs
View File

@ -1,22 +1,40 @@
use std::path::{Path, PathBuf};
use std::{
collections::{hash_map::Entry, HashMap},
fmt::Display,
path::{Path, PathBuf},
sync::Arc,
};
use anyhow::{anyhow, Result};
use tokio::{
sync::{watch, Mutex},
task::JoinHandle,
};
use crate::{
assemble::OciImageAssembler,
fetch::OciImageFetcher,
fetch::{OciImageFetcher, OciResolvedImage},
name::ImageName,
progress::{OciBoundProgress, OciProgress, OciProgressContext},
registry::OciPlatform,
};
use super::{cache::OciPackerCache, OciImagePacked, OciPackedFormat};
use log::{error, info, warn};
use super::{cache::OciPackerCache, OciPackedFormat, OciPackedImage};
pub struct OciPackerTask {
progress: OciBoundProgress,
watch: watch::Sender<Option<Result<OciPackedImage>>>,
task: JoinHandle<()>,
}
#[derive(Clone)]
pub struct OciPackerService {
seed: Option<PathBuf>,
platform: OciPlatform,
cache: OciPackerCache,
tasks: Arc<Mutex<HashMap<OciPackerTaskKey, OciPackerTask>>>,
}
impl OciPackerService {
@ -29,6 +47,7 @@ impl OciPackerService {
seed,
cache: OciPackerCache::new(cache_dir)?,
platform,
tasks: Arc::new(Mutex::new(HashMap::new())),
})
}
@ -36,7 +55,7 @@ impl OciPackerService {
&self,
digest: &str,
format: OciPackedFormat,
) -> Result<Option<OciImagePacked>> {
) -> Result<Option<OciPackedImage>> {
self.cache.recall(digest, format).await
}
@ -45,14 +64,98 @@ impl OciPackerService {
name: ImageName,
format: OciPackedFormat,
progress_context: OciProgressContext,
) -> Result<OciImagePacked> {
) -> Result<OciPackedImage> {
let progress = OciProgress::new();
let progress = OciBoundProgress::new(progress_context.clone(), progress);
let fetcher =
OciImageFetcher::new(self.seed.clone(), self.platform.clone(), progress.clone());
let resolved = fetcher.resolve(name).await?;
let key = OciPackerTaskKey {
digest: resolved.digest.clone(),
format,
};
let (progress_copy_task, mut receiver) = match self.tasks.lock().await.entry(key.clone()) {
Entry::Occupied(entry) => {
let entry = entry.get();
(
Some(entry.progress.also_update(progress_context).await),
entry.watch.subscribe(),
)
}
Entry::Vacant(entry) => {
let task = self
.clone()
.launch(key.clone(), format, resolved, fetcher, progress.clone())
.await;
let (watch, receiver) = watch::channel(None);
let task = OciPackerTask {
progress: progress.clone(),
task,
watch,
};
entry.insert(task);
(None, receiver)
}
};
let _progress_task_guard = scopeguard::guard(progress_copy_task, |task| {
if let Some(task) = task {
task.abort();
}
});
let _task_cancel_guard = scopeguard::guard(self.clone(), |service| {
service.maybe_cancel_task(key);
});
loop {
receiver.changed().await?;
let current = receiver.borrow_and_update();
if current.is_some() {
return current
.as_ref()
.map(|x| x.as_ref().map_err(|err| anyhow!("{}", err)).cloned())
.unwrap();
}
}
}
async fn launch(
self,
key: OciPackerTaskKey,
format: OciPackedFormat,
resolved: OciResolvedImage,
fetcher: OciImageFetcher,
progress: OciBoundProgress,
) -> JoinHandle<()> {
info!("packer task {} started", key);
tokio::task::spawn(async move {
let _task_drop_guard =
scopeguard::guard((key.clone(), self.clone()), |(key, service)| {
service.ensure_task_gone(key);
});
if let Err(error) = self
.task(key.clone(), format, resolved, fetcher, progress)
.await
{
self.finish(&key, Err(error)).await;
}
})
}
async fn task(
&self,
key: OciPackerTaskKey,
format: OciPackedFormat,
resolved: OciResolvedImage,
fetcher: OciImageFetcher,
progress: OciBoundProgress,
) -> Result<()> {
if let Some(cached) = self.cache.recall(&resolved.digest, format).await? {
return Ok(cached);
self.finish(&key, Ok(cached)).await;
return Ok(());
}
let assembler =
OciImageAssembler::new(fetcher, resolved, progress.clone(), None, None).await?;
@ -67,7 +170,7 @@ impl OciPackerService {
packer
.pack(progress, assembled.vfs.clone(), &target)
.await?;
let packed = OciImagePacked::new(
let packed = OciPackedImage::new(
assembled.digest.clone(),
file,
format,
@ -75,6 +178,59 @@ impl OciPackerService {
assembled.manifest.clone(),
);
let packed = self.cache.store(packed).await?;
Ok(packed)
self.finish(&key, Ok(packed)).await;
Ok(())
}
async fn finish(&self, key: &OciPackerTaskKey, result: Result<OciPackedImage>) {
let Some(task) = self.tasks.lock().await.remove(key) else {
error!("packer task {} was not found when task completed", key);
return;
};
match result.as_ref() {
Ok(_) => {
info!("packer task {} completed", key);
}
Err(err) => {
warn!("packer task {} failed: {}", key, err);
}
}
task.watch.send_replace(Some(result));
}
fn maybe_cancel_task(self, key: OciPackerTaskKey) {
tokio::task::spawn(async move {
let tasks = self.tasks.lock().await;
if let Some(task) = tasks.get(&key) {
if task.watch.is_closed() {
task.task.abort();
}
}
});
}
fn ensure_task_gone(self, key: OciPackerTaskKey) {
tokio::task::spawn(async move {
let mut tasks = self.tasks.lock().await;
if let Some(task) = tasks.remove(&key) {
warn!("packer task {} aborted", key);
task.watch.send_replace(Some(Err(anyhow!("task aborted"))));
}
});
}
}
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
struct OciPackerTaskKey {
digest: String,
format: OciPackedFormat,
}
impl Display for OciPackerTaskKey {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_fmt(format_args!("{}:{}", self.digest, self.format.extension()))
}
}

41
crates/oci/src/progress.rs
View File

@ -1,7 +1,11 @@
use std::sync::Arc;
use indexmap::IndexMap;
use tokio::sync::{mpsc::Sender, Mutex};
use std::sync::Arc;
use tokio::{
sync::{broadcast, Mutex},
task::JoinHandle,
};
const OCI_PROGRESS_QUEUE_LEN: usize = 100;
#[derive(Clone, Debug)]
pub struct OciProgress {
@ -99,16 +103,25 @@ pub enum OciProgressLayerPhase {
#[derive(Clone)]
pub struct OciProgressContext {
sender: Sender<OciProgress>,
sender: broadcast::Sender<OciProgress>,
}
impl OciProgressContext {
pub fn new(sender: Sender<OciProgress>) -> OciProgressContext {
pub fn create() -> (OciProgressContext, broadcast::Receiver<OciProgress>) {
let (sender, receiver) = broadcast::channel(OCI_PROGRESS_QUEUE_LEN);
(OciProgressContext::new(sender), receiver)
}
pub fn new(sender: broadcast::Sender<OciProgress>) -> OciProgressContext {
OciProgressContext { sender }
}
pub fn update(&self, progress: &OciProgress) {
let _ = self.sender.try_send(progress.clone());
let _ = self.sender.send(progress.clone());
}
pub fn subscribe(&self) -> broadcast::Receiver<OciProgress> {
self.sender.subscribe()
}
}
@ -137,4 +150,20 @@ impl OciBoundProgress {
function(&mut progress);
self.context.update(&progress);
}
pub async fn also_update(&self, context: OciProgressContext) -> JoinHandle<()> {
let progress = self.instance.lock().await.clone();
context.update(&progress);
let mut receiver = self.context.subscribe();
tokio::task::spawn(async move {
while let Ok(progress) = receiver.recv().await {
match context.sender.send(progress) {
Ok(_) => {}
Err(_) => {
break;
}
}
}
})
}
}

6
crates/runtime/src/cfgblk.rs
View File

@ -1,7 +1,7 @@
use anyhow::Result;
use backhand::{FilesystemWriter, NodeHeader};
use krata::launchcfg::LaunchInfo;
use krataoci::packer::OciImagePacked;
use krataoci::packer::OciPackedImage;
use log::trace;
use std::fs;
use std::fs::File;
@ -9,13 +9,13 @@ use std::path::PathBuf;
use uuid::Uuid;
pub struct ConfigBlock<'a> {
pub image: &'a OciImagePacked,
pub image: &'a OciPackedImage,
pub file: PathBuf,
pub dir: PathBuf,
}
impl ConfigBlock<'_> {
pub fn new<'a>(uuid: &Uuid, image: &'a OciImagePacked) -> Result<ConfigBlock<'a>> {
pub fn new<'a>(uuid: &Uuid, image: &'a OciPackedImage) -> Result<ConfigBlock<'a>> {
let mut dir = std::env::temp_dir().clone();
dir.push(format!("krata-cfg-{}", uuid));
fs::create_dir_all(&dir)?;

4
crates/runtime/src/launch.rs
View File

@ -10,7 +10,7 @@ use krata::launchcfg::{
LaunchInfo, LaunchNetwork, LaunchNetworkIpv4, LaunchNetworkIpv6, LaunchNetworkResolver,
LaunchPackedFormat, LaunchRoot,
};
use krataoci::packer::OciImagePacked;
use krataoci::packer::OciPackedImage;
use tokio::sync::Semaphore;
use uuid::Uuid;
use xenclient::{DomainChannel, DomainConfig, DomainDisk, DomainNetworkInterface};
@ -30,7 +30,7 @@ pub struct GuestLaunchRequest {
pub env: HashMap<String, String>,
pub run: Option<Vec<String>>,
pub debug: bool,
pub image: OciImagePacked,
pub image: OciPackedImage,
}
pub struct GuestLauncher {