diff --git a/Duplicati/Library/Main/Database/LocalRestoreDatabase.cs b/Duplicati/Library/Main/Database/LocalRestoreDatabase.cs index 6c108f483..0a340aa4a 100644 --- a/Duplicati/Library/Main/Database/LocalRestoreDatabase.cs +++ b/Duplicati/Library/Main/Database/LocalRestoreDatabase.cs @@ -2172,7 +2172,7 @@ namespace Duplicati.Library.Main.Database reader.ConvertValueToString(1), reader.ConvertValueToInt64(2), reader.ConvertValueToInt64(3), - false + BlockRequestType.Download ); } finally @@ -2223,7 +2223,7 @@ namespace Duplicati.Library.Main.Database reader.ConvertValueToString(1), reader.ConvertValueToInt64(2), reader.ConvertValueToInt64(3), - false + BlockRequestType.Download ); } } diff --git a/Duplicati/Library/Main/Duplicati.Library.Main.csproj b/Duplicati/Library/Main/Duplicati.Library.Main.csproj index dbed8fa85..007e9a9e0 100644 --- a/Duplicati/Library/Main/Duplicati.Library.Main.csproj +++ b/Duplicati/Library/Main/Duplicati.Library.Main.csproj @@ -8,7 +8,7 @@ - + diff --git a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs index 12f424c73..4ad1d088d 100644 --- a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs @@ -247,7 +247,7 @@ namespace Duplicati.Library.Main.Operation.Restore else if (vol_count == 0) { m_volumecount.Remove(blockRequest.VolumeID); - blockRequest.CacheDecrEvict = true; + blockRequest.RequestType = BlockRequestType.CacheEvict; emit_evict = true; } else // vol_count < 0 @@ -291,6 +291,8 @@ namespace Duplicati.Library.Main.Operation.Restore m_active_readers[block_request.BlockID] = m_active_readers.TryGetValue(block_request.BlockID, out var c) ? c + 1 : 1; } + Logging.Log.WriteExplicitMessage(LOGTAG, "BlockCacheGet", "Active readers for block {0}: {1}", block_request.BlockID, m_active_readers[block_request.BlockID]); + // Check if the block is already in the cache, and return it if it is. if (m_block_cache.TryGetValue(block_request.BlockID, out byte[] value)) return value; @@ -316,6 +318,7 @@ namespace Duplicati.Library.Main.Operation.Restore } else { + Logging.Log.WriteExplicitMessage(LOGTAG, "BlockCacheGet", "Block {0} is already being requested, waiting for it to be available", block_request.BlockID); sw_get_wait?.Stop(); } @@ -332,17 +335,17 @@ namespace Duplicati.Library.Main.Operation.Restore /// /// The block request related to the value. /// The byte[] buffer holding the block data. - public void Set(BlockRequest blockRequest, byte[] value) + public void Set(long blockID, byte[] value) { - Logging.Log.WriteExplicitMessage(LOGTAG, "BlockCacheSet", "Setting block {0} in cache", blockRequest.BlockID); + Logging.Log.WriteExplicitMessage(LOGTAG, "BlockCacheSet", "Setting block {0} in cache", blockID); sw_set_set?.Start(); - m_block_cache.Set(blockRequest.BlockID, value); + m_block_cache.Set(blockID, value); sw_set_set?.Stop(); // Notify any waiters that the block is available. sw_set_wake_get?.Start(); - if (m_waiters.TryRemove(blockRequest.BlockID, out var tcs)) + if (m_waiters.TryRemove(blockID, out var tcs)) { sw_set_wake_get?.Stop(); sw_set_wake_set?.Start(); @@ -447,6 +450,7 @@ namespace Duplicati.Library.Main.Operation.Restore new { Input = channels.DecompressedBlock.AsRead(), + Ack = channels.DecompressionAck.AsWrite(), Output = channels.VolumeRequest.AsWrite() }, async self => @@ -460,17 +464,32 @@ namespace Duplicati.Library.Main.Operation.Restore var volume_consumer = Task.Run(async () => { Stopwatch sw_read = options.InternalProfiling ? new() : null; + Stopwatch sw_ack = options.InternalProfiling ? new() : null; Stopwatch sw_set = options.InternalProfiling ? new() : null; try { while (true) { + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeConsumer", null, "Waiting for block request from volume"); sw_read?.Start(); var (block_request, data) = await self.Input.ReadAsync().ConfigureAwait(false); sw_read?.Stop(); + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeConsumer", null, $"Received block request: {block_request.RequestType} for block {block_request.BlockID} from volume {block_request.VolumeID}"); + sw_ack?.Start(); + block_request.RequestType = BlockRequestType.DecompressAck; + while (true) + { + var did_write = await self.Ack.TryWriteAsync(block_request.VolumeID, TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + if (did_write) + break; + await results.TaskControl.ProgressRendevouz(); + } + sw_ack?.Stop(); + + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeConsumer", null, $"Received data for block {block_request.BlockID} from volume {block_request.VolumeID}"); sw_set?.Start(); - cache.Set(block_request, data); + cache.Set(block_request.BlockID, data); sw_set?.Stop(); } } @@ -480,7 +499,7 @@ namespace Duplicati.Library.Main.Operation.Restore if (options.InternalProfiling) { - Logging.Log.WriteProfilingMessage(LOGTAG, "InternalTimings", $"Volume consumer - Read: {sw_read.ElapsedMilliseconds}ms, Set: {sw_set.ElapsedMilliseconds}ms"); + Logging.Log.WriteProfilingMessage(LOGTAG, "InternalTimings", $"Volume consumer - Read: {sw_read.ElapsedMilliseconds}ms, Ack: {sw_ack.ElapsedMilliseconds}ms, Set: {sw_set.ElapsedMilliseconds}ms"); } // Cancel any remaining readers - although there shouldn't be any. @@ -511,22 +530,27 @@ namespace Duplicati.Library.Main.Operation.Restore sw_req?.Start(); var block_request = await req.ReadAsync().ConfigureAwait(false); sw_req?.Stop(); - if (block_request.CacheDecrEvict) + Logging.Log.WriteExplicitMessage(LOGTAG, "BlockHandler", null, $"Received block request: {block_request.RequestType}"); + switch (block_request.RequestType) { - sw_cache?.Start(); - // Target file already had the block. - await cache.CheckCounts(block_request).ConfigureAwait(false); - sw_cache?.Stop(); - } - else - { - sw_get?.Start(); - var datatask = cache.Get(block_request); - sw_get?.Stop(); + case BlockRequestType.Download: + sw_get?.Start(); + var datatask = cache.Get(block_request); + sw_get?.Stop(); - sw_resp?.Start(); - await res.WriteAsync(datatask).ConfigureAwait(false); - sw_resp?.Stop(); + sw_resp?.Start(); + await res.WriteAsync(datatask).ConfigureAwait(false); + sw_resp?.Stop(); + break; + case BlockRequestType.CacheEvict: + sw_cache?.Start(); + // Target file already had the block. + await cache.CheckCounts(block_request).ConfigureAwait(false); + sw_cache?.Stop(); + break; + case BlockRequestType.DecompressAck: + default: + throw new InvalidOperationException($"Unexpected block request type: {block_request.RequestType}"); } } } diff --git a/Duplicati/Library/Main/Operation/Restore/Channels.cs b/Duplicati/Library/Main/Operation/Restore/Channels.cs index 9ce947488..6327301df 100644 --- a/Duplicati/Library/Main/Operation/Restore/Channels.cs +++ b/Duplicati/Library/Main/Operation/Restore/Channels.cs @@ -29,14 +29,8 @@ namespace Duplicati.Library.Main.Operation.Restore /// /// Named channels for the restore operation. /// - internal class Channels + internal class Channels(Options options) { - /// - /// The buffer size for the channels. The buffer size is the number of - /// messages that can be queued up before the sender blocks. - /// - public static int BufferSize; - /// /// Channel between and . /// @@ -45,30 +39,38 @@ namespace Duplicati.Library.Main.Operation.Restore /// /// Channel between and . /// - public readonly IChannel DownloadRequest = ChannelManager.CreateChannel(buffersize: BufferSize); + public readonly IChannel DownloadRequest = ChannelManager.CreateChannel(buffersize: options.RestoreChannelBufferSize); /// /// Channel between and /// - public readonly IChannel<(long, string, TempFile)> DecryptRequest = ChannelManager.CreateChannel<(long, string, TempFile)>(buffersize: BufferSize); + public readonly IChannel<(long, string, TempFile)> DecryptRequest = ChannelManager.CreateChannel<(long, string, TempFile)>(buffersize: options.RestoreChannelBufferSize); + + /// + /// Channel between and to keep track of volumes that are actively being read. + /// + /// Size is computed as '2 * options.RestoreChannelBufferSize + options.RestoreVolumeDecompressors + 1' to avoid deadlocks. + /// + public readonly IChannel DecompressionAck = ChannelManager.CreateChannel(buffersize: 2 * options.RestoreChannelBufferSize + options.RestoreVolumeDecompressors + 1); /// /// Channel between and /// - public readonly IChannel<(BlockRequest, BlockVolumeReader)> DecompressionRequest = ChannelManager.CreateChannel<(BlockRequest, BlockVolumeReader)>(buffersize: BufferSize); + public readonly IChannel<(BlockRequest, BlockVolumeReader)> DecompressionRequest = ChannelManager.CreateChannel<(BlockRequest, BlockVolumeReader)>(buffersize: options.RestoreChannelBufferSize); /// - /// Channel between and holding the decompressed blocks. - public readonly IChannel<(BlockRequest, byte[])> DecompressedBlock = ChannelManager.CreateChannel<(BlockRequest, byte[])>(buffersize: BufferSize); + /// Channel between and holding the decompressed blocks. + /// + public readonly IChannel<(BlockRequest, byte[])> DecompressedBlock = ChannelManager.CreateChannel<(BlockRequest, byte[])>(buffersize: options.RestoreChannelBufferSize); /// /// Channel between , used for requesting and responding volumes to the manager. /// - public readonly IChannel VolumeResponse = ChannelManager.CreateChannel(buffersize: BufferSize); + public readonly IChannel VolumeResponse = ChannelManager.CreateChannel(buffersize: options.RestoreChannelBufferSize); /// /// Channel between and , used for requesting and responding volumes to the manager. /// - public readonly IChannel VolumeRequest = ChannelManager.CreateChannel(buffersize: BufferSize); + public readonly IChannel VolumeRequest = ChannelManager.CreateChannel(buffersize: options.RestoreChannelBufferSize); } } \ No newline at end of file diff --git a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs index adecbb0ea..f1177e31b 100644 --- a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs +++ b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs @@ -55,6 +55,10 @@ namespace Duplicati.Library.Main.Operation.Restore public static int file_processors_restoring_files; public static object file_processor_continue_lock = new object(); public static TaskCompletionSource file_processor_continue = new(); + /// + /// The current file processor ID. Used for debugging. + /// + private static int processor_id = 0; /// /// Runs the file processor process that restores the files that need to be restored. @@ -93,6 +97,8 @@ namespace Duplicati.Library.Main.Operation.Restore // Indicates whether this FileProcessor is still restoring files var decremented = false; + // ID for this file processor, used for debugging. + var my_id = Interlocked.Increment(ref processor_id); try { @@ -107,6 +113,8 @@ namespace Duplicati.Library.Main.Operation.Restore var file = await self.Input.ReadAsync().ConfigureAwait(false); sw_file?.Stop(); + Logging.Log.WriteExplicitMessage(LOGTAG, "FileRestored", null, $"{my_id} Restoring file {file.TargetPath}"); + if (file.BlocksetID == LocalDatabase.FOLDER_BLOCKSET_ID && !options.SkipMetadata) { // Check if there are other FileProcessor's still restoring files @@ -131,7 +139,7 @@ namespace Duplicati.Library.Main.Operation.Restore sw_block?.Start(); var blocks = await db .GetBlocksFromFile(file.BlocksetID, results.TaskControl.ProgressToken) - .ToArrayAsync() + .ToArrayAsync(results.TaskControl.ProgressToken) .ConfigureAwait(false); sw_block?.Stop(); @@ -175,7 +183,7 @@ namespace Duplicati.Library.Main.Operation.Restore else { verified_blocks.Clear(); - missing_blocks = [.. blocks.Select(x => { x.CacheDecrEvict = false; return x; })]; + missing_blocks = [.. blocks.Select(x => { x.RequestType = BlockRequestType.Download; return x; })]; } file = new_file; } @@ -194,7 +202,7 @@ namespace Duplicati.Library.Main.Operation.Restore block.BlockHash, block.BlockSize, block.VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -239,7 +247,7 @@ namespace Duplicati.Library.Main.Operation.Restore blocks[0].BlockHash, blocks[0].BlockSize, blocks[0].VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); } @@ -287,7 +295,7 @@ namespace Duplicati.Library.Main.Operation.Restore sw_req?.Start(); // Burst the block requests to speed up the restore - int burst = Channels.BufferSize; + int burst = (int)Math.Max(options.RestoreChannelBufferSize, 1); int j = 0; for (int i = 0; i < (int)Math.Min(missing_blocks.Count, burst); i++) { @@ -298,7 +306,7 @@ namespace Duplicati.Library.Main.Operation.Restore missing_blocks[i].BlockHash, missing_blocks[i].BlockSize, missing_blocks[i].VolumeID, - false + BlockRequestType.Download ) ).ConfigureAwait(false); } @@ -330,7 +338,7 @@ namespace Duplicati.Library.Main.Operation.Restore missing_blocks[j + burst].BlockHash, missing_blocks[j + burst].BlockSize, missing_blocks[j + burst].VolumeID, - false + BlockRequestType.Download ) ).ConfigureAwait(false); } @@ -364,7 +372,7 @@ namespace Duplicati.Library.Main.Operation.Restore blocks[i].BlockHash, blocks[i].BlockSize, blocks[i].VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -453,7 +461,7 @@ namespace Duplicati.Library.Main.Operation.Restore } sw_work_results?.Stop(); - Logging.Log.WriteVerboseMessage(LOGTAG, "RestoredFile", "Restored file {0}", file.TargetPath); + Logging.Log.WriteVerboseMessage(LOGTAG, "RestoredFile", $"{my_id} Restored file {{0}}", file.TargetPath); } } catch (RetiredException) @@ -627,7 +635,7 @@ namespace Duplicati.Library.Main.Operation.Restore block.BlockHash, block.BlockSize, block.VolumeID, - false + BlockRequestType.Download ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -648,7 +656,7 @@ namespace Duplicati.Library.Main.Operation.Restore block.BlockHash, block.BlockSize, block.VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -859,7 +867,7 @@ namespace Duplicati.Library.Main.Operation.Restore blocks[j].BlockHash, blocks[j].BlockSize, blocks[j].VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); verified_blocks.Add(blocks[j]); diff --git a/Duplicati/Library/Main/Operation/Restore/Interfaces.cs b/Duplicati/Library/Main/Operation/Restore/Interfaces.cs index f53ca643f..c315dcf18 100644 --- a/Duplicati/Library/Main/Operation/Restore/Interfaces.cs +++ b/Duplicati/Library/Main/Operation/Restore/Interfaces.cs @@ -1,27 +1,37 @@ // Copyright (C) 2025, The Duplicati Team // https://duplicati.com, hello@duplicati.com -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the "Software"), +// to deal in the Software without restriction, including without limitation +// the rights to use, copy, modify, merge, publish, distribute, sublicense, +// and/or sell copies of the Software, and to permit persons to whom the // Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in +// +// The above copyright notice and this permission notice shall be included in // all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. namespace Duplicati.Library.Main.Operation.Restore { + /// + /// Represents the type of block request. + /// + public enum BlockRequestType + { + Download, // Request to download a block + CacheEvict, // Request to evict a block from cache + DecompressAck // Acknowledgment for decompression completion + } + /// /// Represents a block request that the `VolumeDownloader` process will use to download a block from the backend. /// @@ -31,14 +41,14 @@ namespace Duplicati.Library.Main.Operation.Restore /// The size of the block. /// The ID of the volume in which the block is stored remotely. /// Flag indicating that this block request should either decrement the block counter for BlockID (for BlockManager) or evict the VolumeID (for VolumeDownloader). - public class BlockRequest(long blockID, long blockOffset, string blockHash, long blockSize, long volumeID, bool cacheDecrEvict) + public class BlockRequest(long blockID, long blockOffset, string blockHash, long blockSize, long volumeID, BlockRequestType requestType) { // Total = 77 bytes public long BlockID { get; } = blockID; public long BlockOffset { get; } = blockOffset; public string BlockHash { get; } = blockHash; public long BlockSize { get; } = blockSize; public long VolumeID { get; } = volumeID; - public bool CacheDecrEvict { get; set; } = cacheDecrEvict; + public BlockRequestType RequestType { get; set; } = requestType; } /// diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 18d558eca..3c393e15a 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -26,8 +26,11 @@ using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; +using System.Threading; using System.Threading.Tasks; +#nullable enable + namespace Duplicati.Library.Main.Operation.Restore { @@ -53,24 +56,24 @@ namespace Duplicati.Library.Main.Operation.Restore /// /// The first task that is reading from the channels. /// - private Task t1; + private Task? t1; /// /// The second task that is reading from the channels. /// - private Task t2; + private Task? t2; /// /// Reads from either of the two channels asynchronously. /// /// The object read from the channel. - public async Task ReadFromEitherAsync() + public async Task ReadFromEitherAsync(CancellationToken token) { // NOTE: This is not a correct external choice, // as we have actually consumed from both channels, // but we only process one of them // This is safe here, because the shutdown only happens on failure termination - t1 ??= channel1.ReadAsync(); - t2 ??= channel2.ReadAsync(); + t1 ??= channel1.ReadAsync(token); + t2 ??= channel2.ReadAsync(token); var r = await Task.WhenAny(t1, t2).ConfigureAwait(false); if (r == t1) @@ -91,7 +94,7 @@ namespace Duplicati.Library.Main.Operation.Restore /// /// The named channels for the restore operation. /// The restore options. - public static Task Run(Channels channels, Options options) + public static Task Run(Channels channels, Options options, RestoreResults results) { return AutomationExtensions.RunTask( new @@ -99,80 +102,210 @@ namespace Duplicati.Library.Main.Operation.Restore VolumeRequest = channels.VolumeRequest.AsRead(), VolumeResponse = channels.VolumeResponse.AsRead(), DecompressRequest = channels.DecompressionRequest.AsWrite(), + DecompressAck = channels.DecompressionAck.AsRead(), DownloadRequest = channels.DownloadRequest.AsWrite(), }, async self => { + // The maximum number of volumes to have in cache at once. If this is exceeded, we'll try to evict the least recently used volume that is not actively in use. + long cache_max = options.RestoreVolumeCacheHint / options.VolumeSize; + // Cache of volume readers. Dictionary cache = []; + // List of which volume was accessed last. Used for cache eviction. + List cache_last_touched = []; + // Cache of volumes. Dictionary tmpfiles = []; - Dictionary> in_flight = []; + // Dictionary to keep track of active downloads. Used for grouping requests to the same volume. + Dictionary> in_flight_downloads = []; + // Dictionary to keep track of volumes that are actively being accessed. Used for cache eviction. + Dictionary in_flight_decompressing = []; + // Dictionary to keep track of the currently responding request to prevent auto evicting during a response. + long? responding = null; - Stopwatch sw_cache_set = options.InternalProfiling ? new() : null; - Stopwatch sw_cache_evict = options.InternalProfiling ? new() : null; - Stopwatch sw_query = options.InternalProfiling ? new() : null; - Stopwatch sw_backend = options.InternalProfiling ? new() : null; - Stopwatch sw_request = options.InternalProfiling ? new() : null; - Stopwatch sw_wakeup = options.InternalProfiling ? new() : null; + Stopwatch? sw_cache_set = options.InternalProfiling ? new() : null; + Stopwatch? sw_cache_evict = options.InternalProfiling ? new() : null; + Stopwatch? sw_query = options.InternalProfiling ? new() : null; + Stopwatch? sw_backend = options.InternalProfiling ? new() : null; + Stopwatch? sw_request = options.InternalProfiling ? new() : null; + Stopwatch? sw_wakeup = options.InternalProfiling ? new() : null; - var rfa = new ReadFromEither(self.VolumeRequest, self.VolumeResponse); + void handle_evict(long volume_id) + { + sw_cache_evict?.Start(); + cache.Remove(volume_id, out var reader); + cache_last_touched.Remove(volume_id); + reader?.Dispose(); + tmpfiles.Remove(volume_id, out var tmpfile); + tmpfile?.Dispose(); + sw_cache_evict?.Stop(); + } + + void handle_ack(long volume_id) + { + Logging.Log.WriteVerboseMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for volume {0}", volume_id); + if (in_flight_decompressing.TryGetValue(volume_id, out var dcount)) + { + if (dcount <= 1) + { + in_flight_decompressing.Remove(volume_id); + if (volume_id != responding && cache_max == 0) + { + handle_evict(volume_id); + } + } + else + in_flight_decompressing[volume_id] = dcount - 1; + } + else + Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Decompression acknowledgment for volume {0} not found", volume_id); + } + + async Task flush_ack_channel() + { + // Check if we need to flush the ack channel + while (true) + { + var (has_read, ack_msg) = await self.DecompressAck.TryReadAsync().ConfigureAwait(false); + + if (!has_read) + break; + + handle_ack(ack_msg); + } + } + + await results.TaskControl.ProgressRendevouz().ConfigureAwait(false); + + var rfa = new ReadFromEither(self.VolumeResponse, self.VolumeRequest); try { while (true) { + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Flushing ack channel before waiting"); + await flush_ack_channel().ConfigureAwait(false); // TODO: CoCol ReadFromAnyAsync deadlocks, so we use a workaround - var msg = await rfa.ReadFromEitherAsync().ConfigureAwait(false); + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Waiting for volume request or response"); + var msg = await rfa.ReadFromEitherAsync(results.TaskControl.ProgressToken).ConfigureAwait(false); switch (msg) { case BlockRequest request: + switch (request.RequestType) { - if (request.CacheDecrEvict) - { - sw_cache_evict?.Start(); - cache.Remove(request.VolumeID, out var reader); - reader?.Dispose(); - tmpfiles.Remove(request.VolumeID, out var tmpfile); - tmpfile?.Dispose(); - sw_cache_evict?.Stop(); - } - else - { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from volume {1}", request.BlockID, request.VolumeID); - sw_request?.Start(); - if (cache.TryGetValue(request.VolumeID, out BlockVolumeReader reader)) + case BlockRequestType.CacheEvict: { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Block {0} found in cache", request.BlockID); - await self.DecompressRequest.WriteAsync((request, reader)).ConfigureAwait(false); + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Evicting volume {0} from cache by request", request.VolumeID); + handle_evict(request.VolumeID); } - else + break; + case BlockRequestType.DecompressAck: { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Block {0} not found in cache, requesting volume {1}", request.BlockID, request.VolumeID); - if (in_flight.TryGetValue(request.VolumeID, out var waiters)) + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", request.BlockID, request.VolumeID); + handle_ack(request.VolumeID); + } + break; + case BlockRequestType.Download: + { + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Got a request for block {0} from volume {1}", request.BlockID, request.VolumeID); + sw_request?.Start(); + if (cache.TryGetValue(request.VolumeID, out BlockVolumeReader? reader)) { - waiters.Add(request); + cache_last_touched.Remove(request.VolumeID); + cache_last_touched.Add(request.VolumeID); + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Block {0} found in cache", request.BlockID); + while (true) + { + var did_write = await self.DecompressRequest.TryWriteAsync((request, reader), TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + if (did_write) + break; + await results.TaskControl.ProgressRendevouz().ConfigureAwait(false); + } + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting decompression of block {0} from cached volume {1}", request.BlockID, request.VolumeID); + if (in_flight_decompressing.TryGetValue(request.VolumeID, out var count)) + in_flight_decompressing[request.VolumeID] = count + 1; + else + in_flight_decompressing[request.VolumeID] = 1; } else { - await self.DownloadRequest.WriteAsync(request.VolumeID).ConfigureAwait(false); - in_flight[request.VolumeID] = [request]; + // Check if downloading another volume would exceed cache limits. + // Cache_max = 0 auto evicts when all of the requests have been ack'ed. + if (cache_max > 0 && (cache.Count + in_flight_downloads.Count) >= cache_max) + { + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Evicting volume"); + // TODO switch based of the eviction strategy. + // fifo / lifo based on both when they were downloaded and when they were used + // random + // Heuristic based of accesses and recency + // Cache would overflow if we request another; we have to evict something, or store the request for later. + + // LRU + for (int i = 0; i < cache_last_touched.Count; i++) + { + var volume_id = cache_last_touched[i]; + if (!(in_flight_decompressing.TryGetValue(volume_id, out var count) && count > 0)) + { + // Entry can be safely evicted + handle_evict(volume_id); + break; + } + } + } + + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Block {0} not found in cache, requesting volume {1}", request.BlockID, request.VolumeID); + if (in_flight_downloads.TryGetValue(request.VolumeID, out var waiters)) + { + waiters.Add(request); + } + else + { + while (true) + { + var did_write = await self.DownloadRequest.TryWriteAsync(request.VolumeID, TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + if (did_write) + break; + await results.TaskControl.ProgressRendevouz().ConfigureAwait(false); + } + in_flight_downloads[request.VolumeID] = [request]; + } } + sw_request?.Stop(); } - sw_request?.Stop(); - } - break; + break; + default: + throw new InvalidOperationException($"Unexpected request type: {request.RequestType}"); } + break; case (long volume_id, TempFile tmpfile, BlockVolumeReader reader): { sw_cache_set?.Start(); cache[volume_id] = reader; + cache_last_touched.Add(volume_id); tmpfiles[volume_id] = tmpfile; sw_cache_set?.Stop(); sw_wakeup?.Start(); - foreach (var request in in_flight[volume_id]) + responding = volume_id; + foreach (var request in in_flight_downloads[volume_id]) { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from volume {1}", request.BlockID, volume_id); - await self.DecompressRequest.WriteAsync((request, reader)).ConfigureAwait(false); + // Ensure ACK channel is empty to avoid deadlock + await flush_ack_channel().ConfigureAwait(false); + + // Request the decompressions + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from newly cached volume {1}", request.BlockID, volume_id); + while (true) + { + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Writing decompression request for block {0} from volume {1}", request.BlockID, volume_id); + var did_write = await self.DecompressRequest.TryWriteAsync((request, reader), TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + if (did_write) + break; + await results.TaskControl.ProgressRendevouz().ConfigureAwait(false); + } + if (in_flight_decompressing.TryGetValue(request.VolumeID, out var count)) + in_flight_decompressing[request.VolumeID] = count + 1; + else + in_flight_decompressing[request.VolumeID] = 1; } - in_flight.Remove(volume_id); + responding = null; + in_flight_downloads.Remove(volume_id); sw_wakeup?.Stop(); break; } @@ -187,7 +320,7 @@ namespace Duplicati.Library.Main.Operation.Restore if (options.InternalProfiling) { - Logging.Log.WriteProfilingMessage(LOGTAG, "InternalTimings", $"CacheSet: {sw_cache_set.ElapsedMilliseconds}ms, CacheEvict: {sw_cache_evict.ElapsedMilliseconds}ms, Query: {sw_query.ElapsedMilliseconds}ms, Backend: {sw_backend.ElapsedMilliseconds}ms, Request: {sw_request.ElapsedMilliseconds}ms, Wakeup: {sw_wakeup.ElapsedMilliseconds}ms"); + Logging.Log.WriteProfilingMessage(LOGTAG, "InternalTimings", $"CacheSet: {sw_cache_set?.ElapsedMilliseconds}ms, CacheEvict: {sw_cache_evict?.ElapsedMilliseconds}ms, Query: {sw_query?.ElapsedMilliseconds}ms, Backend: {sw_backend?.ElapsedMilliseconds}ms, Request: {sw_request?.ElapsedMilliseconds}ms, Wakeup: {sw_wakeup?.ElapsedMilliseconds}ms"); } } catch (Exception ex) diff --git a/Duplicati/Library/Main/Operation/RestoreHandler.cs b/Duplicati/Library/Main/Operation/RestoreHandler.cs index 2334dc0b6..a29f44ac5 100644 --- a/Duplicati/Library/Main/Operation/RestoreHandler.cs +++ b/Duplicati/Library/Main/Operation/RestoreHandler.cs @@ -344,18 +344,16 @@ namespace Duplicati.Library.Main.Operation .ConfigureAwait(false); using var setup_log_timer = new Logging.Timer(LOGTAG, "RestoreNetworkSetup", "RestoreNetworkSetup"); - // Create the channels between BlockManager and FileProcessor - Restore.Channels.BufferSize = m_options.RestoreChannelBufferSize; - var fileprocessor_requests = Enumerable.Range(0, m_options.RestoreFileProcessors).Select(_ => ChannelManager.CreateChannel(buffersize: Restore.Channels.BufferSize)).ToArray(); - var fileprocessor_responses = Enumerable.Range(0, m_options.RestoreFileProcessors).Select(_ => ChannelManager.CreateChannel>(buffersize: Restore.Channels.BufferSize)).ToArray(); + var fileprocessor_requests = Enumerable.Range(0, m_options.RestoreFileProcessors).Select(_ => ChannelManager.CreateChannel(buffersize: m_options.RestoreChannelBufferSize)).ToArray(); + var fileprocessor_responses = Enumerable.Range(0, m_options.RestoreFileProcessors).Select(_ => ChannelManager.CreateChannel>(buffersize: m_options.RestoreChannelBufferSize)).ToArray(); // Create the process network - Restore.Channels channels = new(); + Restore.Channels channels = new(m_options); var filelister = Restore.FileLister.Run(channels, database, m_options, m_result); Restore.FileProcessor.file_processors_restoring_files = m_options.RestoreFileProcessors; var fileprocessors = Enumerable.Range(0, m_options.RestoreFileProcessors).Select(i => Restore.FileProcessor.Run(channels, database, fileprocessor_requests[i], fileprocessor_responses[i], m_options, m_result)).ToArray(); var blockmanager = Restore.BlockManager.Run(channels, database, fileprocessor_requests, fileprocessor_responses, m_options, m_result); - var volumecache = Restore.VolumeManager.Run(channels, m_options); + var volumecache = Restore.VolumeManager.Run(channels, m_options, m_result); var volumedownloaders = Enumerable.Range(0, m_options.RestoreVolumeDownloaders).Select(i => Restore.VolumeDownloader.Run(channels, database, backendManager, m_options, m_result)).ToArray(); var volumedecryptors = Enumerable.Range(0, m_options.RestoreVolumeDecryptors).Select(i => Restore.VolumeDecryptor.Run(channels, backendManager, m_options)).ToArray(); var volumedecompressors = Enumerable.Range(0, m_options.RestoreVolumeDecompressors).Select(i => Restore.VolumeDecompressor.Run(channels, m_options)).ToArray(); diff --git a/Duplicati/Library/Main/Options.cs b/Duplicati/Library/Main/Options.cs index 76c35659f..f81f6ae26 100644 --- a/Duplicati/Library/Main/Options.cs +++ b/Duplicati/Library/Main/Options.cs @@ -164,6 +164,11 @@ namespace Duplicati.Library.Main /// private static readonly int DEFAULT_RESTORE_FILE_PROCESSORS = Math.Max(1, Environment.ProcessorCount / 2); + /// + /// The default value for the maximum size of the restore volume cache + /// + private const string DEFAULT_RESTORE_VOLUME_CACHE_HINT = "4gb"; + /// /// The default value for the number of volume decryptors during restore /// @@ -553,6 +558,7 @@ namespace Duplicati.Library.Main new CommandLineArgument("restore-file-processors", CommandLineArgument.ArgumentType.Integer, Strings.Options.RestoreFileprocessorsShort, Strings.Options.RestoreFileprocessorsLong, DEFAULT_RESTORE_FILE_PROCESSORS.ToString()), new CommandLineArgument("restore-legacy", CommandLineArgument.ArgumentType.Boolean, Strings.Options.RestoreLegacyShort, Strings.Options.RestoreLegacyLong, "false"), new CommandLineArgument("restore-preallocate-size", CommandLineArgument.ArgumentType.Boolean, Strings.Options.RestorePreallocateSizeShort, Strings.Options.RestorePreallocateSizeLong, "false"), + new CommandLineArgument("restore-volume-cache-hint", CommandLineArgument.ArgumentType.Size, Strings.Options.RestoreVolumeCacheHintShort, Strings.Options.RestoreVolumeCacheHintLong, DEFAULT_RESTORE_VOLUME_CACHE_HINT), new CommandLineArgument("restore-volume-decompressors", CommandLineArgument.ArgumentType.Integer, Strings.Options.RestoreVolumeDecompressorsShort, Strings.Options.RestoreVolumeDecompressorsLong, DEFAULT_RESTORE_VOLUME_DECOMPRESSORS.ToString()), new CommandLineArgument("restore-volume-decryptors", CommandLineArgument.ArgumentType.Integer, Strings.Options.RestoreVolumeDecryptorsShort, Strings.Options.RestoreVolumeDecryptorsLong, DEFAULT_RESTORE_VOLUME_DECRYPTORS.ToString()), new CommandLineArgument("restore-volume-downloaders", CommandLineArgument.ArgumentType.Integer, Strings.Options.RestoreVolumeDownloadersShort, Strings.Options.RestoreVolumeDownloadersLong, DEFAULT_RESTORE_VOLUME_DOWNLOADERS.ToString()), @@ -1715,6 +1721,11 @@ namespace Duplicati.Library.Main /// public bool RestorePreAllocate => GetBool("restore-pre-allocate"); + // + // Gets the size of the volume cache used during restore, in MB. + // + public long RestoreVolumeCacheHint => GetSize("restore-volume-cache-hint", "mb", DEFAULT_RESTORE_VOLUME_CACHE_HINT); + /// /// Gets the number of volume decryptors to use in the restore process /// diff --git a/Duplicati/Library/Main/Strings.cs b/Duplicati/Library/Main/Strings.cs index 9f7e0c913..0c69f06b3 100644 --- a/Duplicati/Library/Main/Strings.cs +++ b/Duplicati/Library/Main/Strings.cs @@ -363,6 +363,8 @@ namespace Duplicati.Library.Main.Strings public static string RestoreLegacyLong { get { return LC.L("Use this option to use the legacy restore method. The legacy restore method is slower than the new method, but may be more reliable in some cases."); } } public static string RestorePreallocateSizeShort { get { return LC.L("Preallocate size of restored files"); } } public static string RestorePreallocateSizeLong { get { return LC.L("Use this option to toggle whether to set the size of the restored files before they are written to disk. This can help to reduce fragmentation and improve performance on some filesystems."); } } + public static string RestoreVolumeCacheHintShort { get { return LC.L("Hints to the desired maximum size of the restore volume cache"); } } + public static string RestoreVolumeCacheHintLong { get { return LC.L("Use this option to hint the maximum desired size of the restore volume cache. The restore volume cache is used to store the volumes from the remote storage on local disk. The number of volumes stored is set to floor(hint_size / volume_size). When this number is exceeded, the client tries to evict the least recently used volume(s) that is not actively being used. This means that the number of volumes kept on disk can exceed this value if no volume can be evicted."); } } public static string RestoreVolumeDecompressorsShort { get { return LC.L("Number of concurrent FileDecompressor processes used during restore"); } } public static string RestoreVolumeDecompressorsLong { get { return LC.L("Use this option to set the number of concurrent FileDecompressor processes used during restore. A FileDecompressor processes one volume at a time, and increasing the number of FileDecompressors may improve restore performance if the bottleneck is decompression."); } } public static string RestoreVolumeDecryptorsShort { get { return LC.L("Number of concurrent FileDecryptor processes used during restore"); } } diff --git a/Duplicati/Library/UsageReporter/Duplicati.Library.UsageReporter.csproj b/Duplicati/Library/UsageReporter/Duplicati.Library.UsageReporter.csproj index 3e945d6ed..a5e2ab7ca 100644 --- a/Duplicati/Library/UsageReporter/Duplicati.Library.UsageReporter.csproj +++ b/Duplicati/Library/UsageReporter/Duplicati.Library.UsageReporter.csproj @@ -7,7 +7,7 @@ - + diff --git a/Duplicati/UnitTest/RestoreHandlerTests.cs b/Duplicati/UnitTest/RestoreHandlerTests.cs index e82945292..637dd7160 100644 --- a/Duplicati/UnitTest/RestoreHandlerTests.cs +++ b/Duplicati/UnitTest/RestoreHandlerTests.cs @@ -1,23 +1,24 @@ // Copyright (C) 2025, The Duplicati Team // https://duplicati.com, hello@duplicati.com -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the "Software"), +// to deal in the Software without restriction, including without limitation +// the rights to use, copy, modify, merge, publish, distribute, sublicense, +// and/or sell copies of the Software, and to permit persons to whom the // Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in +// +// The above copyright notice and this permission notice shall be included in // all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. + using System; using System.Collections.Generic; using System.IO; @@ -28,6 +29,8 @@ using Duplicati.Library.Main; using NUnit.Framework; using Assert = NUnit.Framework.Legacy.ClassicAssert; +#nullable enable + namespace Duplicati.UnitTest { public class RestoreHandlerTests : BasicSetupHelper @@ -161,6 +164,59 @@ namespace Duplicati.UnitTest } } + [Test] + [Category("RestoreHandler")] + public async System.Threading.Tasks.Task RestoreVolumeCache([Values("0b", "1mb", "5mb", "1gb")] string cache_size, [Values("0", "1", null)] string? channel_size) + { + var opts = TestOptions; + opts["dblock-size"] = "1mb"; + opts["blocksize"] = "1kb"; + opts["restore-volume-cache-hint"] = cache_size; + if (channel_size != null) + opts["restore-channel-buffer-size"] = channel_size; + opts["restore-legacy"] = "false"; + opts["restore-file-processors"] = "4"; + opts["restore-volume-decryptors"] = "4"; + opts["restore-volume-decompressors"] = "4"; + opts["restore-volume-downloaders"] = "4"; + opts["restore-path"] = RESTOREFOLDER; + + // Write a bunch of files + Random rng = new(); + byte[] data = new byte[1024]; + for (int i = 0; i < 1000; i++) + { + rng.NextBytes(data); + var filePath = Path.Combine(this.DATAFOLDER, $"file{i}"); + File.WriteAllBytes(filePath, data); + } + + using var c = new Controller("file://" + this.TARGETFOLDER, opts, null); + TestUtils.AssertResults(c.Backup([this.DATAFOLDER])); + + // Start a 30 second timeout + var timeout_task = System.Threading.Tasks.Task.Delay(TimeSpan.FromSeconds(30)); + + var restore_task = System.Threading.Tasks.Task.Run(() => + { + TestUtils.AssertResults(c.Restore(["*"])); + }); + + var t = await System.Threading.Tasks.Task.WhenAny(timeout_task, restore_task); + + if (t == timeout_task) + { + c.Abort(); + await restore_task; // Ensure we wait for the restore task to complete + Assert.Fail("Restore timed out"); + } + else if (t == restore_task) + // Throw any exceptions it might have + t.GetAwaiter().GetResult(); + + TestUtils.AssertDirectoryTreesAreEquivalent(this.DATAFOLDER, this.RESTOREFOLDER, true, "Restoring with different volume cache sizes"); + } + [Test] [Category("RestoreHandler")] public void RestoreWithoutLocalData([Values("true", "false")] string noLocalDb, [Values("true", "false")] string patchWithLocalBlocks)