From ef36a9356bcef93bcd73efe94a81b377febac04f Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Tue, 30 Jan 2018 08:34:57 +0100 Subject: [PATCH] Rewrote the uploader to pass in the index volume blocks, as that is 10x times faster than creating it from the database. This greatly increases performance because the database read is no longer preventing progress. --- .../Main/Operation/Backup/BackendUploader.cs | 31 +++++++++-- .../Operation/Backup/DataBlockProcessor.cs | 13 +++-- .../Backup/RecreateMissingIndexFiles.cs | 2 +- .../Operation/Backup/SpillCollectorProcess.cs | 51 +++++++++++-------- .../Common/BackendHandlerDatabaseGuard.cs | 20 ++++---- .../Main/Operation/Common/DatabaseCommon.cs | 24 ++++++--- .../Common/IIndexVolumeCreatorDatabase.cs | 6 ++- .../Operation/Common/IndexVolumeCreator.cs | 36 +++++++++---- .../Library/Main/Operation/CompactHandler.cs | 2 +- 9 files changed, 130 insertions(+), 55 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Backup/BackendUploader.cs b/Duplicati/Library/Main/Operation/Backup/BackendUploader.cs index 70b4a9c54..805f6983d 100644 --- a/Duplicati/Library/Main/Operation/Backup/BackendUploader.cs +++ b/Duplicati/Library/Main/Operation/Backup/BackendUploader.cs @@ -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 BlocklistData; - public VolumeUploadRequest(BlockVolumeWriter blockvolume, bool createindexvolume) + public VolumeUploadRequest(BlockVolumeWriter blockvolume, bool createindexvolume, IEnumerable 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 DecodeBlockListEntry(string entry) + { + var els = entry.Split(new char[] { ':' }, 3); + return new Tuple( + 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(1, backend.UploadFileAsync(((VolumeUploadRequest)req).BlockVolume, null)); else - task = new KeyValuePair(2, backend.UploadFileAsync(((VolumeUploadRequest)req).BlockVolume, name => IndexVolumeCreator.CreateIndexVolume(name, options, database))); + task = new KeyValuePair(2, backend.UploadFileAsync(((VolumeUploadRequest)req).BlockVolume, name => IndexVolumeCreator.CreateIndexVolume(name, options, database, ((VolumeUploadRequest)req).BlocklistData))); } else if (req is FilesetUploadRequest) task = new KeyValuePair(1, backend.UploadFileAsync(((FilesetUploadRequest)req).Fileset)); diff --git a/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs b/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs index 6863e2bc3..16651578b 100644 --- a/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs +++ b/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs @@ -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)); } } diff --git a/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs b/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs index 78bc917e5..9256009cb 100644 --- a/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs +++ b/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs @@ -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; diff --git a/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs b/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs index bfa56b872..69b6d78a9 100644 --- a/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs +++ b/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs @@ -44,6 +44,7 @@ namespace Duplicati.Library.Main.Operation.Backup async self => { + var useindex = options.IndexfilePolicy == Options.IndexFileStrategy.Full; var lst = new List(); 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() : 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(); + } + + 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) diff --git a/Duplicati/Library/Main/Operation/Common/BackendHandlerDatabaseGuard.cs b/Duplicati/Library/Main/Operation/Common/BackendHandlerDatabaseGuard.cs index fbac8041a..8b4a83c64 100644 --- a/Duplicati/Library/Main/Operation/Common/BackendHandlerDatabaseGuard.cs +++ b/Duplicati/Library/Main/Operation/Common/BackendHandlerDatabaseGuard.cs @@ -171,11 +171,12 @@ namespace Duplicati.Library.Main.Operation.Common /// /// The blocks found in the volume. /// The ID of the volume to examine. - public async Task> GetBlocksAsync(long volumeid) + public async Task GetBlocksAsync(long volumeid, Action callback) { - var res = default(IEnumerable); - await AddToQueue(() => res = m_db.GetBlocks(volumeid)); - return res; + await AddToQueue(() => { + foreach (var n in m_db.GetBlocks(volumeid)) + callback(n); + }); } /// @@ -185,11 +186,13 @@ namespace Duplicati.Library.Main.Operation.Common /// The ID of the volume to get the blocklists for. /// The blocksize setting. /// The size of the hash in bytes. - public async Task>> GetBlocklistsAsync(long volumeid, int blocksize, int hashsize) + /// The callback method called for each result + public Task GetBlocklistsAsync(long volumeid, int blocksize, int hashsize, Action> callback) { - var res = default(IEnumerable>); - 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); + }); } /// @@ -206,7 +209,6 @@ namespace Duplicati.Library.Main.Operation.Common /// /// Processes all pending operations into the database. /// - /// The transaction instance public void ProcessAllPendingOperations() { //if (m_mainthread != System.Threading.Thread.CurrentThread) diff --git a/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs b/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs index 7f797fdf8..99e7b8a29 100644 --- a/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs +++ b/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs @@ -119,10 +119,13 @@ namespace Duplicati.Library.Main.Operation.Common /// /// The blocks found in the volume. /// The ID of the volume to examine. - public Task> GetBlocksAsync(long volumeid) + /// The method called for each item + public Task GetBlocksAsync(long volumeid, Action callback) { - // TODO: Consider AsyncEnumerable - return RunOnMain(() => m_db.GetBlocks(volumeid).ToArray().AsEnumerable()); + return RunOnMain(() => { + foreach (var n in m_db.GetBlocks(volumeid)) + callback(n); + }); } /// @@ -142,12 +145,21 @@ namespace Duplicati.Library.Main.Operation.Common /// The ID of the volume to get the blocklists for. /// The blocksize setting. /// The size of the hash in bytes. - public Task>> GetBlocklistsAsync(long volumeid, int blocksize, int hashsize) + /// The callback method called for each result + public Task GetBlocklistsAsync(long volumeid, int blocksize, int hashsize, Action> 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); + }); } + /// + /// Gets the ID for a remote volume. + /// + /// The remote volume ID. + /// The remote file name. public Task GetRemoteVolumeIDAsync(string remotename) { return RunOnMain(() => m_db.GetRemoteVolumeID(remotename)); diff --git a/Duplicati/Library/Main/Operation/Common/IIndexVolumeCreatorDatabase.cs b/Duplicati/Library/Main/Operation/Common/IIndexVolumeCreatorDatabase.cs index 88af6018a..4e983d284 100644 --- a/Duplicati/Library/Main/Operation/Common/IIndexVolumeCreatorDatabase.cs +++ b/Duplicati/Library/Main/Operation/Common/IIndexVolumeCreatorDatabase.cs @@ -39,7 +39,8 @@ namespace Duplicati.Library.Main.Operation.Common /// /// The blocks found in the volume. /// The ID of the volume to examine. - Task> GetBlocksAsync(long volumeid); + /// The callback method called for each result + Task GetBlocksAsync(long volumeid, Action callback); /// /// Gets the blocklists contained in a remote volume @@ -48,7 +49,8 @@ namespace Duplicati.Library.Main.Operation.Common /// The ID of the volume to get the blocklists for. /// The blocksize setting. /// The size of the hash in bytes. - Task>> GetBlocklistsAsync(long volumeid, int blocksize, int hashsize); + /// The callback method called for each result + Task GetBlocklistsAsync(long volumeid, int blocksize, int hashsize, Action> callback); /// /// Adds a link between a block volume and an index volume diff --git a/Duplicati/Library/Main/Operation/Common/IndexVolumeCreator.cs b/Duplicati/Library/Main/Operation/Common/IndexVolumeCreator.cs index dcf8416e0..f4db10b77 100644 --- a/Duplicati/Library/Main/Operation/Common/IndexVolumeCreator.cs +++ b/Duplicati/Library/Main/Operation/Common/IndexVolumeCreator.cs @@ -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 /// The name of the remote volume to build the index for. /// The job options. /// The database to use. - public static async Task CreateIndexVolume(string blockname, Options options, Common.IIndexVolumeCreatorDatabase database) + /// An optional pre-collected list of blocklist hashes + public static async Task CreateIndexVolume(string blockname, Options options, Common.IIndexVolumeCreatorDatabase database, IEnumerable 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(); diff --git a/Duplicati/Library/Main/Operation/CompactHandler.cs b/Duplicati/Library/Main/Operation/CompactHandler.cs index b939ee182..8d5636659 100644 --- a/Duplicati/Library/Main/Operation/CompactHandler.cs +++ b/Duplicati/Library/Main/Operation/CompactHandler.cs @@ -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(tmp.RemoteFilename, new System.IO.FileInfo(tmp.LocalFilename).Length);