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

Reworked the "Volume loop" of the restore network to handle Backends themselves, rather than using the BackendManager.

Browse Source 
This allows for tighter control over which process performs which action, allowing them more easily to run in parallel.
This commit is contained in:
Carl Johnsen
2025-02-18 09:25:20 +01:00
parent 17e2c0b9cf
commit a0d5096ebc
5 changed files with 110 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Duplicati/Library/Main/Operation/Restore/Channels.cs
+2 -3
View File
@@ -19,7 +19,6 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
using System.Threading.Tasks;
using CoCoL;
using Duplicati.Library.Main.Volumes;
using Duplicati.Library.Utility;
@@ -46,12 +45,12 @@ namespace Duplicati.Library.Main.Operation.Restore
/// <summary>
/// Channel between <see cref="VolumeManager"/> and <see cref="VolumeDownloader"/>.
/// </summary>
public readonly IChannel<(long, Task<TempFile>)> DownloadRequest = ChannelManager.CreateChannel<(long, Task<TempFile>)>(buffersize: BufferSize);
public readonly IChannel<long> DownloadRequest = ChannelManager.CreateChannel<long>(buffersize: BufferSize);
/// <summary>
/// Channel between <see cref="VolumeDownloader"/> and <see cref="VolumeDecryptor"/>
/// </summary>
public readonly IChannel<(long, TempFile)> DecryptRequest = ChannelManager.CreateChannel<(long, TempFile)>(buffersize: BufferSize);
public readonly IChannel<(long, string, TempFile)> DecryptRequest = ChannelManager.CreateChannel<(long, string, TempFile)>(buffersize: BufferSize);
/// <summary>
/// Channel between <see cref="VolumeManager"/> and <see cref="VolumeDecompressor"/>
Duplicati/Library/Main/Operation/Restore/VolumeDecryptor.cs
+88 -2
View File
@@ -21,6 +21,7 @@
using System;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;
using CoCoL;
using Duplicati.Library.Main.Volumes;
@@ -57,15 +58,23 @@ namespace Duplicati.Library.Main.Operation.Restore
Stopwatch sw_decrypt = options.InternalProfiling ? new() : null;
try
{
// Taken from BackendManager.GetOperation
using var encryption = options.NoEncryption
? null
: (DynamicLoader.EncryptionLoader.GetModule(options.EncryptionModule, options.Passphrase, options.RawOptions)
?? throw new Exception(Strings.BackendMananger.EncryptionModuleNotFound(options.EncryptionModule))
);
while (true)
{
// Get the block request and volume from the `VolumeDownloader` process.
sw_read?.Start();
var (volume_id, volume) = await self.Input.ReadAsync();
var (volume_id, volume_name, volume) = await self.Input.ReadAsync().ConfigureAwait(false);
sw_read?.Stop();
sw_decrypt?.Start();
var bvr = new BlockVolumeReader(options.CompressionModule, volume, options);
var tmpfile = DecryptFile(volume, DetectEncryptionModule(volume_name, options, encryption));
var bvr = new BlockVolumeReader(options.CompressionModule, tmpfile, options);
sw_decrypt?.Stop();
sw_write?.Start();
@@ -90,6 +99,83 @@ namespace Duplicati.Library.Main.Operation.Restore
}
});
}
// Taken from BackendManager.GetOperation
public static Library.Utility.TempFile DecryptFile(Library.Utility.TempFile tempFile, Interface.IEncryption? decrypter)
{
// Support no encryption
if (decrypter == null)
return tempFile;
Library.Utility.TempFile? decryptTarget = null;
// Always dispose the source file
using (tempFile)
using (new Logging.Timer(LOGTAG, "DecryptFile", "Decrypting " + tempFile))
{
try
{
decryptTarget = new Library.Utility.TempFile();
try { decrypter.Decrypt(tempFile, decryptTarget); }
// If we fail here, make sure that we throw a crypto exception
catch (System.Security.Cryptography.CryptographicException) { throw; }
catch (Exception ex) { throw new System.Security.Cryptography.CryptographicException(ex.Message, ex); }
var result = decryptTarget;
decryptTarget = null;
return result;
}
finally
{
// Remove temp files on failure
decryptTarget?.Dispose();
}
}
}
private static Interface.IEncryption? DetectEncryptionModule(string remotefilename, Options options, Interface.IEncryption? encryption)
{
try
{
// Auto-guess the encryption module
var ext = (System.IO.Path.GetExtension(remotefilename) ?? "").TrimStart('.');
if (!ext.Equals(encryption?.FilenameExtension, StringComparison.OrdinalIgnoreCase))
{
// Check if the file is not encrypted
if (DynamicLoader.CompressionLoader.Keys.Contains(ext, StringComparer.OrdinalIgnoreCase))
{
if (encryption != null)
Logging.Log.WriteVerboseMessage(LOGTAG, "AutomaticDecryptionDetection", "Filename extension \"{0}\" does not match encryption module \"{1}\", guessing that it is not encrypted", ext, options.EncryptionModule);
return null;
}
// Check if the file is encrypted with something else
else if (DynamicLoader.EncryptionLoader.Keys.Contains(ext, StringComparer.OrdinalIgnoreCase))
{
Logging.Log.WriteVerboseMessage(LOGTAG, "AutomaticDecryptionDetection", "Filename extension \"{0}\" does not match encryption module \"{1}\", attempting to use matching encryption module", ext, options.EncryptionModule);
try
{
return DynamicLoader.EncryptionLoader.GetModule(ext, options.Passphrase, options.RawOptions)
?? encryption;
}
catch (Exception ex)
{
Logging.Log.WriteWarningMessage(LOGTAG, "AutomaticDecryptionDetection", ex, "Failed to load encryption module \"{0}\", using specified encryption module \"{1}\"", ext, options.EncryptionModule);
}
}
// Fallback, lets see what happens...
else
{
Logging.Log.WriteVerboseMessage(LOGTAG, "AutomaticDecryptionDetection", "Filename extension \"{0}\" does not match encryption module \"{1}\", attempting to use specified encryption module as no others match", ext, options.EncryptionModule);
}
}
return encryption;
}
// If we fail here, make sure that we throw a crypto exception
catch (System.Security.Cryptography.CryptographicException) { throw; }
catch (Exception ex) { throw new System.Security.Cryptography.CryptographicException(ex.Message, ex); }
}
}
}
Duplicati/Library/Main/Operation/Restore/VolumeDownloader.cs
+15 -6
View File
@@ -22,8 +22,10 @@
using System;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using CoCoL;
using Duplicati.Library.Interface;
using Duplicati.Library.Main.Database;
using Duplicati.Library.Utility;
@@ -43,9 +45,12 @@ namespace Duplicati.Library.Main.Operation.Restore
/// <summary>
/// Runs the volume downloader process.
/// </summary>
/// <param name="channels">The named channels for the restore operation.</param>
/// <param name="db">The local restore database, used to find volume information given the volume ID.</param>
/// <param name="backend">The backend to use for downloading the volumes.</param>
/// <param name="options">The restore options.</param>
/// <param name="results">The restore results.</param>
public static Task Run(Channels channels, LocalRestoreDatabase db, Options options, RestoreResults results)
public static Task Run(Channels channels, LocalRestoreDatabase db, IBackend backend, Options options, RestoreResults results)
{
return AutomationExtensions.RunTask(
new
@@ -65,19 +70,19 @@ namespace Duplicati.Library.Main.Operation.Restore
{
// Get the block request from the `BlockManager` process.
sw_read?.Start();
var (volume_id, waittask) = await self.Input.ReadAsync();
var volume_id = await self.Input.ReadAsync().ConfigureAwait(false);
sw_read?.Stop();
// Trigger the download.
sw_wait?.Start();
TempFile f = null;
TempFile f = new();
var (volume_name, size, hash) = db.GetVolumeInfo(volume_id).First();
try
{
f = await waittask.ConfigureAwait(false);
await backend.GetAsync(volume_name, f.Name, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception)
{
var (volume_name, _, _) = db.GetVolumeInfo(volume_id).First();
lock (results)
results.BrokenRemoteFiles.Add(volume_name);
@@ -87,7 +92,7 @@ namespace Duplicati.Library.Main.Operation.Restore
// Pass the download handle (which may or may not have downloaded already) to the `VolumeDecryptor` process.
sw_write?.Start();
await self.Output.WriteAsync((volume_id, f));
await self.Output.WriteAsync((volume_id, volume_name, f)).ConfigureAwait(false);
sw_write?.Stop();
}
}
@@ -105,6 +110,10 @@ namespace Duplicati.Library.Main.Operation.Restore
Logging.Log.WriteErrorMessage(LOGTAG, "DownloadError", ex, "Error during download");
throw;
}
finally
{
backend.Dispose();
}
});
}
}
Duplicati/Library/Main/Operation/Restore/VolumeManager.cs
+3 -12
View File
@@ -20,7 +20,6 @@
// DEALINGS IN THE SOFTWARE.
using CoCoL;
using Duplicati.Library.Main.Database;
using Duplicati.Library.Main.Volumes;
using Duplicati.Library.Utility;
using System;
@@ -46,11 +45,9 @@ namespace Duplicati.Library.Main.Operation.Restore
/// <summary>
/// Runs the volume manager process.
/// </summary>
/// <param name="db">The local restore database, used to find the volume where a block is stored.</param>
/// <param name="backend">The backend manager, used to fetch the volumes from the backend.</param>
/// <param name="channels">The named channels for the restore operation.</param>
/// <param name="options">The restore options.</param>
/// <param name="results">The restore results.</param>
public static Task Run(Channels channels, LocalRestoreDatabase db, IBackendManager backend, Options options, RestoreResults results)
public static Task Run(Channels channels, Options options)
{
return AutomationExtensions.RunTask(
new
@@ -106,13 +103,7 @@ namespace Duplicati.Library.Main.Operation.Restore
}
else
{
sw_query?.Start();
var (volume_name, volume_size, volume_hash) = db.GetVolumeInfo(request.VolumeID).First();
sw_query?.Stop();
sw_backend?.Start();
var handle = backend.GetAsync(volume_name, volume_hash, volume_size, results.TaskControl.TransferToken);
sw_backend?.Stop();
await self.DownloadRequest.WriteAsync((request.VolumeID, handle));
await self.DownloadRequest.WriteAsync(request.VolumeID).ConfigureAwait(false);
in_flight[request.VolumeID] = [request];
}
}
Duplicati/Library/Main/Operation/RestoreHandler.cs
+2 -2
View File
@@ -339,8 +339,8 @@ namespace Duplicati.Library.Main.Operation
var filelister = Restore.FileLister.Run(channels, database, m_options, m_result);
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, database, backendManager, m_options, m_result);
var volumedownloaders = Enumerable.Range(0, m_options.RestoreVolumeDownloaders).Select(i => Restore.VolumeDownloader.Run(channels, database, m_options, m_result)).ToArray();
var volumecache = Restore.VolumeManager.Run(channels, m_options);
var volumedownloaders = Enumerable.Range(0, m_options.RestoreVolumeDownloaders).Select(i => Restore.VolumeDownloader.Run(channels, database, backendManager.GetBackend(), m_options, m_result)).ToArray();
var volumedecryptors = Enumerable.Range(0, m_options.RestoreVolumeDecryptors).Select(i => Restore.VolumeDecryptor.Run(channels, m_options)).ToArray();
var volumedecompressors = Enumerable.Range(0, m_options.RestoreVolumeDecompressors).Select(i => Restore.VolumeDecompressor.Run(channels, m_options)).ToArray();