Bandwidth throttling

Dear P.J. van de Sande,

thank you very much for sharing this, it works for me, but I am wondering why you have written a class that inherits from Stream?
You don't call the base constructor and only the methods you have implemented are passed over to the 'parent' object.
If the user would call some other properties or methods that you dont have forwarded, they would fail.
So in my opinion I would just remove the : Stream and all "override" strings.

Best, copa017

The solution is working very much fine. But sometimes if i set to 20 kbps still in resource monitor or some other tool sometimes its showing 30 or 40 kbps. Although its for a fraction of time but still I want that should not happen . Can you please help on this ?

Hello,
Did you work with the HTTPClient if you could share the code how you did it ?.

Thanks

Please Provide a small tutorial how i can configure it using VS2015 Windows 10 (OS).
Thanks for Your Time .

Hi,
I'm trying to understand the bandwidth usage from the class's example, but i never reach more than the 50% of my 1gb network.

I'm transferring files that the size is between 100Gb and 200gb, and also at the worst situation, transferring from a remote machine to another remote machine.

I use the ThrottledStream(DestinyStream, bps); with zero bps and the buffer-size with many different numbers. I had been trying, between 8bytes, 256bytes, 8kb(8192),80kb(81920) and at the end using a function to get all the available free memory to get the max buffer-size possible.

But this had ended in a 0.1% usage of the network. Dead | X|

This is my function to use the ThrottledStream.

C#

