Experiment with batch processing of blocks

Duplicati/Library/Main/Operation/Backup/Channels.cs

@@ -19,7 +19,7 @@
 // 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 CoCoL;
 
 namespace Duplicati.Library.Main.Operation.Backup
@@ -42,6 +42,10 @@ namespace Duplicati.Library.Main.Operation.Backup
         /// </summary>
         public static readonly ChannelMarkerWrapper<DataBlock> OutputBlocks = new ChannelMarkerWrapper<DataBlock>(new ChannelNameAttribute("OutputBlocks"));
         /// <summary>
+        /// All data blocks are sent during the scanning to the <see cref="DataBlockProcessor"/> who bundles them in compressed archives
+        /// </summary>
+        public static readonly ChannelMarkerWrapper<IList<DataBlock>> OutputBlocksChunked = new ChannelMarkerWrapper<IList<DataBlock>>(new ChannelNameAttribute("OutputBlocksChunked"));
+        /// <summary>
         /// If a file has changes in the metadata, it is sent to the <see cref="FileBlockProcessor"/> where it is read
         /// </summary>
         public static readonly ChannelMarkerWrapper<MetadataPreProcess.FileEntry> AcceptedChangedFile = new ChannelMarkerWrapper<MetadataPreProcess.FileEntry>(new ChannelNameAttribute("AcceptedChangedFile"));

Duplicati/Library/Main/Operation/Backup/DataBlock.cs

@@ -23,38 +23,63 @@ using System;
 using Duplicati.Library.Interface;
 using System.Threading.Tasks;
 using CoCoL;
+using System.Buffers;
 
 namespace Duplicati.Library.Main.Operation.Backup
 {
     /// <summary>
     /// The data block represents a single blob of data read from a file
     /// </summary>
-    internal struct DataBlock
+    internal sealed record DataBlock : IDisposable
     {
-        public string HashKey;
-        public byte[] Data;
-        public int Offset;
-        public long Size;
-        public CompressionHint Hint;
-        public bool IsBlocklistHashes;
-        public TaskCompletionSource<bool> TaskCompletion;
+        public DataBlock(ArrayPool<byte> arrayPool)
+            => _arrayPool = arrayPool;
 
-        public static async Task<bool> AddBlockToOutputAsync(IWriteChannel<DataBlock> channel, string hash, byte[] data, int offset, long size, CompressionHint hint, bool isBlocklistHashes)
+        public required string HashKey { get; init; }
+        public required byte[] Data { get; init; }
+        public required int Offset { get; init; }
+        public required long Size { get; init; }
+        public required CompressionHint Hint { get; init; }
+        public required bool IsBlocklistHashes { get; init; }
+        public TaskCompletionSource<bool> TaskCompletion { get; } = new TaskCompletionSource<bool>();
+        private readonly ArrayPool<byte> _arrayPool;
+        private bool _disposed;
+
+        public static async Task AddBlockToOutputAsync(IWriteChannel<DataBlock> channel, string hash, ArrayPool<byte> arrayPool, byte[] data, int offset, long size, CompressionHint hint, bool isBlocklistHashes)
         {
-            var tcs = new TaskCompletionSource<bool>();
-
-            await channel.WriteAsync(new DataBlock() {
+            var b = new DataBlock(arrayPool)
+            {
                 HashKey = hash,
                 Data = data,
                 Offset = offset,
                 Size = size,
                 Hint = hint,
                 IsBlocklistHashes = isBlocklistHashes,
-                TaskCompletion = tcs
-            });
+            };
+            await channel.WriteAsync(b);
 
-            var r = await tcs.Task.ConfigureAwait(false);
-            return r;
+            await b.TaskCompletion.Task.ConfigureAwait(false);
+        }
+
+        public void CompleteSuccess()
+        {
+            Dispose();
+            TaskCompletion.SetResult(true);
+        }
+
+        public void CompleteFailure(Exception ex)
+        {
+            Dispose();
+            TaskCompletion.TrySetException(ex);
+        }
+
+        public void Dispose()
+        {
+            if (!_disposed)
+            {
+                _arrayPool.Return(Data);
+                _disposed = true;
+            }
         }
     }
 }

Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs

