fc::detail::rate_limiting_group_impl Class Reference

Inheritance diagram for fc::detail::rate_limiting_group_impl:

Collaboration diagram for fc::detail::rate_limiting_group_impl:

Public Types
typedef std::list< rate_limited_operation * >	rate_limited_operation_list

Public Member Functions
	rate_limiting_group_impl (uint32_t upload_bytes_per_second, uint32_t download_bytes_per_second, uint32_t burstiness_in_seconds=1)
	~rate_limiting_group_impl ()
virtual size_t	readsome (boost::asio::ip::tcp::socket &socket, char *buffer, size_t length) override
virtual size_t	readsome (boost::asio::ip::tcp::socket &socket, const std::shared_ptr< char > &buffer, size_t length, size_t offset) override
template<typename BufferType >
size_t	readsome_impl (boost::asio::ip::tcp::socket &socket, const BufferType &buffer, size_t length, size_t offset)
virtual size_t	writesome (boost::asio::ip::tcp::socket &socket, const char *buffer, size_t length) override
virtual size_t	writesome (boost::asio::ip::tcp::socket &socket, const std::shared_ptr< const char > &buffer, size_t length, size_t offset) override
template<typename BufferType >
size_t	writesome_impl (boost::asio::ip::tcp::socket &socket, const BufferType &buffer, size_t length, size_t offset)
void	process_pending_reads ()
void	process_pending_writes ()
void	process_pending_operations (time_point &last_iteration_start_time, uint32_t &limit_bytes_per_second, rate_limited_operation_list &operations_in_progress, rate_limited_operation_list &operations_for_next_iteration, uint32_t &tokens, uint32_t &unused_tokens)

Public Attributes
uint32_t	_upload_bytes_per_second
uint32_t	_download_bytes_per_second
uint32_t	_burstiness_in_seconds
microseconds	_granularity
uint32_t	_read_tokens
uint32_t	_unused_read_tokens
uint32_t	_write_tokens
uint32_t	_unused_write_tokens
rate_limited_operation_list	_read_operations_in_progress
rate_limited_operation_list	_read_operations_for_next_iteration
rate_limited_operation_list	_write_operations_in_progress
rate_limited_operation_list	_write_operations_for_next_iteration
time_point	_last_read_iteration_time
time_point	_last_write_iteration_time
future< void >	_process_pending_reads_loop_complete
promise< void >::ptr	_new_read_operation_available_promise
future< void >	_process_pending_writes_loop_complete
promise< void >::ptr	_new_write_operation_available_promise
average_rate_meter	_actual_upload_rate
average_rate_meter	_actual_download_rate

Detailed Description

Definition at line 191 of file rate_limiting.cpp.

Member Typedef Documentation

rate_limited_operation_list

std::list<rate_limited_operation*> fc::detail::rate_limiting_group_impl::rate_limited_operation_list

Definition at line 204 of file rate_limiting.cpp.

Constructor & Destructor Documentation

rate_limiting_group_impl()

fc::detail::rate_limiting_group_impl::rate_limiting_group_impl	(	uint32_t	upload_bytes_per_second,
		uint32_t	download_bytes_per_second,
		uint32_t	burstiness_in_seconds = 1 )

Definition at line 244 of file rate_limiting.cpp.

                                                                                       :
      _upload_bytes_per_second(upload_bytes_per_second),
      _download_bytes_per_second(download_bytes_per_second),
      _burstiness_in_seconds(burstiness_in_seconds),
      _granularity(milliseconds(50)),
      _read_tokens(_download_bytes_per_second),
      _unused_read_tokens(0),
      _write_tokens(_upload_bytes_per_second),
      _unused_write_tokens(0)
    {
    }

~rate_limiting_group_impl()

fc::detail::rate_limiting_group_impl::~rate_limiting_group_impl

(

)

Definition at line 257 of file rate_limiting.cpp.

    {
      try
      {
        _process_pending_reads_loop_complete.cancel_and_wait();
      }
      catch (...)
      {
      }
      try
      {
        _process_pending_writes_loop_complete.cancel_and_wait();
      }
      catch (...)
      {
      }
    }

Member Function Documentation

process_pending_operations()

void fc::detail::rate_limiting_group_impl::process_pending_operations	(	time_point &	last_iteration_start_time,
		uint32_t &	limit_bytes_per_second,
		rate_limited_operation_list &	operations_in_progress,
		rate_limited_operation_list &	operations_for_next_iteration,
		uint32_t &	tokens,
		uint32_t &	unused_tokens )

