The FileProcessor now works in two phases: first it restores the files, then they restore the folder metadata, unless specified otherwise.

This commit is contained in:
Carl Johnsen
2025-03-24 20:24:37 +01:00
parent 11bbc629c3
commit f73d3390cd
4 changed files with 64 additions and 2 deletions
@@ -975,6 +975,22 @@ ORDER BY ""A"".""TargetPath"", ""BB"".""Index""");
yield return new FileRequest(rd.ConvertValueToInt64(0), rd.ConvertValueToString(1), rd.ConvertValueToString(2), rd.ConvertValueToString(3), rd.ConvertValueToInt64(4), rd.ConvertValueToInt64(5));
}
/// <summary>
/// Returns a list of folders to restore. Used to restore folder metadata.
/// </summary>
/// <returns>A list of folders to restore.</returns>
public IEnumerable<FileRequest> GetFolderMetadataToRestore()
{
using var cmd = m_connection.CreateCommand();
using var rd = cmd.ExecuteReader(FormatInvariant($@"
SELECT F.ID, '', F.TargetPath, '', 0, {FOLDER_BLOCKSET_ID}
FROM ""{m_tempfiletable}"" F
WHERE F.BlocksetID = {FOLDER_BLOCKSET_ID} AND F.MetadataID IS NOT NULL AND F.MetadataID >= 0"));
while (rd.Read())
yield return new FileRequest(rd.ConvertValueToInt64(0), rd.ConvertValueToString(1), rd.ConvertValueToString(2), rd.ConvertValueToString(3), rd.ConvertValueToInt64(4), rd.ConvertValueToInt64(5));
}
/// <summary>
/// Returns a list of blocks and their volume IDs. Used by the <see cref="BlockManager"/> to keep track of blocks and volumes to automatically evict them from the respective caches.
/// </summary>
@@ -996,7 +1012,6 @@ ORDER BY ""A"".""TargetPath"", ""BB"".""Index""");
INNER JOIN ""Metadataset"" ON ""{m_tempfiletable}"".""MetadataID"" = ""Metadataset"".""ID""
INNER JOIN ""BlocksetEntry"" ON ""Metadataset"".""BlocksetID"" = ""BlocksetEntry"".""BlocksetID""
INNER JOIN ""Block"" ON ""BlocksetEntry"".""BlockID"" = ""Block"".""ID""
WHERE ""{m_tempfiletable}"".""BlocksetID"" IS NOT {FOLDER_BLOCKSET_ID}
")));
while (reader.Read())
yield return (reader.ConvertValueToInt64(0), reader.ConvertValueToInt64(1));
@@ -71,7 +71,14 @@ namespace Duplicati.Library.Main.Operation.Restore
sw_write?.Start();
foreach (var file in files)
await self.Output.WriteAsync(file).ConfigureAwait(false);
sw_write?.Stop();
if (!options.SkipMetadata)
{
var folders = db.GetFolderMetadataToRestore().ToArray();
foreach (var folder in folders)
await self.Output.WriteAsync(folder).ConfigureAwait(false);
}
}
catch (Exception ex)
{
@@ -24,6 +24,7 @@ using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using CoCoL;
using Duplicati.Library.Common.IO;
@@ -48,6 +49,10 @@ namespace Duplicati.Library.Main.Operation.Restore
/// The log tag for this class.
/// </summary>
private static readonly string LOGTAG = Logging.Log.LogTagFromType<FileProcessor>();
/// <summary>
/// The number of file processors that are still restoring files. It is set by the <see cref="RestoreHandler"/>.
/// </summary>
public static int file_processors_restoring_files;
/// <summary>
/// Runs the file processor process that restores the files that need to be restored.
@@ -84,6 +89,9 @@ namespace Duplicati.Library.Main.Operation.Restore
Stopwatch sw_work_results = options.InternalProfiling ? new() : null;
Stopwatch sw_work_meta = options.InternalProfiling ? new() : null;
// Indicates whether this FileProcessor is still restoring files
var decremented = false;
try
{
using var filehasher = HashFactory.CreateHasher(options.FileHashAlgorithm);
@@ -97,6 +105,20 @@ namespace Duplicati.Library.Main.Operation.Restore
var file = await self.Input.ReadAsync().ConfigureAwait(false);
sw_file?.Stop();
if (file.BlocksetID == LocalDatabase.FOLDER_BLOCKSET_ID && !options.SkipMetadata)
{
// Check if there are other FileProcessor's still restoring files
if (!decremented)
{
await RendesvouzBeforeProcessingFolderMetadata();
decremented = true;
}
await RestoreMetadata(db, file, block_request, block_response, options, sw_meta, sw_work_meta, sw_req, sw_resp);
continue;
}
// Get information about the blocks for the file
// TODO rather than keeping all of the blocks in memory, we could do a single pass over the blocks using a cursor, only keeping the relevant block requests in memory. Maybe even only a single block request at a time.
sw_block?.Start();
@@ -436,6 +458,9 @@ namespace Duplicati.Library.Main.Operation.Restore
}
finally
{
if (!decremented)
Interlocked.Decrement(ref file_processors_restoring_files);
block_request.Retire();
block_response.Retire();
}
@@ -523,6 +548,20 @@ namespace Duplicati.Library.Main.Operation.Restore
return newname;
}
/// <summary>
/// Rendesvouz with the other FileProcessor's before processing folder metadata.
/// </summary>
/// <returns>An awaitable task that completes once all of the FileProcessor's have rendezvoused.</returns>
private static async Task RendesvouzBeforeProcessingFolderMetadata()
{
Interlocked.Decrement(ref file_processors_restoring_files);
while (file_processors_restoring_files > 0)
{
await Task.CompletedTask.ConfigureAwait(ConfigureAwaitOptions.ForceYielding);
}
}
/// <summary>
/// Restores the metadata for a file.
/// </summary>
@@ -335,6 +335,7 @@ namespace Duplicati.Library.Main.Operation
// Create the process network
Restore.Channels channels = new();
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, m_options, fileprocessor_requests, fileprocessor_responses);
var volumecache = Restore.VolumeManager.Run(channels, m_options);