haeaeae/duplicati
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #6476 from carljohnsen/feature/restore-volume-cache-size-parameter

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