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

Use streamable compression

Browse Source
This commit is contained in:
Loporrit
2023-12-18 12:27:22 +00:00
parent 14f0b10244
commit c843af1470
9 changed files with 214 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
MareSynchronos/FileCache/FileCacheManager.cs
View File

@@ -1,4 +1,4 @@
using LZ4;
using K4os.Compression.LZ4.Streams;
using MareSynchronos.Interop;
using MareSynchronos.MareConfiguration;
using MareSynchronos.Services.Mediator;
@@ -221,11 +221,26 @@ public sealed class FileCacheManager : IDisposable
return Path.Combine(_configService.Current.CacheFolder, hash + "." + extension);
}
public async Task<(string, byte[])> GetCompressedFileData(string fileHash, CancellationToken uploadToken)
public async Task<long> GetCompressedFileLength(string fileHash, CancellationToken uploadToken)
{
var fileCache = GetFileCacheByHash(fileHash)!.ResolvedFilepath;
return (fileHash, LZ4Codec.WrapHC(await File.ReadAllBytesAsync(fileCache, uploadToken).ConfigureAwait(false), 0,
(int)new FileInfo(fileCache).Length));
using var fs = File.OpenRead(fileCache);
var cs = new CountedStream(Stream.Null);
using var encstream = LZ4Stream.Encode(cs, new LZ4EncoderSettings(){CompressionLevel=K4os.Compression.LZ4.LZ4Level.L09_HC});
await fs.CopyToAsync(encstream, uploadToken).ConfigureAwait(false);
encstream.Close();
return uploadToken.IsCancellationRequested ? 0 : cs.BytesWritten;
}
public async Task<byte[]> GetCompressedFileData(string fileHash, CancellationToken uploadToken)
{
var fileCache = GetFileCacheByHash(fileHash)!.ResolvedFilepath;
using var fs = File.OpenRead(fileCache);
var ms = new MemoryStream(64 * 1024);
using var encstream = LZ4Stream.Encode(ms, new LZ4EncoderSettings(){CompressionLevel=K4os.Compression.LZ4.LZ4Level.L09_HC});
await fs.CopyToAsync(encstream, uploadToken).ConfigureAwait(false);
encstream.Close();
return ms.ToArray();
}
public FileCacheEntity? GetFileCacheByHash(string hash)

3
MareSynchronos/MareSynchronos.csproj
View File

@@ -33,7 +33,8 @@
<PackageReference Include="Dalamud.ContextMenu" Version="1.3.1" />
<PackageReference Include="DalamudPackager" Version="2.1.12" />
<PackageReference Include="Downloader" Version="3.0.6" />
<PackageReference Include="lz4net" Version="1.0.15.93" />
<PackageReference Include="K4os.Compression.LZ4.Legacy" Version="1.3.6" />
<PackageReference Include="K4os.Compression.LZ4.Streams" Version="1.3.6" />
<PackageReference Include="Meziantou.Analyzer" Version="2.0.92">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

2
MareSynchronos/PlayerData/Export/MareCharaFileManager.cs
View File

@@ -1,5 +1,5 @@
using Dalamud.Game.ClientState.Objects.Types;
using LZ4;
using K4os.Compression.LZ4.Legacy;
using MareSynchronos.API.Data.Enum;
using MareSynchronos.FileCache;
using MareSynchronos.Interop;

6
MareSynchronos/Services/CharacterAnalyzer.cs
View File

@@ -181,16 +181,16 @@ public sealed class CharacterAnalyzer : MediatorSubscriberBase, IDisposable
public bool IsComputed => OriginalSize > 0 && CompressedSize > 0;
public async Task ComputeSizes(FileCacheManager fileCacheManager, CancellationToken token)
{
var compressedsize = await fileCacheManager.GetCompressedFileData(Hash, token).ConfigureAwait(false);
var compressedsize = await fileCacheManager.GetCompressedFileLength(Hash, token).ConfigureAwait(false);
var normalSize = new FileInfo(FilePaths[0]).Length;
var entries = fileCacheManager.GetAllFileCachesByHash(Hash);
foreach (var entry in entries)
{
entry.Size = normalSize;
entry.CompressedSize = compressedsize.Item2.LongLength;
entry.CompressedSize = compressedsize;
}
OriginalSize = normalSize;
CompressedSize = compressedsize.Item2.LongLength;
CompressedSize = compressedsize;
}
public long OriginalSize { get; private set; } = OriginalSize;
public long CompressedSize { get; private set; } = CompressedSize;

2
MareSynchronos/UI/CompactUI.cs
View File