Definition at line 409 of file rate_limiting.cpp.

    {
      // lock here for multithreaded
      std::copy(operations_for_next_iteration.begin(),
                operations_for_next_iteration.end(),
                std::back_inserter(operations_in_progress));
      operations_for_next_iteration.clear();
 
      // find out how much time since our last read/write
      time_point this_iteration_start_time = time_point::now();
      if (limit_bytes_per_second) // the we are limiting up/download speed
      {
        microseconds time_since_last_iteration = this_iteration_start_time - last_iteration_start_time;
        if (time_since_last_iteration > seconds(1))
          time_since_last_iteration = seconds(1);
        else if (time_since_last_iteration < microseconds(0))
          time_since_last_iteration = microseconds(0);
 
        tokens += (uint32_t)((limit_bytes_per_second * time_since_last_iteration.count()) / 1000000);
        tokens += unused_tokens;
        unused_tokens = 0;
        tokens = std::min(tokens, limit_bytes_per_second * _burstiness_in_seconds);
 
        if (tokens)
        {
          // sort the pending reads/writes in order of the number of bytes they need to write, smallest first
          std::vector<rate_limited_operation*> operations_sorted_by_length;
          operations_sorted_by_length.reserve(operations_in_progress.size());
          for (rate_limited_operation* operation_data : operations_in_progress)
            operations_sorted_by_length.push_back(operation_data);
          std::sort(operations_sorted_by_length.begin(), operations_sorted_by_length.end(), is_operation_shorter());
 
          // figure out how many bytes each reader/writer is allowed to read/write
          uint32_t bytes_remaining_to_allocate = tokens;
          while (!operations_sorted_by_length.empty())
          {
            uint32_t bytes_permitted_for_this_operation = bytes_remaining_to_allocate / operations_sorted_by_length.size();
            uint32_t bytes_allocated_for_this_operation = std::min<size_t>(operations_sorted_by_length.back()->length, bytes_permitted_for_this_operation);
            operations_sorted_by_length.back()->permitted_length = bytes_allocated_for_this_operation;
            bytes_remaining_to_allocate -= bytes_allocated_for_this_operation;
            operations_sorted_by_length.pop_back();
          }
          tokens = bytes_remaining_to_allocate;
 
          // kick off the reads/writes in first-come order
          for (auto iter = operations_in_progress.begin(); iter != operations_in_progress.end();)
          {
            if ((*iter)->permitted_length > 0)
            {
              rate_limited_operation* operation_to_perform = *iter;
              iter = operations_in_progress.erase(iter);
              operation_to_perform->perform_operation();
            }
            else
              ++iter;
          }
        }
      }
      else // down/upload speed is unlimited
      {
        // we shouldn't end up here often.  If the rate is unlimited, we should just execute
        // the operation immediately without being queued up.  This should only be hit if
        // we change from a limited rate to unlimited
        for (auto iter = operations_in_progress.begin();
             iter != operations_in_progress.end();)
        {
          rate_limited_operation* operation_to_perform = *iter;
          iter = operations_in_progress.erase(iter);
          operation_to_perform->permitted_length = operation_to_perform->length;
          operation_to_perform->perform_operation();
        }
      }
      last_iteration_start_time = this_iteration_start_time;
    }

Here is the call graph for this function:

Here is the caller graph for this function:

process_pending_reads()

void fc::detail::rate_limiting_group_impl::process_pending_reads

(

)

Definition at line 367 of file rate_limiting.cpp.

    {
      for (;;)
      {
        process_pending_operations(_last_read_iteration_time, _download_bytes_per_second,
                                   _read_operations_in_progress, _read_operations_for_next_iteration, _read_tokens, _unused_read_tokens);
 
        _new_read_operation_available_promise = new promise<void>("rate_limiting_group_impl::process_pending_reads");
        try
        {
          if (_read_operations_in_progress.empty())
            _new_read_operation_available_promise->wait();
          else
            _new_read_operation_available_promise->wait(_granularity);
        }
        catch (const timeout_exception&)
        {
        }
        _new_read_operation_available_promise.reset();
      }
    }

Here is the call graph for this function:

Here is the caller graph for this function:

process_pending_writes()

void fc::detail::rate_limiting_group_impl::process_pending_writes

(

)

