Подтвердить что ты не робот

Регулирование полосы пропускания в С#

Я разрабатываю программу, которая постоянно отправляет поток данных в фоновом режиме, и я хочу разрешить пользователю устанавливать ограничение для ограничения загрузки и загрузки.

Я прочитал в маркер токена и просачивающийся ковш alghorhithms, и, похоже, последнее, похоже, соответствует описанию, поскольку это не вопрос максимизации пропускной способности сети, а скорее как максимально ненавязчивый.

Я, однако, немного не уверен в том, как это реализовать. Естественным подходом является расширение абстрактного класса Stream, чтобы упростить расширение существующего трафика, но не потребует ли этого участия дополнительных потоков для отправки данных при одновременном получении (утечка ведра)? Любые намеки на другие реализации, которые делают то же самое, будут оценены.

Кроме того, хотя я могу изменить, сколько данных получает программа, насколько хорошо работает дросселирование полосы пропускания на уровне С#? Будет ли компьютер получать данные и просто сохранять его, эффективно отменяя эффект дросселирования или он будет ждать, пока я попрошу получить больше?

EDIT: меня интересует дросселирование как входящих, так и исходящих данных, где я не контролирую противоположный конец потока.

4b9b3361

ОТВЕТЫ

Ответ 2

Я придумал другую реализацию класса ThrottledStream, упомянутого arul. В моей версии используется WaitHandle и таймер с интервалом 1 с:

public ThrottledStream(Stream parentStream, int maxBytesPerSecond=int.MaxValue) 
{
    MaxBytesPerSecond = maxBytesPerSecond;
    parent = parentStream;
    processed = 0;
    resettimer = new System.Timers.Timer();
    resettimer.Interval = 1000;
    resettimer.Elapsed += resettimer_Elapsed;
    resettimer.Start();         
}

protected void Throttle(int bytes)
{
    try
    {
        processed += bytes;
        if (processed >= maxBytesPerSecond)
            wh.WaitOne();
    }
    catch
    {
    }
}

private void resettimer_Elapsed(object sender, ElapsedEventArgs e)
{
    processed = 0;
    wh.Set();
}

Всякий раз, когда предел пропускной способности превышает поток, он будет спать до следующей следующей секунды. Нет необходимости рассчитывать оптимальную продолжительность сна.

Полная реализация:

public class ThrottledStream : Stream
{
    #region Properties

    private int maxBytesPerSecond;
    /// <summary>
    /// Number of Bytes that are allowed per second
    /// </summary>
    public int MaxBytesPerSecond
    {
        get { return maxBytesPerSecond; }
        set 
        {
            if (value < 1)
                throw new ArgumentException("MaxBytesPerSecond has to be >0");

            maxBytesPerSecond = value; 
        }
    }

    #endregion


    #region Private Members

    private int processed;
    System.Timers.Timer resettimer;
    AutoResetEvent wh = new AutoResetEvent(true);
    private Stream parent;

    #endregion

    /// <summary>
    /// Creates a new Stream with Databandwith cap
    /// </summary>
    /// <param name="parentStream"></param>
    /// <param name="maxBytesPerSecond"></param>
    public ThrottledStream(Stream parentStream, int maxBytesPerSecond=int.MaxValue) 
    {
        MaxBytesPerSecond = maxBytesPerSecond;
        parent = parentStream;
        processed = 0;
        resettimer = new System.Timers.Timer();
        resettimer.Interval = 1000;
        resettimer.Elapsed += resettimer_Elapsed;
        resettimer.Start();         
    }

    protected void Throttle(int bytes)
    {
        try
        {
            processed += bytes;
            if (processed >= maxBytesPerSecond)
                wh.WaitOne();
        }
        catch
        {
        }
    }

    private void resettimer_Elapsed(object sender, ElapsedEventArgs e)
    {
        processed = 0;
        wh.Set();
    }

    #region Stream-Overrides

    public override void Close()
    {
        resettimer.Stop();
        resettimer.Close();
        base.Close();
    }
    protected override void Dispose(bool disposing)
    {
        resettimer.Dispose();
        base.Dispose(disposing);
    }

    public override bool CanRead
    {
        get { return parent.CanRead; }
    }

    public override bool CanSeek
    {
        get { return parent.CanSeek; }
    }

