Merge pull request #561 from eosnetworkfoundation/http_max_inflight_reqs

[3.2] Add --http-max-in-flight-requests to http_plugin option

plugins/http_plugin/http_plugin.cpp

@@ -130,6 +130,7 @@ namespace eosio {
       struct abstract_conn {
          virtual ~abstract_conn() {}
          virtual bool verify_max_bytes_in_flight() = 0;
+         virtual bool verify_max_requests_in_flight() = 0;
          virtual void handle_exception() = 0;
          virtual void send_response(std::string, int) = 0;
       };
@@ -200,7 +201,9 @@ class http_plugin_impl : public std::enable_shared_from_this<http_plugin_impl> {
          uint16_t                                    thread_pool_size = 2;
          std::optional<eosio::chain::named_thread_pool>   thread_pool;
          std::atomic<size_t>                         bytes_in_flight{0};
+         std::atomic<int32_t>                        requests_in_flight{0};
          size_t                                      max_bytes_in_flight = 0;
+         int32_t                                     max_requests_in_flight = -1;
          fc::microseconds                            max_response_time{30*1000};
 
          std::optional<tcp::endpoint>  https_listen_endpoint;
@@ -325,25 +328,46 @@ class http_plugin_impl : public std::enable_shared_from_this<http_plugin_impl> {
             return true;
          }
 
+         template<typename T>
+         void report_429_error( const T& con, const std::string& what) {
+            error_results::error_info ei;
+            ei.code = websocketpp::http::status_code::too_many_requests;
+            ei.name = "Busy";
+            ei.what = what;
+            error_results results{websocketpp::http::status_code::too_many_requests, "Busy", ei};
+            con->set_body( fc::json::to_string( results, fc::time_point::maximum() ));
+            con->set_status( websocketpp::http::status_code::too_many_requests );
+            con->send_http_response();
+         }
+
          template<typename T>
          bool verify_max_bytes_in_flight( const T& con ) {
             auto bytes_in_flight_size = bytes_in_flight.load();
             if( bytes_in_flight_size > max_bytes_in_flight ) {
                fc_dlog( logger, "429 - too many bytes in flight: ${bytes}", ("bytes", bytes_in_flight_size) );
-               error_results::error_info ei;
-               ei.code = websocketpp::http::status_code::too_many_requests;
-               ei.name = "Busy";
-               ei.what = "Too many bytes in flight: " + std::to_string( bytes_in_flight_size );
-               error_results results{websocketpp::http::status_code::too_many_requests, "Busy", ei};
-               con->set_body( fc::json::to_string( results, fc::time_point::maximum() ));
-               con->set_status( websocketpp::http::status_code::too_many_requests );
-               con->send_http_response();
+               string what = "Too many bytes in flight: " + std::to_string( bytes_in_flight_size ) + ". Try again later.";;
+               report_429_error(con, what);
                return false;
             }
 
             return true;
          }
 
+         template<typename T>
+         bool verify_max_requests_in_flight( const T& con ) {
+            if (max_requests_in_flight < 0)
+                return true;
+
+            auto requests_in_flight_num = requests_in_flight.load();
+            if( requests_in_flight_num > max_requests_in_flight ) {
+               fc_dlog( logger, "429 - too many requests in flight: ${requests}", ("requests", requests_in_flight_num) );
+               string what = "Too many requests in flight: " + std::to_string( requests_in_flight_num ) + ". Try again later.";
+               report_429_error(con, what);
+               return false;
+            }
+            return true;
+         }
+
          /**
           * child struct, implementing abstract connection for various underlying connection types
           * that ties it to an http_plugin_impl
@@ -355,20 +379,28 @@ class http_plugin_impl : public std::enable_shared_from_this<http_plugin_impl> {
             abstract_conn_impl(detail::connection_ptr<T> conn, http_plugin_impl_ptr impl)
             :_conn(std::move(conn))
             ,_impl(std::move(impl))
-            {}
+            {
+                _impl->requests_in_flight += 1;
+            }
+
+            ~abstract_conn_impl() {
+                _impl->requests_in_flight -= 1;
+            }
 
             // No copy constructor and no move
             abstract_conn_impl(const abstract_conn_impl&) = delete;
             abstract_conn_impl(abstract_conn_impl&&) = delete;
             abstract_conn_impl& operator=(const abstract_conn_impl&) = delete;
             abstract_conn_impl& operator=(abstract_conn_impl&&) noexcept = default;
 
-            ~abstract_conn_impl() = default;
-
             bool verify_max_bytes_in_flight() override {
                return _impl->verify_max_bytes_in_flight(_conn);
             }
 
+            bool verify_max_requests_in_flight() override {
+               return _impl->verify_max_requests_in_flight(_conn);
+            }
+
             void handle_exception()override {
                http_plugin_impl::handle_exception<T>(_conn);
             }
@@ -577,7 +609,7 @@ class http_plugin_impl : public std::enable_shared_from_this<http_plugin_impl> {
                con->defer_http_response();
 
                auto abstract_conn_ptr = make_abstract_conn_ptr<T>(con, shared_from_this());
-               if( !verify_max_bytes_in_flight( con )) return;
+               if( !verify_max_bytes_in_flight( con ) || !verify_max_requests_in_flight( con ) ) return;
 
                std::string resource = con->get_uri()->get_resource();
                auto handler_itr = url_handlers.find( resource );
@@ -698,7 +730,9 @@ class http_plugin_impl : public std::enable_shared_from_this<http_plugin_impl> {
             ("max-body-size", bpo::value<uint32_t>()->default_value(my->max_body_size),
              "The maximum body size in bytes allowed for incoming RPC requests")
             ("http-max-bytes-in-flight-mb", bpo::value<uint32_t>()->default_value(500),
-             "Maximum size in megabytes http_plugin should use for processing http requests. 503 error response when exceeded." )
+             "Maximum size in megabytes http_plugin should use for processing http requests. 429 error response when exceeded." )
+            ("http-max-in-flight-requests", bpo::value<int32_t>()->default_value(-1),
+             "Maximum number of requests http_plugin should use for processing http requests. 429 error response when exceeded." )
             ("http-max-response-time-ms", bpo::value<uint32_t>()->default_value(30),
              "Maximum time for processing a request.")
             ("verbose-http-errors", bpo::bool_switch()->default_value(false),
@@ -722,6 +756,7 @@ class http_plugin_impl : public std::enable_shared_from_this<http_plugin_impl> {
                      "http-threads ${num} must be greater than 0", ("num", my->thread_pool_size));
 
          my->max_bytes_in_flight = options.at( "http-max-bytes-in-flight-mb" ).as<uint32_t>() * 1024 * 1024;
+         my->max_requests_in_flight = options.at( "http-max-in-flight-requests" ).as<int32_t>();
          my->max_response_time = fc::microseconds( options.at("http-max-response-time-ms").as<uint32_t>() * 1000 );
 
          my->validate_host = options.at("http-validate-host").as<bool>();