mirror of
https://github.com/Drop-OSS/drop.git
synced 2026-07-24 08:53:04 +10:00
Async downloader, better Proton support (#183)
* feat: async downloader + other fixes * feat: windows command parsing + use library path for install path * feat: better proton support * feat: style fixes and store button now uses in-app * feat: emulator rename + umu emulator fix * feat: bring process creation inline with docs * fix: clippy
This commit is contained in:
DecDuck
GitHub
@@ -11,15 +11,17 @@ use download_manager::util::download_thread_control_flag::{
|
||||
};
|
||||
use download_manager::util::progress_object::{ProgressHandle, ProgressObject, ProgressType};
|
||||
use droplet_rs::manifest::{ChunkData, Manifest};
|
||||
use futures_util::StreamExt;
|
||||
use futures_util::stream::FuturesUnordered;
|
||||
use log::{debug, error, info, warn};
|
||||
use remote::auth::generate_authorization_header;
|
||||
use remote::cache::get_cached_object;
|
||||
use remote::error::RemoteAccessError;
|
||||
use remote::requests::generate_url;
|
||||
use remote::utils::DROP_CLIENT_ASYNC;
|
||||
use serde::Deserialize;
|
||||
use std::collections::HashMap;
|
||||
use std::fmt::Debug;
|
||||
use std::mem;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Instant;
|
||||
@@ -28,7 +30,7 @@ use tokio::sync::mpsc::Sender;
|
||||
use utils::{app_emit, lock, send};
|
||||
|
||||
use crate::downloads::utils::get_disk_available;
|
||||
use crate::library::{on_game_complete, push_game_update, set_partially_installed};
|
||||
use crate::library::{Game, on_game_complete, push_game_update, set_partially_installed};
|
||||
use crate::state::GameStatusManager;
|
||||
|
||||
use super::download_logic::download_game_chunk;
|
||||
@@ -87,9 +89,11 @@ impl GameDownloadAgent {
|
||||
// Don't run by default
|
||||
let control_flag = DownloadThreadControl::new(DownloadThreadControlFlag::Stop);
|
||||
|
||||
let game_name = get_cached_object::<Game>(&format!("game/{}", metadata.id)).map(|v| v.library_path).unwrap_or(metadata.id.clone());
|
||||
|
||||
let base_dir_path = Path::new(&base_dir);
|
||||
info!("base dir {}", base_dir_path.display());
|
||||
let data_base_dir_path = base_dir_path.join(metadata.id.clone());
|
||||
let data_base_dir_path = base_dir_path.join(game_name);
|
||||
info!("data dir path {}", data_base_dir_path.display());
|
||||
|
||||
let stored_manifest = DropData::generate(
|
||||
@@ -166,10 +170,7 @@ impl GameDownloadAgent {
|
||||
|
||||
info!("beginning download for {}...", self.metadata().id);
|
||||
|
||||
let res = self
|
||||
.run()
|
||||
.await
|
||||
.map_err(ApplicationDownloadError::Communication);
|
||||
let res = self.run().await;
|
||||
|
||||
debug!(
|
||||
"{} took {}ms to download",
|
||||
@@ -244,17 +245,16 @@ impl GameDownloadAgent {
|
||||
|
||||
self.download_progress
|
||||
.set_max(dl_info.download_size.try_into().unwrap());
|
||||
self.download_progress
|
||||
.set_size(total_chunks);
|
||||
self.download_progress.set_size(total_chunks);
|
||||
self.download_progress.reset();
|
||||
|
||||
self.disk_progress.set_max(dl_info.install_size.try_into().unwrap());
|
||||
self.disk_progress
|
||||
.set_size(total_chunks);
|
||||
.set_max(dl_info.install_size.try_into().unwrap());
|
||||
self.disk_progress.set_size(total_chunks);
|
||||
self.disk_progress.reset();
|
||||
}
|
||||
|
||||
async fn run(&self) -> Result<bool, RemoteAccessError> {
|
||||
async fn run(&self) -> Result<bool, ApplicationDownloadError> {
|
||||
self.depot_manager.sync_depots().await?;
|
||||
info!("synced depots");
|
||||
self.setup_progress();
|
||||
@@ -278,124 +278,119 @@ impl GameDownloadAgent {
|
||||
completed_chunks.clone()
|
||||
};
|
||||
let chunk_len = manifests_chunks.iter().map(|v| v.1.len()).sum::<usize>();
|
||||
let max_download_threads = borrow_db_checked().settings.max_download_threads;
|
||||
let mut max_download_threads = borrow_db_checked().settings.max_download_threads;
|
||||
if max_download_threads <= 0 {
|
||||
max_download_threads = 1;
|
||||
}
|
||||
|
||||
let (sender, recv) = crossbeam_channel::bounded(16);
|
||||
|
||||
// SAFETY: I pinky-promise
|
||||
// (the scope keeps these in scope)
|
||||
let unsafe_self: &'static GameDownloadAgent = unsafe { mem::transmute(self) };
|
||||
let file_list: &'static HashMap<String, String> = unsafe { mem::transmute(&file_list) };
|
||||
let file_list = &file_list;
|
||||
|
||||
let local_completed_chunks = completed_chunks.clone();
|
||||
|
||||
let download_join_handle = tauri::async_runtime::spawn_blocking(move || {
|
||||
let thread_pool = rayon::ThreadPoolBuilder::new()
|
||||
.num_threads(max_download_threads)
|
||||
.build()
|
||||
.unwrap();
|
||||
thread_pool.scope(move |s| {
|
||||
let mut index = 0;
|
||||
for (version_id, chunks, key) in manifests_chunks.into_iter() {
|
||||
let version_id = &version_id;
|
||||
for (chunk_id, chunk_data) in chunks.into_iter() {
|
||||
let local_sender = sender.clone();
|
||||
let download_progress_handle = ProgressHandle::new(
|
||||
unsafe_self.download_progress.get(index),
|
||||
unsafe_self.download_progress.clone(),
|
||||
);
|
||||
let disk_progress_handle = ProgressHandle::new(
|
||||
unsafe_self.disk_progress.get(index),
|
||||
unsafe_self.disk_progress.clone(),
|
||||
);
|
||||
index += 1;
|
||||
|
||||
let chunk_length = chunk_data.files.iter().map(|v| v.length).sum();
|
||||
|
||||
if *local_completed_chunks.get(&chunk_id).unwrap_or(&false) {
|
||||
download_progress_handle.skip(chunk_length);
|
||||
continue;
|
||||
}
|
||||
|
||||
let sender = unsafe_self.sender.clone();
|
||||
let (depot, permit) = match unsafe_self
|
||||
.depot_manager
|
||||
.next_depot(&unsafe_self.metadata.id, &unsafe_self.metadata.version)
|
||||
{
|
||||
Ok(v) => v,
|
||||
Err(err) => {
|
||||
tauri::async_runtime::spawn(async move {
|
||||
send!(
|
||||
sender,
|
||||
DownloadManagerSignal::Error(
|
||||
ApplicationDownloadError::Communication(err)
|
||||
)
|
||||
);
|
||||
});
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let local_version_id = version_id.clone();
|
||||
s.spawn(move |_| {
|
||||
for i in 0..RETRY_COUNT {
|
||||
let base_path = unsafe_self.dropdata.base_path.clone();
|
||||
match download_game_chunk(
|
||||
&unsafe_self.metadata.id,
|
||||
&local_version_id,
|
||||
&chunk_id,
|
||||
&depot,
|
||||
&key,
|
||||
&chunk_data,
|
||||
file_list,
|
||||
base_path,
|
||||
&unsafe_self.control_flag,
|
||||
&download_progress_handle,
|
||||
&disk_progress_handle,
|
||||
) {
|
||||
Ok(true) => {
|
||||
local_sender.send(chunk_id.clone()).unwrap();
|
||||
drop(permit); // Take ownership
|
||||
return;
|
||||
}
|
||||
Ok(false) => return,
|
||||
Err(e) => {
|
||||
warn!("got error for chunk id {}: {e:?}", chunk_id);
|
||||
|
||||
let retry = true; /*matches!(
|
||||
&e,
|
||||
ApplicationDownloadError::Communication(_)
|
||||
| ApplicationDownloadError::Checksum
|
||||
| ApplicationDownloadError::Lock
|
||||
| ApplicationDownloadError::IoError(_)
|
||||
);*/
|
||||
|
||||
if i == RETRY_COUNT - 1 || !retry {
|
||||
warn!("retry logic failed, not re-attempting.");
|
||||
tauri::async_runtime::spawn(async move {
|
||||
send!(sender, DownloadManagerSignal::Error(e));
|
||||
});
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
drop(sender);
|
||||
});
|
||||
});
|
||||
let mut chunk_completions = FuturesUnordered::new();
|
||||
|
||||
let mut outputs = Vec::new();
|
||||
while let Ok(chunk_id) = recv.recv() {
|
||||
outputs.push(chunk_id);
|
||||
|
||||
let mut handle_output =
|
||||
|value: Result<Option<String>, ApplicationDownloadError>| match value {
|
||||
Ok(value) => {
|
||||
if let Some(chunk_id) = value {
|
||||
outputs.push(chunk_id);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
|
||||
let mut index = 0;
|
||||
for (version_id, chunks, key) in manifests_chunks.into_iter() {
|
||||
let version_id = &version_id;
|
||||
for (chunk_id, chunk_data) in chunks.into_iter() {
|
||||
let download_progress_handle = ProgressHandle::new(
|
||||
self.download_progress.get(index),
|
||||
self.download_progress.clone(),
|
||||
);
|
||||
let disk_progress_handle = ProgressHandle::new(
|
||||
self.disk_progress.get(index),
|
||||
self.disk_progress.clone(),
|
||||
);
|
||||
index += 1;
|
||||
|
||||
let chunk_length = chunk_data.files.iter().map(|v| v.length).sum();
|
||||
|
||||
if *local_completed_chunks.get(&chunk_id).unwrap_or(&false) {
|
||||
download_progress_handle.skip(chunk_length);
|
||||
continue;
|
||||
}
|
||||
|
||||
let (depot, permit) = match self
|
||||
.depot_manager
|
||||
.next_depot(&self.metadata.id, &self.metadata.version)
|
||||
{
|
||||
Ok(v) => v,
|
||||
Err(err) => {
|
||||
return Err(err.into());
|
||||
}
|
||||
};
|
||||
|
||||
let local_version_id = version_id.clone();
|
||||
while chunk_completions.len() >= max_download_threads {
|
||||
handle_output(
|
||||
chunk_completions
|
||||
.next()
|
||||
.await
|
||||
.expect("max download threads is zero?"),
|
||||
)?;
|
||||
}
|
||||
chunk_completions.push(async move {
|
||||
for i in 0..RETRY_COUNT {
|
||||
let base_path = self.dropdata.base_path.clone();
|
||||
match download_game_chunk(
|
||||
&self.metadata.id,
|
||||
&local_version_id,
|
||||
&chunk_id,
|
||||
&depot,
|
||||
&key,
|
||||
&chunk_data,
|
||||
file_list,
|
||||
base_path,
|
||||
&self.control_flag,
|
||||
&download_progress_handle,
|
||||
&disk_progress_handle,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(true) => {
|
||||
drop(permit);
|
||||
return Ok(Some(chunk_id.clone()));
|
||||
}
|
||||
Ok(false) => return Ok(None),
|
||||
Err(e) => {
|
||||
warn!("got error for chunk id {}: {e:?}", chunk_id);
|
||||
|
||||
let retry = true; /*matches!(
|
||||
&e,
|
||||
ApplicationDownloadError::Communication(_)
|
||||
| ApplicationDownloadError::Checksum
|
||||
| ApplicationDownloadError::Lock
|
||||
| ApplicationDownloadError::IoError(_)
|
||||
);*/
|
||||
|
||||
if i == RETRY_COUNT - 1 || !retry {
|
||||
warn!("retry logic failed, not re-attempting.");
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
download_join_handle
|
||||
.await
|
||||
.expect("failed to complete download");
|
||||
while let Some(value) = chunk_completions.next().await {
|
||||
handle_output(value)?
|
||||
}
|
||||
|
||||
for completed_chunk in outputs {
|
||||
completed_chunks.insert(completed_chunk, true);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use std::collections::HashMap;
|
||||
use std::fs::{Permissions, set_permissions};
|
||||
use std::io::{Read, Seek as _, SeekFrom, Write as _};
|
||||
use std::io::SeekFrom;
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::path::PathBuf;
|
||||
@@ -14,19 +14,22 @@ use download_manager::util::download_thread_control_flag::{
|
||||
};
|
||||
use download_manager::util::progress_object::ProgressHandle;
|
||||
use droplet_rs::manifest::ChunkData;
|
||||
use futures_util::StreamExt as _;
|
||||
use log::{debug, info};
|
||||
use remote::auth::generate_authorization_header;
|
||||
use remote::error::{DropServerError, RemoteAccessError};
|
||||
use remote::utils::DROP_CLIENT_SYNC;
|
||||
use remote::utils::DROP_CLIENT_ASYNC;
|
||||
use sha2::Digest;
|
||||
use tauri::Url;
|
||||
use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _, AsyncWriteExt as _};
|
||||
use tokio_util::io::StreamReader;
|
||||
|
||||
const READ_BUF_LEN: usize = 1024 * 1024;
|
||||
|
||||
type Aes128Ctr64LE = ctr::Ctr64LE<aes::Aes128>;
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub fn download_game_chunk(
|
||||
pub async fn download_game_chunk(
|
||||
game_id: &str,
|
||||
version_id: &str,
|
||||
chunk_id: &str,
|
||||
@@ -57,15 +60,16 @@ pub fn download_game_chunk(
|
||||
.join(&format!("content/{}/{}/{}", game_id, version_id, chunk_id))
|
||||
.map_err(|v| ApplicationDownloadError::DownloadError(v.into()))?;
|
||||
|
||||
let response = DROP_CLIENT_SYNC
|
||||
let response = DROP_CLIENT_ASYNC
|
||||
.get(url)
|
||||
.header("Authorization", header)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| ApplicationDownloadError::Communication(e.into()))?;
|
||||
|
||||
if response.status() != 200 {
|
||||
info!("chunk request got status code: {}", response.status());
|
||||
let raw_res = response.text().map_err(|e| {
|
||||
let raw_res = response.text().await.map_err(|e| {
|
||||
ApplicationDownloadError::Communication(RemoteAccessError::FetchErrorLegacy(e.into()))
|
||||
})?;
|
||||
info!("{raw_res}");
|
||||
@@ -89,11 +93,11 @@ pub fn download_game_chunk(
|
||||
|
||||
debug!("took {}ms to start downloading", timestep);
|
||||
|
||||
/*let stream = response
|
||||
let stream = response
|
||||
.bytes_stream()
|
||||
.map(|v| v.map_err(|err| std::io::Error::other(err)));
|
||||
let mut stream_reader = StreamReader::new(stream);*/
|
||||
let mut stream_reader = response;
|
||||
let mut stream_reader = StreamReader::new(stream);
|
||||
//let mut stream_reader = response;
|
||||
|
||||
let mut hasher = sha2::Sha256::new();
|
||||
let mut cipher = Aes128Ctr64LE::new(key.into(), &chunk_data.iv.into());
|
||||
@@ -108,13 +112,14 @@ pub fn download_game_chunk(
|
||||
std::fs::create_dir_all(parent)?;
|
||||
}
|
||||
let mut file_handle = if should_write {
|
||||
let mut file_handle = std::fs::OpenOptions::new()
|
||||
let mut file_handle = tokio::fs::OpenOptions::new()
|
||||
.truncate(false)
|
||||
.write(true)
|
||||
.append(false)
|
||||
.create(true)
|
||||
.open(&path)?;
|
||||
file_handle.seek(SeekFrom::Start(file.start.try_into().unwrap()))?;
|
||||
.open(&path)
|
||||
.await?;
|
||||
file_handle.seek(SeekFrom::Start(file.start.try_into().unwrap())).await?;
|
||||
Some(file_handle)
|
||||
} else {
|
||||
None
|
||||
@@ -122,14 +127,14 @@ pub fn download_game_chunk(
|
||||
|
||||
let mut remaining = file.length;
|
||||
while remaining > 0 {
|
||||
let amount = stream_reader.read(&mut read_buf[0..remaining.min(READ_BUF_LEN)])?;
|
||||
let amount = stream_reader.read(&mut read_buf[0..remaining.min(READ_BUF_LEN)]).await?;
|
||||
download_progress.add(amount);
|
||||
remaining -= amount;
|
||||
|
||||
cipher.apply_keystream(&mut read_buf[0..amount]);
|
||||
//hasher.update(&read_buf[0..amount]);
|
||||
hasher.update(&read_buf[0..amount]);
|
||||
if let Some(file_handle) = &mut file_handle {
|
||||
file_handle.write_all(&read_buf[0..amount])?;
|
||||
file_handle.write_all(&read_buf[0..amount]).await?;
|
||||
disk_progress.add(amount);
|
||||
}
|
||||
}
|
||||
@@ -155,7 +160,7 @@ pub fn download_game_chunk(
|
||||
|
||||
let digest = hex::encode(hasher.finalize());
|
||||
if digest != chunk_data.checksum {
|
||||
//return Err(ApplicationDownloadError::Checksum);
|
||||
return Err(ApplicationDownloadError::Checksum);
|
||||
}
|
||||
|
||||
Ok(true)
|
||||
|
||||
@@ -49,6 +49,7 @@ pub struct Game {
|
||||
pub m_cover_object_id: String,
|
||||
pub m_image_library_object_ids: Vec<String>,
|
||||
pub m_image_carousel_object_ids: Vec<String>,
|
||||
pub library_path: String,
|
||||
}
|
||||
impl Game {
|
||||
pub fn id(&self) -> &String {
|
||||
@@ -246,6 +247,10 @@ pub async fn on_game_complete(
|
||||
.applications
|
||||
.game_statuses
|
||||
.insert(meta.id.clone(), status.clone());
|
||||
db_handle
|
||||
.applications
|
||||
.transient_statuses
|
||||
.remove(meta);
|
||||
drop(db_handle);
|
||||
app_emit!(
|
||||
app_handle,
|
||||
@@ -289,11 +294,6 @@ pub fn push_game_update(
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct FrontendGameOptions {
|
||||
launch_string: String,
|
||||
}
|
||||
|
||||
impl FrontendGameOptions {
|
||||
pub fn launch_string(&self) -> &String {
|
||||
&self.launch_string
|
||||
}
|
||||
pub launch_string: String,
|
||||
pub override_proton_path: Option<String>,
|
||||
}
|
||||
|
||||
@@ -19,7 +19,6 @@ pub fn scan_install_dirs() {
|
||||
if !drop_data_file.exists() {
|
||||
continue;
|
||||
}
|
||||
let game_id = game.file_name().display().to_string();
|
||||
let Ok(drop_data) = DropData::read(&game.path()) else {
|
||||
warn!(
|
||||
".dropdata exists for {}, but couldn't read it. is it corrupted?",
|
||||
@@ -27,7 +26,7 @@ pub fn scan_install_dirs() {
|
||||
);
|
||||
continue;
|
||||
};
|
||||
if db_lock.applications.game_statuses.contains_key(&game_id) {
|
||||
if db_lock.applications.game_statuses.contains_key(&drop_data.game_id) {
|
||||
continue;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user