    public override bool CanWrite
    {
        get { return parent.CanWrite; }
    }

    public override void Flush()
    {
        parent.Flush();
    }

    public override long Length
    {
        get { return parent.Length; }
    }

    public override long Position
    {
        get
        {
            return parent.Position;
        }
        set
        {
            parent.Position = value;
        }
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        Throttle(count);
        return parent.Read(buffer, offset, count);
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        return parent.Seek(offset, origin);
    }

    public override void SetLength(long value)
    {
        parent.SetLength(value);
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        Throttle(count);
        parent.Write(buffer, offset, count);
    }

    #endregion


}

Ответ 3

На основе решения @0xDEADBEEF я создал следующее (проверяемое) решение на основе планировщиков Rx:

public class ThrottledStream : Stream
{
    private readonly Stream parent;
    private readonly int maxBytesPerSecond;
    private readonly IScheduler scheduler;
    private readonly IStopwatch stopwatch;

    private long processed;

    public ThrottledStream(Stream parent, int maxBytesPerSecond, IScheduler scheduler)
    {
        this.maxBytesPerSecond = maxBytesPerSecond;
        this.parent = parent;
        this.scheduler = scheduler;
        stopwatch = scheduler.StartStopwatch();
        processed = 0;
    }

    public ThrottledStream(Stream parent, int maxBytesPerSecond)
        : this (parent, maxBytesPerSecond, Scheduler.Immediate)
    {
    }

    protected void Throttle(int bytes)
    {
        processed += bytes;
        var targetTime = TimeSpan.FromSeconds((double)processed / maxBytesPerSecond);
        var actualTime = stopwatch.Elapsed;
        var sleep = targetTime - actualTime;
        if (sleep > TimeSpan.Zero)
        {
            using (var waitHandle = new AutoResetEvent(initialState: false))
            {
                scheduler.Sleep(sleep).GetAwaiter().OnCompleted(() => waitHandle.Set());
                waitHandle.WaitOne();
            }
        }
    }

    public override bool CanRead
    {
        get { return parent.CanRead; }
    }

    public override bool CanSeek
    {
        get { return parent.CanSeek; }
    }

    public override bool CanWrite
    {
        get { return parent.CanWrite; }
    }

    public override void Flush()
    {
        parent.Flush();
    }

    public override long Length
    {
        get { return parent.Length; }
    }

    public override long Position
    {
        get
        {
            return parent.Position;
        }
        set
        {
            parent.Position = value;
        }
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        var read = parent.Read(buffer, offset, count);
        Throttle(read);
        return read;
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        return parent.Seek(offset, origin);
    }

    public override void SetLength(long value)
    {
        parent.SetLength(value);
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        Throttle(count);
        parent.Write(buffer, offset, count);
    }
}

и некоторые тесты, которые занимают несколько миллисекунд:

[TestMethod]
public void ShouldThrottleReading()
{
    var content = Enumerable
        .Range(0, 1024 * 1024)
        .Select(_ => (byte)'a')
        .ToArray();
    var scheduler = new TestScheduler();
    var source = new ThrottledStream(new MemoryStream(content), content.Length / 8, scheduler);
    var target = new MemoryStream();

    var t = source.CopyToAsync(target);

    t.Wait(10).Should().BeFalse();
    scheduler.AdvanceTo(TimeSpan.FromSeconds(4).Ticks);
    t.Wait(10).Should().BeFalse();
    scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks - 1);
    t.Wait(10).Should().BeFalse();
    scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks);
    t.Wait(10).Should().BeTrue();
}

[TestMethod]
public void ShouldThrottleWriting()
{
    var content = Enumerable
        .Range(0, 1024 * 1024)
        .Select(_ => (byte)'a')
        .ToArray();
    var scheduler = new TestScheduler();
    var source = new MemoryStream(content);
    var target = new ThrottledStream(new MemoryStream(), content.Length / 8, scheduler);

    var t = source.CopyToAsync(target);

    t.Wait(10).Should().BeFalse();
    scheduler.AdvanceTo(TimeSpan.FromSeconds(4).Ticks);
    t.Wait(10).Should().BeFalse();
    scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks - 1);
    t.Wait(10).Should().BeFalse();
    scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks);
    t.Wait(10).Should().BeTrue();
}