diff --git a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs index 722a9437c..5812b9560 100644 --- a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs +++ b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Lesser General Public // 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 CoCoL; using System.IO; using System.Threading.Tasks; @@ -24,9 +24,9 @@ namespace Duplicati.Library.Main.Operation.Backup { internal struct StreamProcessResult { - public string Streamhash; - public long Streamlength; - public long Blocksetid; + public string Streamhash { get; internal set; } + public long Streamlength { get; internal set; } + public long Blocksetid { get; internal set; } } internal struct StreamBlock @@ -40,16 +40,23 @@ namespace Duplicati.Library.Main.Operation.Backup public static async Task ProcessStream(IWriteChannel channel, string path, Stream stream, bool isMetadata, CompressionHint hint) { var tcs = new TaskCompletionSource(); - await channel.WriteAsync(new StreamBlock() + + // limit the stream length to that found now, a fixed point in time + var limitedStream = new Library.Utility.ReadLimitLengthStream(stream, stream.Length); + + var streamBlock = new StreamBlock { Path = path, - Stream = stream, + Stream = limitedStream, IsMetadata = isMetadata, Hint = hint, Result = tcs - }); - + }; + + await channel.WriteAsync(streamBlock); + return await tcs.Task.ConfigureAwait(false); } } + } diff --git a/Duplicati/Library/Utility/Duplicati.Library.Utility.csproj b/Duplicati/Library/Utility/Duplicati.Library.Utility.csproj index 42a947f5b..b554b42b8 100644 --- a/Duplicati/Library/Utility/Duplicati.Library.Utility.csproj +++ b/Duplicati/Library/Utility/Duplicati.Library.Utility.csproj @@ -56,6 +56,7 @@ + diff --git a/Duplicati/Library/Utility/ReadLimitLengthStream.cs b/Duplicati/Library/Utility/ReadLimitLengthStream.cs new file mode 100644 index 000000000..7aab7f2d8 --- /dev/null +++ b/Duplicati/Library/Utility/ReadLimitLengthStream.cs @@ -0,0 +1,151 @@ +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Duplicati.Library.Utility +{ + // StreamReadLimitLengthWrapper() based on code from Matt Smith (Oct 26 '15 at 20:38) + // https://stackoverflow.com/questions/33354822/how-to-set-length-in-stream-without-truncated + + public class ReadLimitLengthStream : Stream + { + readonly Stream m_innerStream; + readonly long m_endPosition; + + public ReadLimitLengthStream(Stream innerStream, long size) + { + if (size < 0) + { + throw new ArgumentOutOfRangeException(nameof(size)); + } + + m_innerStream = innerStream ?? throw new ArgumentNullException(nameof(innerStream)); + m_endPosition = m_innerStream.Position + size; + } + + public override bool CanRead => m_innerStream.CanRead; + + public override bool CanSeek => m_innerStream.CanSeek; + + public override bool CanWrite => false; + + public override void Flush() + { + // NOOP + } + + public override long Length => m_endPosition; + + public override long Position + { + get => m_innerStream.Position; + set => m_innerStream.Position = value; + } + + public override int Read(byte[] buffer, int offset, int count) + { + count = GetAllowedCount(count); + return m_innerStream.Read(buffer, offset, count); + } + + public override long Seek(long offset, SeekOrigin origin) + { + return m_innerStream.Seek(offset, origin); + } + + public override void SetLength(long value) + { + throw new NotSupportedException(); + } + + public override void Write(byte[] buffer, int offset, int count) + { + throw new NotSupportedException(); + } + + public override bool CanTimeout => m_innerStream.CanTimeout; + + public override int ReadTimeout + { + get => m_innerStream.ReadTimeout; + set => m_innerStream.ReadTimeout = value; + } + + public override int WriteTimeout + { + get => m_innerStream.ReadTimeout; + set => m_innerStream.ReadTimeout = value; + } + + public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state) + { + count = GetAllowedCount(count); + return m_innerStream.BeginRead(buffer, offset, count, callback, state); + } + + public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state) + { + throw new NotSupportedException(); + } + + public override void Close() + { + // Since this wrapper does not own the underlying stream, we do not want it to close the underlying stream + } + + public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken) + { + return m_innerStream.CopyToAsync(destination, bufferSize, cancellationToken); + } + + public override int EndRead(IAsyncResult asyncResult) + { + return m_innerStream.EndRead(asyncResult); + } + + public override Task FlushAsync(CancellationToken cancellationToken) + { + return m_innerStream.FlushAsync(cancellationToken); + } + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + count = GetAllowedCount(count); + return m_innerStream.ReadAsync(buffer, offset, count, cancellationToken); + } + + public override int ReadByte() + { + var count = GetAllowedCount(1); + if (count == 0) + { + return -1; + } + + return m_innerStream.ReadByte(); + } + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + throw new NotSupportedException(); + } + + public override void WriteByte(byte value) + { + throw new NotSupportedException(); + } + + private int GetAllowedCount(int count) + { + long pos = m_innerStream.Position; + long maxCount = m_endPosition - pos; + if (count > maxCount) + { + return (int)maxCount; + } + + return count; + } + } +} diff --git a/Duplicati/UnitTest/TestUtils.cs b/Duplicati/UnitTest/TestUtils.cs index 80e301b9b..0c8732d65 100644 --- a/Duplicati/UnitTest/TestUtils.cs +++ b/Duplicati/UnitTest/TestUtils.cs @@ -21,6 +21,8 @@ using Duplicati.Library.Utility; using System.Linq; using Duplicati.Library.Logging; using System.Reflection; +using System.Threading; +using System.Threading.Tasks; using Duplicati.Library.Common.IO; namespace Duplicati.UnitTest @@ -46,6 +48,30 @@ namespace Duplicati.UnitTest } } + public static async Task GrowingFile(string testFile, CancellationToken token) + { + try + { + var str = new string('*', 50); + while (true) + { + if (token.IsCancellationRequested) + { + continue; + } + File.AppendAllText(testFile, str); + await Task.Delay(18, token).ConfigureAwait(false); + } + } + finally + { + if (File.Exists(testFile)) + { + File.Delete(testFile); + } + } + } + public static string GetDefaultTarget(string other = null) { string alttarget = System.IO.Path.Combine(Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location), "unittest_target.txt"); diff --git a/Duplicati/UnitTest/UtilityTests.cs b/Duplicati/UnitTest/UtilityTests.cs index 2ef4c7a59..437889f77 100644 --- a/Duplicati/UnitTest/UtilityTests.cs +++ b/Duplicati/UnitTest/UtilityTests.cs @@ -22,6 +22,8 @@ using System.Collections.Generic; using System.Linq; using System.Globalization; using System.IO; +using System.Threading; +using System.Threading.Tasks; namespace Duplicati.UnitTest { @@ -94,6 +96,35 @@ namespace Duplicati.UnitTest } } + [Test] + [Category("Utility")] + public static void ReadLimitLengthStreamWithGrowingFile() + { + var growingFilename = Path.GetTempFileName(); + long fixedGrowingStreamLength, limitStreamLength, nextGrowingStreamLength; + using (CancellationTokenSource tokenSource = new CancellationTokenSource()) + { + Task task = TestUtils.GrowingFile(growingFilename, tokenSource.Token); + Thread.Sleep(100); + using (FileStream growingStream = new FileStream(growingFilename, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) + { + Thread.Sleep(100); + fixedGrowingStreamLength = growingStream.Length; + Thread.Sleep(100); + using (var limitStream = new ReadLimitLengthStream(growingStream, fixedGrowingStreamLength)) + { + Thread.Sleep(100); + nextGrowingStreamLength = growingStream.Length; + limitStreamLength = limitStream.Length; + } + } + } + Console.WriteLine($"fixedGrowingStreamLength:{fixedGrowingStreamLength} limitStreamLength:{limitStreamLength} nextGrowingStreamLength:{nextGrowingStreamLength}"); + Assert.IsTrue(fixedGrowingStreamLength > 0); + Assert.AreEqual(fixedGrowingStreamLength, limitStreamLength); + Assert.IsTrue(limitStreamLength < nextGrowingStreamLength); + } + [Test] [Category("Utility")] public static void ForceStreamRead()