From 146b98ae4ce9f99733075f85906f715d79da89b9 Mon Sep 17 00:00:00 2001 From: BlueBlock Date: Tue, 27 Aug 2019 14:09:43 -0400 Subject: [PATCH 1/6] initial fix for issue1400 --- .../Main/Operation/Backup/StreamBlock.cs | 161 +++++++++++++++++- 1 file changed, 154 insertions(+), 7 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs index 722a9437c..c1f9bad04 100644 --- a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs +++ b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs @@ -17,6 +17,7 @@ using System; using CoCoL; using System.IO; +using System.Threading; using System.Threading.Tasks; using Duplicati.Library.Interface; @@ -24,9 +25,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 +41,162 @@ 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 StreamReadLimitLengthWrapper(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); } } + + // 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 + sealed class StreamReadLimitLengthWrapper : Stream + { + readonly Stream m_innerStream; + readonly long m_endPosition; + + public StreamReadLimitLengthWrapper(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() + { + m_innerStream.Flush(); + } + + 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) + { + count = (int)maxCount; + } + + return count; + } + } } From c07411acb777082bd073ec231ba0d73b8d9062fb Mon Sep 17 00:00:00 2001 From: BlueBlock Date: Tue, 27 Aug 2019 14:34:34 -0400 Subject: [PATCH 2/6] resolve codacy issues --- Duplicati/Library/Main/Operation/Backup/StreamBlock.cs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs index c1f9bad04..2dccc85cf 100644 --- a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs +++ b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs @@ -69,7 +69,10 @@ namespace Duplicati.Library.Main.Operation.Backup public StreamReadLimitLengthWrapper(Stream innerStream, long size) { - if (size < 0) throw new ArgumentOutOfRangeException(nameof(size)); + if (size < 0) + { + throw new ArgumentOutOfRangeException(nameof(size)); + } m_innerStream = innerStream ?? throw new ArgumentNullException(nameof(innerStream)); m_endPosition = m_innerStream.Position + size; @@ -193,7 +196,7 @@ namespace Duplicati.Library.Main.Operation.Backup long maxCount = m_endPosition - pos; if (count > maxCount) { - count = (int)maxCount; + return (int)maxCount; } return count; From 42342f30be2489c0e8048691ea55a2fa44c1fcfe Mon Sep 17 00:00:00 2001 From: BlueBlock Date: Tue, 27 Aug 2019 23:48:07 -0400 Subject: [PATCH 3/6] move new stream to Utility --- .../Main/Operation/Backup/StreamBlock.cs | 147 +---------------- .../Utility/Duplicati.Library.Utility.csproj | 1 + .../Library/Utility/ReadLimitLengthStream.cs | 151 ++++++++++++++++++ 3 files changed, 154 insertions(+), 145 deletions(-) create mode 100644 Duplicati/Library/Utility/ReadLimitLengthStream.cs diff --git a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs index 2dccc85cf..5812b9560 100644 --- a/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs +++ b/Duplicati/Library/Main/Operation/Backup/StreamBlock.cs @@ -14,10 +14,9 @@ // 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; using System.Threading.Tasks; using Duplicati.Library.Interface; @@ -43,7 +42,7 @@ namespace Duplicati.Library.Main.Operation.Backup var tcs = new TaskCompletionSource(); // limit the stream length to that found now, a fixed point in time - var limitedStream = new StreamReadLimitLengthWrapper(stream, stream.Length); + var limitedStream = new Library.Utility.ReadLimitLengthStream(stream, stream.Length); var streamBlock = new StreamBlock { @@ -60,146 +59,4 @@ namespace Duplicati.Library.Main.Operation.Backup } } - // 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 - sealed class StreamReadLimitLengthWrapper : Stream - { - readonly Stream m_innerStream; - readonly long m_endPosition; - - public StreamReadLimitLengthWrapper(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() - { - m_innerStream.Flush(); - } - - 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/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..7c27046b5 --- /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() + { + m_innerStream.Flush(); + } + + 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; + } + } +} From 33b63cfc530e1b89fae0631fcc690edac38d2605 Mon Sep 17 00:00:00 2001 From: BlueBlock Date: Thu, 29 Aug 2019 06:26:57 -0400 Subject: [PATCH 4/6] Update ReadLimitLengthStream.cs use empty method for flush --- Duplicati/Library/Utility/ReadLimitLengthStream.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Duplicati/Library/Utility/ReadLimitLengthStream.cs b/Duplicati/Library/Utility/ReadLimitLengthStream.cs index 7c27046b5..7aab7f2d8 100644 --- a/Duplicati/Library/Utility/ReadLimitLengthStream.cs +++ b/Duplicati/Library/Utility/ReadLimitLengthStream.cs @@ -32,7 +32,7 @@ namespace Duplicati.Library.Utility public override void Flush() { - m_innerStream.Flush(); + // NOOP } public override long Length => m_endPosition; From 7ce4639a9e36a67ceee061cafea0950353c256fc Mon Sep 17 00:00:00 2001 From: BlueBlock Date: Thu, 29 Aug 2019 17:23:19 -0400 Subject: [PATCH 5/6] unittest added for ReadLimitLengthStream() --- Duplicati/UnitTest/TestUtils.cs | 26 ++++++++++++++++++++++++++ Duplicati/UnitTest/UtilityTests.cs | 29 +++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+) 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..2648e5816 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,33 @@ 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); + using (FileStream growingStream = SystemIO.IO_OS.FileOpenRead(growingFilename)) + { + 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; + } + } + } + Assert.IsTrue(fixedGrowingStreamLength > 0); + Assert.AreEqual(fixedGrowingStreamLength, limitStreamLength); + Assert.IsTrue(limitStreamLength < nextGrowingStreamLength); + } + [Test] [Category("Utility")] public static void ForceStreamRead() From caba4be76f3f10a8b487971850b098376002c22a Mon Sep 17 00:00:00 2001 From: BlueBlock Date: Fri, 30 Aug 2019 11:03:57 -0400 Subject: [PATCH 6/6] shared file access corrected --- Duplicati/UnitTest/UtilityTests.cs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/Duplicati/UnitTest/UtilityTests.cs b/Duplicati/UnitTest/UtilityTests.cs index 2648e5816..437889f77 100644 --- a/Duplicati/UnitTest/UtilityTests.cs +++ b/Duplicati/UnitTest/UtilityTests.cs @@ -105,7 +105,8 @@ namespace Duplicati.UnitTest using (CancellationTokenSource tokenSource = new CancellationTokenSource()) { Task task = TestUtils.GrowingFile(growingFilename, tokenSource.Token); - using (FileStream growingStream = SystemIO.IO_OS.FileOpenRead(growingFilename)) + Thread.Sleep(100); + using (FileStream growingStream = new FileStream(growingFilename, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) { Thread.Sleep(100); fixedGrowingStreamLength = growingStream.Length; @@ -118,6 +119,7 @@ namespace Duplicati.UnitTest } } } + Console.WriteLine($"fixedGrowingStreamLength:{fixedGrowingStreamLength} limitStreamLength:{limitStreamLength} nextGrowingStreamLength:{nextGrowingStreamLength}"); Assert.IsTrue(fixedGrowingStreamLength > 0); Assert.AreEqual(fixedGrowingStreamLength, limitStreamLength); Assert.IsTrue(limitStreamLength < nextGrowingStreamLength);