public bool fnGlobFCopyBandwidthWithProgress(string strSourceFileName, string strDesntinyFileName, int BufferSize, long bps, out string strResult)
{
    try
    {
        FileInfo file = new FileInfo(strSourceFileName);

        if (!file.Exists)
        {
            throw new DirectoryNotFoundException("Source file does not exist or could not be found: " + strSourceFileName);
            //return false;
        }

           using (FileStream sourceStream = new FileStream(file.FullName, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
            {

                //Get the total size of the download
                Int64 dataLength = sourceStream.Length;

                //To control the progress for the transfer.
                ByteArgs byteArgs = new ByteArgs();
                byteArgs.downloaded = 0;
                byteArgs.total = file.Length;
                byteArgs.FileName = strDesntinyFileName;
               //-------

                //we need to test for a null as if an event is not consumed we will get an exception
                if (bytesDownloaded != null) bytesDownloaded(byteArgs);

                using (FileStream DestinyStream = new FileStream(strDesntinyFileName, FileMode.Create, FileAccess.Write, FileShare.Write))
                {

                    // Create throttled destination stream.
                    ThrottledStream destinationStream = new ThrottledStream(DestinyStream, bps);

                    byte[] buffer = new byte[BufferSize];
                    int readCount = sourceStream.Read(buffer, 0, BufferSize);

                    while (readCount > 0)
                    {
                        destinationStream.Write(buffer, 0, readCount);
                        readCount = sourceStream.Read(buffer, 0, BufferSize);

                        //To control the progress for the transfer.
                        byteArgs.downloaded += readCount;
                        byteArgs.total = dataLength;
                        if (bytesDownloaded != null) bytesDownloaded(byteArgs);
                        //----------
                    }
                }
            }

    }
    catch (Exception Ex)
    {
        strResult = "Error: InnerException: " + Ex.InnerException + "#Message: " + Ex.Message + "#Source: " + Ex.Source + "#StackTrace: " + Ex.StackTrace;
        return false;
    }

    strResult = "";
    return true;
}

and the call:

C#

        private const int KILOBYTE = 1024;
        private const int MEGABYTE = 1024*1024;
        private const int BufferSize_256b = 256;
        private const int BufferSize_8kb = 8192;
        private const int BufferSize_4kb = 4096;
        private const int BufferSize_80kb = 81920;


int intBuffSize = BufferSize_8kb;

if (bpsRemote == 0)
{
    bpsRemote = ThrottledStream.Infinite;
    GMS_LIB_GLOBAL_COMPUTER objComputer = new GMS_LIB_GLOBAL_COMPUTER();
    //intBuffSize = MEGABYTE;
    //intBuffSize = BufferSize_256b;
    //intBuffSize = BufferSize_8kb;
    //intBuffSize = BufferSize_4kb;
    //intBuffSize = BufferSize_80kb;
    //intBuffSize = (int)objComputer.GetMemoryInMiB("total") * KILOBYTE; //Will crash.
    intBuffSize = (int)objComputer.GetMemoryInMiB("Available") * KILOBYTE;
}

bolFileCopied = objGlobalFiles.fnGlobFCopyBandwidthWithProgress(KeyOrigin.ToString(), strNewPathDestiny.ToString(), intBuffSize, bpsRemote, out strResultado);

I will continue trying with different approaches.

Maybe, this class is not intended for a file transfer over lan network and control the bandwidth... but i spent useful hours and it doesn't matter... Thanks a lot, for ThrottledStream class and the example!

I will learn how to use the windows shadow copy and try to use other approaches to wrap the "FileSystem.CopyTo()" function.

Hi, i've made the following changes:
* use StopWatch instead of Environment.TickCount (because Environment.TickCount failed sometimes on my machine)
* added property BlockSize (default value is 512)
Packets larger than BlockSize are splitted to make the throttling much smoother
* override ReadByte() and WriteByte()

C#

/// <summary>
/// Class for streaming data with throttling support.
/// </summary>
/// <remarks>http://www.codeproject.com/Articles/18243/Bandwidth-throttling</remarks>
public class ThrottledStream : Stream
{
    /// <summary>
    /// A constant used to specify an infinite number of bytes that can be transferred per second.
    /// </summary>
    public const long Infinite = 0;

    /// <summary>
    /// The base stream.
    /// </summary>
    private Stream _baseStream;

    /// <summary>
    /// The maximum bytes per second that can be transferred through the base stream.
    /// </summary>
    private long _maximumBytesPerSecond;

    /// <summary>
    /// The number of bytes that has been transferred since the last throttle.
    /// </summary>
    private long _byteCount;

    protected Stopwatch _stopWatch;

    /// <summary>
    /// Gets or sets the maximum bytes per second that can be transferred through the base stream.
    /// </summary>
    /// <value>The maximum bytes per second.</value>
    /// <exception cref="T:System.ArgumentOutOfRangeException">Value is negative. </exception>
    public long MaximumBytesPerSecond
    {
        get
        {
            return _maximumBytesPerSecond;
        }
        set
        {
            if (value < 0)
            {
                throw new ArgumentOutOfRangeException("MaximumBytesPerSecond",
                    value, "The maximum number of bytes per second can't be negative.");
            }
            if (_maximumBytesPerSecond != value)
            {
                _maximumBytesPerSecond = value;
                Reset();
            }
        }
    }

    public int BlockSize { get; set; }

    /// <summary>
    /// Gets a value indicating whether the current stream supports reading.
    /// </summary>
    /// <returns>true if the stream supports reading; otherwise, false.</returns>
    public override bool CanRead
    {
        get
        {
            return _baseStream.CanRead;
        }
    }

    /// <summary>
    /// Gets a value indicating whether the current stream supports seeking.
    /// </summary>
    /// <value></value>
    /// <returns>true if the stream supports seeking; otherwise, false.</returns>
    public override bool CanSeek
    {
        get
        {
            return _baseStream.CanSeek;
        }
    }

    /// <summary>
    /// Gets a value indicating whether the current stream supports writing.
    /// </summary>
    /// <value></value>
    /// <returns>true if the stream supports writing; otherwise, false.</returns>
    public override bool CanWrite
    {
        get
        {
            return _baseStream.CanWrite;
        }
    }

    /// <summary>
    /// Gets the length in bytes of the stream.
    /// </summary>
    /// <value></value>
    /// <returns>A long value representing the length of the stream in bytes.</returns>
    /// <exception cref="T:System.NotSupportedException">The base stream does not support seeking. </exception>
    /// <exception cref="T:System.ObjectDisposedException">Methods were called after the stream was closed. </exception>
    public override long Length
    {
        get
        {
            return _baseStream.Length;
        }
    }

    /// <summary>
    /// Gets or sets the position within the current stream.
    /// </summary>
    /// <value></value>
    /// <returns>The current position within the stream.</returns>
    /// <exception cref="T:System.IO.IOException">An I/O error occurs. </exception>
    /// <exception cref="T:System.NotSupportedException">The base stream does not support seeking. </exception>
    /// <exception cref="T:System.ObjectDisposedException">Methods were called after the stream was closed. </exception>
    public override long Position
    {
        get
        {
            return _baseStream.Position;
        }
        set
        {
            _baseStream.Position = value;
        }
    }

    /// <summary>
    /// Initializes a new instance of the <see cref="T:ThrottledStream"/> class with an
    /// infinite amount of bytes that can be processed.
    /// </summary>
    /// <param name="baseStream">The base stream.</param>
    public ThrottledStream(Stream baseStream)
        : this(baseStream, ThrottledStream.Infinite)
    {
    }

    /// <summary>
    /// Initializes a new instance of the <see cref="T:ThrottledStream"/> class.
    /// </summary>
    /// <param name="baseStream">The base stream.</param>
    /// <param name="maximumBytesPerSecond">The maximum bytes per second that can be transferred through the base stream.</param>
    /// <exception cref="ArgumentNullException">Thrown when <see cref="baseStream"/> is a null reference.</exception>
    /// <exception cref="ArgumentOutOfRangeException">Thrown when <see cref="maximumBytesPerSecond"/> is a negative value.</exception>
    public ThrottledStream(Stream baseStream, long maximumBytesPerSecond)
    {
        if (baseStream == null)
        {
            throw new ArgumentNullException("baseStream");
        }

        _baseStream = baseStream;
        _stopWatch = Stopwatch.StartNew();
        MaximumBytesPerSecond = maximumBytesPerSecond;
        BlockSize = 512;
        _byteCount = 0;
    }

    /// <summary>
    /// Clears all buffers for this stream and causes any buffered data to be written to the underlying device.
    /// </summary>
    /// <exception cref="T:System.IO.IOException">An I/O error occurs.</exception>
    public override void Flush()
    {
        _baseStream.Flush();
    }

    /// <summary>
    /// Reads a sequence of bytes from the current stream and advances the position within the stream by the number of bytes read.
    /// </summary>
    /// <param name="buffer">An array of bytes. When this method returns, the buffer contains the specified byte array with the values between offset and (offset + count - 1) replaced by the bytes read from the current source.</param>
    /// <param name="offset">The zero-based byte offset in buffer at which to begin storing the data read from the current stream.</param>
    /// <param name="count">The maximum number of bytes to be read from the current stream.</param>
    /// <returns>
    /// The total number of bytes read into the buffer. This can be less than the number of bytes requested if that many bytes are not currently available, or zero (0) if the end of the stream has been reached.
    /// </returns>
    /// <exception cref="T:System.ArgumentException">The sum of offset and count is larger than the buffer length. </exception>
    /// <exception cref="T:System.ObjectDisposedException">Methods were called after the stream was closed. </exception>
    /// <exception cref="T:System.NotSupportedException">The base stream does not support reading. </exception>
    /// <exception cref="T:System.ArgumentNullException">buffer is null. </exception>
    /// <exception cref="T:System.IO.IOException">An I/O error occurs. </exception>
    /// <exception cref="T:System.ArgumentOutOfRangeException">offset or count is negative. </exception>
    public override int Read(byte[] buffer, int offset, int count)
    {
        int total = 0;
        while (count > BlockSize)
        {
            Throttle(BlockSize);
            int rb = _baseStream.Read(buffer, offset, BlockSize);
            total += rb;
            if (rb < BlockSize)
                return total;
            offset += BlockSize;
            count -= BlockSize;
        }
        Throttle(count);
        return total + _baseStream.Read(buffer, offset, count);
    }

    /// <summary>
    /// Sets the position within the current stream.
    /// </summary>
    /// <param name="offset">A byte offset relative to the origin parameter.</param>
    /// <param name="origin">A value of type <see cref="T:System.IO.SeekOrigin"></see> indicating the reference point used to obtain the new position.</param>
    /// <returns>
    /// The new position within the current stream.
    /// </returns>
    /// <exception cref="T:System.IO.IOException">An I/O error occurs. </exception>
    /// <exception cref="T:System.NotSupportedException">The base stream does not support seeking, such as if the stream is constructed from a pipe or console output. </exception>
    /// <exception cref="T:System.ObjectDisposedException">Methods were called after the stream was closed. </exception>
    public override long Seek(long offset, SeekOrigin origin)
    {
        return _baseStream.Seek(offset, origin);
    }

    /// <summary>
    /// Sets the length of the current stream.
    /// </summary>
    /// <param name="value">The desired length of the current stream in bytes.</param>
    /// <exception cref="T:System.NotSupportedException">The base stream does not support both writing and seeking, such as if the stream is constructed from a pipe or console output. </exception>
    /// <exception cref="T:System.IO.IOException">An I/O error occurs. </exception>
    /// <exception cref="T:System.ObjectDisposedException">Methods were called after the stream was closed. </exception>
    public override void SetLength(long value)
    {
        _baseStream.SetLength(value);
    }

    /// <summary>
    /// Writes a sequence of bytes to the current stream and advances the current position within this stream by the number of bytes written.
    /// </summary>
    /// <param name="buffer">An array of bytes. This method copies count bytes from buffer to the current stream.</param>
    /// <param name="offset">The zero-based byte offset in buffer at which to begin copying bytes to the current stream.</param>
    /// <param name="count">The number of bytes to be written to the current stream.</param>
    /// <exception cref="T:System.IO.IOException">An I/O error occurs. </exception>
    /// <exception cref="T:System.NotSupportedException">The base stream does not support writing. </exception>
    /// <exception cref="T:System.ObjectDisposedException">Methods were called after the stream was closed. </exception>
    /// <exception cref="T:System.ArgumentNullException">buffer is null. </exception>
    /// <exception cref="T:System.ArgumentException">The sum of offset and count is greater than the buffer length. </exception>
    /// <exception cref="T:System.ArgumentOutOfRangeException">offset or count is negative. </exception>
    public override void Write(byte[] buffer, int offset, int count)
    {
        while (count > BlockSize)
        {
            Throttle(BlockSize);
            _baseStream.Write(buffer, offset, BlockSize);
            offset += BlockSize;
            count -= BlockSize;
        }
        Throttle(count);
        _baseStream.Write(buffer, offset, count);
    }

    public override int ReadByte()
    {
        Throttle(1);
        return _baseStream.ReadByte();
    }

    public override void WriteByte(byte value)
    {
        Throttle(1);
        _baseStream.WriteByte(value);
    }

    /// <summary>
    /// Returns a <see cref="T:System.String"></see> that represents the current <see cref="T:System.Object"></see>.
    /// </summary>
    /// <returns>
    /// A <see cref="T:System.String"></see> that represents the current <see cref="T:System.Object"></see>.
    /// </returns>
    public override string ToString()
    {
        return _baseStream.ToString();
    }

    /// <summary>
    /// Throttles for the specified buffer size in bytes.
    /// </summary>
    /// <param name="bufferSizeInBytes">The buffer size in bytes.</param>
    protected void Throttle(int bufferSizeInBytes)
    {
        // Make sure the buffer isn't empty.
        if (_maximumBytesPerSecond <= 0 || bufferSizeInBytes <= 0)
        {
            return;
        }

        _byteCount += bufferSizeInBytes;
        if (_byteCount < 16)
            return;

        long elapsedMilliseconds = _stopWatch.ElapsedMilliseconds;

        if (elapsedMilliseconds > 0)
        {
            // Calculate the current bps.
            long bps = _byteCount * 1000L / elapsedMilliseconds;

            // If the bps are more then the maximum bps, try to throttle.
            if (bps > _maximumBytesPerSecond)
            {
                // Calculate the time to sleep.
                long wakeElapsed = _byteCount * 1000L / _maximumBytesPerSecond;
                int toSleep = (int)(wakeElapsed - elapsedMilliseconds);

                if (toSleep > 1)
                {
                    try
                    {
                        // The time to sleep is more then a millisecond, so sleep.
                        Thread.Sleep(toSleep);
                    }
                    catch (ThreadAbortException)
                    {
                        // Eatup ThreadAbortException.
                    }

                    // A sleep has been done, reset.
                    Reset();
                }
            }
        }
    }

    /// <summary>
    /// Will reset the bytecount to 0 and reset the start time to the current time.
    /// </summary>
    protected void Reset()
    {
        long difference = _stopWatch.ElapsedMilliseconds;

        // Only reset counters when a known history is available of more then 1 second.
        if (difference > 1000)
        {
            _byteCount = 0;
            _stopWatch.Restart();
        }
    }

}

Thanks! Work's great!

Thanks! Your update is a better solution. Environment.TickCount is not accurate and not very suitable for speed control. StopWatch should be used in place of TickCount.

How Throtle class can be used with System.Net.HttpClient with Task async approach?
For example I have:

C#

public async Task Demo(Uri requestUri, CancellationToken cancellationToken)
        {
            using (var httpClient = new HttpClient())
            {
                using (var request = new HttpRequestMessage(HttpMethod.Get, requestUri))
                {
                    using (var responseMessage = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false))
                    {
                        const int bufferSize = 1024; // 1 KB buffer
                        using (var httpStream = await responseMessage.Content.ReadAsStreamAsync().ConfigureAwait(false))
                        {
                            using (var filestream = new FileStream("filename.bin", FileMode.Create, FileAccess.Write, FileShare.None, bufferSize, true))
                            {
                                await httpStream.CopyToAsync(filestream);
                                await Task.Yield();
                            }
                        }
                    }
                }
            }
        }

How to make download of:

C#

responseMessage.Content.ReadAsStreamAsync()

be limited for 512 kbps for example?

Still useful after all these years Smile | :)

