t0w0bi/ClubPenguinClient
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Single file transfers (#56)

Browse Source 
* move to single file transfer and extraction per server download

* clean up downloads

---------

Co-authored-by: rootdarkarchon <root.darkarchon@outlook.com>
This commit is contained in:
rootdarkarchon
2023-08-01 16:48:16 +02:00
committed by GitHub
parent 1c66c33daa
commit 4e2df4bbc4
4 changed files with 127 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
MareAPI

Submodule MareAPI updated: a5373bca24...4aacbb78bb

2
MareSynchronos/FileCache/PeriodicFileScanner.cs
View File

@@ -63,7 +63,7 @@ public sealed class PeriodicFileScanner : DisposableMediatorSubscriberBase
_scanCancellationTokenSource?.Cancel(); _scanCancellationTokenSource?.Cancel();
_scanCancellationTokenSource = new CancellationTokenSource(); _scanCancellationTokenSource = new CancellationTokenSource();
var token = _scanCancellationTokenSource.Token; var token = _scanCancellationTokenSource.Token;
Task.Run(async () => _ = Task.Run(async () =>
{ {
while (!token.IsCancellationRequested) while (!token.IsCancellationRequested)
{ {

1
MareSynchronos/PlayerData/Handlers/PairHandler.cs
View File

@@ -59,6 +59,7 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
Mediator.Subscribe<FrameworkUpdateMessage>(this, (_) => FrameworkUpdate()); Mediator.Subscribe<FrameworkUpdateMessage>(this, (_) => FrameworkUpdate());
Mediator.Subscribe<ZoneSwitchStartMessage>(this, (_) => Mediator.Subscribe<ZoneSwitchStartMessage>(this, (_) =>
{ {
_downloadCancellationTokenSource?.CancelDispose();
MediatorUnsubscribeFromCharacterChanged(); MediatorUnsubscribeFromCharacterChanged();
_charaHandler?.Invalidate(); _charaHandler?.Invalidate();
IsVisible = false; IsVisible = false;

200
MareSynchronos/WebAPI/Files/FileDownloadManager.cs
View File

@@ -77,20 +77,41 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
base.Dispose(disposing); base.Dispose(disposing);
} }
private async Task DownloadFileHttpClient(string downloadGroup, DownloadFileTransfer fileTransfer, string tempPath, IProgress<long> progress, CancellationToken ct) private static (string fileHash, long fileLengthBytes) ReadBlockFileHeader(FileStream fileBlockStream)
{ {
var requestId = await GetQueueRequest(fileTransfer, ct).ConfigureAwait(false); List<char> hashName = new();
List<char> fileLength = new();
var separator = (char)fileBlockStream.ReadByte();
if (separator != '#') throw new InvalidDataException("Data is invalid, first char is not #");
Logger.LogDebug("GUID {requestId} for file {hash} on server {uri}", requestId, fileTransfer.Hash, fileTransfer.DownloadUri); bool readHash = false;
while (true)
{
var readChar = (char)fileBlockStream.ReadByte();
if (readChar == ':')
{
readHash = true;
continue;
}
if (readChar == '#') break;
if (!readHash) hashName.Add(readChar);
else fileLength.Add(readChar);
}
return (string.Join("", hashName), long.Parse(string.Join("", fileLength)));
}
private async Task DownloadFileHttpClient(string downloadGroup, Guid requestId, List<DownloadFileTransfer> fileTransfer, string tempPath, IProgress<long> progress, CancellationToken ct)
{
Logger.LogDebug("GUID {requestId} on server {uri} for files {files}", requestId, fileTransfer[0].DownloadUri, string.Join(", ", fileTransfer.Select(c => c.Hash).ToList()));
await WaitForDownloadReady(fileTransfer, requestId, ct).ConfigureAwait(false); await WaitForDownloadReady(fileTransfer, requestId, ct).ConfigureAwait(false);
_downloadStatus[downloadGroup].DownloadStatus = DownloadStatus.Downloading; _downloadStatus[downloadGroup].DownloadStatus = DownloadStatus.Downloading;
HttpResponseMessage response = null!; HttpResponseMessage response = null!;
var requestUrl = MareFiles.CacheGetFullPath(fileTransfer.DownloadUri, requestId); var requestUrl = MareFiles.CacheGetFullPath(fileTransfer[0].DownloadUri, requestId);
Logger.LogDebug("Downloading {requestUrl} for file {hash}", requestUrl, fileTransfer.Hash); Logger.LogDebug("Downloading {requestUrl} for request {id}", requestUrl, requestId);
try try
{ {
response = await _orchestrator.SendRequestAsync(HttpMethod.Get, requestUrl, ct).ConfigureAwait(false); response = await _orchestrator.SendRequestAsync(HttpMethod.Get, requestUrl, ct).ConfigureAwait(false);
@@ -144,7 +165,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
private async Task DownloadFilesInternal(GameObjectHandler gameObjectHandler, List<FileReplacementData> fileReplacement, CancellationToken ct) private async Task DownloadFilesInternal(GameObjectHandler gameObjectHandler, List<FileReplacementData> fileReplacement, CancellationToken ct)
{ {
Logger.LogDebug("Downloading files for {id}", gameObjectHandler.Name); Logger.LogDebug("Download start: {id}", gameObjectHandler.Name);
List<DownloadFileDto> downloadFileInfoFromService = new(); List<DownloadFileDto> downloadFileInfoFromService = new();
downloadFileInfoFromService.AddRange(await FilesGetSizes(fileReplacement.Select(f => f.Hash).ToList(), ct).ConfigureAwait(false)); downloadFileInfoFromService.AddRange(await FilesGetSizes(fileReplacement.Select(f => f.Hash).ToList(), ct).ConfigureAwait(false));
@@ -156,7 +177,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
foreach (var dto in downloadFileInfoFromService.Where(c => c.IsForbidden)) foreach (var dto in downloadFileInfoFromService.Where(c => c.IsForbidden))
{ {
if (!_orchestrator.ForbiddenTransfers.Any(f => string.Equals(f.Hash, dto.Hash, StringComparison.Ordinal))) if (!_orchestrator.ForbiddenTransfers.Exists(f => string.Equals(f.Hash, dto.Hash, StringComparison.Ordinal)))
{ {
_orchestrator.ForbiddenTransfers.Add(new DownloadFileTransfer(dto)); _orchestrator.ForbiddenTransfers.Add(new DownloadFileTransfer(dto));
} }
@@ -170,7 +191,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
{ {
DownloadStatus = DownloadStatus.Initializing, DownloadStatus = DownloadStatus.Initializing,
TotalBytes = downloadGroup.Sum(c => c.Total), TotalBytes = downloadGroup.Sum(c => c.Total),
TotalFiles = downloadGroup.Count(), TotalFiles = 1,
TransferredBytes = 0, TransferredBytes = 0,
TransferredFiles = 0 TransferredFiles = 0
}; };
@@ -186,88 +207,96 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
async (fileGroup, token) => async (fileGroup, token) =>
{ {
// let server predownload files // let server predownload files
await _orchestrator.SendRequestAsync(HttpMethod.Post, MareFiles.RequestEnqueueFullPath(fileGroup.First().DownloadUri), var requestIdResponse = await _orchestrator.SendRequestAsync(HttpMethod.Post, MareFiles.RequestEnqueueFullPath(fileGroup.First().DownloadUri),
fileGroup.Select(c => c.Hash), token).ConfigureAwait(false); fileGroup.Select(c => c.Hash), token).ConfigureAwait(false);
Logger.LogDebug("Sent request for {n} files on server {uri} with result {result}", fileGroup.Count(), fileGroup.First().DownloadUri, requestIdResponse.Content.ReadAsStringAsync().Result);
foreach (var file in fileGroup) Guid requestId = Guid.Parse(requestIdResponse.Content.ReadAsStringAsync().Result.Trim('"'));
_downloadReady[requestId] = false;
Logger.LogDebug("GUID {requestId} for {n} files on server {uri}", requestId, fileGroup.Count(), fileGroup.First().DownloadUri);
var blockFile = _fileDbManager.GetCacheFilePath(requestId.ToString("N"), "blk", true);
try
{ {
var ext = fileReplacement.First(f => string.Equals(f.Hash, file.Hash, StringComparison.OrdinalIgnoreCase)).GamePaths.First().Split(".").Last(); _downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.WaitingForSlot;
var tempPath = _fileDbManager.GetCacheFilePath(file.Hash, ext, isTemporaryFile: true); await _orchestrator.WaitForDownloadSlotAsync(token).ConfigureAwait(false);
_downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.WaitingForQueue;
Progress<long> progress = new((bytesDownloaded) => Progress<long> progress = new((bytesDownloaded) =>
{ {
try try
{ {
if (!_downloadStatus.ContainsKey(fileGroup.Key)) return; if (!_downloadStatus.ContainsKey(fileGroup.Key)) return;
_downloadStatus[fileGroup.Key].TransferredBytes += bytesDownloaded; _downloadStatus[fileGroup.Key].TransferredBytes += bytesDownloaded;
file.Transferred += bytesDownloaded;
} }
catch (Exception ex) catch (Exception ex)
{ {
Logger.LogWarning(ex, "Could not set download progress"); Logger.LogWarning(ex, "Could not set download progress");
} }
}); });
await DownloadFileHttpClient(fileGroup.Key, requestId, fileGroup.ToList(), blockFile, progress, token).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
_orchestrator.ReleaseDownloadSlot();
File.Delete(blockFile);
Logger.LogDebug("Detected cancellation, removing {id}", gameObjectHandler);
CancelDownload();
return;
}
catch (Exception ex)
{
_orchestrator.ReleaseDownloadSlot();
File.Delete(blockFile);
Logger.LogError(ex, "Error during download of {id}", requestId);
CancelDownload();
return;
}
try FileStream? fileBlockStream = null;
{ try
_downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.WaitingForSlot; {
await _orchestrator.WaitForDownloadSlotAsync(token).ConfigureAwait(false); _downloadStatus[fileGroup.Key].TransferredFiles = 1;
_downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.WaitingForQueue;
await DownloadFileHttpClient(fileGroup.Key, file, tempPath, progress, token).ConfigureAwait(false);
_downloadStatus[fileGroup.Key].TransferredFiles += 1;
}
catch (OperationCanceledException)
{
File.Delete(tempPath);
Logger.LogDebug("Detected cancellation, removing {id}", gameObjectHandler);
CancelDownload();
return;
}
catch (Exception ex)
{
Logger.LogError(ex, "Error during download of {hash}", file.Hash);
continue;
}
finally
{
_orchestrator.ReleaseDownloadSlot();
}
_downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.Decompressing; _downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.Decompressing;
var tempFileData = await File.ReadAllBytesAsync(tempPath, token).ConfigureAwait(false); fileBlockStream = File.OpenRead(blockFile);
var extractedFile = LZ4Codec.Unwrap(tempFileData); while (fileBlockStream.Position < fileBlockStream.Length)
File.Delete(tempPath);
var filePath = _fileDbManager.GetCacheFilePath(file.Hash, ext, isTemporaryFile: false);
await File.WriteAllBytesAsync(filePath, extractedFile, token).ConfigureAwait(false);
var fi = new FileInfo(filePath);
Func<DateTime> RandomDayInThePast()
{ {
DateTime start = new(1995, 1, 1); (string fileHash, long fileLengthBytes) = ReadBlockFileHeader(fileBlockStream);
Random gen = new();
int range = (DateTime.Today - start).Days;
return () => start.AddDays(gen.Next(range));
}
fi.CreationTime = RandomDayInThePast().Invoke(); try
fi.LastAccessTime = DateTime.Today;
fi.LastWriteTime = RandomDayInThePast().Invoke();
try
{
var entry = _fileDbManager.CreateCacheEntry(filePath);
if (!string.Equals(entry?.Hash, file.Hash, StringComparison.OrdinalIgnoreCase))
{ {
Logger.LogError("Hash mismatch after extracting, got {hash}, expected {expectedHash}, deleting file", entry?.Hash, file.Hash); Logger.LogDebug("Found file {file} with length {le}, decompressing download", fileHash, fileLengthBytes);
File.Delete(filePath); var fileExtension = fileReplacement.First(f => string.Equals(f.Hash, fileHash, StringComparison.OrdinalIgnoreCase)).GamePaths[0].Split(".").Last();
_fileDbManager.RemoveHashedFile(entry);
byte[] compressedFileContent = new byte[fileLengthBytes];
_ = await fileBlockStream.ReadAsync(compressedFileContent, token).ConfigureAwait(false);
var decompressedFile = LZ4Codec.Unwrap(compressedFileContent);
var filePath = _fileDbManager.GetCacheFilePath(fileHash, fileExtension, false);
await File.WriteAllBytesAsync(filePath, decompressedFile, token).ConfigureAwait(false);
PersistFileToStorage(fileHash, filePath);
}
catch (Exception e)
{
Logger.LogWarning(e, "Error during decompression");
} }
} }
catch (Exception ex) }
{ catch (Exception ex)
Logger.LogWarning(ex, "Issue creating cache entry"); {
} Logger.LogError(ex, "Error during block file read");
}
finally
{
_orchestrator.ReleaseDownloadSlot();
fileBlockStream?.Dispose();
File.Delete(blockFile);
} }
}).ConfigureAwait(false); }).ConfigureAwait(false);
Logger.LogDebug("Download for {id} complete", gameObjectHandler); Logger.LogDebug("Download end: {id}", gameObjectHandler);
CancelDownload(); CancelDownload();
} }
@@ -278,19 +307,37 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
return await response.Content.ReadFromJsonAsync<List<DownloadFileDto>>(cancellationToken: ct).ConfigureAwait(false) ?? new List<DownloadFileDto>(); return await response.Content.ReadFromJsonAsync<List<DownloadFileDto>>(cancellationToken: ct).ConfigureAwait(false) ?? new List<DownloadFileDto>();
} }
private async Task<Guid> GetQueueRequest(DownloadFileTransfer downloadFileTransfer, CancellationToken ct) private void PersistFileToStorage(string fileHash, string filePath)
{ {
var response = await _orchestrator.SendRequestAsync(HttpMethod.Get, MareFiles.RequestRequestFileFullPath(downloadFileTransfer.DownloadUri, downloadFileTransfer.Hash), ct).ConfigureAwait(false); var fi = new FileInfo(filePath);
var responseString = await response.Content.ReadAsStringAsync(ct).ConfigureAwait(false); Func<DateTime> RandomDayInThePast()
var requestId = Guid.Parse(responseString.Trim('"'));
if (!_downloadReady.ContainsKey(requestId))
{ {
_downloadReady[requestId] = false; DateTime start = new(1995, 1, 1);
Random gen = new();
int range = (DateTime.Today - start).Days;
return () => start.AddDays(gen.Next(range));
}
fi.CreationTime = RandomDayInThePast().Invoke();
fi.LastAccessTime = DateTime.Today;
fi.LastWriteTime = RandomDayInThePast().Invoke();
try
{
var entry = _fileDbManager.CreateCacheEntry(filePath);
if (!string.Equals(entry?.Hash, fileHash, StringComparison.OrdinalIgnoreCase))
{
Logger.LogError("Hash mismatch after extracting, got {hash}, expected {expectedHash}, deleting file", entry?.Hash, fileHash);
File.Delete(filePath);
_fileDbManager.RemoveHashedFile(entry);
}
}
catch (Exception ex)
{
Logger.LogWarning(ex, "Error creating cache entry");
} }
return requestId;
} }
private async Task WaitForDownloadReady(DownloadFileTransfer downloadFileTransfer, Guid requestId, CancellationToken downloadCt) private async Task WaitForDownloadReady(List<DownloadFileTransfer> downloadFileTransfer, Guid requestId, CancellationToken downloadCt)
{ {
bool alreadyCancelled = false; bool alreadyCancelled = false;
try try
@@ -309,7 +356,8 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
{ {
if (downloadCt.IsCancellationRequested) throw; if (downloadCt.IsCancellationRequested) throw;
var req = await _orchestrator.SendRequestAsync(HttpMethod.Get, MareFiles.RequestCheckQueueFullPath(downloadFileTransfer.DownloadUri, requestId, downloadFileTransfer.Hash), downloadCt).ConfigureAwait(false); var req = await _orchestrator.SendRequestAsync(HttpMethod.Get, MareFiles.RequestCheckQueueFullPath(downloadFileTransfer[0].DownloadUri, requestId),
downloadFileTransfer.Select(c => c.Hash).ToList(), downloadCt).ConfigureAwait(false);
req.EnsureSuccessStatusCode(); req.EnsureSuccessStatusCode();
localTimeoutCts.Dispose(); localTimeoutCts.Dispose();
composite.Dispose(); composite.Dispose();
@@ -328,7 +376,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
{ {
try try
{ {
await _orchestrator.SendRequestAsync(HttpMethod.Get, MareFiles.RequestCancelFullPath(downloadFileTransfer.DownloadUri, requestId)).ConfigureAwait(false); await _orchestrator.SendRequestAsync(HttpMethod.Get, MareFiles.RequestCancelFullPath(downloadFileTransfer[0].DownloadUri, requestId)).ConfigureAwait(false);
alreadyCancelled = true; alreadyCancelled = true;
} }
catch catch
@@ -344,7 +392,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
{ {
try try
{ {
await _orchestrator.SendRequestAsync(HttpMethod.Get, MareFiles.RequestCancelFullPath(downloadFileTransfer.DownloadUri, requestId)).ConfigureAwait(false); await _orchestrator.SendRequestAsync(HttpMethod.Get, MareFiles.RequestCancelFullPath(downloadFileTransfer[0].DownloadUri, requestId)).ConfigureAwait(false);
} }
catch catch
{ {