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

Rewrote the uploader to pass in the index volume blocks, as that is 10x times faster than creating it from the database.

Browse Source 
This greatly increases performance because the database read is no longer preventing progress.
This commit is contained in:
Kenneth Skovhede
2018-01-30 08:34:57 +01:00
parent ece4ff2021
commit ef36a9356b
9 changed files with 130 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Duplicati/Library/Main/Operation/Backup/BackendUploader.cs
+27 -4
View File
@@ -60,13 +60,36 @@ namespace Duplicati.Library.Main.Operation.Backup
internal class VolumeUploadRequest : IUploadRequest
{
public BlockVolumeWriter BlockVolume { get; private set; }
public bool CreateIndexVolume { get; private set; }
public readonly BlockVolumeWriter BlockVolume;
public readonly bool CreateIndexVolume;
public readonly IEnumerable<string> BlocklistData;
public VolumeUploadRequest(BlockVolumeWriter blockvolume, bool createindexvolume)
public VolumeUploadRequest(BlockVolumeWriter blockvolume, bool createindexvolume, IEnumerable<string> blocklistdata)
{
BlockVolume = blockvolume;
CreateIndexVolume = createindexvolume;
BlocklistData = blocklistdata;
}
public static string EncodeBlockListEntry(string hash, long size, byte[] data)
{
return hash + ":" + size + ":" + Convert.ToBase64String(data);
}
public static string DecodeBlockListEntryHash(string entry)
{
var ix = entry.IndexOf(':');
return entry.Substring(0, ix);
}
public static Tuple<string, long, byte[]> DecodeBlockListEntry(string entry)
{
var els = entry.Split(new char[] { ':' }, 3);
return new Tuple<string, long, byte[]>(
els[0],
long.Parse(els[1]),
Convert.FromBase64String(els[2])
);
}
}
@@ -108,7 +131,7 @@ namespace Duplicati.Library.Main.Operation.Backup
if (noIndexFiles || !((VolumeUploadRequest)req).CreateIndexVolume)
task = new KeyValuePair<int, Task>(1, backend.UploadFileAsync(((VolumeUploadRequest)req).BlockVolume, null));
else
task = new KeyValuePair<int, Task>(2, backend.UploadFileAsync(((VolumeUploadRequest)req).BlockVolume, name => IndexVolumeCreator.CreateIndexVolume(name, options, database)));
task = new KeyValuePair<int, Task>(2, backend.UploadFileAsync(((VolumeUploadRequest)req).BlockVolume, name => IndexVolumeCreator.CreateIndexVolume(name, options, database, ((VolumeUploadRequest)req).BlocklistData)));
}
else if (req is FilesetUploadRequest)
task = new KeyValuePair<int, Task>(1, backend.UploadFileAsync(((FilesetUploadRequest)req).Fileset));
Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs
+10 -3
View File
@@ -22,6 +22,7 @@ using Duplicati.Library.Main.Volumes;
using Duplicati.Library.Main.Operation.Common;
using System.IO;
using System.Linq;
using System.Collections.Generic;
namespace Duplicati.Library.Main.Operation.Backup
{
@@ -46,7 +47,8 @@ namespace Duplicati.Library.Main.Operation.Backup
async self =>
{
BlockVolumeWriter blockvolume = null;
var log = new LogWrapper(self.LogChannel);
var useindex = options.IndexfilePolicy == Options.IndexFileStrategy.Full;
var indexdata = useindex ? new Library.Utility.FileBackedStringList() : null;
try
{
@@ -94,10 +96,12 @@ namespace Duplicati.Library.Main.Operation.Backup
// Close this volume, and send it to upload
blockvolume.Close();
await database.CommitTransactionAsync("CommitAddBlockToOutputFlush");
await self.Output.WriteAsync(new VolumeUploadRequest(blockvolume, true));
await self.Output.WriteAsync(new VolumeUploadRequest(blockvolume, true, indexdata));
// Continue with the freshly created volume
blockvolume = tmpvolume;
if (useindex)
indexdata = new Library.Utility.FileBackedStringList();
}
catch
{
@@ -112,6 +116,9 @@ namespace Duplicati.Library.Main.Operation.Backup
// Now add the block to the current volume, as we know there is space for it
blockvolume.AddBlock(b.HashKey, b.Data, b.Offset, (int)b.Size, b.Hint);
if (b.IsBlocklistHashes && useindex)
indexdata.Add(VolumeUploadRequest.EncodeBlockListEntry(b.HashKey, b.Size, b.Data));
}
// We ignore the stop signal, but not the pause and terminate
@@ -125,7 +132,7 @@ namespace Duplicati.Library.Main.Operation.Backup
// If we have collected data, merge all pending volumes into a single volume
if (blockvolume != null && blockvolume.SourceSize > 0)
{
await self.SpillPickup.WriteAsync(new VolumeUploadRequest(blockvolume, true));
await self.SpillPickup.WriteAsync(new VolumeUploadRequest(blockvolume, true, indexdata));
}
}
Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs
+1 -1
View File
@@ -43,7 +43,7 @@ namespace Duplicati.Library.Main.Operation.Backup
return;
await log.WriteInformationAsync(string.Format("Re-creating missing index file for {0}", blockfile));
var w = await Common.IndexVolumeCreator.CreateIndexVolume(blockfile, options, database);
var w = await Common.IndexVolumeCreator.CreateIndexVolume(blockfile, options, database, null);
if (!await taskreader.ProgressAsync)
return;
Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs
+31 -20
View File
@@ -44,6 +44,7 @@ namespace Duplicati.Library.Main.Operation.Backup
async self =>
{
var useindex = options.IndexfilePolicy == Options.IndexFileStrategy.Full;
var lst = new List<VolumeUploadRequest>();
while(!await self.Input.IsRetiredAsync)
@@ -64,28 +65,41 @@ namespace Duplicati.Library.Main.Operation.Backup
// We ignore the stop signal, but not the pause and terminate
await taskreader.ProgressAsync;
VolumeUploadRequest target = null;
// The top-level item is the output target
var target = lst[0];
lst.RemoveAt(0);
// The next item is copied into the target
var source = lst[0];
// Finalize the current work
source.BlockVolume.Close();
// Remove it from the list of active operations
lst.RemoveAt(0);
var buffer = new byte[options.Blocksize];
var indexblocks = useindex ? new HashSet<string>() : null;
using(var rd = new BlockVolumeReader(options.CompressionModule, source.BlockVolume.LocalFilename, options))
{
foreach(var file in rd.Blocks)
{
// Grab a target
if (target == null)
var len = rd.ReadBlock(file.Key, buffer);
// If we do not have enough space, finish this one
if (target.BlockVolume.Filesize + len > options.VolumeSize)
{
if (indexblocks != null && target.BlocklistData != null && source.BlocklistData != null)
{
foreach(var s in source.BlocklistData.Where(x => indexblocks.Contains(VolumeUploadRequest.DecodeBlockListEntryHash(x))))
((Library.Utility.FileBackedStringList)target.BlocklistData).Add(s);
indexblocks = new HashSet<string>();
}
target.BlockVolume.Close();
await self.Output.WriteAsync(target);
if (lst.Count == 0)
{
// No more targets, make one
target = new VolumeUploadRequest(new BlockVolumeWriter(options), true);
target = new VolumeUploadRequest(new BlockVolumeWriter(options), true, useindex ? new Library.Utility.FileBackedStringList() : null);
target.BlockVolume.VolumeID = await database.RegisterRemoteVolumeAsync(target.BlockVolume.RemoteFilename, RemoteVolumeType.Blocks, RemoteVolumeState.Temporary);
}
else
@@ -96,17 +110,10 @@ namespace Duplicati.Library.Main.Operation.Backup
}
}
var len = rd.ReadBlock(file.Key, buffer);
target.BlockVolume.AddBlock(file.Key, buffer, 0, len, Duplicati.Library.Interface.CompressionHint.Default);
await database.MoveBlockToVolumeAsync(file.Key, len, source.BlockVolume.VolumeID, target.BlockVolume.VolumeID);
if (target.BlockVolume.Filesize > options.VolumeSize - options.Blocksize)
{
target.BlockVolume.Close();
await self.Output.WriteAsync(target);
target = null;
}
if (indexblocks != null)
indexblocks.Add(file.Key);
}
}
@@ -114,10 +121,14 @@ namespace Duplicati.Library.Main.Operation.Backup
System.IO.File.Delete(source.BlockVolume.LocalFilename);
await database.SafeDeleteRemoteVolumeAsync(source.BlockVolume.RemoteFilename);
// Re-inject the target if it has content
if (target != null)
lst.Insert(lst.Count == 0 ? 0 : 1, target);
if (indexblocks != null && target.BlocklistData != null && source.BlocklistData != null)
{
foreach(var s in source.BlocklistData.Where(x => indexblocks.Contains(VolumeUploadRequest.DecodeBlockListEntryHash(x))))
((Library.Utility.FileBackedStringList)target.BlocklistData).Add(s);
}
// Re-inject the target
lst.Insert(0, target);
}
foreach(var n in lst)
Duplicati/Library/Main/Operation/Common/BackendHandlerDatabaseGuard.cs
+11 -9
View File
@@ -171,11 +171,12 @@ namespace Duplicati.Library.Main.Operation.Common
/// </summary>
/// <returns>The blocks found in the volume.</returns>
/// <param name="volumeid">The ID of the volume to examine.</param>
public async Task<IEnumerable<Database.LocalDatabase.IBlock>> GetBlocksAsync(long volumeid)
public async Task GetBlocksAsync(long volumeid, Action<Database.LocalDatabase.IBlock> callback)
{
var res = default(IEnumerable<Database.LocalDatabase.IBlock>);
await AddToQueue(() => res = m_db.GetBlocks(volumeid));
return res;
await AddToQueue(() => {
foreach (var n in m_db.GetBlocks(volumeid))
callback(n);
});
}
/// <summary>
@@ -185,11 +186,13 @@ namespace Duplicati.Library.Main.Operation.Common
/// <param name="volumeid">The ID of the volume to get the blocklists for.</param>
/// <param name="blocksize">The blocksize setting.</param>
/// <param name="hashsize">The size of the hash in bytes.</param>
public async Task<IEnumerable<Tuple<string, byte[], int>>> GetBlocklistsAsync(long volumeid, int blocksize, int hashsize)
/// <param name="callback">The callback method called for each result</param>
public Task GetBlocklistsAsync(long volumeid, int blocksize, int hashsize, Action<Tuple<string, byte[], int>> callback)
{
var res = default(IEnumerable<Tuple<string, byte[], int>>);
await AddToQueue(() => res = m_db.GetBlocklists(volumeid, blocksize, hashsize));
return res;
return AddToQueue(() => {
foreach (var e in m_db.GetBlocklists(volumeid, blocksize, hashsize))
callback(e);
});
}
/// <summary>
@@ -206,7 +209,6 @@ namespace Duplicati.Library.Main.Operation.Common
/// <summary>
/// Processes all pending operations into the database.
/// </summary>
/// <param name="transaction">The transaction instance</param>
public void ProcessAllPendingOperations()
{
//if (m_mainthread != System.Threading.Thread.CurrentThread)
Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs
+18 -6
View File
@@ -119,10 +119,13 @@ namespace Duplicati.Library.Main.Operation.Common
/// </summary>
/// <returns>The blocks found in the volume.</returns>
/// <param name="volumeid">The ID of the volume to examine.</param>
public Task<IEnumerable<LocalDatabase.IBlock>> GetBlocksAsync(long volumeid)
/// <param name="callback">The method called for each item</param>
public Task GetBlocksAsync(long volumeid, Action<LocalDatabase.IBlock> callback)
{
// TODO: Consider AsyncEnumerable
return RunOnMain(() => m_db.GetBlocks(volumeid).ToArray().AsEnumerable());
return RunOnMain(() => {
foreach (var n in m_db.GetBlocks(volumeid))
callback(n);
});
}
/// <summary>
@@ -142,12 +145,21 @@ namespace Duplicati.Library.Main.Operation.Common
/// <param name="volumeid">The ID of the volume to get the blocklists for.</param>
/// <param name="blocksize">The blocksize setting.</param>
/// <param name="hashsize">The size of the hash in bytes.</param>
public Task<IEnumerable<Tuple<string, byte[], int>>> GetBlocklistsAsync(long volumeid, int blocksize, int hashsize)
/// <param name="callback">The callback method called for each result</param>
public Task GetBlocklistsAsync(long volumeid, int blocksize, int hashsize, Action<Tuple<string, byte[], int>> callback)
{
// TODO: Consider AsyncEnumerable
return RunOnMain(() => m_db.GetBlocklists(volumeid, blocksize, hashsize).ToArray().AsEnumerable());
return RunOnMain(() =>
{
foreach (var n in m_db.GetBlocklists(volumeid, blocksize, hashsize))
callback(n);
});
}
/// <summary>
/// Gets the ID for a remote volume.
/// </summary>
/// <returns>The remote volume ID.</returns>
/// <param name="remotename">The remote file name.</param>
public Task<long> GetRemoteVolumeIDAsync(string remotename)
{
return RunOnMain(() => m_db.GetRemoteVolumeID(remotename));
Duplicati/Library/Main/Operation/Common/IIndexVolumeCreatorDatabase.cs
+4 -2
View File
@@ -39,7 +39,8 @@ namespace Duplicati.Library.Main.Operation.Common
/// </summary>
/// <returns>The blocks found in the volume.</returns>
/// <param name="volumeid">The ID of the volume to examine.</param>
Task<IEnumerable<Database.LocalDatabase.IBlock>> GetBlocksAsync(long volumeid);
/// <param name="callback">The callback method called for each result</param>
Task GetBlocksAsync(long volumeid, Action<Database.LocalDatabase.IBlock> callback);
/// <summary>
/// Gets the blocklists contained in a remote volume
@@ -48,7 +49,8 @@ namespace Duplicati.Library.Main.Operation.Common
/// <param name="volumeid">The ID of the volume to get the blocklists for.</param>
/// <param name="blocksize">The blocksize setting.</param>
/// <param name="hashsize">The size of the hash in bytes.</param>
Task<IEnumerable<Tuple<string, byte[], int>>> GetBlocklistsAsync(long volumeid, int blocksize, int hashsize);
/// <param name="callback">The callback method called for each result</param>
Task GetBlocklistsAsync(long volumeid, int blocksize, int hashsize, Action<Tuple<string, byte[], int>> callback);
/// <summary>
/// Adds a link between a block volume and an index volume
Duplicati/Library/Main/Operation/Common/IndexVolumeCreator.cs
+27 -9
View File
@@ -15,6 +15,8 @@
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Duplicati.Library.Main.Volumes;
@@ -32,7 +34,8 @@ namespace Duplicati.Library.Main.Operation.Common
/// <param name="blockname">The name of the remote volume to build the index for.</param>
/// <param name="options">The job options.</param>
/// <param name="database">The database to use.</param>
public static async Task<IndexVolumeWriter> CreateIndexVolume(string blockname, Options options, Common.IIndexVolumeCreatorDatabase database)
/// <param name="blocklist">An optional pre-collected list of blocklist hashes</param>
public static async Task<IndexVolumeWriter> CreateIndexVolume(string blockname, Options options, Common.IIndexVolumeCreatorDatabase database, IEnumerable<string> blocklist)
{
using(var h = Duplicati.Library.Utility.HashAlgorithmHelper.Create(options.BlockHashAlgorithm))
{
@@ -41,19 +44,34 @@ namespace Duplicati.Library.Main.Operation.Common
var blockvolume = await database.GetVolumeInfoAsync(blockname);
var st = DateTime.Now;
w.StartVolume(blockname);
foreach(var b in await database.GetBlocksAsync(blockvolume.ID))
w.AddBlock(b.Hash, b.Size);
await database.GetBlocksAsync(blockvolume.ID, b => w.AddBlock(b.Hash, b.Size));
w.FinishVolume(blockvolume.Hash, blockvolume.Size);
if (options.IndexfilePolicy == Options.IndexFileStrategy.Full)
foreach(var b in await database.GetBlocklistsAsync(blockvolume.ID, options.Blocksize, options.BlockhashSize))
using (new Library.Logging.Timer("Create index file"))
{
var bh = Convert.ToBase64String(h.ComputeHash(b.Item2, 0, b.Item3));
if (bh != b.Item1)
throw new Exception(string.Format("Internal consistency check failed, generated index block has wrong hash, {0} vs {1}", bh, b.Item1));
w.WriteBlocklist(b.Item1, b.Item2, 0, b.Item3);
if (blocklist != null)
{
foreach (var b in blocklist.Select(x => Backup.VolumeUploadRequest.DecodeBlockListEntry(x)))
{
var bh = Convert.ToBase64String(h.ComputeHash(b.Item3, 0, (int)b.Item2));
if (bh != b.Item1)
throw new Exception(string.Format("Internal consistency check failed, generated index block has wrong hash, {0} vs {1}", bh, b.Item1));
w.WriteBlocklist(b.Item1, b.Item3, 0, (int)b.Item2);
}
}
else
{
await database.GetBlocklistsAsync(blockvolume.ID, options.Blocksize, options.BlockhashSize, b =>
{
var bh = Convert.ToBase64String(h.ComputeHash(b.Item2, 0, b.Item3));
if (bh != b.Item1)
throw new Exception(string.Format("Internal consistency check failed, generated index block has wrong hash, {0} vs {1}", bh, b.Item1));
w.WriteBlocklist(b.Item1, b.Item2, 0, b.Item3);
});
}
}
w.Close();
Duplicati/Library/Main/Operation/CompactHandler.cs
+1 -1
View File
@@ -121,7 +121,7 @@ namespace Duplicati.Library.Main.Operation
if (m_options.IndexfilePolicy != Options.IndexFileStrategy.None)
{
indexCreator = async (name) => {
var tmp = await Common.IndexVolumeCreator.CreateIndexVolume(name, m_options, backend.Database);
var tmp = await Common.IndexVolumeCreator.CreateIndexVolume(name, m_options, backend.Database, null);
// We assign to a dictionary, to prevent counting multiple items if the upload fails
// and the index file is recreated
uploadedIndexVolumes[name] = new KeyValuePair<string, long>(tmp.RemoteFilename, new System.IO.FileInfo(tmp.LocalFilename).Length);