Merge pull request #28 from logiclrd/JDG_ImproveSpeedCalculation

Improve speed calculation

src/Client/Client/ProgressStreamContent.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 
 using Bytewizer.Backblaze.Models;
+using Bytewizer.Backblaze.Utility;
 
 namespace Bytewizer.Backblaze.Client
 {
@@ -67,15 +68,17 @@ protected override Task SerializeToStreamAsync(Stream stream, TransportContext c
             return Task.Run(() =>
             {
                 var totalTime = new System.Diagnostics.Stopwatch();
-                var singleTime = new System.Diagnostics.Stopwatch();
                 totalTime.Start();
-                singleTime.Start();
 
                 var buffer = new byte[_bufferSize];
                 long streamLength = _content.CanSeek ? _content.Length : 0;
                 long size = _expectedContentLength > 0 ? _expectedContentLength : streamLength;
                 long uploaded = 0;
 
+                var speedCalculator = new SpeedCalculator();
+
+                speedCalculator.AddSample(0);
+
                 while (true)
                 {
                     var length = _content.Read(buffer, 0, buffer.Length);
@@ -84,10 +87,9 @@ protected override Task SerializeToStreamAsync(Stream stream, TransportContext c
 
                     stream.Write(buffer, 0, length);
 
-                    long singleElapsed = Math.Max(1, singleTime.ElapsedTicks);
-                    singleTime.Restart();
+                    speedCalculator.AddSample(uploaded);
 
-                    _progressReport?.Report(new CopyProgress(totalTime.Elapsed, length * TimeSpan.TicksPerSecond / singleElapsed, uploaded, size));
+                    _progressReport?.Report(new CopyProgress(totalTime.Elapsed, speedCalculator.CalculateBytesPerSecond(), uploaded, size));
                 }
             });
         }

src/Client/Client/Utility/SpeedCalculator.cs

@@ -0,0 +1,82 @@
+using System;
+using System.Collections.Generic;
+
+namespace Bytewizer.Backblaze.Utility
+{
+    /// <summary>
+    /// Calculates transfer speed as a rolling average. Every time the position changes,
+    /// the consumer notifies us and a timestamped sample is logged.
+    /// </summary>
+    public class SpeedCalculator
+    {
+        List<Sample> _samples = new List<Sample>();
+
+        struct Sample
+        {
+            public long Position;
+            public DateTime DateTimeUTC;
+        }
+
+        /// <summary>
+        /// The length in seconds of the window across which speed is averaged.
+        /// </summary>
+        public const int WindowSeconds = 10;
+
+        /// <summary>
+        /// Adds a position sample to the set. It is automatically timestamped. Samples
+        /// should be monotonically increasing. If, for whatever reason, they are not,
+        /// previously-added samples that are later in the file are discarded so that
+        /// the set remains strictly increasing.
+        /// </summary>
+        /// <param name="position">The updated position of the operation.</param>
+        public void AddSample(long position)
+        {
+            var sample = new Sample();
+
+            sample.Position = position;
+            sample.DateTimeUTC = DateTime.UtcNow;
+
+            // If we have walked backward for whatever reason, discard any samples past this
+            // point so that we maintain the invariant of the sample set increasing position
+            // monotonically.
+            while ((_samples.Count > 0) && (_samples[_samples.Count - 1].Position > position))
+                _samples.RemoveAt(_samples.Count - 1);
+
+            _samples.Add(sample);
+        }
+
+        /// <summary>
+        /// Calculates the current speed based on samples previously added by calls to
+        /// <see cref="AddSample" />. The value of this function will change over time,
+        /// even with no changes to the state of the <see cref="SpeedCalculator" />
+        /// instance, because the value is relative to the current date/time, and the
+        /// samples with which the calculation is being made are timestamped.
+        /// </summary>
+        /// <returns>The average number of bytes per second being processed.</returns>
+        public long CalculateBytesPerSecond()
+        {
+            var cutoff = DateTime.UtcNow.AddSeconds(-WindowSeconds);
+
+            // Discard any samples that are outside of the averaging window. We will never
+            // need them again.
+            while ((_samples.Count > 0) && (_samples[0].DateTimeUTC < cutoff))
+                _samples.RemoveAt(0);
+
+            if (_samples.Count < 2)
+                return 0;
+
+            var firstSample = _samples[0];
+            var lastSample = _samples[_samples.Count - 1];
+
+            long bytes = lastSample.Position - firstSample.Position;
+            double seconds = (lastSample.DateTimeUTC - firstSample.DateTimeUTC).TotalSeconds;
+
+            // If we don't have a meaningful span of time, clamp it. The number wouldn't
+            // be terribly meaningful anyway.
+            if (seconds < 0.01)
+                seconds = 0.01;
+
+            return (long)Math.Round(bytes / seconds);
+        }
+    }
+}

src/Client/Extensions/StreamExtensions.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 
 using Bytewizer.Backblaze.Models;
+using Bytewizer.Backblaze.Utility;
 
 namespace Bytewizer.Backblaze.Extensions
 {
@@ -37,17 +38,18 @@ public static async Task CopyToAsync(this Stream source, Stream destination, int
             int bytesRead;
 
             var totalTime = new System.Diagnostics.Stopwatch();
-            var singleTime = new System.Diagnostics.Stopwatch();
             totalTime.Start();
-            singleTime.Start();
+
+            var speedCalculator = new SpeedCalculator();
+
+            speedCalculator.AddSample(0);
 
             while ((bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancel).ConfigureAwait(false)) != 0)
             {
                 await destination.WriteAsync(buffer, 0, bytesRead, cancel).ConfigureAwait(false);
                 totalBytesRead += bytesRead;
-                long singleTicks = Math.Max(1, singleTime.ElapsedTicks);
-                progressReport?.Report(new CopyProgress(totalTime.Elapsed, bytesRead * TimeSpan.TicksPerSecond / singleTicks, totalBytesRead, expectedTotalBytes));
-                singleTime.Restart();
+                speedCalculator.AddSample(totalBytesRead);
+                progressReport?.Report(new CopyProgress(totalTime.Elapsed, speedCalculator.CalculateBytesPerSecond(), totalBytesRead, expectedTotalBytes));
 
                 if (cancel.IsCancellationRequested) { break; }
             }