Change default max-transaction-time

docs/01_nodeos/03_plugins/producer_plugin/index.md

@@ -27,10 +27,10 @@ Config Options for eosio::producer_plugin:
                                         chain is stale.
   -x [ --pause-on-startup ]             Start this node in a state where
                                         production is paused
-  --max-transaction-time arg (=30)      Limits the maximum time (in
-                                        milliseconds) that is allowed a pushed
-                                        transaction's code to execute before
-                                        being considered invalid
+  --max-transaction-time arg (=499)     Locally lowers the max_transaction_cpu_
+                                        usage limit (in milliseconds) that an
+                                        input transaction is allowed to execute
+                                        before being considered invalid
   --max-irreversible-block-age arg (=-1)
                                         Limits the maximum age (in seconds) of
                                         the DPOS Irreversible Block for a chain

libraries/custom_appbase/include/eosio/chain/application.hpp

@@ -3,6 +3,7 @@
 #include <appbase/application_base.hpp>
 #include <eosio/chain/exec_pri_queue.hpp>
 #include <chrono>
+#include <optional>
 #include <mutex>
 
 /*
@@ -23,15 +24,21 @@ enum class exec_window {
 
 enum class exec_queue {
    read_only,          // the queue storing tasks which are safe to execute
-                       // in parallel with other read-only tasks in the read-only
+                       // in parallel with other read-only & read_exclusive tasks in the read-only
                        // thread pool as well as on the main app thread.
                        // Multi-thread safe as long as nothing is executed from the read_write queue.
-   read_write          // the queue storing tasks which can be only executed
+   read_write,         // the queue storing tasks which can be only executed
                        // on the app thread while read-only tasks are
                        // not being executed in read-only threads. Single threaded.
+   read_exclusive      // the queue storing tasks which should only be executed
+                       // in parallel with other read_exclusive or read_only tasks in the
+                       // read-only thread pool. Should never be executed on the main thread.
+                       // If no read-only thread pool is available this queue grows unbounded
+                       // as tasks will never execute. User is responsible for not queueing
+                       // read_exclusive tasks if no read-only thread pool is available.
 };
 
-class two_queue_executor {
+class three_queue_executor {
 public:
 
    // Trade off on returning to appbase exec() loop as the overhead of poll/run can be measurable for small running tasks.
@@ -42,8 +49,10 @@ class two_queue_executor {
    auto post( int priority, exec_queue q, Func&& func ) {
       if ( q == exec_queue::read_write )
          return boost::asio::post(io_serv_, read_write_queue_.wrap(priority, --order_, std::forward<Func>(func)));
-      else
+      else if ( q == exec_queue::read_only )
          return boost::asio::post( io_serv_, read_only_queue_.wrap( priority, --order_, std::forward<Func>( func)));
+      else
+         return boost::asio::post( io_serv_, read_exclusive_queue_.wrap( priority, --order_, std::forward<Func>( func)));
    }
 
    // Legacy and deprecated. To be removed after cleaning up its uses in base appbase
@@ -56,6 +65,7 @@ class two_queue_executor {
 
    boost::asio::io_service& get_io_service() { return io_serv_; }
 
+   // called from main thread
    bool execute_highest() {
       // execute for at least minimum runtime
       const auto end = std::chrono::high_resolution_clock::now() + std::chrono::milliseconds(minimum_runtime_ms);
@@ -81,13 +91,20 @@ class two_queue_executor {
       return more;
    }
 
-   bool execute_highest_read_only() {
+   bool execute_highest_read() {
       // execute for at least minimum runtime
       const auto end = std::chrono::high_resolution_clock::now() + std::chrono::milliseconds(minimum_runtime_ms);
 
       bool more = false;
       while (true) {
-         more = read_only_queue_.execute_highest_locked( true );
+         std::optional<bool> exec_read_only_queue = read_only_queue_.compare_queues_locked(read_exclusive_queue_);
+         if (!exec_read_only_queue) {
+            more = read_only_queue_.execute_highest_locked( true );
+         } else if (*exec_read_only_queue) {
+            more = read_only_queue_.execute_highest_locked( false );
+         } else {
+            more = read_exclusive_queue_.execute_highest_locked( false );
+         }
          if (!more || std::chrono::high_resolution_clock::now() > end)
             break;
       }
@@ -102,20 +119,29 @@ class two_queue_executor {
       else
          return read_only_queue_.wrap( priority, --order_, std::forward<Function>( func));
    }
+
+   void stop() {
+      read_only_queue_.stop();
+      read_write_queue_.stop();
+      read_exclusive_queue_.stop();
+   }
 
    void clear() {
       read_only_queue_.clear();
       read_write_queue_.clear();
+      read_exclusive_queue_.clear();
    }
 
    void set_to_read_window(uint32_t num_threads, std::function<bool()> should_exit) {
       exec_window_ = exec_window::read;
-      read_only_queue_.enable_locking(num_threads, std::move(should_exit));
+      read_only_queue_.enable_locking(num_threads, should_exit);
+      read_exclusive_queue_.enable_locking(num_threads, std::move(should_exit));
    }
 
    void set_to_write_window() {
       exec_window_ = exec_window::write;
       read_only_queue_.disable_locking();
+      read_exclusive_queue_.disable_locking();
    }
 
    bool is_read_window() const {
@@ -127,19 +153,20 @@ class two_queue_executor {
    }
 
    auto& read_only_queue() { return read_only_queue_; }
-
    auto& read_write_queue() { return read_write_queue_; }
+   auto& read_exclusive_queue() { return read_exclusive_queue_; }
 
    // members are ordered taking into account that the last one is destructed first
 private:
    boost::asio::io_service            io_serv_;
    appbase::exec_pri_queue            read_only_queue_;
    appbase::exec_pri_queue            read_write_queue_;
+   appbase::exec_pri_queue            read_exclusive_queue_;
    std::atomic<std::size_t>           order_ { std::numeric_limits<size_t>::max() }; // to maintain FIFO ordering in both queues within priority
    exec_window                        exec_window_ { exec_window::write };
 };
 
-using application = application_t<two_queue_executor>;
+using application = application_t<three_queue_executor>;
 }
 
 #include <appbase/application_instance.hpp>

libraries/custom_appbase/include/eosio/chain/exec_pri_queue.hpp

@@ -3,6 +3,7 @@
 
 #include <condition_variable>
 #include <mutex>
+#include <optional>
 #include <queue>
 
 namespace appbase {
@@ -13,6 +14,12 @@ class exec_pri_queue : public boost::asio::execution_context
 {
 public:
 
+   void stop() {
+      std::lock_guard g( mtx_ );
+      exiting_blocking_ = true;
+      cond_.notify_all();
+   }
+
    void enable_locking(uint32_t num_threads, std::function<bool()> should_exit) {
       assert(num_threads > 0 && num_waiting_ == 0);
       lock_enabled_ = true;
@@ -105,6 +112,14 @@ class exec_pri_queue : public boost::asio::execution_context
    // Only call when locking disabled
    const auto& top() const { return handlers_.top(); }
 
+   // return empty optional if both queues empty.
+   //        true if this queue has the highest priority task to execute
+   std::optional<bool> compare_queues_locked( const exec_pri_queue& rhs ) {
+      std::scoped_lock g(mtx_, rhs.mtx_);
+      if (empty() && rhs.empty()) return {};
+      return !empty() && (rhs.empty() || *rhs.top() < *top());
+   }
+
    class executor
    {
    public:
@@ -220,7 +235,7 @@ class exec_pri_queue : public boost::asio::execution_context
    };
 
    bool lock_enabled_ = false;
-   std::mutex mtx_;
+   mutable std::mutex mtx_;
    std::condition_variable cond_;
    uint32_t num_waiting_{0};
    uint32_t max_waiting_{0};

plugins/producer_plugin/producer_plugin.cpp

@@ -807,9 +807,12 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
                                       bool                                 return_failure_traces,
                                       next_function<transaction_trace_ptr> next) {
       if (trx_type == transaction_metadata::trx_type::read_only) {
+         EOS_ASSERT( _ro_thread_pool_size > 0, unsupported_feature,
+                     "read-only transactions execution not enabled on API node. Set read-only-threads > 0" );
+
          // Post all read only trxs to read_only queue for execution.
          auto trx_metadata = transaction_metadata::create_no_recover_keys(trx, transaction_metadata::trx_type::read_only);
-         app().executor().post(priority::low, exec_queue::read_only, [this, trx{std::move(trx_metadata)}, next{std::move(next)}]() mutable {
+         app().executor().post(priority::low, exec_queue::read_exclusive, [this, trx{std::move(trx_metadata)}, next{std::move(next)}]() mutable {
             push_read_only_transaction(std::move(trx), std::move(next));
          });
          return;
@@ -1047,8 +1050,8 @@ void producer_plugin::set_program_options(
    producer_options.add_options()
          ("enable-stale-production,e", boost::program_options::bool_switch()->notifier([this](bool e){my->_production_enabled = e;}), "Enable block production, even if the chain is stale.")
          ("pause-on-startup,x", boost::program_options::bool_switch()->notifier([this](bool p){my->_pause_production = p;}), "Start this node in a state where production is paused")
-         ("max-transaction-time", bpo::value<int32_t>()->default_value(30),
-          "Limits the maximum time (in milliseconds) that is allowed a pushed transaction's code to execute before being considered invalid")
+         ("max-transaction-time", bpo::value<int32_t>()->default_value(config::block_interval_ms-1),
+          "Locally lowers the max_transaction_cpu_usage limit (in milliseconds) that an input transaction is allowed to execute before being considered invalid")
          ("max-irreversible-block-age", bpo::value<int32_t>()->default_value( -1 ),
           "Limits the maximum age (in seconds) of the DPOS Irreversible Block for a chain this node will produce blocks on (use negative value to indicate unlimited)")
          ("producer-name,p", boost::program_options::value<vector<string>>()->composing()->multitoken(),
@@ -1270,28 +1273,25 @@ void producer_plugin_impl::plugin_initialize(const boost::program_options::varia
                  ("read", _ro_read_window_time_us)("min", _ro_read_window_minimum_time_us));
       _ro_read_window_effective_time_us = _ro_read_window_time_us - _ro_read_window_minimum_time_us;
 
-      // Make sure a read-only transaction can finish within the read
-      // window if scheduled at the very beginning of the window.
-      // Add _ro_read_window_minimum_time_us for safety margin.
-      if (_max_transaction_time_ms.load() > 0) {
-         EOS_ASSERT(
-            _ro_read_window_time_us > (fc::milliseconds(_max_transaction_time_ms.load()) + _ro_read_window_minimum_time_us),
-            plugin_config_exception,
-            "read-only-read-window-time-us (${read} us) must be greater than max-transaction-time (${trx_time} us) "
-            "plus ${min} us, required: ${read} us > (${trx_time} us + ${min} us).",
-            ("read", _ro_read_window_time_us)("trx_time", _max_transaction_time_ms.load() * 1000)("min", _ro_read_window_minimum_time_us));
-      }
       ilog("read-only-write-window-time-us: ${ww} us, read-only-read-window-time-us: ${rw} us, effective read window time to be used: ${w} us",
            ("ww", _ro_write_window_time_us)("rw", _ro_read_window_time_us)("w", _ro_read_window_effective_time_us));
    }
 
-   // Make sure _ro_max_trx_time_us is alwasys set.
+   // Make sure _ro_max_trx_time_us is always set.
+   // Make sure a read-only transaction can finish within the read
+   // window if scheduled at the very beginning of the window.
+   // Add _ro_read_window_minimum_time_us for safety margin.
    if (_max_transaction_time_ms.load() > 0) {
       _ro_max_trx_time_us = fc::milliseconds(_max_transaction_time_ms.load());
    } else {
       // max-transaction-time can be set to negative for unlimited time
       _ro_max_trx_time_us = fc::microseconds::maximum();
    }
+   if (_ro_max_trx_time_us > _ro_read_window_effective_time_us) {
+      _ro_max_trx_time_us = _ro_read_window_effective_time_us;
+   }
+   ilog("Read-only max transaction time ${rot}us set to fit in the effective read-only window ${row}us.",
+        ("rot", _ro_max_trx_time_us)("row", _ro_read_window_effective_time_us));
    ilog("read-only-threads ${s}, max read-only trx time to be enforced: ${t} us", ("s", _ro_thread_pool_size)("t", _ro_max_trx_time_us));
 
    _incoming_block_sync_provider = app().get_method<incoming::methods::block_sync>().register_provider(
@@ -1427,6 +1427,10 @@ void producer_plugin::plugin_startup() {
 void producer_plugin_impl::plugin_shutdown() {
    boost::system::error_code ec;
    _timer.cancel(ec);
+   boost::system::error_code ro_ec;
+   _ro_timer.cancel(ro_ec);
+   app().executor().stop();
+   _ro_thread_pool.stop();
    _thread_pool.stop();
    _unapplied_transactions.clear();
 
@@ -2841,7 +2845,7 @@ void producer_plugin_impl::switch_to_read_window() {
    _time_tracker.pause();
 
    // we are in write window, so no read-only trx threads are processing transactions.
-   if (app().executor().read_only_queue().empty()) { // no read-only tasks to process. stay in write window
+   if (app().executor().read_only_queue().empty() && app().executor().read_exclusive_queue().empty()) { // no read-only tasks to process. stay in write window
       start_write_window();                          // restart write window timer for next round
       return;
    }
@@ -2892,7 +2896,7 @@ bool producer_plugin_impl::read_only_execution_task(uint32_t pending_block_num)
    // 2. net_plugin receives a block
    // 3. no read-only tasks to execute
    while (fc::time_point::now() < _ro_window_deadline && _received_block < pending_block_num) {
-      bool more = app().executor().execute_highest_read_only(); // blocks until all read only threads are idle
+      bool more = app().executor().execute_highest_read(); // blocks until all read only threads are idle
       if (!more) {
          break;
       }
@@ -2909,7 +2913,7 @@ bool producer_plugin_impl::read_only_execution_task(uint32_t pending_block_num)
       // last thread post any exhausted back into read_only queue with slightly higher priority (low+1) so they are executed first
       ro_trx_t t;
       while (_ro_exhausted_trx_queue.pop_front(t)) {
-         app().executor().post(priority::low + 1, exec_queue::read_only, [this, trx{std::move(t.trx)}, next{std::move(t.next)}]() mutable {
+         app().executor().post(priority::low + 1, exec_queue::read_exclusive, [this, trx{std::move(t.trx)}, next{std::move(t.next)}]() mutable {
             push_read_only_transaction(std::move(trx), std::move(next));
          });
       }
@@ -2927,7 +2931,7 @@ void producer_plugin_impl::repost_exhausted_transactions(const fc::time_point& d
       // post any exhausted back into read_only queue with slightly higher priority (low+1) so they are executed first
       ro_trx_t t;
       while (!should_interrupt_start_block(deadline, pending_block_num) && _ro_exhausted_trx_queue.pop_front(t)) {
-         app().executor().post(priority::low + 1, exec_queue::read_only, [this, trx{std::move(t.trx)}, next{std::move(t.next)}]() mutable {
+         app().executor().post(priority::low + 1, exec_queue::read_exclusive, [this, trx{std::move(t.trx)}, next{std::move(t.next)}]() mutable {
             push_read_only_transaction(std::move(trx), std::move(next));
          });
       }
@@ -2947,21 +2951,10 @@ bool producer_plugin_impl::push_read_only_transaction(transaction_metadata_ptr t
          return true;
       }
 
-      // When executing a read-only trx on the main thread while in the write window,
-      // need to switch db mode to read only.
-      auto db_read_only_mode_guard = fc::make_scoped_exit([&] {
-         if (chain.is_write_window())
-            chain.unset_db_read_only_mode();
-      });
+      assert(!chain.is_write_window());
 
-      std::optional<block_time_tracker::trx_time_tracker> trx_tracker;
-      if ( chain.is_write_window() ) {
-         chain.set_db_read_only_mode();
-         trx_tracker.emplace(_time_tracker.start_trx(true, start));
-      }
-
-      // use read-window/write-window deadline if there are read/write windows, otherwise use block_deadline if only the app thead
-      auto window_deadline = (_ro_thread_pool_size != 0) ? _ro_window_deadline : _pending_block_deadline;
+      // use read-window/write-window deadline
+      auto window_deadline = _ro_window_deadline;
 
       // Ensure the trx to finish by the end of read-window or write-window or block_deadline depending on
       auto trace = chain.push_transaction(trx, window_deadline, _ro_max_trx_time_us, 0, false, 0);
@@ -2979,9 +2972,6 @@ bool producer_plugin_impl::push_read_only_transaction(transaction_metadata_ptr t
          _ro_exhausted_trx_queue.push_front({std::move(trx), std::move(next)});
       }
 
-      if ( chain.is_write_window() && !pr.failed ) {
-         trx_tracker->trx_success();
-      }
    } catch (const guard_exception& e) {
       chain_plugin::handle_guard_exception(e);
    } catch (boost::interprocess::bad_alloc&) {

tests/CMakeLists.txt

@@ -131,11 +131,11 @@ add_test(NAME nodeos_protocol_feature_test COMMAND tests/nodeos_protocol_feature
 set_property(TEST nodeos_protocol_feature_test PROPERTY LABELS nonparallelizable_tests)
 add_test(NAME compute_transaction_test COMMAND tests/compute_transaction_test.py -v -p 2 -n 3 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST compute_transaction_test PROPERTY LABELS nonparallelizable_tests)
-add_test(NAME read-only-trx-basic-test COMMAND tests/read_only_trx_test.py -v -p 2 -n 3 --read-only-threads 0 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+add_test(NAME read-only-trx-basic-test COMMAND tests/read_only_trx_test.py -v -p 2 -n 3 --read-only-threads 1 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST read-only-trx-basic-test PROPERTY LABELS nonparallelizable_tests)
-add_test(NAME read-only-trx-parallel-test COMMAND tests/read_only_trx_test.py -v -p 2 -n 3 --read-only-threads 128 --num-test-runs 3 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+add_test(NAME read-only-trx-parallel-test COMMAND tests/read_only_trx_test.py -v -p 2 -n 3 --read-only-threads 16 --num-test-runs 3 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST read-only-trx-parallel-test PROPERTY LABELS nonparallelizable_tests)
-add_test(NAME read-only-trx-parallel-eos-vm-oc-test COMMAND tests/read_only_trx_test.py -v -p 2 -n 3 --eos-vm-oc-enable all --read-only-threads 128 --num-test-runs 3 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+add_test(NAME read-only-trx-parallel-eos-vm-oc-test COMMAND tests/read_only_trx_test.py -v -p 2 -n 3 --eos-vm-oc-enable all --read-only-threads 16 --num-test-runs 3 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST read-only-trx-parallel-eos-vm-oc-test PROPERTY LABELS nonparallelizable_tests)
 add_test(NAME read-only-trx-parallel-no-oc-test COMMAND tests/read_only_trx_test.py -v -p 2 -n 3 --eos-vm-oc-enable none --read-only-threads 6 --num-test-runs 2 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST read-only-trx-parallel-no-oc-test PROPERTY LABELS nonparallelizable_tests)