@@ -23,6 +23,8 @@ using CoCoL;
 using Duplicati.Library.Main.Operation.Common;
 using Duplicati.Library.Main.Volumes;
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using static Duplicati.Library.Main.Operation.Common.BackendHandler;
 
@@ -35,12 +37,54 @@ namespace Duplicati.Library.Main.Operation.Backup
     /// </summary>
     internal static class DataBlockProcessor
     {
+        public static Task RunBatcher(int bufferSize)
+        {
+            return RunBuffer(Channels.OutputBlocks.ForRead, Channels.OutputBlocksChunked.ForWrite, bufferSize);
+        }
+
+        private static Task RunBuffer<T>(IReadChannel<T> readChannel, IWriteChannel<IList<T>> writeChannel, int bufferSize)
+            where T : class
+        {
+            return AutomationExtensions.RunTask(new { readChannel, writeChannel }, async self =>
+            {
+                var buffer = new List<T>(bufferSize);
+
+                try
+                {
+                    while (true)
+                    {
+                        var r = await self.readChannel.ReadAsync();
+
+                        // Null messages are flush messages
+                        if (r != null)
+                            buffer.Add(r);
+
+                        if (buffer.Count >= bufferSize || (r == null && buffer.Count != 0))
+                        {
+                            await self.writeChannel.WriteAsync(buffer);
+                            buffer = new List<T>(bufferSize);
+                        }
+                    }
+                }
+                finally
+                {
+                    try
+                    {
+                        // Flush the last buffer
+                        if (buffer.Count > 0)
+                            await self.writeChannel.WriteAsync(buffer);
+                    }
+                    catch { }
+                }
+            });
+        }
+
         public static Task Run(BackupDatabase database, Options options, ITaskReader taskreader)
         {
             return AutomationExtensions.RunTask(
             new
             {
-                Input = Channels.OutputBlocks.ForRead,
+                Input = Channels.OutputBlocksChunked.ForRead,
                 Output = Channels.BackendRequest.ForWrite,
                 SpillPickup = Channels.SpillPickup.ForWrite,
             },
@@ -53,74 +97,76 @@ namespace Duplicati.Library.Main.Operation.Backup
                 BlockVolumeWriter blockvolume = null;
                 TemporaryIndexVolume indexvolume = null;
 
+                IEnumerable<DataBlock> blockChunks = null;
+
                 try
                 {
                     while (true)
                     {
-                        var b = await self.Input.ReadAsync();
+                        blockChunks = null;
+                        blockChunks = await self.Input.ReadAsync();
 
-                        // Lazy-start a new block volume
-                        if (blockvolume == null)
+                        if (blockChunks == null || blockChunks.Count() == 0)
+                            continue;
+
+                        foreach (var b in blockChunks)
                         {
-                            // Before we start a new volume, probe to see if it exists
-                            // This will delay creation of volumes for differential backups
-                            // There can be a race, such that two workers determine that
-                            // the block is missing, but this will be solved by the AddBlock call
-                            // which runs atomically
-                            if (await database.FindBlockIDAsync(b.HashKey, b.Size) >= 0)
+                            // Make sure we have a volume to write to
+                            if (blockvolume == null)
                             {
-                                b.TaskCompletion.TrySetResult(false);
-                                continue;
-                            }
-
-                            blockvolume = new BlockVolumeWriter(options);
-                            blockvolume.VolumeID = await database.RegisterRemoteVolumeAsync(blockvolume.RemoteFilename, RemoteVolumeType.Blocks, RemoteVolumeState.Temporary);
-
-                            indexvolume = noIndexFiles ? null : new TemporaryIndexVolume(options);
-                        }
-
-                        var newBlock = await database.AddBlockAsync(b.HashKey, b.Size, blockvolume.VolumeID);
-                        b.TaskCompletion.TrySetResult(newBlock);
-
-                        if (newBlock)
-                        {
-                            blockvolume.AddBlock(b.HashKey, b.Data, b.Offset, (int)b.Size, b.Hint);
-                            if (indexvolume != null)
-                            {
-                                indexvolume.AddBlock(b.HashKey, b.Size);
-                                if (b.IsBlocklistHashes && fullIndexFiles)
-                                    indexvolume.AddBlockListHash(b.HashKey, b.Size, b.Data);
-                            }
-
-                            // If the volume is full, send to upload
-                            if (blockvolume.Filesize > options.VolumeSize - options.Blocksize)
-                            {
-                                //When uploading a new volume, we register the volumes and then flush the transaction
-                                // this ensures that the local database and remote storage are as closely related as possible
-                                await database.UpdateRemoteVolumeAsync(blockvolume.RemoteFilename, RemoteVolumeState.Uploading, -1, null);
-
-                                blockvolume.Close();
-
-                                await database.CommitTransactionAsync("CommitAddBlockToOutputFlush");
-
-                                FileEntryItem blockEntry = blockvolume.CreateFileEntryForUpload(options);
-
-                                TemporaryIndexVolume indexVolumeCopy = null;
-                                if (indexvolume != null)
+                                if (await database.FindBlockIDAsync(b.HashKey, b.Size) >= 0)
                                 {
-                                    indexVolumeCopy = new TemporaryIndexVolume(options);
-                                    indexvolume.CopyTo(indexVolumeCopy, false);
+                                    b.CompleteSuccess();
+                                    continue;
                                 }
 
-                                var uploadRequest = new VolumeUploadRequest(blockvolume, blockEntry, indexVolumeCopy, options, database);
+                                blockvolume = new BlockVolumeWriter(options);
+                                blockvolume.VolumeID = await database.RegisterRemoteVolumeAsync(blockvolume.RemoteFilename, RemoteVolumeType.Blocks, RemoteVolumeState.Temporary);
 
-                                blockvolume = null;
-                                indexvolume = null;
-
-                                // Write to output at the end here to prevent sending a full volume to the SpillCollector
-                                await self.Output.WriteAsync(uploadRequest);
+                                indexvolume = noIndexFiles ? null : new TemporaryIndexVolume(options);
                             }
 
+                            if (await database.AddBlockAsync(b.HashKey, b.Size, blockvolume.VolumeID))
+                            {
+                                blockvolume.AddBlock(b.HashKey, b.Data, b.Offset, (int)b.Size, b.Hint);
+                                if (indexvolume != null)
+                                {
+                                    indexvolume.AddBlock(b.HashKey, b.Size);
+                                    if (b.IsBlocklistHashes && fullIndexFiles)
+                                        indexvolume.AddBlockListHash(b.HashKey, b.Size, b.Data);
+                                }
+
+                                // If the volume is full, send to upload
+                                if (blockvolume.Filesize > options.VolumeSize - options.Blocksize)
+                                {
+                                    //When uploading a new volume, we register the volumes and then flush the transaction
+                                    // this ensures that the local database and remote storage are as closely related as possible
+                                    await database.UpdateRemoteVolumeAsync(blockvolume.RemoteFilename, RemoteVolumeState.Uploading, -1, null);
+
+                                    blockvolume.Close();
+
+                                    await database.CommitTransactionAsync("CommitAddBlockToOutputFlush");
+
+                                    FileEntryItem blockEntry = blockvolume.CreateFileEntryForUpload(options);
+
+                                    TemporaryIndexVolume indexVolumeCopy = null;
+                                    if (indexvolume != null)
+                                    {
+                                        indexVolumeCopy = new TemporaryIndexVolume(options);
+                                        indexvolume.CopyTo(indexVolumeCopy, false);
+                                    }
+
+                                    var uploadRequest = new VolumeUploadRequest(blockvolume, blockEntry, indexVolumeCopy, options, database);
+
+                                    blockvolume = null;
+                                    indexvolume = null;
+
+                                    // Write to output at the end here to prevent sending a full volume to the SpillCollector
+                                    await self.Output.WriteAsync(uploadRequest);
+                                }
+                            }
+
+                            b.CompleteSuccess();
                         }
 
                         // We ignore the stop signal, but not the pause and terminate
@@ -129,6 +175,10 @@ namespace Duplicati.Library.Main.Operation.Backup
                 }
                 catch (Exception ex)
                 {
+                    if (blockChunks != null)
+                        foreach (var b in blockChunks)
+                            b.CompleteFailure(ex);
+
                     if (ex.IsRetiredException())
                     {
                         // If we have collected data, merge all pending volumes into a single volume

Duplicati/Library/Main/Operation/Backup/StreamBlockSplitter.cs

@@ -25,10 +25,9 @@ using Duplicati.Library.Main.Operation.Common;
 using System.Threading.Tasks;
 using System.Collections.Generic;
 using Duplicati.Library.Utility;
-using System.Linq;
 using Duplicati.Library.Interface;
 using System.IO;
-using System.Security.Cryptography;
+using System.Buffers;
 
 namespace Duplicati.Library.Main.Operation.Backup
 {
@@ -37,10 +36,10 @@ namespace Duplicati.Library.Main.Operation.Backup
         /// <summary>
         /// The tag used for log messages
         /// </summary>
-        private static readonly string LOGTAG = Logging.Log.LogTagFromType(typeof(StreamBlockSplitter)) ;
+        private static readonly string LOGTAG = Logging.Log.LogTagFromType(typeof(StreamBlockSplitter));
         private static readonly string FILELOGTAG = LOGTAG + ".FileEntry";
 
-        public static Task Run(Options options, BackupDatabase database, ITaskReader taskreader)
+        public static Task Run(Options options, BackupDatabase database, ITaskReader taskreader, ArrayPool<byte> arrayPool)
         {
             return AutomationExtensions.RunTask(
             new
@@ -54,10 +53,10 @@ namespace Duplicati.Library.Main.Operation.Backup
             {
                 var blocksize = options.Blocksize;
                 var emptymetadata = Utility.WrapMetadata(new Dictionary<string, string>(), options);
-                var maxmetadatasize = (options.Blocksize / (long) options.BlockhashSize) * options.Blocksize;
+                var maxmetadatasize = (options.Blocksize / (long)options.BlockhashSize) * options.Blocksize;
 
-                using(var filehasher = HashFactory.CreateHasher(options.FileHashAlgorithm))
-                using(var blockhasher = HashFactory.CreateHasher(options.BlockHashAlgorithm))
+                using (var filehasher = HashFactory.CreateHasher(options.FileHashAlgorithm))
+                using (var blockhasher = HashFactory.CreateHasher(options.BlockHashAlgorithm))
                 using (var empty_metadata_stream = new MemoryStream(emptymetadata.Blob))
                 {
                     while (await taskreader.ProgressAsync)
@@ -67,6 +66,7 @@ namespace Duplicati.Library.Main.Operation.Backup
 
                         var e = await self.Input.ReadAsync();
                         var cur = e.Result;
+                        var tasks = new List<Task>();
 
                         try
                         {
@@ -75,7 +75,7 @@ namespace Duplicati.Library.Main.Operation.Backup
                             using (var blocklisthashes = new Library.Utility.FileBackedStringList())
                             using (var hashcollector = new Library.Utility.FileBackedStringList())
                             {
-                                var blocklistbuffer = new byte[blocksize];
+                                var blocklistbuffer = arrayPool.Rent(blocksize);
                                 var blocklistoffset = 0L;
 
                                 long fslen = -1;
@@ -105,13 +105,13 @@ namespace Duplicati.Library.Main.Operation.Backup
                                 // Don't send progress reports for metadata
                                 if (!e.IsMetadata)
                                 {
-                                    await self.ProgressChannel.WriteAsync(new ProgressEvent() {Filepath = e.Path, Length = fslen, Type = EventType.FileStarted});
+                                    await self.ProgressChannel.WriteAsync(new ProgressEvent() { Filepath = e.Path, Length = fslen, Type = EventType.FileStarted });
                                     send_close = true;
                                 }
 
                                 filehasher.Initialize();
                                 var lastread = 0;
-                                var buf = new byte[blocksize];
+                                var buf = arrayPool.Rent(blocksize);
                                 var lastupdate = DateTime.Now;
 
                                 // Core processing loop, read blocks of data and hash individually
@@ -124,13 +124,13 @@ namespace Duplicati.Library.Main.Operation.Backup
                                     var hashkey = Convert.ToBase64String(hashdata);
 
                                     // If we have too many hashes, flush the blocklist
-                                    if (blocklistbuffer.Length - blocklistoffset < hashdata.Length)
+                                    if (blocksize - blocklistoffset < hashdata.Length)
                                     {
-                                        var blkey = Convert.ToBase64String(blockhasher.ComputeHash(blocklistbuffer, 0, (int) blocklistoffset));
+                                        var blkey = Convert.ToBase64String(blockhasher.ComputeHash(blocklistbuffer, 0, (int)blocklistoffset));
                                         blocklisthashes.Add(blkey);
-                                        await DataBlock.AddBlockToOutputAsync(self.BlockOutput, blkey, blocklistbuffer, 0, blocklistoffset, CompressionHint.Noncompressible, true);
+                                        tasks.Add(DataBlock.AddBlockToOutputAsync(self.BlockOutput, blkey, arrayPool, blocklistbuffer, 0, blocklistoffset, CompressionHint.Noncompressible, true));
                                         blocklistoffset = 0;
-                                        blocklistbuffer = new byte[blocksize];
+                                        blocklistbuffer = arrayPool.Rent(blocksize);
                                     }
 
                                     // Store the current hash in the blocklist
@@ -142,28 +142,33 @@ namespace Duplicati.Library.Main.Operation.Backup
                                     // Don't spam updates
                                     if (send_close && (DateTime.Now - lastupdate).TotalSeconds > 5)
                                     {
-                                        await self.ProgressChannel.WriteAsync(new ProgressEvent() {Filepath = e.Path, Length = filesize, Type = EventType.FileProgressUpdate});
+                                        await self.ProgressChannel.WriteAsync(new ProgressEvent() { Filepath = e.Path, Length = filesize, Type = EventType.FileProgressUpdate });
                                         lastupdate = DateTime.Now;
                                     }
 
                                     // Make sure the filehasher is done with the buf instance before we pass it on
                                     await pftask.ConfigureAwait(false);
-                                    await DataBlock.AddBlockToOutputAsync(self.BlockOutput, hashkey, buf, 0, lastread, e.Hint, false);
-                                    buf = new byte[blocksize];
+                                    tasks.Add(DataBlock.AddBlockToOutputAsync(self.BlockOutput, hashkey, arrayPool, buf, 0, lastread, e.Hint, false));
+                                    buf = arrayPool.Rent(blocksize);
                                 }
 
                                 // If we have more than a single block of data, output the (trailing) blocklist
                                 if (hashcollector.Count > 1)
                                 {
-                                    var blkey = Convert.ToBase64String(blockhasher.ComputeHash(blocklistbuffer, 0, (int) blocklistoffset));
+                                    var blkey = Convert.ToBase64String(blockhasher.ComputeHash(blocklistbuffer, 0, (int)blocklistoffset));
                                     blocklisthashes.Add(blkey);
-                                    await DataBlock.AddBlockToOutputAsync(self.BlockOutput, blkey, blocklistbuffer, 0, blocklistoffset, CompressionHint.Noncompressible, true);
+                                    tasks.Add(DataBlock.AddBlockToOutputAsync(self.BlockOutput, blkey, arrayPool, blocklistbuffer, 0, blocklistoffset, CompressionHint.Noncompressible, true));
                                 }
 
+                                // TOOD: A bit clunky, but all hashes must be added to the database before we can add the blockset,
+                                // and the blocks can be split across multiple volumes
+                                await self.BlockOutput.WriteAsync(null);
+                                await Task.WhenAll(tasks);
+
                                 filehasher.TransformFinalBlock(new byte[0], 0, 0);
                                 var filehash = Convert.ToBase64String(filehasher.Hash);
                                 var blocksetid = await database.AddBlocksetAsync(filehash, filesize, blocksize, hashcollector, blocklisthashes);
-                                cur.SetResult(new StreamProcessResult() {Streamlength = filesize, Streamhash = filehash, Blocksetid = blocksetid});
+                                cur.SetResult(new StreamProcessResult() { Streamlength = filesize, Streamhash = filehash, Blocksetid = blocksetid });
                                 cur = null;
                             }
                         }
@@ -198,7 +203,7 @@ namespace Duplicati.Library.Main.Operation.Backup
                             }
 
                             if (send_close)
-                                await self.ProgressChannel.WriteAsync(new ProgressEvent() {Filepath = e.Path, Length = filesize, Type = EventType.FileClosed});
+                                await self.ProgressChannel.WriteAsync(new ProgressEvent() { Filepath = e.Path, Length = filesize, Type = EventType.FileClosed });
                             send_close = false;
                         }
                     }

Duplicati/Library/Main/Operation/BackupHandler.cs

@@ -31,8 +31,8 @@ using System.Threading;
 using Duplicati.Library.Snapshots;
 using Duplicati.Library.Utility;
 using Duplicati.Library.Common.IO;
-using Duplicati.Library.Common;
 using Duplicati.Library.Logging;
+using System.Buffers;
 
 namespace Duplicati.Library.Main.Operation
 {
@@ -190,16 +190,19 @@ namespace Duplicati.Library.Main.Operation
             {
                 // Make sure the CompressionHints table is initialized, otherwise all workers will initialize it
                 var unused = options.CompressionHints.Count;
+                var arrayPool = ArrayPool<byte>.Shared;
 
                 Task all;
+
                 using (new ChannelScope())
                 {
                     all = Task.WhenAll(
                         new[]
                             {
+                                    Backup.DataBlockProcessor.RunBatcher(options.ConcurrencyDataBlocks),
                                     Backup.DataBlockProcessor.Run(database, options, taskreader),
                                     Backup.FileBlockProcessor.Run(snapshot, options, database, stats, taskreader, token),
-                                    Backup.StreamBlockSplitter.Run(options, database, taskreader),
+                                    Backup.StreamBlockSplitter.Run(options, database, taskreader, arrayPool),
                                     Backup.FileEnumerationProcess.Run(sources, snapshot, journalService,
                                         options.FileAttributeFilter, sourcefilter, filter, options.SymlinkPolicy,
                                         options.HardlinkPolicy, options.ExcludeEmptyFolders, options.IgnoreFilenames,
@@ -212,7 +215,7 @@ namespace Duplicati.Library.Main.Operation
                             // Spawn additional block hashers
                             .Union(
                                 Enumerable.Range(0, options.ConcurrencyBlockHashers - 1).Select(x =>
-                                    Backup.StreamBlockSplitter.Run(options, database, taskreader))
+                                    Backup.StreamBlockSplitter.Run(options, database, taskreader, arrayPool))
                             )
                             // Spawn additional compressors
                             .Union(

Duplicati/Library/Main/Options.cs

@@ -77,6 +77,11 @@ namespace Duplicati.Library.Main
         /// </summary>
         private readonly int DEFAULT_BLOCK_HASHERS = Math.Max(1, Environment.ProcessorCount / 2);
 
+        /// <summary>
+        /// The default number of data blocks to batch
+        /// </summary>
+        private readonly int DEFAULT_CONCURRENCY_DATA_BLOCKS = 1000;
+
         /// <summary>
         /// The default threshold for warning about coming close to quota
         /// </summary>
@@ -401,6 +406,7 @@ namespace Duplicati.Library.Main
                     new CommandLineArgument("concurrency-max-threads", CommandLineArgument.ArgumentType.Integer, Strings.Options.ConcurrencymaxthreadsShort, Strings.Options.ConcurrencymaxthreadsLong, "0"),
                     new CommandLineArgument("concurrency-block-hashers", CommandLineArgument.ArgumentType.Integer, Strings.Options.ConcurrencyblockhashersShort, Strings.Options.ConcurrencyblockhashersLong, DEFAULT_BLOCK_HASHERS.ToString()),
                     new CommandLineArgument("concurrency-compressors", CommandLineArgument.ArgumentType.Integer, Strings.Options.ConcurrencycompressorsShort, Strings.Options.ConcurrencycompressorsLong, DEFAULT_COMPRESSORS.ToString()),
+                    new CommandLineArgument("concurrency-data-blocks", CommandLineArgument.ArgumentType.Integer, Strings.Options.ConcurrencydatablocksShort, Strings.Options.ConcurrencydatablocksLong, DEFAULT_CONCURRENCY_DATA_BLOCKS.ToString()),
 
                     new CommandLineArgument("auto-vacuum", CommandLineArgument.ArgumentType.Boolean, Strings.Options.AutoVacuumShort, Strings.Options.AutoVacuumLong, "false"),
                     new CommandLineArgument("disable-file-scanner", CommandLineArgument.ArgumentType.Boolean, Strings.Options.DisablefilescannerShort, Strings.Options.DisablefilescannerLong, "false"),
@@ -1867,6 +1873,21 @@ namespace Duplicati.Library.Main
             }
         }
 
+        public int ConcurrencyDataBlocks
+        {
+            get
+            {
+                string value;
+                if (!m_options.TryGetValue("concurrency-data-blocks", out value))
+                    value = null;
+
+                if (string.IsNullOrEmpty(value))
+                    return DEFAULT_CONCURRENCY_DATA_BLOCKS;
+                else
+                    return int.Parse(value);
+            }
+        }
+
         /// <summary>
         /// Gets the number of concurrent block hashers
         /// </summary>

Duplicati/Library/Main/Strings.cs

@@ -260,6 +260,8 @@ namespace Duplicati.Library.Main.Strings
         public static string ConcurrencyblockhashersLong { get { return LC.L(@"Use this option to set the number of processes that perform hashing of data."); } }
         public static string ConcurrencycompressorsShort { get { return LC.L(@"Specify the number of concurrent compression processes"); } }
         public static string ConcurrencycompressorsLong { get { return LC.L(@"Use this option to set the number of processes that perform compression of output data."); } }
+        public static string ConcurrencydatablocksShort { get { return LC.L(@"Specify the number of concurrent data block processing processes"); } }
+        public static string ConcurrencydatablocksLong { get { return LC.L(@"Use this option to set the number of blocks that are batch processed."); } }
         public static string DisablesyntehticfilelistLong { get { return LC.L(@"If Duplicati detects that the previous backup did not complete, it will generate a filelist that is a merge of the last completed backup and the contents that were uploaded in the incomplete backup session."); } }
         public static string DisablesyntheticfilelistShort { get { return LC.L(@"Disables synthetic filelist"); } }
         public static string CheckfiletimeonlyLong { get { return LC.L(@"This flag instructs Duplicati to not look at metadata or filesize when deciding to scan a file for changes. Use this option if you have a large number of files and notice that the scanning takes a long time with unmodified files."); } }