Merge branch 'phoenix' of https://github.com/InvictusInnovations/fc into phoenix

.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*

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)
+    target_compile_options(fc PUBLIC -std=c++11 -Wall -fnon-call-exceptions)
-    SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-unused-local-typedefs -fmax-errors=3 ")
+    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

include/fc/asio.hpp

@@ -79,11 +79,17 @@ namespace asio {
      *  @return the number of bytes read.
      */
     template<typename AsyncReadStream, typename MutableBufferSequence>
-    size_t read_some( AsyncReadStream& s, const MutableBufferSequence& buf ) 
+    size_t read_some(AsyncReadStream& s, const MutableBufferSequence& buf)
     {
-        promise<size_t>::ptr p(new promise<size_t>("fc::asio::async_read_some"));
+      promise<size_t>::ptr p(new promise<size_t>("fc::asio::async_read_some"));
-        s.async_read_some( buf, boost::bind( detail::read_write_handler, p, _1, _2 ) );
+      s.async_read_some(buf, boost::bind(detail::read_write_handler, p, _1, _2));
-        return p->wait();
+      return p->wait();
+    }
+
+    template<typename AsyncReadStream, typename MutableBufferSequence>
+    void async_read_some(AsyncReadStream& s, const MutableBufferSequence& buf, promise<size_t>::ptr completion_promise)
+    {
+      s.async_read_some(buf, boost::bind(detail::read_write_handler, completion_promise, _1, _2));
     }
 
     template<typename AsyncReadStream>
@@ -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<typename AsyncWriteStream, typename ConstBufferSequence>
+    void async_write_some(AsyncWriteStream& s, const ConstBufferSequence& buf, promise<size_t>::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;

include/fc/crypto/ripemd160.hpp

@@ -77,7 +77,7 @@ class ripemd160
   typedef ripemd160 uint160_t;
   typedef ripemd160 uint160;
 
-  template<> struct get_typename<uint160>    { static const char* name()  { return "uint160";  } };
+  template<> struct get_typename<uint160_t>    { static const char* name()  { return "uint160_t";  } };
 
 } // namespace fc
 

include/fc/network/rate_limiting.hpp

@@ -0,0 +1,35 @@
+#pragma once
+#include <stdint.h>
+
+#include <memory>
+
+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<detail::rate_limiting_group_impl> my;
+  };
+  typedef std::shared_ptr<rate_limiting_group> rate_limiting_group_ptr;
+
+} // namesapce fc
+

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<impl,0x68> my;
+      fc::fwd<impl,0x70> my;
       #else
-      fc::fwd<impl,0x44> my;
+      fc::fwd<impl,0x4c> my;
       #endif
   };
   typedef std::shared_ptr<tcp_socket> tcp_socket_ptr;

include/fc/network/tcp_socket_io_hooks.hpp

@@ -0,0 +1,12 @@
+#include <boost/asio.hpp>
+
+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

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); }

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;

src/network/rate_limiting.cpp

@@ -0,0 +1,344 @@
+#include <fc/network/rate_limiting.hpp>
+#include <fc/network/tcp_socket_io_hooks.hpp>
+#include <fc/network/tcp_socket.hpp>
+#include <fc/network/ip.hpp>
+#include <fc/fwd_impl.hpp>
+#include <fc/asio.hpp>
+#include <fc/log/logger.hpp>
+#include <fc/io/stdio.hpp>
+#include <fc/exception/exception.hpp>
+#include <fc/thread/thread.hpp>
+
+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<size_t>::ptr          completion_promise;
+
+      rate_limited_operation(size_t length,
+                             promise<size_t>::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<size_t>::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<size_t>::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*> 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<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;
+
+      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<size_t>::ptr completion_promise(new promise<size_t>());
+        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<size_t>::ptr completion_promise(new promise<size_t>());
+        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<void>();
+        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<void>();
+        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<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(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 

src/network/tcp_socket.cpp

@@ -1,5 +1,6 @@
 #include <fc/network/tcp_socket.hpp>
 #include <fc/network/ip.hpp>
+#include <fc/network/tcp_socket_io_hooks.hpp>
 #include <fc/fwd_impl.hpp>
 #include <fc/asio.hpp>
 #include <fc/log/logger.hpp>
@@ -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() )
+        _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 ) {
+  size_t tcp_socket::writesome(const char* buf, size_t len) {
-    return fc::asio::write_some( my->_sock, boost::asio::buffer( buf, 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 my->_io_hooks->readsome(my->_sock, buf, len);
-    return r;
   }
 
   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());

tests/rate_limiting.cpp

@@ -0,0 +1,34 @@
+#include <fc/network/http/connection.hpp>
+#include <fc/network/rate_limiting.hpp>
+#include <fc/network/ip.hpp>
+#include <fc/time.hpp>
+#include <fc/thread/thread.hpp>
+
+#include <iostream>
+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<fc::future<void> > 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;
+}