friendlynamazu/client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

extract partial downloads

Browse Source
This commit is contained in:
Stanley Dimant
2025-02-21 21:29:54 +01:00
parent f3e9f90f9d
commit d94cb6f2f4
3 changed files with 60 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
MareSynchronos/Interop/DalamudLogger.cs
View File

@@ -38,7 +38,8 @@ internal sealed class DalamudLogger : ILogger
else else
{ {
StringBuilder sb = new(); StringBuilder sb = new();
sb.AppendLine($"{unsupported}[{_name}]{{{(int)logLevel}}} {state}{(_hasModifiedGameFiles ? "." : string.Empty)} {exception?.Message}"); sb.Append($"{unsupported}[{_name}]{{{(int)logLevel}}} {state}{(_hasModifiedGameFiles ? "." : string.Empty)} {exception?.Message}");
if (!string.IsNullOrWhiteSpace(exception?.StackTrace))
sb.AppendLine(exception?.StackTrace); sb.AppendLine(exception?.StackTrace);
var innerException = exception?.InnerException; var innerException = exception?.InnerException;
while (innerException != null) while (innerException != null)

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

@@ -223,7 +223,6 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
base.Dispose(disposing); base.Dispose(disposing);
SetUploading(isUploading: false); SetUploading(isUploading: false);
_downloadManager.Dispose();
var name = PlayerName; var name = PlayerName;
Logger.LogDebug("Disposing {name} ({user})", name, Pair); Logger.LogDebug("Disposing {name} ({user})", name, Pair);
try try
@@ -233,6 +232,7 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
_applicationCancellationTokenSource = null; _applicationCancellationTokenSource = null;
_downloadCancellationTokenSource?.CancelDispose(); _downloadCancellationTokenSource?.CancelDispose();
_downloadCancellationTokenSource = null; _downloadCancellationTokenSource = null;
_downloadManager.Dispose();
_charaHandler?.Dispose(); _charaHandler?.Dispose();
_charaHandler = null; _charaHandler = null;
@@ -255,7 +255,7 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
} }
else else
{ {
var cts = new CancellationTokenSource(); using var cts = new CancellationTokenSource();
cts.CancelAfter(TimeSpan.FromSeconds(60)); cts.CancelAfter(TimeSpan.FromSeconds(60));
Logger.LogInformation("[{applicationId}] CachedData is null {isNull}, contains things: {contains}", applicationId, _cachedData == null, _cachedData?.FileReplacements.Any() ?? false); Logger.LogInformation("[{applicationId}] CachedData is null {isNull}, contains things: {contains}", applicationId, _cachedData == null, _cachedData?.FileReplacements.Any() ?? false);
@@ -272,8 +272,6 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
break; break;
} }
} }
cts.CancelDispose();
} }
} }
} }
@@ -386,6 +384,8 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
_ = DownloadAndApplyCharacterAsync(applicationBase, charaData, updatedData, updateModdedPaths, updateManip, downloadToken).ConfigureAwait(false); _ = DownloadAndApplyCharacterAsync(applicationBase, charaData, updatedData, updateModdedPaths, updateManip, downloadToken).ConfigureAwait(false);
} }
private Task? _pairDownloadTask;
private async Task DownloadAndApplyCharacterAsync(Guid applicationBase, CharacterData charaData, Dictionary<ObjectKind, HashSet<PlayerChanges>> updatedData, private async Task DownloadAndApplyCharacterAsync(Guid applicationBase, CharacterData charaData, Dictionary<ObjectKind, HashSet<PlayerChanges>> updatedData,
bool updateModdedPaths, bool updateManip, CancellationToken downloadToken) bool updateModdedPaths, bool updateManip, CancellationToken downloadToken)
{ {
@@ -398,7 +398,12 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
while (toDownloadReplacements.Count > 0 && attempts++ <= 10 && !downloadToken.IsCancellationRequested) while (toDownloadReplacements.Count > 0 && attempts++ <= 10 && !downloadToken.IsCancellationRequested)
{ {
_downloadManager.CancelDownload(); if (_pairDownloadTask != null && !_pairDownloadTask.IsCompleted)
{
Logger.LogDebug("[BASE-{appBase}] Finishing prior running download task for player {name}, {kind}", applicationBase, PlayerName, updatedData);
await _pairDownloadTask.ConfigureAwait(false);
}
Logger.LogDebug("[BASE-{appBase}] Downloading missing files for player {name}, {kind}", applicationBase, PlayerName, updatedData); Logger.LogDebug("[BASE-{appBase}] Downloading missing files for player {name}, {kind}", applicationBase, PlayerName, updatedData);
Mediator.Publish(new EventMessage(new Event(PlayerName, Pair.UserData, nameof(PairHandler), EventSeverity.Informational, Mediator.Publish(new EventMessage(new Event(PlayerName, Pair.UserData, nameof(PairHandler), EventSeverity.Informational,
@@ -407,17 +412,17 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
if (!_playerPerformanceService.ComputeAndAutoPauseOnVRAMUsageThresholds(this, charaData, toDownloadFiles)) if (!_playerPerformanceService.ComputeAndAutoPauseOnVRAMUsageThresholds(this, charaData, toDownloadFiles))
{ {
_downloadManager.CancelDownload(); _downloadManager.ClearDownload();
return; return;
} }
await _downloadManager.DownloadFiles(_charaHandler!, toDownloadReplacements, downloadToken).ConfigureAwait(false); _pairDownloadTask = Task.Run(async () => await _downloadManager.DownloadFiles(_charaHandler!, toDownloadReplacements, downloadToken).ConfigureAwait(false));
_downloadManager.CancelDownload();
await _pairDownloadTask.ConfigureAwait(false);
if (downloadToken.IsCancellationRequested) if (downloadToken.IsCancellationRequested)
{ {
Logger.LogTrace("[BASE-{appBase}] Detected cancellation", applicationBase); Logger.LogTrace("[BASE-{appBase}] Detected cancellation", applicationBase);
_downloadManager.CancelDownload();
return; return;
} }
@@ -428,7 +433,7 @@ public sealed class PairHandler : DisposableMediatorSubscriberBase
break; break;
} }
await Task.Delay(TimeSpan.FromSeconds(2)).ConfigureAwait(false); await Task.Delay(TimeSpan.FromSeconds(2), downloadToken).ConfigureAwait(false);
} }
if (!await _playerPerformanceService.CheckBothThresholds(this, charaData).ConfigureAwait(false)) if (!await _playerPerformanceService.CheckBothThresholds(this, charaData).ConfigureAwait(false))

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

@@ -57,7 +57,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
} }
} }
public void CancelDownload() public void ClearDownload()
{ {
CurrentDownloads.Clear(); CurrentDownloads.Clear();
_downloadStatus.Clear(); _downloadStatus.Clear();
@@ -72,7 +72,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
} }
catch catch
{ {
CancelDownload(); ClearDownload();
} }
finally finally
{ {
@@ -83,8 +83,8 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
protected override void Dispose(bool disposing) protected override void Dispose(bool disposing)
{ {
CancelDownload(); ClearDownload();
foreach (var stream in _activeDownloadStreams) foreach (var stream in _activeDownloadStreams.ToList())
{ {
try try
{ {
@@ -119,7 +119,11 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
bool readHash = false; bool readHash = false;
while (true) while (true)
{ {
var readChar = (char)MungeByte(fileBlockStream.ReadByte()); int readByte = fileBlockStream.ReadByte();
if (readByte == -1)
throw new EndOfStreamException();
var readChar = (char)MungeByte(readByte);
if (readChar == ':') if (readChar == ':')
{ {
readHash = true; readHash = true;
@@ -169,7 +173,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
var bytesRead = 0; var bytesRead = 0;
var limit = _orchestrator.DownloadLimitPerSlot(); var limit = _orchestrator.DownloadLimitPerSlot();
Logger.LogTrace("Starting Download of {id} with a speed limit of {limit}", requestId, limit); Logger.LogTrace("Starting Download of {id} with a speed limit of {limit} to {tempPath}", requestId, limit, tempPath);
stream = new ThrottledStream(await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false), limit); stream = new ThrottledStream(await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false), limit);
_activeDownloadStreams.Add(stream); _activeDownloadStreams.Add(stream);
while ((bytesRead = await stream.ReadAsync(buffer, ct).ConfigureAwait(false)) > 0) while ((bytesRead = await stream.ReadAsync(buffer, ct).ConfigureAwait(false)) > 0)
@@ -186,9 +190,12 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
Logger.LogDebug("{requestUrl} downloaded to {tempPath}", requestUrl, tempPath); Logger.LogDebug("{requestUrl} downloaded to {tempPath}", requestUrl, tempPath);
} }
} }
catch (OperationCanceledException)
{
throw;
}
catch (Exception ex) catch (Exception ex)
{ {
Logger.LogWarning(ex, "Error during file download of {requestUrl}", requestUrl);
try try
{ {
if (!tempPath.IsNullOrEmpty()) if (!tempPath.IsNullOrEmpty())
@@ -271,6 +278,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
Logger.LogDebug("GUID {requestId} for {n} files on server {uri}", requestId, fileGroup.Count(), fileGroup.First().DownloadUri); Logger.LogDebug("GUID {requestId} for {n} files on server {uri}", requestId, fileGroup.Count(), fileGroup.First().DownloadUri);
var blockFile = _fileDbManager.GetCacheFilePath(requestId.ToString("N"), "blk"); var blockFile = _fileDbManager.GetCacheFilePath(requestId.ToString("N"), "blk");
FileInfo fi = new(blockFile);
try try
{ {
_downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.WaitingForSlot; _downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.WaitingForSlot;
@@ -292,26 +300,25 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
} }
catch (OperationCanceledException) catch (OperationCanceledException)
{ {
_orchestrator.ReleaseDownloadSlot(); Logger.LogDebug("{dlName}: Detected cancellation of download, partially extracting files for {id}", fi.Name, gameObjectHandler);
File.Delete(blockFile);
Logger.LogDebug("Detected cancellation, removing {id}", gameObjectHandler);
CancelDownload();
return;
} }
catch (Exception ex) catch (Exception ex)
{ {
_orchestrator.ReleaseDownloadSlot(); _orchestrator.ReleaseDownloadSlot();
File.Delete(blockFile); File.Delete(blockFile);
Logger.LogError(ex, "Error during download of {id}", requestId); Logger.LogError(ex, "{dlName}: Error during download of {id}", fi.Name, requestId);
CancelDownload(); ClearDownload();
return; return;
} }
FileStream? fileBlockStream = null; FileStream? fileBlockStream = null;
try try
{ {
_downloadStatus[fileGroup.Key].TransferredFiles = 1; if (_downloadStatus.TryGetValue(fileGroup.Key, out var status))
_downloadStatus[fileGroup.Key].DownloadStatus = DownloadStatus.Decompressing; {
status.TransferredFiles = 1;
status.DownloadStatus = DownloadStatus.Decompressing;
}
fileBlockStream = File.OpenRead(blockFile); fileBlockStream = File.OpenRead(blockFile);
while (fileBlockStream.Position < fileBlockStream.Length) while (fileBlockStream.Position < fileBlockStream.Length)
{ {
@@ -319,28 +326,40 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
try try
{ {
Logger.LogDebug("Found file {file} with length {le}, decompressing download", fileHash, fileLengthBytes);
var fileExtension = fileReplacement.First(f => string.Equals(f.Hash, fileHash, StringComparison.OrdinalIgnoreCase)).GamePaths[0].Split(".")[^1]; var fileExtension = fileReplacement.First(f => string.Equals(f.Hash, fileHash, StringComparison.OrdinalIgnoreCase)).GamePaths[0].Split(".")[^1];
var filePath = _fileDbManager.GetCacheFilePath(fileHash, fileExtension);
Logger.LogDebug("{dlName}: Decompressing {file}:{le} => {dest}", fi.Name, fileHash, fileLengthBytes, filePath);
byte[] compressedFileContent = new byte[fileLengthBytes]; byte[] compressedFileContent = new byte[fileLengthBytes];
_ = await fileBlockStream.ReadAsync(compressedFileContent, token).ConfigureAwait(false); var readBytes = await fileBlockStream.ReadAsync(compressedFileContent, CancellationToken.None).ConfigureAwait(false);
if (readBytes != fileLengthBytes)
{
throw new EndOfStreamException();
}
MungeBuffer(compressedFileContent); MungeBuffer(compressedFileContent);
var decompressedFile = LZ4Wrapper.Unwrap(compressedFileContent); var decompressedFile = LZ4Wrapper.Unwrap(compressedFileContent);
var filePath = _fileDbManager.GetCacheFilePath(fileHash, fileExtension); await _fileCompactor.WriteAllBytesAsync(filePath, decompressedFile, CancellationToken.None).ConfigureAwait(false);
await _fileCompactor.WriteAllBytesAsync(filePath, decompressedFile, token).ConfigureAwait(false);
PersistFileToStorage(fileHash, filePath); PersistFileToStorage(fileHash, filePath);
} }
catch (EndOfStreamException)
{
Logger.LogWarning("{dlName}: Failure to extract file {fileHash}, stream ended prematurely", fi.Name, fileHash);
}
catch (Exception e) catch (Exception e)
{ {
Logger.LogWarning(e, "Error during decompression"); Logger.LogWarning(e, "{dlName}: Error during decompression", fi.Name);
} }
} }
} }
catch (EndOfStreamException)
{
Logger.LogDebug("{dlName}: Failure to extract file header data, stream ended", fi.Name);
}
catch (Exception ex) catch (Exception ex)
{ {
Logger.LogError(ex, "Error during block file read"); Logger.LogError(ex, "{dlName}: Error during block file read", fi.Name);
} }
finally finally
{ {
@@ -353,7 +372,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
Logger.LogDebug("Download end: {id}", gameObjectHandler); Logger.LogDebug("Download end: {id}", gameObjectHandler);
CancelDownload(); ClearDownload();
} }
private async Task<List<DownloadFileDto>> FilesGetSizes(List<string> hashes, CancellationToken ct) private async Task<List<DownloadFileDto>> FilesGetSizes(List<string> hashes, CancellationToken ct)