Definition at line 388 of file rate_limiting.cpp.

    {
      for (;;)
      {
        process_pending_operations(_last_write_iteration_time, _upload_bytes_per_second,
                                   _write_operations_in_progress, _write_operations_for_next_iteration, _write_tokens, _unused_write_tokens);
 
        _new_write_operation_available_promise = new promise<void>("rate_limiting_group_impl::process_pending_writes");
        try
        {
          if (_write_operations_in_progress.empty())
            _new_write_operation_available_promise->wait();
          else
            _new_write_operation_available_promise->wait(_granularity);
        }
        catch (const timeout_exception&)
        {
        }
        _new_write_operation_available_promise.reset();
      }
    }

Here is the call graph for this function:

Here is the caller graph for this function:

readsome() [1/2]

size_t fc::detail::rate_limiting_group_impl::readsome ( boost::asio::ip::tcp::socket & socket, char * buffer, size_t length )

overridevirtual

Definition at line 280 of file rate_limiting.cpp.

    {
      return readsome_impl(socket, buffer, length, 0);
    }

Here is the call graph for this function:

readsome() [2/2]

size_t fc::detail::rate_limiting_group_impl::readsome ( boost::asio::ip::tcp::socket & socket, const std::shared_ptr< char > & buffer, size_t length, size_t offset )

overridevirtual

Definition at line 275 of file rate_limiting.cpp.

    {
      return readsome_impl(socket, buffer, length, offset);
    }

Here is the call graph for this function:

readsome_impl()

template<typename BufferType >

size_t fc::detail::rate_limiting_group_impl::readsome_impl	(	boost::asio::ip::tcp::socket &	socket,
		const BufferType &	buffer,
		size_t	length,
		size_t	offset )

Definition at line 286 of file rate_limiting.cpp.

    {
      size_t bytes_read;
      if (_download_bytes_per_second)
      {
        promise<size_t>::ptr completion_promise(new promise<size_t>("rate_limiting_group_impl::readsome"));
        rate_limited_tcp_read_operation read_operation(socket, buffer, length, offset, completion_promise);
        _read_operations_for_next_iteration.push_back(&read_operation);
 
        // launch the read processing loop it if isn't running, or signal it to resume if it's paused.
        if (!_process_pending_reads_loop_complete.valid() || _process_pending_reads_loop_complete.ready())
          _process_pending_reads_loop_complete = async([=](){ process_pending_reads(); }, "process_pending_reads" );
        else if (_new_read_operation_available_promise)
          _new_read_operation_available_promise->set_value();
 
        try
        {
          bytes_read = completion_promise->wait();
        }
        catch (...)
        {
          _read_operations_for_next_iteration.remove(&read_operation);
          _read_operations_in_progress.remove(&read_operation);
          throw;
        }
        _unused_read_tokens += read_operation.permitted_length - bytes_read;
      }
      else
        bytes_read = asio::read_some(socket, buffer, length, offset);
 
      _actual_download_rate.update(bytes_read);
 
      return bytes_read;
    }

Here is the call graph for this function:

Here is the caller graph for this function:

writesome() [1/2]

size_t fc::detail::rate_limiting_group_impl::writesome ( boost::asio::ip::tcp::socket & socket, const char * buffer, size_t length )

overridevirtual

Definition at line 321 of file rate_limiting.cpp.

    {
      return writesome_impl(socket, buffer, length, 0);
    }

Here is the call graph for this function:

writesome() [2/2]

size_t fc::detail::rate_limiting_group_impl::writesome ( boost::asio::ip::tcp::socket & socket, const std::shared_ptr< const char > & buffer, size_t length, size_t offset )

overridevirtual

Definition at line 326 of file rate_limiting.cpp.

    {
      return writesome_impl(socket, buffer, length, offset);
    }

Here is the call graph for this function:

writesome_impl()

template<typename BufferType >

size_t fc::detail::rate_limiting_group_impl::writesome_impl	(	boost::asio::ip::tcp::socket &	socket,
		const BufferType &	buffer,
		size_t	length,
		size_t	offset )