@@ -76,7 +76,7 @@ public class CompactUi : WindowMediatorSubscriberBase
#if DEBUG
string dev = "Dev Build";
var ver = Assembly.GetExecutingAssembly().GetName().Version!;
WindowName = $"Loporrit Sync {dev} ({ver.Major}.{ver.Minor}.{ver.Build}-lop{ver.Revision})###LoporritSyncMainUI";
WindowName = $"Loporrit Sync {dev} ({ver.Major}.{ver.Minor}.{ver.Build}-lop{ver.Revision})###LoporritSyncMainUIDev";
Toggle();
#else
var ver = Assembly.GetExecutingAssembly().GetName().Version;

72
MareSynchronos/Utils/CountedStream.cs Normal file
View File

@@ -0,0 +1,72 @@
namespace MareSynchronos.Utils;
// Counts the number of bytes read/written to an underlying stream
public class CountedStream : Stream
{
private readonly Stream _stream;
public long BytesRead { get; private set; }
public long BytesWritten { get; private set; }
public bool DisposeUnderlying = true;
public Stream UnderlyingStream { get => _stream; }
public CountedStream(Stream underlyingStream)
{
_stream = underlyingStream;
}
protected override void Dispose(bool disposing)
{
if (!DisposeUnderlying)
return;
_stream.Dispose();
}
public override bool CanRead => _stream.CanRead;
public override bool CanSeek => _stream.CanSeek;
public override bool CanWrite => _stream.CanWrite;
public override long Length => _stream.Length;
public override long Position { get => _stream.Position; set => _stream.Position = value; }
public override void Flush()
{
_stream.Flush();
}
public override int Read(byte[] buffer, int offset, int count)
{
int n = _stream.Read(buffer, offset, count);
BytesRead += n;
return n;
}
public async override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
int n = await _stream.ReadAsync(buffer, offset, count, cancellationToken);
BytesRead += n;
return n;
}
public override long Seek(long offset, SeekOrigin origin)
{
return _stream.Seek(offset, origin);
}
public override void SetLength(long value)
{
_stream.SetLength(value);
}
public override void Write(byte[] buffer, int offset, int count)
{
_stream.Write(buffer, offset, count);
BytesWritten += count;
}
public async override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
await _stream.WriteAsync(buffer, offset, count, cancellationToken);
BytesWritten += count;
}
}

100
MareSynchronos/Utils/LimitedStream.cs Normal file
View File

@@ -0,0 +1,100 @@
namespace MareSynchronos.Utils;
// Limits the number of bytes read/written to an underlying stream
public class LimitedStream : Stream
{
private readonly Stream _stream;
public long _estimatedPosition = 0;
public long MaxPosition { get; private init; }
public bool DisposeUnderlying = true;
public Stream UnderlyingStream { get => _stream; }
public LimitedStream(Stream underlyingStream, long byteLimit)
{
_stream = underlyingStream;
try
{
_estimatedPosition = Position;
}
catch { }
MaxPosition = _estimatedPosition + byteLimit;
}
protected override void Dispose(bool disposing)
{
if (!DisposeUnderlying)
return;
_stream.Dispose();
}
public override bool CanRead => _stream.CanRead;
public override bool CanSeek => _stream.CanSeek;
public override bool CanWrite => _stream.CanWrite;
public override long Length => _stream.Length;
public override long Position { get => _stream.Position; set => _stream.Position = _estimatedPosition = value; }
public override void Flush()
{
_stream.Flush();
}
public override int Read(byte[] buffer, int offset, int count)
{
int remainder = (int)long.Clamp(MaxPosition - _estimatedPosition, 0, int.MaxValue);
if (count > remainder)
count = remainder;
int n = _stream.Read(buffer, offset, count);
_estimatedPosition += n;
return n;
}
public async override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
int remainder = (int)long.Clamp(MaxPosition - _estimatedPosition, 0, int.MaxValue);
if (count > remainder)
count = remainder;
int n = await _stream.ReadAsync(buffer, offset, count, cancellationToken);
_estimatedPosition += n;
return n;
}
public override long Seek(long offset, SeekOrigin origin)
{
long result = _stream.Seek(offset, origin);
_estimatedPosition = result;
return result;
}
public override void SetLength(long value)
{
_stream.SetLength(value);
}
public override void Write(byte[] buffer, int offset, int count)
{
int remainder = (int)long.Clamp(MaxPosition - _estimatedPosition, 0, int.MaxValue);
if (count > remainder)
count = remainder;
_stream.Write(buffer, offset, count);
_estimatedPosition += count;
}
public async override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
int remainder = (int)long.Clamp(MaxPosition - _estimatedPosition, 0, int.MaxValue);
if (count > remainder)
count = remainder;
await _stream.WriteAsync(buffer, offset, count, cancellationToken);
_estimatedPosition += count;
}
}

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

