Merge pull request #2221 from AntelopeIO/merge-main-02-07-2024

IF: Merge main 02-07-2024
AntelopeIO · Feb 7, 2024 · 47cbaf2 · 47cbaf2
2 parents 55606cb + 841e537
commit 47cbaf2
Show file tree

Hide file tree

Showing 10 changed files with 164 additions and 13 deletions.
diff --git a/libraries/chainbase b/libraries/chainbase
diff --git a/plugins/http_plugin/http_plugin.cpp b/plugins/http_plugin/http_plugin.cpp
@@ -357,9 +357,9 @@ namespace eosio {
             ("max-body-size", bpo::value<uint32_t>()->default_value(my->plugin_state->max_body_size),
              "The maximum body size in bytes allowed for incoming RPC requests")
             ("http-max-bytes-in-flight-mb", bpo::value<int64_t>()->default_value(500),
-             "Maximum size in megabytes http_plugin should use for processing http requests. -1 for unlimited. 429 error response when exceeded." )
+             "Maximum size in megabytes http_plugin should use for processing http requests. -1 for unlimited. 503 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." )
+             "Maximum number of requests http_plugin should use for processing http requests. 503 error response when exceeded." )
             ("http-max-response-time-ms", bpo::value<int64_t>()->default_value(15),
              "Maximum time on main thread for processing a request, -1 for unlimited")
             ("verbose-http-errors", bpo::bool_switch()->default_value(false),

diff --git a/plugins/http_plugin/include/eosio/http_plugin/beast_http_session.hpp b/plugins/http_plugin/include/eosio/http_plugin/beast_http_session.hpp
@@ -219,18 +219,18 @@ class beast_http_session : public detail::abstract_conn,
 
    virtual void send_busy_response(std::string&& what) final {
       error_results::error_info ei;
-      ei.code = static_cast<int64_t>(http::status::too_many_requests);
+      ei.code = static_cast<int64_t>(http::status::service_unavailable);
       ei.name = "Busy";
       ei.what = std::move(what);
-      error_results results{static_cast<uint16_t>(http::status::too_many_requests), "Busy", ei};
+      error_results results{static_cast<uint16_t>(http::status::service_unavailable), "Busy", ei};
       send_response(fc::json::to_string(results, fc::time_point::maximum()),
-                    static_cast<unsigned int>(http::status::too_many_requests) );
+                    static_cast<unsigned int>(http::status::service_unavailable) );
    }
 
    virtual std::string verify_max_bytes_in_flight(size_t extra_bytes) final {
       auto bytes_in_flight_size = plugin_state_->bytes_in_flight.load() + extra_bytes;
       if(bytes_in_flight_size > plugin_state_->max_bytes_in_flight) {
-         fc_dlog(plugin_state_->get_logger(), "429 - too many bytes in flight: ${bytes}", ("bytes", bytes_in_flight_size));
+         fc_dlog(plugin_state_->get_logger(), "503 - too many bytes in flight: ${bytes}", ("bytes", bytes_in_flight_size));
          return "Too many bytes in flight: " + std::to_string( bytes_in_flight_size );
       }
       return {};
@@ -242,7 +242,7 @@ class beast_http_session : public detail::abstract_conn,
 
       auto requests_in_flight_num = plugin_state_->requests_in_flight.load();
       if(requests_in_flight_num > plugin_state_->max_requests_in_flight) {
-         fc_dlog(plugin_state_->get_logger(), "429 - too many requests in flight: ${requests}", ("requests", requests_in_flight_num));
+         fc_dlog(plugin_state_->get_logger(), "503 - too many requests in flight: ${requests}", ("requests", requests_in_flight_num));
          return "Too many requests in flight: " + std::to_string( requests_in_flight_num );
       }
       return {};
@@ -492,8 +492,9 @@ class beast_http_session : public detail::abstract_conn,
 
    void run_session() {
       if(auto error_str = verify_max_requests_in_flight(); !error_str.empty()) {
+         res_->keep_alive(false);
          send_busy_response(std::move(error_str));
-         return do_eof();
+         return;
       }
 
       do_read_header();

diff --git a/plugins/http_plugin/include/eosio/http_plugin/common.hpp b/plugins/http_plugin/include/eosio/http_plugin/common.hpp
@@ -161,13 +161,14 @@ inline auto make_http_response_handler(http_plugin_state& plugin_state, detail::
       // post back to an HTTP thread to allow the response handler to be called from any thread
       boost::asio::dispatch(plugin_state.thread_pool.get_executor(),
                         [&plugin_state, session_ptr{std::move(session_ptr)}, code, payload_size, response = std::move(response), content_type]() {
+                           auto on_exit = fc::scoped_exit<std::function<void()>>([&](){plugin_state.bytes_in_flight -= payload_size;});
+
                            if(auto error_str = session_ptr->verify_max_bytes_in_flight(0); !error_str.empty()) {
                               session_ptr->send_busy_response(std::move(error_str));
                               return;
                            }
 
                            try {
-                              plugin_state.bytes_in_flight -= payload_size;
                               if (response.has_value()) {
                                  std::string json = (content_type == http_content_type::plaintext) ? response->as_string() : fc::json::to_string(*response, fc::time_point::maximum());
                                  if (auto error_str = session_ptr->verify_max_bytes_in_flight(json.size()); error_str.empty())

diff --git a/plugins/http_plugin/tests/CMakeLists.txt b/plugins/http_plugin/tests/CMakeLists.txt
@@ -11,3 +11,4 @@ target_include_directories( http_plugin_unit_tests PUBLIC
                             ${CMAKE_SOURCE_DIR}/plugins/http_plugin/include )
 
 add_test( NAME http_plugin_unit_tests COMMAND http_plugin_unit_tests )
+set_property(TEST http_plugin_unit_tests PROPERTY LABELS nonparallelizable_tests)
diff --git a/plugins/http_plugin/tests/unit_tests.cpp b/plugins/http_plugin/tests/unit_tests.cpp
@@ -10,6 +10,7 @@
 #include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/local/stream_protocol.hpp>
 #include <fc/scoped_exit.hpp>
+#include <fc/crypto/rand.hpp>
 
 #define BOOST_TEST_MODULE http_plugin unit tests
 #include <boost/test/included/unit_test.hpp>
@@ -572,4 +573,145 @@ BOOST_AUTO_TEST_CASE(test_on_loopback) {
    BOOST_CHECK(on_loopback({"test", "--plugin=eosio::http_plugin", "--http-server-address", "localhost:8888"}));
    BOOST_CHECK(!on_loopback({"test", "--plugin=eosio::http_plugin", "--http-server-address", ":8888"}));
    BOOST_CHECK(!on_loopback({"test", "--plugin=eosio::http_plugin", "--http-server-address", "example.com:8888"}));
+}
+
+BOOST_FIXTURE_TEST_CASE(bytes_in_flight, http_plugin_test_fixture) {
+   http_plugin* http_plugin = init({"--plugin=eosio::http_plugin",
+                                    "--http-server-address=127.0.0.1:8888",
+                                    "--http-max-bytes-in-flight-mb=64"});
+   BOOST_REQUIRE(http_plugin);
+
+   http_plugin->add_api({{std::string("/4megabyte"), api_category::node,
+                          [&](string&&, string&& body, url_response_callback&& cb) {
+                             fc::blob b;
+                             b.data.resize(4*1024*1024);
+                             fc::rand_bytes(b.data.data(), b.data.size());
+                             cb(200, b);
+                          }}}, appbase::exec_queue::read_write);
+
+   boost::asio::io_context ctx;
+   boost::asio::ip::tcp::resolver resolver(ctx);
+
+   std::list<boost::asio::ip::tcp::socket> connections;
+
+   auto send_4mb_requests = [&](unsigned count) {
+      for(unsigned i = 0; i < count; ++i) {
+         boost::asio::ip::tcp::socket& s = connections.emplace_back(ctx, boost::asio::ip::tcp::v4());
+         //we can't control http_plugin's send buffer, but at least we can control our receive buffer size to help increase
+         // chance of server blocking
+         s.set_option(boost::asio::socket_base::receive_buffer_size(8*1024));
+         s.connect(resolver.resolve("127.0.0.1", "8888")->endpoint());
+         boost::beast::http::request<boost::beast::http::empty_body> req(boost::beast::http::verb::get, "/4megabyte", 11);
+         req.keep_alive(true);
+         req.set(http::field::host, "127.0.0.1:8888");
+         boost::beast::http::write(s, req);
+      }
+   };
+
+   auto drain_http_replies = [&](unsigned max = std::numeric_limits<unsigned>::max()) {
+      std::unordered_map<boost::beast::http::status, size_t> count_of_status_replies;
+      while(connections.size() && max--) {
+         boost::beast::http::response<boost::beast::http::string_body> resp;
+         boost::beast::flat_buffer buffer;
+         boost::beast::http::read(connections.front(), buffer, resp);
+
+         count_of_status_replies[resp.result()]++;
+
+         connections.erase(connections.begin());
+      }
+      return count_of_status_replies;
+   };
+
+
+   //send a single request to start with
+   send_4mb_requests(1u);
+   std::unordered_map<boost::beast::http::status, size_t> r = drain_http_replies();
+   BOOST_REQUIRE_EQUAL(r[boost::beast::http::status::ok], 1u);
+
+   //load up 32, this should exceed max
+   send_4mb_requests(32u);
+   r = drain_http_replies();
+   BOOST_REQUIRE_GT(r[boost::beast::http::status::ok], 0u);
+   BOOST_REQUIRE_GT(r[boost::beast::http::status::service_unavailable], 0u);
+   BOOST_REQUIRE_EQUAL(r[boost::beast::http::status::service_unavailable] + r[boost::beast::http::status::ok], 32u);
+
+   //send some more requests
+   send_4mb_requests(10u);
+   r = drain_http_replies();
+   BOOST_REQUIRE_EQUAL(r[boost::beast::http::status::ok], 10u);
+
+   //load up some more requests that exceed max
+   send_4mb_requests(32u);
+   //make sure got to the point http threads had responses queued
+   std::this_thread::sleep_for(std::chrono::seconds(1));
+   //now rip these connections out before the responses are completely sent
+   connections.clear();
+   //send some requests that should work still
+   send_4mb_requests(8u);
+   r = drain_http_replies();
+   BOOST_REQUIRE_EQUAL(r[boost::beast::http::status::ok], 8u);
+}
+
+BOOST_FIXTURE_TEST_CASE(requests_in_flight, http_plugin_test_fixture) {
+   http_plugin* http_plugin = init({"--plugin=eosio::http_plugin",
+                                    "--http-server-address=127.0.0.1:8888",
+                                    "--http-max-in-flight-requests=16"});
+   BOOST_REQUIRE(http_plugin);
+
+   http_plugin->add_api({{std::string("/doit"), api_category::node,
+                          [&](string&&, string&& body, url_response_callback&& cb) {
+                             cb(200, "hello");
+                          }}}, appbase::exec_queue::read_write);
+
+   boost::asio::io_context ctx;
+   boost::asio::ip::tcp::resolver resolver(ctx);
+
+   std::list<boost::asio::ip::tcp::socket> connections;
+
+   auto send_requests = [&](unsigned count) {
+      for(unsigned i = 0; i < count; ++i) {
+         boost::asio::ip::tcp::socket& s = connections.emplace_back(ctx, boost::asio::ip::tcp::v4());
+         boost::asio::connect(s, resolver.resolve("127.0.0.1", "8888"));
+         boost::beast::http::request<boost::beast::http::empty_body> req(boost::beast::http::verb::get, "/doit", 11);
+         req.keep_alive(true);
+         req.set(http::field::host, "127.0.0.1:8888");
+         boost::beast::http::write(s, req);
+      }
+   };
+
+   auto scan_http_replies = [&]() {
+      std::unordered_map<boost::beast::http::status, size_t> count_of_status_replies;
+      for(boost::asio::ip::tcp::socket& c : connections) {
+         boost::beast::http::response<boost::beast::http::string_body> resp;
+         boost::beast::flat_buffer buffer;
+         boost::beast::http::read(c, buffer, resp);
+
+         count_of_status_replies[resp.result()]++;
+
+         if(resp.result() == boost::beast::http::status::ok)
+            BOOST_REQUIRE(resp.keep_alive());
+      }
+      return count_of_status_replies;
+   };
+
+
+   //8 requests to start with
+   send_requests(8u);
+   std::unordered_map<boost::beast::http::status, size_t> r = scan_http_replies();
+   BOOST_REQUIRE_EQUAL(r[boost::beast::http::status::ok], 8u);
+   connections.clear();
+
+   //24 requests will exceed threshold
+   send_requests(24u);
+   r = scan_http_replies();
+   BOOST_REQUIRE_GT(r[boost::beast::http::status::ok], 0u);
+   BOOST_REQUIRE_GT(r[boost::beast::http::status::service_unavailable], 0u);
+   BOOST_REQUIRE_EQUAL(r[boost::beast::http::status::service_unavailable] + r[boost::beast::http::status::ok], 24u);
+   connections.clear();
+
+   //requests should still work
+   send_requests(8u);
+   r = scan_http_replies();
+   BOOST_REQUIRE_EQUAL(r[boost::beast::http::status::ok], 8u);
+   connections.clear();
 }
diff --git a/plugins/producer_plugin/producer_plugin.cpp b/plugins/producer_plugin/producer_plugin.cpp
@@ -591,6 +591,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
    // threads use 11TB (128 * 11 * 8GB)). It is about 11.7% of total VM space
    // in a 64-bit Linux machine (about 128TB).
    static constexpr uint32_t         _ro_max_threads_allowed{128};
+   static constexpr uint32_t         _ro_default_threads_nonproducer{3};
    named_thread_pool<struct read>    _ro_thread_pool;
    fc::microseconds                  _ro_write_window_time_us{200000};
    fc::microseconds                  _ro_read_window_time_us{60000};
@@ -1031,6 +1032,8 @@ void producer_plugin::set_program_options(
    boost::program_options::options_description& command_line_options,
    boost::program_options::options_description& config_file_options)
 {
+   using namespace std::string_literals;
+
    auto default_priv_key = private_key_type::regenerate<fc::ecc::private_key_shim>(fc::sha256::hash(std::string("nathan")));
    auto private_key_default = std::make_pair(default_priv_key.get_public_key(), default_priv_key );
 
@@ -1079,7 +1082,7 @@ void producer_plugin::set_program_options(
          ("snapshots-dir", bpo::value<std::filesystem::path>()->default_value("snapshots"),
           "the location of the snapshots directory (absolute path or relative to application data dir)")
          ("read-only-threads", bpo::value<uint32_t>(),
-          "Number of worker threads in read-only execution thread pool. Max 8.")
+         ("Number of worker threads in read-only execution thread pool. Defaults to 0 if configured as producer, otherwise defaults to "s + std::to_string(producer_plugin_impl::_ro_default_threads_nonproducer) + ". Max "s + std::to_string(producer_plugin_impl::_ro_max_threads_allowed) + "."s).c_str())
          ("read-only-write-window-time-us", bpo::value<uint32_t>()->default_value(my->_ro_write_window_time_us.count()),
           "Time in microseconds the write window lasts.")
          ("read-only-read-window-time-us", bpo::value<uint32_t>()->default_value(my->_ro_read_window_time_us.count()),
@@ -1235,7 +1238,7 @@ void producer_plugin_impl::plugin_initialize(const boost::program_options::varia
          auto i = std::find_if(v.cbegin(), v.cend(), [](const std::string& p) { return p.find("eosio::chain_api_plugin") != std::string::npos; });
          if (i != v.cend()) {
             // default to 3 threads for non producer nodes running chain_api_plugin if not specified
-            _ro_thread_pool_size = 3;
+            _ro_thread_pool_size = _ro_default_threads_nonproducer;
             ilog("chain_api_plugin configured, defaulting read-only-threads to ${t}", ("t", _ro_thread_pool_size));
          }
       }

diff --git a/plugins/state_history_plugin/tests/CMakeLists.txt b/plugins/state_history_plugin/tests/CMakeLists.txt
@@ -2,4 +2,5 @@ add_executable( test_state_history main.cpp session_test.cpp plugin_config_test.
 target_link_libraries(test_state_history state_history_plugin eosio_testing eosio_chain_wrap)
 target_include_directories( test_state_history PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/../include" )
 
-add_test(test_state_history test_state_history)
+add_test(test_state_history test_state_history)
+set_property(TEST test_state_history PROPERTY LABELS nonparallelizable_tests)
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
@@ -16,6 +16,7 @@ target_include_directories( plugin_test PUBLIC
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/p2p_tests/dawn_515/test.sh ${CMAKE_CURRENT_BINARY_DIR}/p2p_tests/dawn_515/test.sh COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/block_log_util_test.py ${CMAKE_CURRENT_BINARY_DIR}/block_log_util_test.py COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/block_log_retain_blocks_test.py ${CMAKE_CURRENT_BINARY_DIR}/block_log_retain_blocks_test.py COPYONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/bridge_for_fork_test_shape.json ${CMAKE_CURRENT_BINARY_DIR}/bridge_for_fork_test_shape.json COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/cluster_launcher.py ${CMAKE_CURRENT_BINARY_DIR}/cluster_launcher.py COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/distributed-transactions-test.py ${CMAKE_CURRENT_BINARY_DIR}/distributed-transactions-test.py COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/leap_util_bls_test.py ${CMAKE_CURRENT_BINARY_DIR}/leap_util_bls_test.py COPYONLY)

diff --git a/tests/trx_generator/CMakeLists.txt b/tests/trx_generator/CMakeLists.txt
@@ -8,3 +8,4 @@ add_executable(trx_generator_tests trx_generator_tests.cpp trx_provider.cpp trx_
 target_link_libraries(trx_generator_tests PRIVATE eosio_chain fc Boost::program_options ${CMAKE_DL_LIBS} ${PLATFORM_SPECIFIC_LIBS})
 target_include_directories(trx_generator_tests PUBLIC ${CMAKE_CURRENT_BINARY_DIR} ${CMAKE_CURRENT_SOURCE_DIR})
 add_test(trx_generator_tests trx_generator_tests)
+set_property(TEST trx_generator_tests PROPERTY LABELS nonparallelizable_tests)