how can i create Socket for mySocket, i used this:
Socket s = new Socket(AddressFamily.InterNetwork,
SocketType.Stream, ProtocolType.Tcp);
but it tells me:
The operation is not allowed on non-connected sockets.

the code works perfect when download , but when upload it works strange , for example if i upload a 3 mb file at 512kB limit , it must last 6 seconds and it does , but the first secnods he uses almost all the bandwith , then it just sleep. Its that ok?

here is my code:

C#

public void EscribeDataOriginal(Stream input,Stream output,long bps)
   {
       using (input)
       {
           // Create throttled destination stream.
           ThrottledStream destinationStream = new ThrottledStream(output, bps);

           byte[] buffer = new byte[BufferSize];
           int readCount = input.Read(buffer, 0, BufferSize);

           while (readCount > 0)
           {
               destinationStream.Write(buffer, 0, readCount);
               readCount = input.Read(buffer, 0, BufferSize);
           }
           output.Close();
           destinationStream.Close();

       }

can any one tell me how can i use this class "ThrottledStream"
to control bandwidth on all internet connection
on my computer
and i'm sorry for bad English

If I have a list of connection TCP/UDP [see Getting active TCP/UDP connections on a box[^]], how do can limit each connection of that list using this code? I don't have idea to "intercep and limit" each connection of the list

any ideas?. Thanks!

This class is not perfect.
You must delegate all overrides from Stream since some streams override these methods:

CanTimeout
ReadTimeout
WriteTimeout

BeginRead
EndRead
BeginWrite
EndWrite
ReadByte
Close
WriteByte

very nice piece of code, thank you man

What with asp.net worker threadpool, using thread.sleep can be low performance, especially i you have a file hosting site like rapidshare with premium and normal users, would it be better to make this asynchronous ? so don't make sleep instead code instructs site to make another end (in begin.. end async scenario, where end is responsible for transferring another data) after coupe seconds hmm ?

This is helping me in a C++ server application where I want to implement QoS.

Easy Profiler : a compile-time profiler for C++
www.potatosoftware.com

I have integrated your solution in one of my multi threaded network applications and it works fine.
Your article was not extensive but most useful, nontheless.

Thank you. Thumbs Up | :thumbsup:

Ooops, although it seems to work at first, I now have to admit that I couldn't make it work in a multi threaded application of mine. The problem is, I cannot control the overall bandwidth of, say, 20 downloading threads. The maximum bandwidth is divided by the number of threads, of course, but it simply does not work. I've tested it with numerous bandwidth meters (DU meter etc) but no luck. I believe, that something is wrong with the calculation of the sleep time... Confused | :confused:

Any ideas?

Thank you.

Yes... i'm getting same problem when i reach about 20 downloadings threads, and also dividing bandwidth by the number of threads D'Oh! | :doh:

I have no idea of what cause it... maybe priority of threads change, or maybe code should calculate the sleep time relating the total threads...

Ideas?

J.P.?? any solution?

By the way... congrats and thanks for your usefull code!!

Using your code, can I adjust the bit rate on the fly based on some external indicator? For example, say I have a message broker that will deliver my application messages and provide feedback based on the current state of the connection. If some other application that is monitoring the connection publishes a message that I need to slow down, can I adjust the bit rate and move on until it stops sending me the message?

Thanks!

Yes, I am wondering the same thing. I have a windows forms application on hundreds of office computers that sync up to a single server. They all do their sync late at night so the office's networks will be pretty quiet. Obviously if I have each installation use maximum bandwidth on the office network, then it'll crush the destination server.

I am wondering if there is some way I can use your class in conjunction with a server throttling message that way the server can tell the client your using too much bandwidth on the server.

I thought about using a "ping" command or some external bandwidth meter/latency test to check the server's responsiveness but not sure about how to go about it. I don't think "ping" is going to work because of firewall/routers will not respond to ICMP messages.

Just wondering if anyone had any ideas how to set the threshold correctly.

In the source code in the throttling class you need to change _maximumBytesPerSecond from private to public

using the sample code in the TestWebsite you would add these before the using statement:

double duration = media_duration; //seconds

FileInfo fs = new FileInfo(source);
int maxLen = (int)fs.Length;

long byterate = (long)Math.Round(fs.Length / duration);

//set throttle init so that fast buffering occurs on the player
int init_throt = (maxLen - ((int)byterate * 15));
bool set_throt = true;
long bps = byterate * 10;

--------------------------------

then inside the using statement the while statement should be something like this:

while (readCount > 0)
{
destinationStream.Write(buffer, 0, readCount);
readCount = sourceStream.Read(buffer, 0, BufferSize);
maxLen -= readCount;

if (set_throt)
{
if (maxLen < init_throt)
{ set_throt = false; destinationStream._maximumBytesPerSecond = byterate + 5; }
}
}

--------------------------------

i'm using this to stream audio/video straight from the source file to a wmp plugin embedded in a browser

there are some variable that are changed from the sample or that you'll need to define or change, like source and/or duration and so on...

modified on Friday, February 5, 2010 10:23 AM

What happens if you didn't use the ThrottledStream Write or Read functions immediately after creating the object. Seems to me that the time would increase and allow for data transfer rates above what you are hoping for. Maybe a fix would be to reset the _byteCount and _start variables after the elapsedMilliseconds is greater than X seconds? Let me know what you think.
Thanks,
Nick

Bandwidth throttling

Introduction

Using the code

Points of interest

License

Comments and Discussions