@@ -1,11 +1,12 @@
using Dalamud.Utility;
using LZ4;
using K4os.Compression.LZ4.Streams;
using MareSynchronos.API.Data;
using MareSynchronos.API.Dto.Files;
using MareSynchronos.API.Routes;
using MareSynchronos.FileCache;
using MareSynchronos.PlayerData.Handlers;
using MareSynchronos.Services.Mediator;
using MareSynchronos.Utils;
using MareSynchronos.WebAPI.Files.Models;
using Microsoft.Extensions.Logging;
using System.Net;
@@ -49,14 +50,6 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
public bool IsDownloading => !CurrentDownloads.Any();
public static void MungeBuffer(Span<byte> buffer)
{
for (int i = 0; i < buffer.Length; ++i)
{
buffer[i] ^= 42;
}
}
public void CancelDownload()
{
CurrentDownloads.Clear();
@@ -95,27 +88,27 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
base.Dispose(disposing);
}
private static byte MungeByte(int byteOrEof)
private static byte ConvertReadByte(int byteOrEof)
{
if (byteOrEof == -1)
{
throw new EndOfStreamException();
}
return (byte)(byteOrEof ^ 42);
return (byte)byteOrEof;
}
private static (string fileHash, long fileLengthBytes) ReadBlockFileHeader(FileStream fileBlockStream)
{
List<char> hashName = [];
List<char> fileLength = [];
var separator = (char)MungeByte(fileBlockStream.ReadByte());
var separator = (char)ConvertReadByte(fileBlockStream.ReadByte());
if (separator != '#') throw new InvalidDataException("Data is invalid, first char is not #");
bool readHash = false;
while (true)
{
var readChar = (char)MungeByte(fileBlockStream.ReadByte());
var readChar = (char)ConvertReadByte(fileBlockStream.ReadByte());
if (readChar == ':')
{
readHash = true;
@@ -172,8 +165,6 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
{
ct.ThrowIfCancellationRequested();
MungeBuffer(buffer.AsSpan(0, bytesRead));
await fileStream.WriteAsync(buffer.AsMemory(0, bytesRead), ct).ConfigureAwait(false);
progress.Report(bytesRead);
@@ -313,13 +304,14 @@ public partial class FileDownloadManager : DisposableMediatorSubscriberBase
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];
byte[] compressedFileContent = new byte[fileLengthBytes];
_ = await fileBlockStream.ReadAsync(compressedFileContent, token).ConfigureAwait(false);
MungeBuffer(compressedFileContent);
using var decompressedFile = new MemoryStream(64 * 1024);
using var innerFileStream = new LimitedStream(fileBlockStream, fileLengthBytes);
innerFileStream.DisposeUnderlying = false;
using var decStream = LZ4Stream.Decode(innerFileStream, 0, true);
await decStream.CopyToAsync(decompressedFile, token);
var decompressedFile = LZ4Codec.Unwrap(compressedFileContent);
var filePath = _fileDbManager.GetCacheFilePath(fileHash, fileExtension);
await _fileCompactor.WriteAllBytesAsync(filePath, decompressedFile, token).ConfigureAwait(false);
await _fileCompactor.WriteAllBytesAsync(filePath, decompressedFile.ToArray(), token).ConfigureAwait(false);
PersistFileToStorage(fileHash, filePath);
}

10
MareSynchronos/WebAPI/Files/FileUploadManager.cs
View File

@@ -11,6 +11,7 @@ using Microsoft.Extensions.Logging;
using System.Net.Http.Headers;
using System.Net.Http.Json;
namespace MareSynchronos.WebAPI.Files;
public sealed class FileUploadManager : DisposableMediatorSubscriberBase
@@ -164,10 +165,7 @@ public sealed class FileUploadManager : DisposableMediatorSubscriberBase
private async Task UploadFileStream(byte[] compressedFile, string fileHash, bool munged, CancellationToken uploadToken)
{
if (munged)
{
throw new NotImplementedException();
FileDownloadManager.MungeBuffer(compressedFile.AsSpan());
}
using var ms = new MemoryStream(compressedFile);
@@ -234,10 +232,10 @@ public sealed class FileUploadManager : DisposableMediatorSubscriberBase
{
Logger.LogDebug("[{hash}] Compressing", file);
var data = await _fileDbManager.GetCompressedFileData(file.Hash, uploadToken).ConfigureAwait(false);
CurrentUploads.Single(e => string.Equals(e.Hash, data.Item1, StringComparison.Ordinal)).Total = data.Item2.Length;
Logger.LogDebug("[{hash}] Starting upload for {filePath}", data.Item1, _fileDbManager.GetFileCacheByHash(data.Item1)!.ResolvedFilepath);
CurrentUploads.Single(e => string.Equals(e.Hash, file.Hash, StringComparison.Ordinal)).Total = data.Length;
Logger.LogDebug("[{hash}] Starting upload for {filePath}", file.Hash, _fileDbManager.GetFileCacheByHash(file.Hash)!.ResolvedFilepath);
await uploadTask.ConfigureAwait(false);
uploadTask = UploadFile(data.Item2, file.Hash, uploadToken);
uploadTask = UploadFile(data, file.Hash, uploadToken);
uploadToken.ThrowIfCancellationRequested();
}