Definition at line 332 of file rate_limiting.cpp.

    {
      size_t bytes_written;
      if (_upload_bytes_per_second)
      {
        promise<size_t>::ptr completion_promise(new promise<size_t>("rate_limiting_group_impl::writesome"));
        rate_limited_tcp_write_operation write_operation(socket, buffer, length, offset, completion_promise);
        _write_operations_for_next_iteration.push_back(&write_operation);
 
        // launch the write processing loop it if isn't running, or signal it to resume if it's paused.
        if (!_process_pending_writes_loop_complete.valid() || _process_pending_writes_loop_complete.ready())
          _process_pending_writes_loop_complete = async([=](){ process_pending_writes(); }, "process_pending_writes");
        else if (_new_write_operation_available_promise)
          _new_write_operation_available_promise->set_value();
 
        try
        {
          bytes_written = completion_promise->wait();
        }
        catch (...)
        {
          _write_operations_for_next_iteration.remove(&write_operation);
          _write_operations_in_progress.remove(&write_operation);
          throw;
        }
        _unused_write_tokens += write_operation.permitted_length - bytes_written;
      }
      else
        bytes_written = asio::write_some(socket, buffer, length, offset);
 
      _actual_upload_rate.update(bytes_written);
 
      return bytes_written;
    }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

_actual_download_rate

average_rate_meter fc::detail::rate_limiting_group_impl::_actual_download_rate

Definition at line 219 of file rate_limiting.cpp.

_actual_upload_rate

average_rate_meter fc::detail::rate_limiting_group_impl::_actual_upload_rate

Definition at line 218 of file rate_limiting.cpp.

_burstiness_in_seconds

uint32_t fc::detail::rate_limiting_group_impl::_burstiness_in_seconds

Definition at line 196 of file rate_limiting.cpp.

_download_bytes_per_second

uint32_t fc::detail::rate_limiting_group_impl::_download_bytes_per_second

Definition at line 195 of file rate_limiting.cpp.

_granularity

microseconds fc::detail::rate_limiting_group_impl::_granularity

Definition at line 198 of file rate_limiting.cpp.

_last_read_iteration_time

time_point fc::detail::rate_limiting_group_impl::_last_read_iteration_time

Definition at line 210 of file rate_limiting.cpp.

_last_write_iteration_time

time_point fc::detail::rate_limiting_group_impl::_last_write_iteration_time

Definition at line 211 of file rate_limiting.cpp.

_new_read_operation_available_promise

promise<void>::ptr fc::detail::rate_limiting_group_impl::_new_read_operation_available_promise

Definition at line 214 of file rate_limiting.cpp.

_new_write_operation_available_promise

promise<void>::ptr fc::detail::rate_limiting_group_impl::_new_write_operation_available_promise

Definition at line 216 of file rate_limiting.cpp.

_process_pending_reads_loop_complete

future<void> fc::detail::rate_limiting_group_impl::_process_pending_reads_loop_complete

Definition at line 213 of file rate_limiting.cpp.

_process_pending_writes_loop_complete

future<void> fc::detail::rate_limiting_group_impl::_process_pending_writes_loop_complete

Definition at line 215 of file rate_limiting.cpp.

_read_operations_for_next_iteration

rate_limited_operation_list fc::detail::rate_limiting_group_impl::_read_operations_for_next_iteration

Definition at line 206 of file rate_limiting.cpp.

_read_operations_in_progress

rate_limited_operation_list fc::detail::rate_limiting_group_impl::_read_operations_in_progress

Definition at line 205 of file rate_limiting.cpp.

_read_tokens

uint32_t fc::detail::rate_limiting_group_impl::_read_tokens

Definition at line 199 of file rate_limiting.cpp.

_unused_read_tokens

uint32_t fc::detail::rate_limiting_group_impl::_unused_read_tokens

Definition at line 200 of file rate_limiting.cpp.

_unused_write_tokens

uint32_t fc::detail::rate_limiting_group_impl::_unused_write_tokens

Definition at line 202 of file rate_limiting.cpp.

_upload_bytes_per_second

uint32_t fc::detail::rate_limiting_group_impl::_upload_bytes_per_second

Definition at line 194 of file rate_limiting.cpp.

_write_operations_for_next_iteration

rate_limited_operation_list fc::detail::rate_limiting_group_impl::_write_operations_for_next_iteration

Definition at line 208 of file rate_limiting.cpp.

_write_operations_in_progress

rate_limited_operation_list fc::detail::rate_limiting_group_impl::_write_operations_in_progress

Definition at line 207 of file rate_limiting.cpp.

_write_tokens

uint32_t fc::detail::rate_limiting_group_impl::_write_tokens

Definition at line 201 of file rate_limiting.cpp.

---

The documentation for this class was generated from the following file:

• libraries/fc/src/network/rate_limiting.cpp