Lightless-Sync/LightlessClient
8
1
Fork 2
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases 53 Wiki Activity

"improving" pair handler clean up and some other stuff

Browse Source
This commit is contained in:
Essie
2026-01-01 00:33:24 +09:00
parent bbb3375661
commit cfc5c1e0f3
17 changed files with 780 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
LightlessSync/WebAPI/Files/FileDownloadManager.cs
View File

@@ -28,8 +28,6 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
private readonly TextureMetadataHelper _textureMetadataHelper;
private readonly ConcurrentDictionary<ThrottledStream, byte> _activeDownloadStreams;
private readonly SemaphoreSlim _decompressGate =
new(Math.Max(1, Environment.ProcessorCount / 2), Math.Max(1, Environment.ProcessorCount / 2));
private volatile bool _disableDirectDownloads;
private int _consecutiveDirectDownloadFailures;
@@ -502,18 +500,11 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
}
}
private void RemoveStatus(string key)
{
lock (_downloadStatusLock)
{
_downloadStatus.Remove(key);
}
}
private async Task DecompressBlockFileAsync(
string downloadStatusKey,
string blockFilePath,
Dictionary<string, (string Extension, string GamePath)> replacementLookup,
IReadOnlyDictionary<string, long> rawSizeLookup,
string downloadLabel,
CancellationToken ct,
bool skipDownscale)
@@ -542,7 +533,9 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
if (!replacementLookup.TryGetValue(fileHash, out var repl))
{
Logger.LogWarning("{dlName}: No replacement mapping for {fileHash}", downloadLabel, fileHash);
fileBlockStream.Seek(len, SeekOrigin.Current);
// still need to skip bytes:
var skip = checked((int)fileLengthBytes);
fileBlockStream.Position += skip;
continue;
}
@@ -552,37 +545,23 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
// read compressed data
var compressed = new byte[len];
await ReadExactlyAsync(fileBlockStream, compressed.AsMemory(0, len), ct).ConfigureAwait(false);
if (len == 0)
MungeBuffer(compressed);
var decompressed = LZ4Wrapper.Unwrap(compressed);
if (rawSizeLookup.TryGetValue(fileHash, out var expectedRawSize)
&& expectedRawSize > 0
&& decompressed.LongLength != expectedRawSize)
{
await _fileCompactor.WriteAllBytesAsync(filePath, Array.Empty<byte>(), ct).ConfigureAwait(false);
PersistFileToStorage(fileHash, filePath, repl.GamePath, skipDownscale);
Logger.LogWarning("{dlName}: Decompressed size mismatch for {fileHash} (expected {expected}, got {actual})",
downloadLabel, fileHash, expectedRawSize, decompressed.LongLength);
continue;
}
MungeBuffer(compressed);
// limit concurrent decompressions
await _decompressGate.WaitAsync(ct).ConfigureAwait(false);
try
{
var sw = System.Diagnostics.Stopwatch.StartNew();
// decompress
var decompressed = LZ4Wrapper.Unwrap(compressed);
Logger.LogTrace("{dlName}: Unwrap {fileHash} took {ms}ms (compressed {c} bytes, decompressed {d} bytes)",
downloadLabel, fileHash, sw.ElapsedMilliseconds, compressed.Length, decompressed?.Length ?? -1);
// write to file
await _fileCompactor.WriteAllBytesAsync(filePath, decompressed, ct).ConfigureAwait(false);
PersistFileToStorage(fileHash, filePath, repl.GamePath, skipDownscale);
}
finally
{
_decompressGate.Release();
}
await _fileCompactor.WriteAllBytesAsync(filePath, decompressed, ct).ConfigureAwait(false);
PersistFileToStorage(fileHash, filePath, repl.GamePath, skipDownscale);
}
catch (EndOfStreamException)
{
@@ -594,6 +573,8 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
}
}
}
SetStatus(downloadStatusKey, DownloadStatus.Completed);
}
catch (EndOfStreamException)
{
@@ -603,10 +584,6 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
{
Logger.LogError(ex, "{dlName}: Error during block file read", downloadLabel);
}
finally
{
RemoveStatus(downloadStatusKey);
}
}
public async Task<List<DownloadFileTransfer>> InitiateDownloadList(
@@ -644,21 +621,25 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
.. await FilesGetSizes(hashes, ct).ConfigureAwait(false),
];
Logger.LogDebug("Files with size 0 or less: {files}",
string.Join(", ", downloadFileInfoFromService.Where(f => f.Size <= 0).Select(f => f.Hash)));
foreach (var dto in downloadFileInfoFromService.Where(c => c.IsForbidden))
{
if (!_orchestrator.ForbiddenTransfers.Exists(f => string.Equals(f.Hash, dto.Hash, StringComparison.Ordinal)))
_orchestrator.ForbiddenTransfers.Add(new DownloadFileTransfer(dto));
}
CurrentDownloads = [.. downloadFileInfoFromService
CurrentDownloads = downloadFileInfoFromService
.Distinct()
.Select(d => new DownloadFileTransfer(d))
.Where(d => d.CanBeTransferred)];
.Where(d => d.CanBeTransferred)
.ToList();
return CurrentDownloads;
}
private sealed record BatchChunk(string Key, List<DownloadFileTransfer> Items);
private sealed record BatchChunk(string HostKey, string StatusKey, List<DownloadFileTransfer> Items);
private static IEnumerable<List<T>> ChunkList<T>(List<T> items, int chunkSize)
{
@@ -684,6 +665,20 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
var allowDirectDownloads = ShouldUseDirectDownloads();
var replacementLookup = BuildReplacementLookup(fileReplacement);
var rawSizeLookup = new Dictionary<string, long>(StringComparer.OrdinalIgnoreCase);
foreach (var download in CurrentDownloads)
{
if (string.IsNullOrWhiteSpace(download.Hash))
{
continue;
}
if (!rawSizeLookup.TryGetValue(download.Hash, out var existing) || existing <= 0)
{
rawSizeLookup[download.Hash] = download.TotalRaw;
}
}
var directDownloads = new List<DownloadFileTransfer>();
var batchDownloads = new List<DownloadFileTransfer>();
@@ -708,7 +703,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
var chunkSize = (int)Math.Ceiling(list.Count / (double)chunkCount);
return ChunkList(list, chunkSize)
.Select(chunk => new BatchChunk(g.Key, chunk));
.Select((chunk, index) => new BatchChunk(g.Key, $"{g.Key}#{index + 1}", chunk));
})
.ToArray();
@@ -722,7 +717,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
{
_downloadStatus[d.DirectDownloadUrl!] = new FileDownloadStatus
{
DownloadStatus = DownloadStatus.Initializing,
DownloadStatus = DownloadStatus.WaitingForSlot,
TotalBytes = d.Total,
TotalFiles = 1,
TransferredBytes = 0,
@@ -730,12 +725,12 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
};
}
foreach (var g in batchChunks.GroupBy(c => c.Key, StringComparer.Ordinal))
foreach (var chunk in batchChunks)
{
_downloadStatus[g.Key] = new FileDownloadStatus
_downloadStatus[chunk.StatusKey] = new FileDownloadStatus
{
DownloadStatus = DownloadStatus.Initializing,
TotalBytes = g.SelectMany(x => x.Items).Sum(x => x.Total),
DownloadStatus = DownloadStatus.WaitingForQueue,
TotalBytes = chunk.Items.Sum(x => x.Total),
TotalFiles = 1,
TransferredBytes = 0,
TransferredFiles = 0
@@ -759,13 +754,13 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
Task batchTask = batchChunks.Length == 0
? Task.CompletedTask
: Parallel.ForEachAsync(batchChunks, new ParallelOptions { MaxDegreeOfParallelism = workerDop, CancellationToken = ct },
async (chunk, token) => await ProcessBatchChunkAsync(chunk, replacementLookup, token, skipDownscale).ConfigureAwait(false));
async (chunk, token) => await ProcessBatchChunkAsync(chunk, replacementLookup, rawSizeLookup, token, skipDownscale).ConfigureAwait(false));
// direct downloads
Task directTask = directDownloads.Count == 0
? Task.CompletedTask
: Parallel.ForEachAsync(directDownloads, new ParallelOptions { MaxDegreeOfParallelism = workerDop, CancellationToken = ct },
async (d, token) => await ProcessDirectAsync(d, replacementLookup, token, skipDownscale).ConfigureAwait(false));
async (d, token) => await ProcessDirectAsync(d, replacementLookup, rawSizeLookup, token, skipDownscale).ConfigureAwait(false));
await Task.WhenAll(batchTask, directTask).ConfigureAwait(false);
@@ -773,9 +768,14 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
ClearDownload();
}
private async Task ProcessBatchChunkAsync(BatchChunk chunk, Dictionary<string, (string Extension, string GamePath)> replacementLookup, CancellationToken ct, bool skipDownscale)
private async Task ProcessBatchChunkAsync(
BatchChunk chunk,
Dictionary<string, (string Extension, string GamePath)> replacementLookup,
IReadOnlyDictionary<string, long> rawSizeLookup,
CancellationToken ct,
bool skipDownscale)
{
var statusKey = chunk.Key;
var statusKey = chunk.StatusKey;
// enqueue (no slot)
SetStatus(statusKey, DownloadStatus.WaitingForQueue);
@@ -803,10 +803,11 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
if (!File.Exists(blockFile))
{
Logger.LogWarning("{dlName}: Block file missing before extraction, skipping", fi.Name);
SetStatus(statusKey, DownloadStatus.Completed);
return;
}
await DecompressBlockFileAsync(statusKey, blockFile, replacementLookup, fi.Name, ct, skipDownscale).ConfigureAwait(false);
await DecompressBlockFileAsync(statusKey, blockFile, replacementLookup, rawSizeLookup, fi.Name, ct, skipDownscale).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
@@ -823,7 +824,12 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
}
}
private async Task ProcessDirectAsync(DownloadFileTransfer directDownload, Dictionary<string, (string Extension, string GamePath)> replacementLookup, CancellationToken ct, bool skipDownscale)
private async Task ProcessDirectAsync(
DownloadFileTransfer directDownload,
Dictionary<string, (string Extension, string GamePath)> replacementLookup,
IReadOnlyDictionary<string, long> rawSizeLookup,
CancellationToken ct,
bool skipDownscale)
{
var progress = CreateInlineProgress(bytes =>
{
@@ -833,7 +839,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
if (!ShouldUseDirectDownloads() || string.IsNullOrEmpty(directDownload.DirectDownloadUrl))
{
await ProcessDirectAsQueuedFallbackAsync(directDownload, replacementLookup, progress, ct, skipDownscale).ConfigureAwait(false);
await ProcessDirectAsQueuedFallbackAsync(directDownload, replacementLookup, rawSizeLookup, progress, ct, skipDownscale).ConfigureAwait(false);
return;
}
@@ -861,6 +867,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
if (!replacementLookup.TryGetValue(directDownload.Hash, out var repl))
{
Logger.LogWarning("{hash}: No replacement data found for direct download.", directDownload.Hash);
SetStatus(directDownload.DirectDownloadUrl!, DownloadStatus.Completed);
return;
}
@@ -873,13 +880,18 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
byte[] compressedBytes = await File.ReadAllBytesAsync(tempFilename, ct).ConfigureAwait(false);
var decompressedBytes = LZ4Wrapper.Unwrap(compressedBytes);
if (directDownload.TotalRaw > 0 && decompressedBytes.LongLength != directDownload.TotalRaw)
{
throw new InvalidDataException(
$"{directDownload.Hash}: Decompressed size mismatch (expected {directDownload.TotalRaw}, got {decompressedBytes.LongLength})");
}
await _fileCompactor.WriteAllBytesAsync(finalFilename, decompressedBytes, ct).ConfigureAwait(false);
PersistFileToStorage(directDownload.Hash, finalFilename, repl.GamePath, skipDownscale);
MarkTransferredFiles(directDownload.DirectDownloadUrl!, 1);
SetStatus(directDownload.DirectDownloadUrl!, DownloadStatus.Completed);
Logger.LogDebug("Finished direct download of {hash}.", directDownload.Hash);
RemoveStatus(directDownload.DirectDownloadUrl!);
}
catch (OperationCanceledException ex)
{
@@ -902,7 +914,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
try
{
await ProcessDirectAsQueuedFallbackAsync(directDownload, replacementLookup, progress, ct, skipDownscale).ConfigureAwait(false);
await ProcessDirectAsQueuedFallbackAsync(directDownload, replacementLookup, rawSizeLookup, progress, ct, skipDownscale).ConfigureAwait(false);
if (!expectedDirectDownloadFailure && failureCount >= 3 && !_disableDirectDownloads)
{
@@ -929,6 +941,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
private async Task ProcessDirectAsQueuedFallbackAsync(
DownloadFileTransfer directDownload,
Dictionary<string, (string Extension, string GamePath)> replacementLookup,
IReadOnlyDictionary<string, long> rawSizeLookup,
IProgress<long> progress,
CancellationToken ct,
bool skipDownscale)
@@ -956,7 +969,7 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
if (!File.Exists(blockFile))
throw new FileNotFoundException("Block file missing after direct download fallback.", blockFile);
await DecompressBlockFileAsync(statusKey, blockFile, replacementLookup, $"fallback-{directDownload.Hash}", ct, skipDownscale)
await DecompressBlockFileAsync(statusKey, blockFile, replacementLookup, rawSizeLookup, $"fallback-{directDownload.Hash}", ct, skipDownscale)
.ConfigureAwait(false);
}
finally
@@ -1003,11 +1016,15 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
try
{
var entry = _fileDbManager.CreateCacheEntry(filePath);
var mapKind = _textureMetadataHelper.DetermineMapKind(gamePath, filePath);
var entry = _fileDbManager.CreateCacheEntryWithKnownHash(filePath, fileHash);
if (!skipDownscale)
_textureDownscaleService.ScheduleDownscale(fileHash, filePath, mapKind);
if (!skipDownscale && _textureDownscaleService.ShouldScheduleDownscale(filePath))
{
_textureDownscaleService.ScheduleDownscale(
fileHash,
filePath,
() => _textureMetadataHelper.DetermineMapKind(gamePath, filePath));
}
if (entry != null && !string.Equals(entry.Hash, fileHash, StringComparison.OrdinalIgnoreCase))
{

3
LightlessSync/WebAPI/Files/Models/DownloadStatus.cs
View File

@@ -6,5 +6,6 @@ public enum DownloadStatus
WaitingForSlot,
WaitingForQueue,
Downloading,
Decompressing
Decompressing,
Completed
}