haeaeae/duplicati
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/fix_throttle'

Browse Source
This commit is contained in:
Kenneth Skovhede
2016-10-18 11:03:27 +02:00
parent 857c25d3e8 6e8f371e36
commit f2fddc6327
2 changed files with 137 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Duplicati/Library/UsageReporter/OSInfoHelper.cs
+18 -1
View File
@@ -19,17 +19,28 @@ using System.Linq;
namespace Duplicati.Library.UsageReporter
{
/// <summary>
/// Helper class for extracting operating system info
/// </summary>
internal static class OSInfoHelper
{
/// <summary>
/// Runs a commandline program and returns stdout as a string.
/// </summary>
/// <returns>The stdout value.</returns>
/// <param name="cmd">The command to run.</param>
/// <param name="args">The commandline arguments.</param>
private static string RunProgramAndReadOutput(string cmd, string args)
{
System.Diagnostics.Process pi = null;
try
{
var psi = new System.Diagnostics.ProcessStartInfo(cmd, args);
psi.RedirectStandardOutput = true;
psi.RedirectStandardError = true; // Suppress error messages
psi.UseShellExecute = false;
var pi = System.Diagnostics.Process.Start(psi);
pi = System.Diagnostics.Process.Start(psi);
pi.WaitForExit(5000);
if (pi.HasExited)
return pi.StandardOutput.ReadToEnd().Trim();
@@ -37,6 +48,12 @@ namespace Duplicati.Library.UsageReporter
catch
{
}
finally
{
if (pi != null && !pi.HasExited)
try { pi.Kill(); }
catch { }
}
return null;
}
Duplicati/Library/Utility/ThrottledStream.cs
+119 -152
View File
@@ -33,16 +33,6 @@ namespace Duplicati.Library.Utility
/// </summary>
public class ThrottledStream : OverrideableStream
{
/// <summary>
/// The delegate type for the callback
/// </summary>
public delegate void ThrottledStreamCallback(ThrottledStream sender);
/// <summary>
/// An event that is raised while the stream is active
/// </summary>
public event ThrottledStreamCallback Callback;
/// <summary>
/// The max number of bytes pr. second to write
/// </summary>
@@ -52,41 +42,37 @@ namespace Duplicati.Library.Utility
/// </summary>
private long m_readspeed;
/// <summary>
/// This is a list of the most recent reads. The key is the tick at the time, and the value is the number of bytes.
/// </summary>
List<KeyValuePair<long, long>> m_dataread;
/// <summary>
/// This is a list of the most recent writes. The key is the tick at the time, and the value is the number of bytes.
/// </summary>
List<KeyValuePair<long, long>> m_datawritten;
/// <summary>
/// This is the sum of all bytes in the m_dataread table, summed for fast access.
/// </summary>
private long m_bytesread;
/// <summary>
/// This is the sum of all bytes in the m_datawritten table, summed for fast access.
/// </summary>
private long m_byteswritten;
/// <summary>
/// The time the last read was sampled
/// </summary>
private DateTime m_last_read_sample;
/// <summary>
/// The bytes-read counter for this period
/// </summary>
private long m_current_read_counter;
/// <summary>
/// The current measured read speed
/// </summary>
private double m_current_read_speed;
/// <summary>
/// The number of bytes transfered without raising an event
/// </summary>
private long m_progresscounter = 0;
/// <summary>
/// The time the last read was sampled
/// </summary>
private DateTime m_last_write_sample;
/// <summary>
/// The bytes-written counter for this period
/// </summary>
private long m_current_write_counter;
/// <summary>
/// The current measured read speed
/// </summary>
private double m_current_write_speed;
/// <summary>
/// The number of reads or writes to keep track of
/// <summary>
/// The number of ticks to have passed before a sample is taken
/// </summary>
private const long STATISTICS_SIZE = 500;
/// <summary>
/// The number of ticks to have passed before the throttle begins
/// </summary>
private const long MIN_DURATION = TimeSpan.TicksPerSecond / 4;
/// <summary>
/// The number of sub chunks to perform when throttling
/// </summary>
private const int DELAY_CHUNK_SIZE = 1024;
private const long SAMPLE_PERIOD = TimeSpan.TicksPerSecond / 4;
/// <summary>
/// The number of bytes to process without raising an event
/// </summary>
@@ -104,44 +90,59 @@ namespace Duplicati.Library.Utility
m_readspeed = readspeed;
m_writespeed = writespeed;
if (m_basestream.CanRead && m_readspeed > 0)
m_dataread = new List<KeyValuePair<long, long>>();
if (m_basestream.CanWrite && m_writespeed > 0)
m_datawritten = new List<KeyValuePair<long, long>>();
m_last_read_sample = m_last_write_sample = new DateTime(0);
}
/// <summary>
/// Read the specified buffer, offset and count.
/// </summary>
/// <param name="buffer">The buffer to read from.</param>
/// <param name="offset">The offset into the buffer.</param>
/// <param name="count">The number of bytes to read.</param>
public override int Read(byte[] buffer, int offset, int count)
{
int bytesRead = DelayIfRequired(true, buffer, ref offset, ref count);
if (count != 0)
bytesRead += m_basestream.Read(buffer, offset, count);
var remaining = count;
m_progresscounter += bytesRead;
while (remaining > 0)
{
// To avoid excessive waiting, the delay will wait at most 2 seconds,
// so we split the blocks to limit the number of seconds we can wait
var chunksize = (int)Math.Min(remaining, m_readspeed <= 0 ? remaining : m_readspeed * 2);
DelayIfRequired(ref m_readspeed, chunksize, ref m_last_read_sample, ref m_current_read_counter, ref m_current_read_speed);
if (m_progresscounter > REPORT_DISTANCE_SIZE)
{
m_progresscounter %= REPORT_DISTANCE_SIZE;
if (Callback != null)
Callback(this);
}
var actual = m_basestream.Read(buffer, offset, chunksize);
return bytesRead;
if (actual <= 0)
break;
m_current_read_counter += actual;
remaining -= actual;
}
return count - remaining;
}
/// <summary>
/// Write the specified buffer, offset and count.
/// </summary>
/// <param name="buffer">The buffer to write to.</param>
/// <param name="offset">The offset into the buffer.</param>
/// <param name="count">The number of bytes to write.</param>
public override void Write(byte[] buffer, int offset, int count)
{
m_progresscounter += count;
while (count > 0)
{
// To avoid excessive waiting, the delay will wait at most 2 seconds,
// so we split the blocks to limit the number of seconds we can wait
var chunksize = (int)Math.Min(count, m_writespeed <= 0 ? count : m_writespeed * 2);
DelayIfRequired(ref m_writespeed, chunksize, ref m_last_write_sample, ref m_current_write_counter, ref m_current_write_speed);
m_basestream.Write(buffer, offset, chunksize);
DelayIfRequired(false, buffer, ref offset, ref count);
if (count > 0)
m_basestream.Write(buffer, offset, count);
m_current_write_counter += chunksize;
if (m_progresscounter > REPORT_DISTANCE_SIZE)
{
m_progresscounter %= REPORT_DISTANCE_SIZE;
if (Callback != null)
Callback(this);
}
count -= chunksize;
}
}
/// <summary>
@@ -164,100 +165,66 @@ namespace Duplicati.Library.Utility
set { m_writespeed = value; }
}
/// <summary>
/// Calculates the speed, and inserts appropriate delays
/// </summary>
/// <param name="read">True if the operation is read, false otherwise</param>
/// <param name="buffer">The data buffer</param>
/// <param name="offset">The offset into the buffer</param>
/// <param name="count">The number of bytes to read or write</param>
/// <returns>The number of bytes processed while delaying</returns>
private int DelayIfRequired(bool read, byte[] buffer, ref int offset, ref int count)
{
if (count <= 0)
return 0;
/// <summary>
/// Gets the actual measured read speed.
/// </summary>
public double MeasuredReadSpeed { get { return m_current_read_speed; } }
List<KeyValuePair<long, long>> table = read ? m_dataread : m_datawritten;
int bytesprocessed = 0;
/// <summary>
/// Gets the actual measured write speed.
/// </summary>
public double MeasuredWriteSpeed { get { return m_current_write_speed; } }
if (table != null)
{
long maxspeed = read ? m_readspeed : m_writespeed;
Stream stream = m_basestream;
long bytecount = read ? m_bytesread : m_byteswritten;
/// <summary>
/// Calculates the speed, and inserts appropriate delays
/// </summary>
/// <param name="read">True if the operation is read, false otherwise</param>
/// <param name="buffer">The data buffer</param>
/// <param name="offset">The offset into the buffer</param>
/// <param name="count">The number of bytes to read or write</param>
/// <returns>The number of bytes processed while delaying</returns>
private void DelayIfRequired(ref long limit, int count, ref DateTime last_sample, ref long last_count, ref double current_speed)
{
var now = DateTime.Now;
//Add this access
table.Add(new KeyValuePair<long,long>(DateTime.Now.Ticks, count));
bytecount += count;
if (count <= 0 || limit <= 0)
return;
//Prevent too large tables
while (table.Count > STATISTICS_SIZE)
{
bytecount -= table[0].Value;
table.RemoveAt(0);
}
// If we are just starting, set the timer and counter
if (last_sample.Ticks == 0)
{
last_count = 0;
last_sample = now;
current_speed = limit;
return;
}
if (table.Count != 0 && bytecount != 0)
{
TimeSpan duration = new TimeSpan(table[table.Count - 1].Key - table[0].Key);
// Compute the current duration
var duration = now - last_sample;
//Bail if we are too slow
if (duration.Ticks < MIN_DURATION || bytecount <= 0)
return 0;
// Update speed in intervals
if (duration.Ticks > SAMPLE_PERIOD || last_count > limit)
{
// After a sample period, measure how far ahead we are
var target_delay = TimeSpan.FromSeconds(last_count / (double)limit) - duration;
//TODO: The resolution is too low in "TotalSeconds", so the speed is a little higher
double speed = bytecount / duration.TotalSeconds;
if (speed > maxspeed)
{
//We are too fast, delay the access. Calculating how much wait we need.
double secondsNeeded = (bytecount / (double)maxspeed) - duration.TotalSeconds;
long delayTicks = (long)(secondsNeeded * TimeSpan.TicksPerSecond);
// If we are actually ahead, delay for a little while
if (target_delay.Ticks > 1000)
{
// With large changes, we avoid sleeping for several minutes
// This makes the throttling more resposive when increasing the
// throughput, even with large changes
var ms = (int)Math.Min(target_delay.TotalMilliseconds, 2 * 1000);
System.Threading.Thread.Sleep(ms);
//Calculate the time we should finish, to obey the limit
long targetTime = DateTime.Now.Ticks + delayTicks;
// When we compute how fast this sample was, we include the delay
now = DateTime.Now;
}
if (delayTicks > 0)
{
while (count > DELAY_CHUNK_SIZE && delayTicks > 0)
{
int bytes = read ? stream.Read(buffer, offset, DELAY_CHUNK_SIZE) : DELAY_CHUNK_SIZE;
if (!read)
stream.Write(buffer, offset, DELAY_CHUNK_SIZE);
delayTicks = targetTime - DateTime.Now.Ticks;
long ticksToDelay = (delayTicks / count) * bytes;
if (ticksToDelay > 0)
System.Threading.Thread.Sleep(new TimeSpan(ticksToDelay));
//Reset to include the waited time
delayTicks = targetTime - DateTime.Now.Ticks;
offset += bytes;
count -= bytes;
bytesprocessed += bytes;
if (bytes == 0)
break;
}
if (delayTicks > 0)
System.Threading.Thread.Sleep(new TimeSpan(delayTicks));
//Add a marker, indicating that we already slowed down
table.Add(new KeyValuePair<long, long>(DateTime.Now.Ticks, 0));
}
}
}
if (read)
m_bytesread = bytecount;
else
m_byteswritten = bytecount;
}
return bytesprocessed;
current_speed = last_count / (now - last_sample).TotalSeconds;
last_sample = now;
last_count = 0;
}
}
}
}