diff --git a/.gitignore b/.gitignore index bbccfeb..5fc5eea 100644 --- a/.gitignore +++ b/.gitignore @@ -15,7 +15,6 @@ *.a *.lib - #CMake->MSVC artifacts *.sln *.vcxproj @@ -41,5 +40,7 @@ libfc.a libfc_debug.a fc_automoc.cpp -*.swp +git_revision.cpp GitSHA3.cpp + +*.sw* diff --git a/CMakeLists.txt b/CMakeLists.txt index 4f306a1..9f3e8f5 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -38,10 +38,6 @@ LIST(APPEND BOOST_COMPONENTS thread date_time system filesystem program_options IF( WIN32 ) MESSAGE(STATUS "Configuring fc to build on Win32") - #You need to set OPENSSL_ROOT environment variable for your system on WIN32 - message(STATUS "Setting up OpenSSL root and include vars on Win32 platform") - set( OPENSSL_ROOT_DIR $ENV{OPENSSL_ROOT} ) - set( RPCRT4 Rpcrt4 ) #boost @@ -77,7 +73,13 @@ ELSE(WIN32) ENDIF(NOT APPLE) ENDIF(WIN32) -find_package( OpenSSL ) +#IF($ENV{OPENSSL_ROOT_DIR}) + set(OPENSSL_ROOT_DIR $ENV{OPENSSL_ROOT_DIR} ) + set(OPENSSL_INCLUDE_DIR ${OPENSSL_ROOT_DIR}/include) + message(STATUS "Setting up OpenSSL root and include vars to ${OPENSSL_ROOT_DIR}, ${OPENSSL_INCLUDE_DIR}") +#ENDIF() + +find_package(OpenSSL) set( CMAKE_FIND_LIBRARY_SUFFIXES ${ORIGINAL_LIB_SUFFIXES} ) @@ -147,6 +149,7 @@ set( fc_sources src/network/http/http_connection.cpp src/network/http/http_server.cpp src/network/ip.cpp + src/network/rate_limiting.cpp src/network/resolve.cpp src/network/url.cpp src/compress/smaz.cpp @@ -186,13 +189,13 @@ IF(WIN32) # Add /U options to be sure settings specific to dynamic boost link are ineffective target_compile_options(fc PUBLIC /EHsc /UBOOST_ALL_DYN_LINK /UBOOST_LINKING_PYTHON /UBOOST_DEBUG_PYTHON) ELSE() - SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -std=c99 -Wall" ) + SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -std=c99 -Wall") IF(APPLE) - SET(CMAKE_CXX_FLAGS "${CMAKE_C_FLAGS} -std=c++11 -stdlib=libc++ -Wall" ) + SET(CMAKE_CXX_FLAGS "${CMAKE_C_FLAGS} -std=c++11 -stdlib=libc++ -Wall") ELSE() - target_compile_options(fc PUBLIC -std=c++11 -Wall -fnon-call-exceptions -Wno-unused-local-typedefs -fmax-errors=3) - SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-unused-local-typedefs -fmax-errors=3 ") + target_compile_options(fc PUBLIC -std=c++11 -Wall -fnon-call-exceptions) + SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wall") ENDIF() ENDIF() @@ -217,6 +220,8 @@ target_link_libraries( fc PUBLIC easylzma_static ${Boost_LIBRARIES} ${OPENSSL_LI #target_link_libraries( test_aes fc ${rt_library} ${pthread_library} ) #add_executable( test_sleep tests/sleep.cpp ) #target_link_libraries( test_sleep fc ) +#add_executable( test_rate_limiting tests/rate_limiting.cpp ) +#target_link_libraries( test_rate_limiting fc ) if(WIN32) # add addtional import library on windows platform diff --git a/include/fc/asio.hpp b/include/fc/asio.hpp index 36a2727..8723290 100644 --- a/include/fc/asio.hpp +++ b/include/fc/asio.hpp @@ -79,11 +79,17 @@ namespace asio { * @return the number of bytes read. */ template - size_t read_some( AsyncReadStream& s, const MutableBufferSequence& buf ) + size_t read_some(AsyncReadStream& s, const MutableBufferSequence& buf) { - promise::ptr p(new promise("fc::asio::async_read_some")); - s.async_read_some( buf, boost::bind( detail::read_write_handler, p, _1, _2 ) ); - return p->wait(); + promise::ptr p(new promise("fc::asio::async_read_some")); + s.async_read_some(buf, boost::bind(detail::read_write_handler, p, _1, _2)); + return p->wait(); + } + + template + void async_read_some(AsyncReadStream& s, const MutableBufferSequence& buf, promise::ptr completion_promise) + { + s.async_read_some(buf, boost::bind(detail::read_write_handler, completion_promise, _1, _2)); } template @@ -117,6 +123,16 @@ namespace asio { return p->wait(); } + /** + * @pre s.non_blocking() == true + * @brief wraps boost::asio::async_write_some + * @return the number of bytes written + */ + template + void async_write_some(AsyncWriteStream& s, const ConstBufferSequence& buf, promise::ptr completion_promise) { + s.async_write_some(buf, boost::bind(detail::read_write_handler, completion_promise, _1, _2)); + } + namespace tcp { typedef boost::asio::ip::tcp::endpoint endpoint; typedef boost::asio::ip::tcp::resolver::iterator resolver_iterator; diff --git a/include/fc/crypto/ripemd160.hpp b/include/fc/crypto/ripemd160.hpp index e2d13a9..998b01d 100644 --- a/include/fc/crypto/ripemd160.hpp +++ b/include/fc/crypto/ripemd160.hpp @@ -77,7 +77,7 @@ class ripemd160 typedef ripemd160 uint160_t; typedef ripemd160 uint160; - template<> struct get_typename { static const char* name() { return "uint160"; } }; + template<> struct get_typename { static const char* name() { return "uint160_t"; } }; } // namespace fc diff --git a/include/fc/network/rate_limiting.hpp b/include/fc/network/rate_limiting.hpp new file mode 100644 index 0000000..7694789 --- /dev/null +++ b/include/fc/network/rate_limiting.hpp @@ -0,0 +1,35 @@ +#pragma once +#include + +#include + +namespace fc +{ + namespace detail + { + class rate_limiting_group_impl; + } + + class tcp_socket; + + class rate_limiting_group + { + public: + rate_limiting_group(uint32_t upload_bytes_per_second, uint32_t download_bytes_per_second); + ~rate_limiting_group(); + + void set_upload_limit(uint32_t upload_bytes_per_second); + uint32_t get_upload_limit() const; + + void set_download_limit(uint32_t download_bytes_per_second); + uint32_t get_download_limit() const; + + void add_tcp_socket(tcp_socket* tcp_socket_to_limit); + void remove_tcp_socket(tcp_socket* tcp_socket_to_stop_limiting); + private: + std::unique_ptr my; + }; + typedef std::shared_ptr rate_limiting_group_ptr; + +} // namesapce fc + diff --git a/include/fc/network/tcp_socket.hpp b/include/fc/network/tcp_socket.hpp index 44b384a..a2fc8f4 100644 --- a/include/fc/network/tcp_socket.hpp +++ b/include/fc/network/tcp_socket.hpp @@ -6,6 +6,9 @@ namespace fc { namespace ip { class endpoint; } + + class tcp_socket_io_hooks; + class tcp_socket : public virtual iostream { public: @@ -15,7 +18,8 @@ namespace fc { void connect_to( const fc::ip::endpoint& remote_endpoint ); void connect_to( const fc::ip::endpoint& remote_endpoint, const fc::ip::endpoint& local_endpoint ); void enable_keep_alives(const fc::microseconds& interval); - void set_reuse_address(bool enable = true); // set SO_REUSEADDR + void set_io_hooks(tcp_socket_io_hooks* new_hooks); + void set_reuse_address(bool enable = true); // set SO_REUSEADDR fc::ip::endpoint remote_endpoint() const; fc::ip::endpoint local_endpoint() const; @@ -45,9 +49,9 @@ namespace fc { friend class tcp_server; class impl; #ifdef _WIN64 - fc::fwd my; + fc::fwd my; #else - fc::fwd my; + fc::fwd my; #endif }; typedef std::shared_ptr tcp_socket_ptr; diff --git a/include/fc/network/tcp_socket_io_hooks.hpp b/include/fc/network/tcp_socket_io_hooks.hpp new file mode 100644 index 0000000..a317ed1 --- /dev/null +++ b/include/fc/network/tcp_socket_io_hooks.hpp @@ -0,0 +1,12 @@ +#include + +namespace fc +{ + class tcp_socket_io_hooks + { + public: + virtual ~tcp_socket_io_hooks() {} + virtual size_t readsome(boost::asio::ip::tcp::socket& socket, char* buffer, size_t length) = 0; + virtual size_t writesome(boost::asio::ip::tcp::socket& socket, const char* buffer, size_t length) = 0; + }; +} // namesapce fc diff --git a/include/fc/time.hpp b/include/fc/time.hpp index 888eefd..22e577f 100644 --- a/include/fc/time.hpp +++ b/include/fc/time.hpp @@ -95,6 +95,7 @@ namespace fc { friend bool operator != ( const time_point_sec& a, const time_point_sec& b ) { return a.utc_seconds != b.utc_seconds; } time_point_sec& operator += ( uint32_t m ) { utc_seconds+=m; return *this; } time_point_sec& operator -= ( uint32_t m ) { utc_seconds-=m; return *this; } + time_point_sec operator+( uint32_t offset ) { return time_point_sec(utc_seconds + offset); } friend time_point operator - ( const time_point_sec& t, const microseconds& m ) { return time_point(t) - m; } friend microseconds operator - ( const time_point_sec& t, const time_point_sec& m ) { return time_point(t) - time_point(m); } diff --git a/src/crypto/aes.cpp b/src/crypto/aes.cpp index 3f0933f..954f0b4 100644 --- a/src/crypto/aes.cpp +++ b/src/crypto/aes.cpp @@ -38,7 +38,7 @@ void aes_encoder::init( const fc::sha256& key, const fc::uint128& init_value ) * is 128 bits */ if(1 != EVP_EncryptInit_ex(my->ctx, EVP_aes_256_cbc(), NULL, (unsigned char*)&key, (unsigned char*)&init_value)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption init", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption init", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } EVP_CIPHER_CTX_set_padding( my->ctx, 0 ); @@ -52,7 +52,7 @@ uint32_t aes_encoder::encode( const char* plaintxt, uint32_t plaintext_len, char * */ if(1 != EVP_EncryptUpdate(my->ctx, (unsigned char*)ciphertxt, &ciphertext_len, (const unsigned char*)plaintxt, plaintext_len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption update", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption update", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } FC_ASSERT( ciphertext_len == plaintext_len, "", ("ciphertext_len",ciphertext_len)("plaintext_len",plaintext_len) ); @@ -67,7 +67,7 @@ uint32_t aes_encoder::final_encode( char* ciphertxt ) * */ if(1 != EVP_EncryptFinal_ex(my->ctx, (unsigned char*)ciphertxt, &ciphertext_len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption final", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption final", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } return ciphertext_len; @@ -98,7 +98,7 @@ void aes_decoder::init( const fc::sha256& key, const fc::uint128& init_value ) * is 128 bits */ if(1 != EVP_DecryptInit_ex(my->ctx, EVP_aes_256_cbc(), NULL, (unsigned char*)&key, (unsigned char*)&init_value)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption init", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption init", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } EVP_CIPHER_CTX_set_padding( my->ctx, 0 ); @@ -115,7 +115,7 @@ uint32_t aes_decoder::decode( const char* ciphertxt, uint32_t plaintext_len, cha * */ if(1 != EVP_DecryptUpdate(my->ctx, (unsigned char*)plaintext, &ciphertext_len, (const unsigned char*)ciphertxt, plaintext_len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption update", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption update", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } FC_ASSERT( ciphertext_len == plaintext_len, "", ("ciphertext_len",ciphertext_len)("plaintext_len",plaintext_len) ); @@ -131,7 +131,7 @@ uint32_t aes_decoder::final_decode( char* plaintext ) * */ if(1 != EVP_DecryptFinal_ex(my->ctx, (unsigned char*)plaintext, &ciphertext_len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption final", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption final", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } return ciphertext_len; @@ -172,7 +172,7 @@ int aes_encrypt(unsigned char *plaintext, int plaintext_len, unsigned char *key, * is 128 bits */ if(1 != EVP_EncryptInit_ex(ctx, EVP_aes_256_cbc(), NULL, key, iv)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption init", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption init", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } @@ -181,7 +181,7 @@ int aes_encrypt(unsigned char *plaintext, int plaintext_len, unsigned char *key, * */ if(1 != EVP_EncryptUpdate(ctx, ciphertext, &len, plaintext, plaintext_len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption update", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption update", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } ciphertext_len = len; @@ -191,7 +191,7 @@ int aes_encrypt(unsigned char *plaintext, int plaintext_len, unsigned char *key, * */ if(1 != EVP_EncryptFinal_ex(ctx, ciphertext + len, &len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc encryption final", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc encryption final", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } ciphertext_len += len; @@ -220,7 +220,7 @@ int aes_decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char *ke * * is 128 bits */ if(1 != EVP_DecryptInit_ex(ctx, EVP_aes_256_cbc(), NULL, key, iv)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc decrypt init", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc decrypt init", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } @@ -229,7 +229,7 @@ int aes_decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char *ke * */ if(1 != EVP_DecryptUpdate(ctx, plaintext, &len, ciphertext, ciphertext_len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc decrypt update", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc decrypt update", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } @@ -240,7 +240,7 @@ int aes_decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char *ke * */ if(1 != EVP_DecryptFinal_ex(ctx, plaintext + len, &len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc decrypt final", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc decrypt final", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } plaintext_len += len; @@ -269,7 +269,7 @@ int aes_cfb_decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char * * is 128 bits */ if(1 != EVP_DecryptInit_ex(ctx, EVP_aes_256_cfb128(), NULL, key, iv)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc decrypt init", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc decrypt init", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } @@ -278,7 +278,7 @@ int aes_cfb_decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char * */ if(1 != EVP_DecryptUpdate(ctx, plaintext, &len, ciphertext, ciphertext_len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc decrypt update", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc decrypt update", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } @@ -289,7 +289,7 @@ int aes_cfb_decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char * */ if(1 != EVP_DecryptFinal_ex(ctx, plaintext + len, &len)) { - FC_THROW_EXCEPTION( exception, "error durring aes 256 cbc decrypt final", + FC_THROW_EXCEPTION( exception, "error during aes 256 cbc decrypt final", ("s", ERR_error_string( ERR_get_error(), nullptr) ) ); } plaintext_len += len; diff --git a/src/network/rate_limiting.cpp b/src/network/rate_limiting.cpp new file mode 100644 index 0000000..0055abb --- /dev/null +++ b/src/network/rate_limiting.cpp @@ -0,0 +1,344 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace fc +{ + + namespace detail + { + // data about a read or write we're managing + class rate_limited_operation + { + public: + size_t length; + size_t permitted_length; + promise::ptr completion_promise; + + rate_limited_operation(size_t length, + promise::ptr&& completion_promise) : + length(length), + permitted_length(0), + completion_promise(completion_promise) + {} + + virtual void perform_operation() = 0; + }; + + class rate_limited_tcp_write_operation : public rate_limited_operation + { + public: + boost::asio::ip::tcp::socket& socket; + const char* buffer; + + rate_limited_tcp_write_operation(boost::asio::ip::tcp::socket& socket, + const char* buffer, + size_t length, + promise::ptr completion_promise) : + rate_limited_operation(length, std::move(completion_promise)), + socket(socket), + buffer(buffer) + {} + virtual void perform_operation() override + { + asio::async_write_some(socket, + boost::asio::buffer(buffer, permitted_length), + completion_promise); + } + }; + class rate_limited_tcp_read_operation : public rate_limited_operation + { + public: + boost::asio::ip::tcp::socket& socket; + char* buffer; + + rate_limited_tcp_read_operation(boost::asio::ip::tcp::socket& socket, + char* buffer, + size_t length, + promise::ptr completion_promise) : + rate_limited_operation(length, std::move(completion_promise)), + socket(socket), + buffer(buffer) + {} + virtual void perform_operation() override + { + asio::async_read_some(socket, + boost::asio::buffer(buffer, permitted_length), + completion_promise); + } + }; + + struct is_operation_shorter + { + // less than operator designed to bring the shortest operations to the end + bool operator()(const rate_limited_operation* lhs, const rate_limited_operation* rhs) + { + return lhs->length > rhs->length; + } + }; + + class rate_limiting_group_impl : public tcp_socket_io_hooks + { + public: + uint32_t _upload_bytes_per_second; + uint32_t _download_bytes_per_second; + + microseconds _granularity; // how often to add tokens to the bucket + uint32_t _read_tokens; + uint32_t _unused_read_tokens; // gets filled with tokens for unused bytes (if I'm allowed to read 200 bytes and I try to read 200 bytes, but can only read 50, tokens for the other 150 get returned here) + uint32_t _write_tokens; + uint32_t _unused_write_tokens; + + typedef std::list rate_limited_operation_list; + 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 _process_pending_reads_loop_complete; + promise::ptr _new_read_operation_available_promise; + future _process_pending_writes_loop_complete; + promise::ptr _new_write_operation_available_promise; + + rate_limiting_group_impl(uint32_t upload_bytes_per_second, uint32_t download_bytes_per_second); + + virtual size_t readsome(boost::asio::ip::tcp::socket& socket, char* buffer, size_t length) override; + virtual size_t writesome(boost::asio::ip::tcp::socket& socket, const char* buffer, size_t length) override; + + 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); + }; + + rate_limiting_group_impl::rate_limiting_group_impl(uint32_t upload_bytes_per_second, uint32_t download_bytes_per_second) : + _upload_bytes_per_second(upload_bytes_per_second), + _download_bytes_per_second(download_bytes_per_second), + _granularity(milliseconds(50)), + _read_tokens(_download_bytes_per_second), + _unused_read_tokens(0), + _write_tokens(_upload_bytes_per_second), + _unused_write_tokens(0) + { + } + + size_t rate_limiting_group_impl::readsome(boost::asio::ip::tcp::socket& socket, char* buffer, size_t length) + { + if (_download_bytes_per_second) + { + promise::ptr completion_promise(new promise()); + rate_limited_tcp_read_operation read_operation(socket, buffer, length, 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(); }); + else if (_new_read_operation_available_promise) + _new_read_operation_available_promise->set_value(); + + size_t bytes_read = completion_promise->wait(); + _unused_read_tokens += read_operation.permitted_length - bytes_read; + return bytes_read; + } + else + return asio::read_some(socket, boost::asio::buffer(buffer, length)); + } + size_t rate_limiting_group_impl::writesome(boost::asio::ip::tcp::socket& socket, const char* buffer, size_t length) + { + if (_upload_bytes_per_second) + { + promise::ptr completion_promise(new promise()); + rate_limited_tcp_write_operation write_operation(socket, buffer, length, 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(); }); + else if (_new_write_operation_available_promise) + _new_write_operation_available_promise->set_value(); + + size_t bytes_written = completion_promise->wait(); + _unused_write_tokens += write_operation.permitted_length - bytes_written; + return bytes_written; + } + else + return asio::write_some(socket, boost::asio::buffer(buffer, length)); + } + void rate_limiting_group_impl::process_pending_reads() + { + 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(); + 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(); + } + } + void rate_limiting_group_impl::process_pending_writes() + { + 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(); + 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(); + } + } + void 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) + { + // 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); + + if (tokens) + { + // sort the pending reads/writes in order of the number of bytes they need to write, smallest first + std::vector 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(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) + { + (*iter)->perform_operation(); + iter = operations_in_progress.erase(iter); + } + 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(); + ++iter) + { + (*iter)->permitted_length = (*iter)->length; + (*iter)->perform_operation(); + } + operations_in_progress.clear(); + } + last_iteration_start_time = this_iteration_start_time; + } + + } + + rate_limiting_group::rate_limiting_group(uint32_t upload_bytes_per_second, uint32_t download_bytes_per_second) : + my(new detail::rate_limiting_group_impl(upload_bytes_per_second, download_bytes_per_second)) + { + } + + rate_limiting_group::~rate_limiting_group() + { + } + + void rate_limiting_group::set_upload_limit(uint32_t upload_bytes_per_second) + { + my->_upload_bytes_per_second = upload_bytes_per_second; + } + + uint32_t rate_limiting_group::get_upload_limit() const + { + return my->_upload_bytes_per_second; + } + + void rate_limiting_group::set_download_limit(uint32_t download_bytes_per_second) + { + my->_download_bytes_per_second = download_bytes_per_second; + } + + uint32_t rate_limiting_group::get_download_limit() const + { + return my->_download_bytes_per_second; + } + + void rate_limiting_group::add_tcp_socket(tcp_socket* tcp_socket_to_limit) + { + tcp_socket_to_limit->set_io_hooks(my.get()); + } + + void rate_limiting_group::remove_tcp_socket(tcp_socket* tcp_socket_to_stop_limiting) + { + tcp_socket_to_stop_limiting->set_io_hooks(NULL); + } + + +} // namespace fc diff --git a/src/network/tcp_socket.cpp b/src/network/tcp_socket.cpp index c4b0e11..1363347 100644 --- a/src/network/tcp_socket.cpp +++ b/src/network/tcp_socket.cpp @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -12,18 +13,34 @@ namespace fc { - class tcp_socket::impl { + class tcp_socket::impl : public tcp_socket_io_hooks{ public: - impl() : - _sock( fc::asio::default_io_service() ) + impl() : + _sock(fc::asio::default_io_service()), + _io_hooks(this) {} - ~impl(){ + ~impl() + { if( _sock.is_open() ) _sock.close(); } + virtual size_t readsome(boost::asio::ip::tcp::socket& socket, char* buffer, size_t length) override; + virtual size_t writesome(boost::asio::ip::tcp::socket& socket, const char* buffer, size_t length) override; + boost::asio::ip::tcp::socket _sock; + tcp_socket_io_hooks* _io_hooks; }; + size_t tcp_socket::impl::readsome(boost::asio::ip::tcp::socket& socket, char* buffer, size_t length) + { + return fc::asio::read_some(socket, boost::asio::buffer(buffer, length)); + } + size_t tcp_socket::impl::writesome(boost::asio::ip::tcp::socket& socket, const char* buffer, size_t length) + { + return fc::asio::write_some(socket, boost::asio::buffer(buffer, length)); + } + + void tcp_socket::open() { my->_sock.open(boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), 0).protocol()); @@ -51,8 +68,8 @@ namespace fc { return !my->_sock.is_open(); } - size_t tcp_socket::writesome( const char* buf, size_t len ) { - return fc::asio::write_some( my->_sock, boost::asio::buffer( buf, len ) ); + size_t tcp_socket::writesome(const char* buf, size_t len) { + return my->_io_hooks->writesome(my->_sock, buf, len); } fc::ip::endpoint tcp_socket::remote_endpoint()const @@ -61,6 +78,7 @@ namespace fc { return fc::ip::endpoint(rep.address().to_v4().to_ulong(), rep.port() ); } + fc::ip::endpoint tcp_socket::local_endpoint() const { auto boost_local_endpoint = my->_sock.local_endpoint(); @@ -68,8 +86,7 @@ namespace fc { } size_t tcp_socket::readsome( char* buf, size_t len ) { - auto r = fc::asio::read_some( my->_sock, boost::asio::buffer( buf, len ) ); - return r; + return my->_io_hooks->readsome(my->_sock, buf, len); } void tcp_socket::connect_to( const fc::ip::endpoint& remote_endpoint ) { @@ -131,6 +148,11 @@ namespace fc { } } + void tcp_socket::set_io_hooks(tcp_socket_io_hooks* new_hooks) + { + my->_io_hooks = new_hooks ? new_hooks : &*my; + } + void tcp_socket::set_reuse_address(bool enable /* = true */) { FC_ASSERT(my->_sock.is_open()); diff --git a/tests/rate_limiting.cpp b/tests/rate_limiting.cpp new file mode 100644 index 0000000..dd454a1 --- /dev/null +++ b/tests/rate_limiting.cpp @@ -0,0 +1,34 @@ +#include +#include +#include +#include +#include + +#include +fc::rate_limiting_group rate_limiter(1000000, 1000000); + +void download_url(const std::string& ip_address, const std::string& url) +{ + fc::http::connection http_connection; + rate_limiter.add_tcp_socket(&http_connection.get_socket()); + http_connection.connect_to(fc::ip::endpoint(fc::ip::address(ip_address.c_str()), 80)); + std::cout << "Starting download...\n"; + fc::time_point start_time(fc::time_point::now()); + fc::http::reply reply = http_connection.request("GET", "http://mirror.cs.vt.edu/pub/cygwin/glibc/releases/glibc-2.9.tar.gz"); + fc::time_point end_time(fc::time_point::now()); + + std::cout << "HTTP return code: " << reply.status << "\n"; + std::cout << "Retreived " << reply.body.size() << " bytes in " << ((end_time - start_time).count() / fc::milliseconds(1).count()) << "ms\n"; + std::cout << "Average speed " << ((1000 * (uint64_t)reply.body.size()) / ((end_time - start_time).count() / fc::milliseconds(1).count())) << " bytes per second"; +} + +int main( int argc, char** argv ) +{ + std::vector > download_futures; + download_futures.push_back(fc::async([](){ download_url("198.82.184.145", "http://mirror.cs.vt.edu/pub/cygwin/glibc/releases/glibc-2.9.tar.gz"); })); + download_futures.push_back(fc::async([](){ download_url("198.82.184.145", "http://mirror.cs.vt.edu/pub/cygwin/glibc/releases/glibc-2.7.tar.gz"); })); + + for (int i = 0; i < download_futures.size(); ++i) + download_futures[i].wait(); + return 0; +}