From 3c012331e5b6feddf9852efd6244f05a905344f2 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 12:36:33 +0200 Subject: [PATCH 01/28] Removed whitespace --- .../Main/Operation/Restore/Interfaces.cs | 30 +++++++++---------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/Interfaces.cs b/Duplicati/Library/Main/Operation/Restore/Interfaces.cs index f53ca643f..453a72800 100644 --- a/Duplicati/Library/Main/Operation/Restore/Interfaces.cs +++ b/Duplicati/Library/Main/Operation/Restore/Interfaces.cs @@ -1,22 +1,22 @@ // 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 From 44c8793eaaca580e724f0c198eaf57d0d0704499 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 12:38:08 +0200 Subject: [PATCH 02/28] Changed the cache evict flag in BlockRequest to be an enum, as there are now more than 2 types of block request --- .../Main/Database/LocalRestoreDatabase.cs | 4 +-- .../Main/Operation/Restore/BlockManager.cs | 34 +++++++++++-------- .../Main/Operation/Restore/FileProcessor.cs | 18 +++++----- .../Main/Operation/Restore/Interfaces.cs | 14 ++++++-- 4 files changed, 42 insertions(+), 28 deletions(-) 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/Operation/Restore/BlockManager.cs b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs index 12f424c73..7ed4e8224 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 @@ -511,22 +511,26 @@ namespace Duplicati.Library.Main.Operation.Restore sw_req?.Start(); var block_request = await req.ReadAsync().ConfigureAwait(false); sw_req?.Stop(); - if (block_request.CacheDecrEvict) + 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/FileProcessor.cs b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs index adecbb0ea..29a246f3e 100644 --- a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs +++ b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs @@ -175,7 +175,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 +194,7 @@ namespace Duplicati.Library.Main.Operation.Restore block.BlockHash, block.BlockSize, block.VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -239,7 +239,7 @@ namespace Duplicati.Library.Main.Operation.Restore blocks[0].BlockHash, blocks[0].BlockSize, blocks[0].VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); } @@ -298,7 +298,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 +330,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 +364,7 @@ namespace Duplicati.Library.Main.Operation.Restore blocks[i].BlockHash, blocks[i].BlockSize, blocks[i].VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -627,7 +627,7 @@ namespace Duplicati.Library.Main.Operation.Restore block.BlockHash, block.BlockSize, block.VolumeID, - false + BlockRequestType.Download ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -648,7 +648,7 @@ namespace Duplicati.Library.Main.Operation.Restore block.BlockHash, block.BlockSize, block.VolumeID, - true + BlockRequestType.CacheEvict ) ).ConfigureAwait(false); sw_req?.Stop(); @@ -859,7 +859,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 453a72800..c315dcf18 100644 --- a/Duplicati/Library/Main/Operation/Restore/Interfaces.cs +++ b/Duplicati/Library/Main/Operation/Restore/Interfaces.cs @@ -22,6 +22,16 @@ 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; } /// From 0f950b2102200b2b30404e0473a3ab56b4261ad2 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 12:41:17 +0200 Subject: [PATCH 03/28] Updated BlockManager.VolumeConsumer and VolumeManager to ack the decompression requests. Also made VolumeManager keep track of which volumes are safe to dispose. --- .../Main/Operation/Restore/BlockManager.cs | 17 ++- .../Main/Operation/Restore/Channels.cs | 8 +- .../Main/Operation/Restore/VolumeManager.cs | 127 ++++++++++++++---- 3 files changed, 117 insertions(+), 35 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs index 7ed4e8224..431961e05 100644 --- a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs @@ -332,17 +332,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 +447,7 @@ namespace Duplicati.Library.Main.Operation.Restore new { Input = channels.DecompressedBlock.AsRead(), + Ack = channels.DecompressionAck.AsWrite(), Output = channels.VolumeRequest.AsWrite() }, async self => @@ -460,6 +461,7 @@ 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 { @@ -469,8 +471,13 @@ namespace Duplicati.Library.Main.Operation.Restore var (block_request, data) = await self.Input.ReadAsync().ConfigureAwait(false); sw_read?.Stop(); + sw_ack?.Start(); + block_request.RequestType = BlockRequestType.DecompressAck; + await self.Ack.WriteAsync(block_request).ConfigureAwait(false); + sw_ack?.Stop(); + sw_set?.Start(); - cache.Set(block_request, data); + cache.Set(block_request.BlockID, data); sw_set?.Stop(); } } diff --git a/Duplicati/Library/Main/Operation/Restore/Channels.cs b/Duplicati/Library/Main/Operation/Restore/Channels.cs index 9ce947488..04320ac48 100644 --- a/Duplicati/Library/Main/Operation/Restore/Channels.cs +++ b/Duplicati/Library/Main/Operation/Restore/Channels.cs @@ -52,13 +52,19 @@ namespace Duplicati.Library.Main.Operation.Restore /// public readonly IChannel<(long, string, TempFile)> DecryptRequest = ChannelManager.CreateChannel<(long, string, TempFile)>(buffersize: BufferSize); + /// + /// Channel between and to keep track of volumes that are actively being read. + /// + public readonly IChannel DecompressionAck = ChannelManager.CreateChannel(buffersize: BufferSize); + /// /// Channel between and /// public readonly IChannel<(BlockRequest, BlockVolumeReader)> DecompressionRequest = ChannelManager.CreateChannel<(BlockRequest, BlockVolumeReader)>(buffersize: BufferSize); /// - /// Channel between and holding the decompressed blocks. + /// Channel between and holding the decompressed blocks. + /// public readonly IChannel<(BlockRequest, byte[])> DecompressedBlock = ChannelManager.CreateChannel<(BlockRequest, byte[])>(buffersize: BufferSize); /// diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 18d558eca..e8cdbd94b 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -99,13 +99,23 @@ 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. + long cache_max = options.RestoreVolumeCacheMax / 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 = []; Stopwatch sw_cache_set = options.InternalProfiling ? new() : null; Stopwatch sw_cache_evict = options.InternalProfiling ? new() : null; @@ -114,7 +124,7 @@ namespace Duplicati.Library.Main.Operation.Restore Stopwatch sw_request = options.InternalProfiling ? new() : null; Stopwatch sw_wakeup = options.InternalProfiling ? new() : null; - var rfa = new ReadFromEither(self.VolumeRequest, self.VolumeResponse); + var rfa = new ReadFromEitherPrioritized(self.DecompressAck, self.VolumeResponse, self.VolumeRequest); try { while (true) @@ -124,55 +134,114 @@ namespace Duplicati.Library.Main.Operation.Restore 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); + sw_cache_evict?.Start(); + cache.Remove(request.VolumeID, out var reader); + cache_last_touched.Remove(request.VolumeID); + reader?.Dispose(); + tmpfiles.Remove(request.VolumeID, out var tmpfile); + tmpfile?.Dispose(); + sw_cache_evict?.Stop(); } - 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); + if (in_flight_decompressing.TryGetValue(request.VolumeID, out var count)) { - waiters.Add(request); + if (count <= 1) + in_flight_decompressing.Remove(request.VolumeID); + else + in_flight_decompressing[request.VolumeID] = count - 1; + } + else + Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Decompression acknowledgment for block {0} from volume {1} not found", request.BlockID, request.VolumeID); + } + break; + case BlockRequestType.Download: + { + 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)) + { + cache_last_touched.Remove(request.VolumeID); + cache_last_touched.Add(request.VolumeID); + 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", "Requesting block {0} from 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. + if ((cache.Count + in_flight_downloads.Count) >= cache_max) + { + Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "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 + cache.Remove(volume_id); + cache_last_touched.RemoveAt(i); + tmpfiles.Remove(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 + { + await self.DownloadRequest.WriteAsync(request.VolumeID).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]) + foreach (var request in in_flight_downloads[volume_id]) { + // Request the decompressions Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from volume {1}", request.BlockID, volume_id); await self.DecompressRequest.WriteAsync((request, reader)).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); + in_flight_downloads.Remove(volume_id); sw_wakeup?.Stop(); break; } From 8dfe328ea8482cd965574c16e2a9cef07666bc03 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 12:42:50 +0200 Subject: [PATCH 04/28] Updated the ReadFromEither workaround to include another channel, and to be prioritized based on the order of the channels. --- .../Main/Operation/Restore/Channels.cs | 1 + .../Main/Operation/Restore/VolumeManager.cs | 32 +++++++++++++++---- 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/Channels.cs b/Duplicati/Library/Main/Operation/Restore/Channels.cs index 04320ac48..93ae32fc4 100644 --- a/Duplicati/Library/Main/Operation/Restore/Channels.cs +++ b/Duplicati/Library/Main/Operation/Restore/Channels.cs @@ -52,6 +52,7 @@ namespace Duplicati.Library.Main.Operation.Restore /// public readonly IChannel<(long, string, TempFile)> DecryptRequest = ChannelManager.CreateChannel<(long, string, TempFile)>(buffersize: BufferSize); + // TODO DecompressionAck could be made into a channel of `long`, but because of the ReadFromEitherPrioritized workaround, it has to be an `object`. /// /// Channel between and to keep track of volumes that are actively being read. /// diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index e8cdbd94b..8fb9a832c 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -43,12 +43,13 @@ namespace Duplicati.Library.Main.Operation.Restore private static readonly string LOGTAG = Logging.Log.LogTagFromType(); /// - /// Helper class to read from either of two channels. + /// Helper class to read from either of three channels, in a prioritized manner, where channel1 is consumed first. /// This is a workaround for the fact that CoCoL seems to deadlock on ReadFroAnyAsync /// /// The first channel to read from. /// The second channel to read from. - private sealed class ReadFromEither(IReadChannel channel1, IReadChannel channel2) + /// The third channel to read from. + private sealed class ReadFromEitherPrioritized(IReadChannel channel1, IReadChannel channel2, IReadChannel channel3) { /// /// The first task that is reading from the channels. @@ -58,6 +59,10 @@ namespace Duplicati.Library.Main.Operation.Restore /// The second task that is reading from the channels. /// private Task t2; + /// + /// The third task that is reading from the channels. + /// + private Task t3; /// /// Reads from either of the two channels asynchronously. @@ -71,17 +76,30 @@ namespace Duplicati.Library.Main.Operation.Restore // This is safe here, because the shutdown only happens on failure termination t1 ??= channel1.ReadAsync(); t2 ??= channel2.ReadAsync(); + t3 ??= channel3.ReadAsync(); - var r = await Task.WhenAny(t1, t2).ConfigureAwait(false); - if (r == t1) + // Wait for any of the tasks to complete. + await Task.WhenAny(t1, t2, t3).ConfigureAwait(false); + + // Check the channels in priority + // If t1 is completed, we process it first + if (t1.IsCompletedSuccessfully) { + var result = await t1.ConfigureAwait(false); t1 = null; - return await r.ConfigureAwait(false); + return result; } - else + else if (t2.IsCompletedSuccessfully) { + var result = await t2.ConfigureAwait(false); t2 = null; - return await r.ConfigureAwait(false); + return result; + } + else // t3.IsCompletedSuccessfully + { + var result = await t3.ConfigureAwait(false); + t3 = null; + return result; } } } From dc4d8fa58730a597f34411d99818c024e4564d3f Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 12:44:59 +0200 Subject: [PATCH 05/28] Added the option to control the size of the volume cache during restore --- Duplicati/Library/Main/Options.cs | 11 +++++++++++ Duplicati/Library/Main/Strings.cs | 2 ++ 2 files changed, 13 insertions(+) diff --git a/Duplicati/Library/Main/Options.cs b/Duplicati/Library/Main/Options.cs index 7dbad81b7..ef22878d8 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_MAX = "4gb"; + /// /// The default value for the number of volume decryptors during restore /// @@ -549,6 +554,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-max", CommandLineArgument.ArgumentType.Size, Strings.Options.RestoreVolumeCacheMaxShort, Strings.Options.RestoreVolumeCacheMaxLong, DEFAULT_RESTORE_VOLUME_CACHE_MAX), 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()), @@ -1697,6 +1703,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 RestoreVolumeCacheMax => GetSize("restore-volume-cache-max", "mb", DEFAULT_RESTORE_VOLUME_CACHE_MAX); + /// /// 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 2ecd00e91..1280518c8 100644 --- a/Duplicati/Library/Main/Strings.cs +++ b/Duplicati/Library/Main/Strings.cs @@ -356,6 +356,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 RestoreVolumeCacheMaxShort { get { return LC.L("Maximum size of the restore volume cache"); } } + public static string RestoreVolumeCacheMaxLong { get { return LC.L("Use this option to set the maximum size of the restore volume cache. The restore volume cache is used to store the volumes that are downloaded from the remote storage. The number of volumes stored is set to floor(max_size / volume_size), except for when it's set to 0, which disables the volume cache."); } } 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"); } } From 2ad6a9e73002ae54c00f367eb3ca279d2de07b66 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 14:08:41 +0200 Subject: [PATCH 06/28] Reverted ReadFromEither back to two channels alongside an explicit flushing of the ack channel to prevent deadlock. --- .../Main/Operation/Restore/VolumeManager.cs | 61 +++++++++++-------- 1 file changed, 35 insertions(+), 26 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 8fb9a832c..4bd699a8e 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -43,13 +43,12 @@ namespace Duplicati.Library.Main.Operation.Restore private static readonly string LOGTAG = Logging.Log.LogTagFromType(); /// - /// Helper class to read from either of three channels, in a prioritized manner, where channel1 is consumed first. + /// Helper class to read from either of two channels. /// This is a workaround for the fact that CoCoL seems to deadlock on ReadFroAnyAsync /// /// The first channel to read from. /// The second channel to read from. - /// The third channel to read from. - private sealed class ReadFromEitherPrioritized(IReadChannel channel1, IReadChannel channel2, IReadChannel channel3) + private sealed class ReadFromEither(IReadChannel channel1, IReadChannel channel2) { /// /// The first task that is reading from the channels. @@ -59,10 +58,6 @@ namespace Duplicati.Library.Main.Operation.Restore /// The second task that is reading from the channels. /// private Task t2; - /// - /// The third task that is reading from the channels. - /// - private Task t3; /// /// Reads from either of the two channels asynchronously. @@ -76,30 +71,17 @@ namespace Duplicati.Library.Main.Operation.Restore // This is safe here, because the shutdown only happens on failure termination t1 ??= channel1.ReadAsync(); t2 ??= channel2.ReadAsync(); - t3 ??= channel3.ReadAsync(); - // Wait for any of the tasks to complete. - await Task.WhenAny(t1, t2, t3).ConfigureAwait(false); - - // Check the channels in priority - // If t1 is completed, we process it first - if (t1.IsCompletedSuccessfully) + var r = await Task.WhenAny(t1, t2).ConfigureAwait(false); + if (r == t1) { - var result = await t1.ConfigureAwait(false); t1 = null; - return result; + return await r.ConfigureAwait(false); } - else if (t2.IsCompletedSuccessfully) + else { - var result = await t2.ConfigureAwait(false); t2 = null; - return result; - } - else // t3.IsCompletedSuccessfully - { - var result = await t3.ConfigureAwait(false); - t3 = null; - return result; + return await r.ConfigureAwait(false); } } } @@ -142,11 +124,35 @@ namespace Duplicati.Library.Main.Operation.Restore Stopwatch sw_request = options.InternalProfiling ? new() : null; Stopwatch sw_wakeup = options.InternalProfiling ? new() : null; - var rfa = new ReadFromEitherPrioritized(self.DecompressAck, self.VolumeResponse, self.VolumeRequest); + 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); + var ack_request = ack_msg as BlockRequest; + if (!has_read) + break; + Logging.Log.WriteVerboseMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", ack_request.BlockID, ack_request.VolumeID); + if (in_flight_decompressing.TryGetValue(ack_request.VolumeID, out var dcount)) + { + if (dcount <= 1) + in_flight_decompressing.Remove(ack_request.VolumeID); + else + in_flight_decompressing[ack_request.VolumeID] = dcount - 1; + } + else + Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Decompression acknowledgment for block {0} from volume {1} not found", ack_request.BlockID, ack_request.VolumeID); + } + } + + + var rfa = new ReadFromEither(self.VolumeResponse, self.VolumeRequest); try { while (true) { + await flush_ack_channel().ConfigureAwait(false); // TODO: CoCol ReadFromAnyAsync deadlocks, so we use a workaround var msg = await rfa.ReadFromEitherAsync().ConfigureAwait(false); switch (msg) @@ -251,6 +257,9 @@ namespace Duplicati.Library.Main.Operation.Restore sw_wakeup?.Start(); foreach (var request in in_flight_downloads[volume_id]) { + // 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 volume {1}", request.BlockID, volume_id); await self.DecompressRequest.WriteAsync((request, reader)).ConfigureAwait(false); From 149c22a9b8a326e84dff0f3ea8fbb907914423ac Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 14:10:04 +0200 Subject: [PATCH 07/28] Added the new timer in the volume consumer to the profiling output. --- Duplicati/Library/Main/Operation/Restore/BlockManager.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs index 431961e05..91236ee8d 100644 --- a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs @@ -487,7 +487,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. From 9d9dd950030de18f0053dc19dd31527a97150561 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 14:11:21 +0200 Subject: [PATCH 08/28] Made some of the debug logging more differentiable --- Duplicati/Library/Main/Operation/Restore/VolumeManager.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 4bd699a8e..e63ea2eef 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -188,7 +188,7 @@ namespace Duplicati.Library.Main.Operation.Restore break; case BlockRequestType.Download: { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from volume {1}", request.BlockID, request.VolumeID); + 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)) { @@ -196,7 +196,7 @@ namespace Duplicati.Library.Main.Operation.Restore cache_last_touched.Add(request.VolumeID); 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", "Requesting block {0} from volume {1}", request.BlockID, request.VolumeID); + 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 @@ -261,7 +261,7 @@ namespace Duplicati.Library.Main.Operation.Restore await flush_ack_channel().ConfigureAwait(false); // Request the decompressions - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from volume {1}", request.BlockID, volume_id); + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from newly cached volume {1}", request.BlockID, volume_id); await self.DecompressRequest.WriteAsync((request, reader)).ConfigureAwait(false); if (in_flight_decompressing.TryGetValue(request.VolumeID, out var count)) in_flight_decompressing[request.VolumeID] = count + 1; From f54ccbcca65ab66328491080c1ec94dfde0b8122 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 15:44:23 +0200 Subject: [PATCH 09/28] Made VolumeManager nullable --- .../Main/Operation/Restore/VolumeManager.cs | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index e63ea2eef..83113b053 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -28,6 +28,8 @@ using System.Diagnostics; using System.Linq; using System.Threading.Tasks; +#nullable enable + namespace Duplicati.Library.Main.Operation.Restore { @@ -53,11 +55,11 @@ 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. @@ -117,12 +119,12 @@ namespace Duplicati.Library.Main.Operation.Restore // Dictionary to keep track of volumes that are actively being accessed. Used for cache eviction. Dictionary in_flight_decompressing = []; - 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; async Task flush_ack_channel() { @@ -190,7 +192,7 @@ namespace Duplicati.Library.Main.Operation.Restore { 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)) + if (cache.TryGetValue(request.VolumeID, out BlockVolumeReader? reader)) { cache_last_touched.Remove(request.VolumeID); cache_last_touched.Add(request.VolumeID); @@ -283,7 +285,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) From 8087823ee549d9ce2cc9ae5b3699b6faabb2b6b4 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 15:45:00 +0200 Subject: [PATCH 10/28] Moved internal VolumeManager functionality to functions --- .../Main/Operation/Restore/VolumeManager.cs | 73 +++++++++++-------- 1 file changed, 43 insertions(+), 30 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 83113b053..34aac8b13 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -126,25 +126,52 @@ namespace Duplicati.Library.Main.Operation.Restore Stopwatch? sw_request = options.InternalProfiling ? new() : null; Stopwatch? sw_wakeup = options.InternalProfiling ? new() : null; + 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 block_id, long volume_id) + { + Logging.Log.WriteVerboseMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", block_id, volume_id); + if (in_flight_decompressing.TryGetValue(volume_id, out var dcount)) + { + if (dcount <= 1) + { + in_flight_decompressing.Remove(volume_id); + if (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 block {0} from volume {1} not found", block_id, 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); - var ack_request = ack_msg as BlockRequest; if (!has_read) break; - Logging.Log.WriteVerboseMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", ack_request.BlockID, ack_request.VolumeID); - if (in_flight_decompressing.TryGetValue(ack_request.VolumeID, out var dcount)) + if (ack_msg is not BlockRequest ack_request) { - if (dcount <= 1) - in_flight_decompressing.Remove(ack_request.VolumeID); - else - in_flight_decompressing[ack_request.VolumeID] = dcount - 1; + Logging.Log.WriteErrorMessage(LOGTAG, "VolumeRequest", null, "Unexpected message type in ack channel: {0}", ack_msg?.GetType().Name); + break; } - else - Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Decompression acknowledgment for block {0} from volume {1} not found", ack_request.BlockID, ack_request.VolumeID); + + handle_ack(ack_request.BlockID, ack_request.VolumeID); } } @@ -164,28 +191,16 @@ namespace Duplicati.Library.Main.Operation.Restore { case BlockRequestType.CacheEvict: { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Evicting volume {0} from cache by request", request.VolumeID); - sw_cache_evict?.Start(); - cache.Remove(request.VolumeID, out var reader); - cache_last_touched.Remove(request.VolumeID); - reader?.Dispose(); - tmpfiles.Remove(request.VolumeID, out var tmpfile); - tmpfile?.Dispose(); - sw_cache_evict?.Stop(); + if (cache_max > 0) + { + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Evicting volume {0} from cache by request", request.VolumeID); + handle_evict(request.VolumeID); + } } break; case BlockRequestType.DecompressAck: { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", request.BlockID, request.VolumeID); - if (in_flight_decompressing.TryGetValue(request.VolumeID, out var count)) - { - if (count <= 1) - in_flight_decompressing.Remove(request.VolumeID); - else - in_flight_decompressing[request.VolumeID] = count - 1; - } - else - Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Decompression acknowledgment for block {0} from volume {1} not found", request.BlockID, request.VolumeID); + handle_ack(request.BlockID, request.VolumeID); } break; case BlockRequestType.Download: @@ -223,9 +238,7 @@ namespace Duplicati.Library.Main.Operation.Restore if (!(in_flight_decompressing.TryGetValue(volume_id, out var count) && count > 0)) { // Entry can be safely evicted - cache.Remove(volume_id); - cache_last_touched.RemoveAt(i); - tmpfiles.Remove(volume_id); + handle_evict(volume_id); break; } } From e8f7244ac2903120d20fbeb22a13eedf1064c3d9 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 15:45:17 +0200 Subject: [PATCH 11/28] A log message was a warning, when it shouldn't be. --- Duplicati/Library/Main/Operation/Restore/VolumeManager.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 34aac8b13..99a630ddc 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -224,7 +224,7 @@ namespace Duplicati.Library.Main.Operation.Restore // Check if downloading another volume would exceed cache limits. if ((cache.Count + in_flight_downloads.Count) >= cache_max) { - Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Evicting volume"); + Logging.Log.WriteVerboseMessage(LOGTAG, "VolumeRequest", null, "Evicting volume"); // TODO switch based of the eviction strategy. // fifo / lifo based on both when they were downloaded and when they were used // random From 309ae187827ae123a0fc28e4016add08064d7a39 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 15:45:35 +0200 Subject: [PATCH 12/28] Removed whitespace --- Duplicati/UnitTest/RestoreHandlerTests.cs | 31 ++++++++++++----------- 1 file changed, 16 insertions(+), 15 deletions(-) diff --git a/Duplicati/UnitTest/RestoreHandlerTests.cs b/Duplicati/UnitTest/RestoreHandlerTests.cs index e82945292..d3cc9007d 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; From 4c9f78f839528df488478a9bd2b253e6f25fd10c Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Fri, 8 Aug 2025 15:45:52 +0200 Subject: [PATCH 13/28] Added first try at a test of the restore volume cache --- Duplicati/UnitTest/RestoreHandlerTests.cs | 51 +++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/Duplicati/UnitTest/RestoreHandlerTests.cs b/Duplicati/UnitTest/RestoreHandlerTests.cs index d3cc9007d..4d69ca1b3 100644 --- a/Duplicati/UnitTest/RestoreHandlerTests.cs +++ b/Duplicati/UnitTest/RestoreHandlerTests.cs @@ -162,6 +162,57 @@ namespace Duplicati.UnitTest } } + [Test] + [Category("RestoreHandler")] + public void 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-max"] = 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 Random(); + 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(() => + { + using var c = new Controller("file://" + this.TARGETFOLDER, opts, null); + TestUtils.AssertResults(c.Restore(["*"])); + }); + + System.Threading.Tasks.Task.WhenAny(timeout_task, restore_task).ContinueWith(t => + { + if (t.Result == timeout_task) + Assert.Fail("Restore timed out"); + else if (t.Result == restore_task) + // Throw any exceptions it might have + t.Result.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) From 537ee9c4cb5bc4445b6909ccd68e2c5c7e202de9 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 07:39:22 +0200 Subject: [PATCH 14/28] Final eviction should always happen, and some caching is always needed in the new restore flow --- .../Library/Main/Operation/Restore/VolumeManager.cs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 99a630ddc..c25674399 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -191,11 +191,8 @@ namespace Duplicati.Library.Main.Operation.Restore { case BlockRequestType.CacheEvict: { - if (cache_max > 0) - { - Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Evicting volume {0} from cache by request", request.VolumeID); - handle_evict(request.VolumeID); - } + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Evicting volume {0} from cache by request", request.VolumeID); + handle_evict(request.VolumeID); } break; case BlockRequestType.DecompressAck: @@ -222,7 +219,8 @@ namespace Duplicati.Library.Main.Operation.Restore else { // Check if downloading another volume would exceed cache limits. - if ((cache.Count + in_flight_downloads.Count) >= cache_max) + // 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.WriteVerboseMessage(LOGTAG, "VolumeRequest", null, "Evicting volume"); // TODO switch based of the eviction strategy. From 9e7c37fb33c6db0ec077976c2d4eefaa0b670b6b Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 07:41:02 +0200 Subject: [PATCH 15/28] Fixed timeout logic in restore volume cache tests --- Duplicati/UnitTest/RestoreHandlerTests.cs | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/Duplicati/UnitTest/RestoreHandlerTests.cs b/Duplicati/UnitTest/RestoreHandlerTests.cs index 4d69ca1b3..a7e36bce4 100644 --- a/Duplicati/UnitTest/RestoreHandlerTests.cs +++ b/Duplicati/UnitTest/RestoreHandlerTests.cs @@ -164,7 +164,7 @@ namespace Duplicati.UnitTest [Test] [Category("RestoreHandler")] - public void RestoreVolumeCache([Values("0b", "1mb", "5mb", "1gb")] string cache_size, [Values("0", "1", null)] string? channel_size) + 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"; @@ -201,14 +201,13 @@ namespace Duplicati.UnitTest TestUtils.AssertResults(c.Restore(["*"])); }); - System.Threading.Tasks.Task.WhenAny(timeout_task, restore_task).ContinueWith(t => - { - if (t.Result == timeout_task) - Assert.Fail("Restore timed out"); - else if (t.Result == restore_task) - // Throw any exceptions it might have - t.Result.GetAwaiter().GetResult(); - }); + var t = await System.Threading.Tasks.Task.WhenAny(timeout_task, restore_task); + + if (t == timeout_task) + 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"); } From fd94ae9f08390b2369fb013fa1f68dc56c3c4806 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 07:41:29 +0200 Subject: [PATCH 16/28] Added additional explicit logging --- Duplicati/Library/Main/Operation/Restore/VolumeManager.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index c25674399..d1c152bd1 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -197,6 +197,7 @@ namespace Duplicati.Library.Main.Operation.Restore break; case BlockRequestType.DecompressAck: { + Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", request.BlockID, request.VolumeID); handle_ack(request.BlockID, request.VolumeID); } break; From f85727bf3a73050423b752477d0dd2e10a3e074d Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:34:26 +0200 Subject: [PATCH 17/28] Added explicit debug printing --- .../Main/Operation/Restore/BlockManager.cs | 7 +++++++ .../Main/Operation/Restore/FileProcessor.cs | 18 +++++++++++++++++- .../Main/Operation/Restore/VolumeManager.cs | 5 +++-- 3 files changed, 27 insertions(+), 3 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs index 91236ee8d..b45c6ba49 100644 --- a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs @@ -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(); } @@ -467,15 +470,18 @@ namespace Duplicati.Library.Main.Operation.Restore { 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; await self.Ack.WriteAsync(block_request).ConfigureAwait(false); 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.BlockID, data); sw_set?.Stop(); @@ -518,6 +524,7 @@ namespace Duplicati.Library.Main.Operation.Restore sw_req?.Start(); var block_request = await req.ReadAsync().ConfigureAwait(false); sw_req?.Stop(); + Logging.Log.WriteExplicitMessage(LOGTAG, "BlockHandler", null, $"Received block request: {block_request.RequestType}"); switch (block_request.RequestType) { case BlockRequestType.Download: diff --git a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs index 29a246f3e..4f0aaad7f 100644 --- a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs +++ b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs @@ -55,6 +55,14 @@ 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(); + /// + /// Lock for the file processor ID. + /// + public static object id_lock = new(); + /// + /// The current file processor ID. Used for debugging. + /// + public static int id = 0; /// /// Runs the file processor process that restores the files that need to be restored. @@ -93,6 +101,12 @@ 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. + int my_id = -1; + lock (id_lock) + { + my_id = id++; + } try { @@ -107,6 +121,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 @@ -453,7 +469,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) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index d1c152bd1..6016e7780 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -181,9 +181,10 @@ namespace Duplicati.Library.Main.Operation.Restore { 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"); switch (msg) { case BlockRequest request: @@ -223,7 +224,7 @@ namespace Duplicati.Library.Main.Operation.Restore // 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.WriteVerboseMessage(LOGTAG, "VolumeRequest", null, "Evicting volume"); + 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 From 5129958f4f765c214de03c60644ad3677ae7260e Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:48:49 +0200 Subject: [PATCH 18/28] Changed the restore channels to use a constructor to avoid static variables --- .../Main/Operation/Restore/Channels.cs | 20 +++++++------------ .../Library/Main/Operation/RestoreHandler.cs | 8 +++----- 2 files changed, 10 insertions(+), 18 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/Channels.cs b/Duplicati/Library/Main/Operation/Restore/Channels.cs index 93ae32fc4..cb4634edc 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,12 +39,12 @@ 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); // TODO DecompressionAck could be made into a channel of `long`, but because of the ReadFromEitherPrioritized workaround, it has to be an `object`. /// @@ -61,21 +55,21 @@ namespace Duplicati.Library.Main.Operation.Restore /// /// 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); + 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/RestoreHandler.cs b/Duplicati/Library/Main/Operation/RestoreHandler.cs index 2334dc0b6..c147bcb45 100644 --- a/Duplicati/Library/Main/Operation/RestoreHandler.cs +++ b/Duplicati/Library/Main/Operation/RestoreHandler.cs @@ -344,13 +344,11 @@ 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(); From fa1c109ad3cd324a7545eb34c501d5c45e4eedfd Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:50:50 +0200 Subject: [PATCH 19/28] Burst in Restore.FileProcessor has to be at least 1 --- Duplicati/Library/Main/Operation/Restore/FileProcessor.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs index 4f0aaad7f..3e5697dfc 100644 --- a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs +++ b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs @@ -303,7 +303,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++) { From dfc92c71b15690cb6de5b0f31483789d2b8a907f Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:54:02 +0200 Subject: [PATCH 20/28] Added more rendevouz points to allow for aborting during deadlocks --- .../Main/Operation/Restore/BlockManager.cs | 8 ++++- .../Main/Operation/Restore/FileProcessor.cs | 2 +- .../Main/Operation/Restore/VolumeManager.cs | 35 +++++++++++++++---- .../Library/Main/Operation/RestoreHandler.cs | 2 +- 4 files changed, 37 insertions(+), 10 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs index b45c6ba49..5c0ee4c7a 100644 --- a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs @@ -478,7 +478,13 @@ namespace Duplicati.Library.Main.Operation.Restore 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; - await self.Ack.WriteAsync(block_request).ConfigureAwait(false); + while (true) + { + var did_write = await self.Ack.TryWriteAsync(block_request, 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}"); diff --git a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs index 3e5697dfc..72bcbc60f 100644 --- a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs +++ b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs @@ -147,7 +147,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(); diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 6016e7780..086e20a31 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -65,14 +65,14 @@ namespace Duplicati.Library.Main.Operation.Restore /// 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) @@ -93,7 +93,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 @@ -175,6 +175,7 @@ namespace Duplicati.Library.Main.Operation.Restore } } + await results.TaskControl.ProgressRendevouz().ConfigureAwait(false); var rfa = new ReadFromEither(self.VolumeResponse, self.VolumeRequest); try @@ -185,6 +186,7 @@ namespace Duplicati.Library.Main.Operation.Restore await flush_ack_channel().ConfigureAwait(false); // TODO: CoCol ReadFromAnyAsync deadlocks, so we use a workaround 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: @@ -211,7 +213,13 @@ namespace Duplicati.Library.Main.Operation.Restore cache_last_touched.Remove(request.VolumeID); cache_last_touched.Add(request.VolumeID); Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Block {0} found in cache", request.BlockID); - await self.DecompressRequest.WriteAsync((request, reader)).ConfigureAwait(false); + 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; @@ -251,7 +259,13 @@ namespace Duplicati.Library.Main.Operation.Restore } else { - await self.DownloadRequest.WriteAsync(request.VolumeID).ConfigureAwait(false); + 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]; } } @@ -277,7 +291,14 @@ namespace Duplicati.Library.Main.Operation.Restore // Request the decompressions Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Requesting block {0} from newly cached volume {1}", request.BlockID, volume_id); - await self.DecompressRequest.WriteAsync((request, reader)).ConfigureAwait(false); + 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 diff --git a/Duplicati/Library/Main/Operation/RestoreHandler.cs b/Duplicati/Library/Main/Operation/RestoreHandler.cs index c147bcb45..a29f44ac5 100644 --- a/Duplicati/Library/Main/Operation/RestoreHandler.cs +++ b/Duplicati/Library/Main/Operation/RestoreHandler.cs @@ -353,7 +353,7 @@ namespace Duplicati.Library.Main.Operation 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(); From 0fd3e515dac87287c520782f59bf840733c56218 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:55:02 +0200 Subject: [PATCH 21/28] Ensured that an actively read volume in the volumemanager cannot be disposed --- Duplicati/Library/Main/Operation/Restore/VolumeManager.cs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 086e20a31..0d8e11d1c 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -26,6 +26,7 @@ using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; +using System.Threading; using System.Threading.Tasks; #nullable enable @@ -118,6 +119,8 @@ namespace Duplicati.Library.Main.Operation.Restore 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; @@ -145,7 +148,7 @@ namespace Duplicati.Library.Main.Operation.Restore if (dcount <= 1) { in_flight_decompressing.Remove(volume_id); - if (cache_max == 0) + if (volume_id != responding && cache_max == 0) { handle_evict(volume_id); } @@ -284,6 +287,7 @@ namespace Duplicati.Library.Main.Operation.Restore tmpfiles[volume_id] = tmpfile; sw_cache_set?.Stop(); sw_wakeup?.Start(); + responding = volume_id; foreach (var request in in_flight_downloads[volume_id]) { // Ensure ACK channel is empty to avoid deadlock @@ -304,6 +308,7 @@ namespace Duplicati.Library.Main.Operation.Restore else in_flight_decompressing[request.VolumeID] = 1; } + responding = null; in_flight_downloads.Remove(volume_id); sw_wakeup?.Stop(); break; From 4edbbda7f989ec1e35b4ef60ea3fa6505c2a6a38 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:56:05 +0200 Subject: [PATCH 22/28] Cleaned up RestoreVolumeCache test and send an abort request on timeout (deadlock) --- Duplicati/UnitTest/RestoreHandlerTests.cs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/Duplicati/UnitTest/RestoreHandlerTests.cs b/Duplicati/UnitTest/RestoreHandlerTests.cs index a7e36bce4..e32d5ba6e 100644 --- a/Duplicati/UnitTest/RestoreHandlerTests.cs +++ b/Duplicati/UnitTest/RestoreHandlerTests.cs @@ -180,7 +180,7 @@ namespace Duplicati.UnitTest opts["restore-path"] = RESTOREFOLDER; // Write a bunch of files - Random rng = new Random(); + Random rng = new(); byte[] data = new byte[1024]; for (int i = 0; i < 1000; i++) { @@ -189,22 +189,25 @@ namespace Duplicati.UnitTest File.WriteAllBytes(filePath, data); } - using (var c = new Controller("file://" + this.TARGETFOLDER, opts, null)) - TestUtils.AssertResults(c.Backup([this.DATAFOLDER])); + 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(() => { - using var c = new Controller("file://" + this.TARGETFOLDER, opts, null); 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(); From 4091d8bbc02a4d46e99939c70f38a17cc6db8ba4 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:56:55 +0200 Subject: [PATCH 23/28] Set the buffer size of the ack channel to allow for enough space for all possible in flight decompression requests --- Duplicati/Library/Main/Operation/Restore/Channels.cs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/Duplicati/Library/Main/Operation/Restore/Channels.cs b/Duplicati/Library/Main/Operation/Restore/Channels.cs index cb4634edc..e7fffba1d 100644 --- a/Duplicati/Library/Main/Operation/Restore/Channels.cs +++ b/Duplicati/Library/Main/Operation/Restore/Channels.cs @@ -49,8 +49,10 @@ namespace Duplicati.Library.Main.Operation.Restore // TODO DecompressionAck could be made into a channel of `long`, but because of the ReadFromEitherPrioritized workaround, it has to be an `object`. /// /// 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: BufferSize); + public readonly IChannel DecompressionAck = ChannelManager.CreateChannel(buffersize: 2 * options.RestoreChannelBufferSize + options.RestoreVolumeDecompressors + 1); /// /// Channel between and From 1a7321556e46e67c5764f16b56b1e0e7198891e3 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 20:59:33 +0200 Subject: [PATCH 24/28] Changed the decompression ack channel type to long --- .../Main/Operation/Restore/BlockManager.cs | 2 +- .../Library/Main/Operation/Restore/Channels.cs | 3 +-- .../Main/Operation/Restore/VolumeManager.cs | 16 ++++++---------- 3 files changed, 8 insertions(+), 13 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs index 5c0ee4c7a..4ad1d088d 100644 --- a/Duplicati/Library/Main/Operation/Restore/BlockManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/BlockManager.cs @@ -480,7 +480,7 @@ namespace Duplicati.Library.Main.Operation.Restore block_request.RequestType = BlockRequestType.DecompressAck; while (true) { - var did_write = await self.Ack.TryWriteAsync(block_request, TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + var did_write = await self.Ack.TryWriteAsync(block_request.VolumeID, TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); if (did_write) break; await results.TaskControl.ProgressRendevouz(); diff --git a/Duplicati/Library/Main/Operation/Restore/Channels.cs b/Duplicati/Library/Main/Operation/Restore/Channels.cs index e7fffba1d..6327301df 100644 --- a/Duplicati/Library/Main/Operation/Restore/Channels.cs +++ b/Duplicati/Library/Main/Operation/Restore/Channels.cs @@ -46,13 +46,12 @@ namespace Duplicati.Library.Main.Operation.Restore /// public readonly IChannel<(long, string, TempFile)> DecryptRequest = ChannelManager.CreateChannel<(long, string, TempFile)>(buffersize: options.RestoreChannelBufferSize); - // TODO DecompressionAck could be made into a channel of `long`, but because of the ReadFromEitherPrioritized workaround, it has to be an `object`. /// /// 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); + public readonly IChannel DecompressionAck = ChannelManager.CreateChannel(buffersize: 2 * options.RestoreChannelBufferSize + options.RestoreVolumeDecompressors + 1); /// /// Channel between and diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 0d8e11d1c..50c5f1a9c 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -140,9 +140,9 @@ namespace Duplicati.Library.Main.Operation.Restore sw_cache_evict?.Stop(); } - void handle_ack(long block_id, long volume_id) + void handle_ack(long volume_id) { - Logging.Log.WriteVerboseMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", block_id, 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) @@ -157,7 +157,7 @@ namespace Duplicati.Library.Main.Operation.Restore in_flight_decompressing[volume_id] = dcount - 1; } else - Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Decompression acknowledgment for block {0} from volume {1} not found", block_id, volume_id); + Logging.Log.WriteWarningMessage(LOGTAG, "VolumeRequest", null, "Decompression acknowledgment for volume {0} not found", volume_id); } async Task flush_ack_channel() @@ -166,15 +166,11 @@ namespace Duplicati.Library.Main.Operation.Restore while (true) { var (has_read, ack_msg) = await self.DecompressAck.TryReadAsync().ConfigureAwait(false); + if (!has_read) break; - if (ack_msg is not BlockRequest ack_request) - { - Logging.Log.WriteErrorMessage(LOGTAG, "VolumeRequest", null, "Unexpected message type in ack channel: {0}", ack_msg?.GetType().Name); - break; - } - handle_ack(ack_request.BlockID, ack_request.VolumeID); + handle_ack(ack_msg); } } @@ -204,7 +200,7 @@ namespace Duplicati.Library.Main.Operation.Restore case BlockRequestType.DecompressAck: { Logging.Log.WriteExplicitMessage(LOGTAG, "VolumeRequest", "Decompression acknowledgment for block {0} from volume {1}", request.BlockID, request.VolumeID); - handle_ack(request.BlockID, request.VolumeID); + handle_ack(request.VolumeID); } break; case BlockRequestType.Download: From 356eb0326463dbc14dc5114b3222b07e34a40f23 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Sun, 10 Aug 2025 21:00:48 +0200 Subject: [PATCH 25/28] Handled nullable warning in RestoreHandlerTests --- Duplicati/UnitTest/RestoreHandlerTests.cs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Duplicati/UnitTest/RestoreHandlerTests.cs b/Duplicati/UnitTest/RestoreHandlerTests.cs index e32d5ba6e..ed3eb0414 100644 --- a/Duplicati/UnitTest/RestoreHandlerTests.cs +++ b/Duplicati/UnitTest/RestoreHandlerTests.cs @@ -29,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 From 966e8b5d87f42c638c0ac9e892fe536058540546 Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Thu, 14 Aug 2025 16:57:27 +0200 Subject: [PATCH 26/28] Bumped CoCoL version --- Duplicati/Library/Main/Duplicati.Library.Main.csproj | 2 +- .../UsageReporter/Duplicati.Library.UsageReporter.csproj | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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/UsageReporter/Duplicati.Library.UsageReporter.csproj b/Duplicati/Library/UsageReporter/Duplicati.Library.UsageReporter.csproj index 39d26139b..a5e2ab7ca 100644 --- a/Duplicati/Library/UsageReporter/Duplicati.Library.UsageReporter.csproj +++ b/Duplicati/Library/UsageReporter/Duplicati.Library.UsageReporter.csproj @@ -7,7 +7,7 @@ - + From 3e10b8446af961db75fd0d7abc91a2894b5d2b9d Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Thu, 14 Aug 2025 17:10:56 +0200 Subject: [PATCH 27/28] Changed the wording of the new volume cache parameter --- Duplicati/Library/Main/Operation/Restore/VolumeManager.cs | 4 ++-- Duplicati/Library/Main/Options.cs | 6 +++--- Duplicati/Library/Main/Strings.cs | 4 ++-- Duplicati/UnitTest/RestoreHandlerTests.cs | 2 +- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs index 50c5f1a9c..3c393e15a 100644 --- a/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs +++ b/Duplicati/Library/Main/Operation/Restore/VolumeManager.cs @@ -107,8 +107,8 @@ namespace Duplicati.Library.Main.Operation.Restore }, async self => { - // The maximum number of volumes to have in cache at once. - long cache_max = options.RestoreVolumeCacheMax / options.VolumeSize; + // 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. diff --git a/Duplicati/Library/Main/Options.cs b/Duplicati/Library/Main/Options.cs index ef22878d8..991edd031 100644 --- a/Duplicati/Library/Main/Options.cs +++ b/Duplicati/Library/Main/Options.cs @@ -167,7 +167,7 @@ namespace Duplicati.Library.Main /// /// The default value for the maximum size of the restore volume cache /// - private const string DEFAULT_RESTORE_VOLUME_CACHE_MAX = "4gb"; + private const string DEFAULT_RESTORE_VOLUME_CACHE_HINT = "4gb"; /// /// The default value for the number of volume decryptors during restore @@ -554,7 +554,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-max", CommandLineArgument.ArgumentType.Size, Strings.Options.RestoreVolumeCacheMaxShort, Strings.Options.RestoreVolumeCacheMaxLong, DEFAULT_RESTORE_VOLUME_CACHE_MAX), + 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()), @@ -1706,7 +1706,7 @@ namespace Duplicati.Library.Main // // Gets the size of the volume cache used during restore, in MB. // - public long RestoreVolumeCacheMax => GetSize("restore-volume-cache-max", "mb", DEFAULT_RESTORE_VOLUME_CACHE_MAX); + 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 1280518c8..4c1c45f28 100644 --- a/Duplicati/Library/Main/Strings.cs +++ b/Duplicati/Library/Main/Strings.cs @@ -356,8 +356,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 RestoreVolumeCacheMaxShort { get { return LC.L("Maximum size of the restore volume cache"); } } - public static string RestoreVolumeCacheMaxLong { get { return LC.L("Use this option to set the maximum size of the restore volume cache. The restore volume cache is used to store the volumes that are downloaded from the remote storage. The number of volumes stored is set to floor(max_size / volume_size), except for when it's set to 0, which disables the volume cache."); } } + 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/UnitTest/RestoreHandlerTests.cs b/Duplicati/UnitTest/RestoreHandlerTests.cs index ed3eb0414..637dd7160 100644 --- a/Duplicati/UnitTest/RestoreHandlerTests.cs +++ b/Duplicati/UnitTest/RestoreHandlerTests.cs @@ -171,7 +171,7 @@ namespace Duplicati.UnitTest var opts = TestOptions; opts["dblock-size"] = "1mb"; opts["blocksize"] = "1kb"; - opts["restore-volume-cache-max"] = cache_size; + opts["restore-volume-cache-hint"] = cache_size; if (channel_size != null) opts["restore-channel-buffer-size"] = channel_size; opts["restore-legacy"] = "false"; From 4f524eff68ca42db36ad899febdd8f8ea8cde863 Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Fri, 15 Aug 2025 10:29:15 +0200 Subject: [PATCH 28/28] Fixed increment without lock --- .../Library/Main/Operation/Restore/FileProcessor.cs | 12 ++---------- 1 file changed, 2 insertions(+), 10 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs index 72bcbc60f..f1177e31b 100644 --- a/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs +++ b/Duplicati/Library/Main/Operation/Restore/FileProcessor.cs @@ -56,13 +56,9 @@ namespace Duplicati.Library.Main.Operation.Restore public static object file_processor_continue_lock = new object(); public static TaskCompletionSource file_processor_continue = new(); /// - /// Lock for the file processor ID. - /// - public static object id_lock = new(); - /// /// The current file processor ID. Used for debugging. /// - public static int id = 0; + private static int processor_id = 0; /// /// Runs the file processor process that restores the files that need to be restored. @@ -102,11 +98,7 @@ 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. - int my_id = -1; - lock (id_lock) - { - my_id = id++; - } + var my_id = Interlocked.Increment(ref processor_id); try {