Fix merge issue

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

@@ -1,30 +1,33 @@
 #pragma once
 
 #include <appbase/application_base.hpp>
-#include <appbase/execution_priority_queue.hpp>
+#include <eosio/chain/exec_pri_queue.hpp>
+#include <mutex>
 
 /*
- * Custmomize appbase to support two-queue exector.
+ * Customize appbase to support two-queue executor.
 */
 namespace appbase { 
 
 enum class exec_window {
-   read,  // the window during which operations from read_only_trx_safe queue
-          // can be executed in app thread in parallel with read-only transactions
-          // in read-only transaction excuting threads.
-   write, // the window during which operations from both general and
-          // read_only_trx_safe queues can be executed in app thread,
-          // while read-only transactions are not executed in read-only
-          // transaction excuting threads.
+   read,  // the window during which operations from parallel queue
+          // can be executed in parallel in the read-only thread pool
+          // as well as in the app thread.
+   write, // the window during which operations from both read_write and
+          // parallel queues can be executed in app thread,
+          // while read-only operations are not executed in read-only
+          // thread pool. The read-only thread pool is not active; only
+          // the main app thread is active.
 };
 
 enum class exec_queue {
-   read_only_trx_safe, // the queue storing operations which are safe to execute
-                       // on app thread in parallel with read-only transactions
-                       // in read-only transaction excuting threads.
-   general             // the queue storing operations which can be only executed
-                       // on the app thread while read-only transactions are
-                       // not being executed in read-only threads
+   read_only,          // the queue storing tasks which are safe to execute
+                       // in parallel with other read-only 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
+                       // on the app thread while read-only tasks are
+                       // not being executed in read-only threads. Single threaded.
 };
 
 class two_queue_executor {
@@ -32,56 +35,64 @@ public:
 
    template <typename Func>
    auto post( int priority, exec_queue q, Func&& func ) {
-      if ( q == exec_queue::general )
-         return boost::asio::post(io_serv_, general_queue_.wrap(priority, --order_, std::forward<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
-         return boost::asio::post(io_serv_, read_only_trx_safe_queue_.wrap(priority, --order_, std::forward<Func>(func)));
+         return boost::asio::post( io_serv_, read_only_queue_.wrap( priority, --order_, std::forward<Func>( func)));
    }
 
    // Legacy and deprecated. To be removed after cleaning up its uses in base appbase
    template <typename Func>
    auto post( int priority, Func&& func ) {
-      // safer to use general queue for unknown type of operation since operations
-      // from general queue are not executed in parallel with read-only transactions
-      return boost::asio::post(io_serv_, general_queue_.wrap(priority, --order_, std::forward<Func>(func)));
+      // safer to use read_write queue for unknown type of operation since operations
+      // from read_write queue are not executed in parallel with read-only operations
+      return boost::asio::post(io_serv_, read_write_queue_.wrap(priority, --order_, std::forward<Func>(func)));
    }
 
    boost::asio::io_service& get_io_service() { return io_serv_; }
 
    bool execute_highest() {
       if ( exec_window_ == exec_window::write ) {
-         if( !general_queue_.empty() && ( read_only_trx_safe_queue_.empty() || *read_only_trx_safe_queue_.top() < *general_queue_.top()) )  {
-            // general_queue_'s top function's priority greater than read_only_trx_safe_queue_'s top function's, or general_queue_ empty
-            general_queue_.execute_highest();
-         } else if( !read_only_trx_safe_queue_.empty() ) {
-            read_only_trx_safe_queue_.execute_highest();
+         // During write window only main thread is accessing anything in two_queue_executor, no locking required
+         if( !read_write_queue_.empty() && (read_only_queue_.empty() || *read_only_queue_.top() < *read_write_queue_.top()) )  {
+            // read_write_queue_'s top function's priority greater than read_only_queue_'s top function's, or read_only_queue_ empty
+            read_write_queue_.execute_highest();
+         } else if( !read_only_queue_.empty() ) {
+            read_only_queue_.execute_highest();
          }
-         return !read_only_trx_safe_queue_.empty() || !general_queue_.empty();
+         return !read_only_queue_.empty() || !read_write_queue_.empty();
       } else {
-         return read_only_trx_safe_queue_.execute_highest();
+         // When in read window, multiple threads including main app thread are accessing two_queue_executor, locking required
+         return read_only_queue_.execute_highest_locked();
       }
    }
 
+   bool execute_highest_read_only() {
+      return read_only_queue_.execute_highest_locked();
+   }
+
    template <typename Function>
-   boost::asio::executor_binder<Function, appbase::execution_priority_queue::executor>
+   boost::asio::executor_binder<Function, appbase::exec_pri_queue::executor>
    wrap(int priority, exec_queue q, Function&& func ) {
-      if ( q == exec_queue::general )
-         return general_queue_.wrap(priority, --order_, std::forward<Function>(func));
+      if ( q == exec_queue::read_write )
+         return read_write_queue_.wrap(priority, --order_, std::forward<Function>(func));
       else
-         return read_only_trx_safe_queue_.wrap(priority, --order_, std::forward<Function>(func));
+         return read_only_queue_.wrap( priority, --order_, std::forward<Function>( func));
    }
      
    void clear() {
-      read_only_trx_safe_queue_.clear();
-      general_queue_.clear();
+      read_only_queue_.clear();
+      read_write_queue_.clear();
    }
 
    void set_to_read_window() {
       exec_window_ = exec_window::read;
+      read_only_queue_.enable_locking();
    }
 
    void set_to_write_window() {
       exec_window_ = exec_window::write;
+      read_only_queue_.disable_locking();
    }
 
    bool is_read_window() const {
@@ -92,15 +103,15 @@ public:
       return exec_window_ == exec_window::write;
    }
 
-   auto& read_only_trx_safe_queue() { return read_only_trx_safe_queue_; }
+   auto& read_only_queue() { return read_only_queue_; }
 
-   auto& general_queue() { return general_queue_; }
+   auto& read_write_queue() { return read_write_queue_; }
 
    // members are ordered taking into account that the last one is destructed first
 private:
    boost::asio::io_service            io_serv_;
-   appbase::execution_priority_queue  read_only_trx_safe_queue_;
-   appbase::execution_priority_queue  general_queue_;
+   appbase::exec_pri_queue            read_only_queue_;
+   appbase::exec_pri_queue            read_write_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 };
 };

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

@@ -0,0 +1,203 @@
+#pragma once
+#include <boost/asio.hpp>
+
+#include <mutex>
+#include <queue>
+
+namespace appbase {
+// adapted from: https://www.boost.org/doc/libs/1_69_0/doc/html/boost_asio/example/cpp11/invocation/prioritised_handlers.cpp
+
+// Locking has to be coordinated by caller, use with care.
+class exec_pri_queue : public boost::asio::execution_context
+{
+public:
+
+   void enable_locking() {
+      lock_enabled_ = true;
+   }
+
+   void disable_locking() {
+      lock_enabled_ = false;
+   }
+
+   // called from appbase::application_base::exec poll_one() or run_one()
+   template <typename Function>
+   void add(int priority, size_t order, Function function)
+   {
+      std::unique_ptr<queued_handler_base> handler(new queued_handler<Function>(priority, order, std::move(function)));
+      if (lock_enabled_) {
+         std::lock_guard g( mtx_ );
+         handlers_.push( std::move( handler ) );
+      } else {
+         handlers_.push( std::move( handler ) );
+      }
+   }
+
+   // only call when no lock required
+   void clear()
+   {
+      handlers_ = prio_queue();
+   }
+
+   // only call when no lock required
+   bool execute_highest()
+   {
+      if( !handlers_.empty() ) {
+         handlers_.top()->execute();
+         handlers_.pop();
+      }
+
+      return !handlers_.empty();
+   }
+
+private:
+   // has to be defined before use, auto return type
+   auto pop() {
+      auto t = std::move(const_cast<std::unique_ptr<queued_handler_base>&>(handlers_.top()));
+      handlers_.pop();
+      return t;
+   }
+
+public:
+
+   bool execute_highest_locked() {
+      std::unique_lock g(mtx_);
+      if( handlers_.empty() )
+         return false;
+      auto t = pop();
+      g.unlock();
+      t->execute();
+      g.lock();
+      return !handlers_.empty();
+   }
+
+   // Only call when locking disabled
+   size_t size() const { return handlers_.size(); }
+
+   // Only call when locking disabled
+   bool empty() const { return handlers_.empty(); }
+
+   // Only call when locking disabled
+   const auto& top() const { return handlers_.top(); }
+
+   class executor
+   {
+   public:
+      executor(exec_pri_queue& q, int p, size_t o)
+            : context_(q), priority_(p), order_(o)
+      {
+      }
+
+      exec_pri_queue& context() const noexcept
+      {
+         return context_;
+      }
+
+      template <typename Function, typename Allocator>
+      void dispatch(Function f, const Allocator&) const
+      {
+         context_.add(priority_, order_, std::move(f));
+      }
+
+      template <typename Function, typename Allocator>
+      void post(Function f, const Allocator&) const
+      {
+         context_.add(priority_, order_, std::move(f));
+      }
+
+      template <typename Function, typename Allocator>
+      void defer(Function f, const Allocator&) const
+      {
+         context_.add(priority_, order_, std::move(f));
+      }
+
+      void on_work_started() const noexcept {}
+      void on_work_finished() const noexcept {}
+
+      bool operator==(const executor& other) const noexcept
+      {
+         return order_ == other.order_ && &context_ == &other.context_ && priority_ == other.priority_;
+      }
+
+      bool operator!=(const executor& other) const noexcept
+      {
+         return !operator==(other);
+      }
+
+   private:
+      exec_pri_queue& context_;
+      int priority_;
+      size_t order_;
+   };
+
+   template <typename Function>
+   boost::asio::executor_binder<Function, executor>
+   wrap(int priority, size_t order, Function&& func)
+   {
+      return boost::asio::bind_executor( executor(*this, priority, order), std::forward<Function>(func) );
+   }
+
+private:
+   class queued_handler_base
+   {
+   public:
+      queued_handler_base( int p, size_t order )
+            : priority_( p )
+            , order_( order )
+      {
+      }
+
+      virtual ~queued_handler_base() = default;
+
+      virtual void execute() = 0;
+
+      int priority() const { return priority_; }
+      // C++20
+      // friend std::weak_ordering operator<=>(const queued_handler_base&,
+      //                                       const queued_handler_base&) noexcept = default;
+      friend bool operator<(const queued_handler_base& a,
+                            const queued_handler_base& b) noexcept
+      {
+         return std::tie( a.priority_, a.order_ ) < std::tie( b.priority_, b.order_ );
+      }
+
+   private:
+      int priority_;
+      size_t order_;
+   };
+
+   template <typename Function>
+   class queued_handler : public queued_handler_base
+   {
+   public:
+      queued_handler(int p, size_t order, Function f)
+            : queued_handler_base( p, order )
+            , function_( std::move(f) )
+      {
+      }
+
+      void execute() override
+      {
+         function_();
+      }
+
+   private:
+      Function function_;
+   };
+
+   struct deref_less
+   {
+      template<typename Pointer>
+      bool operator()(const Pointer& a, const Pointer& b) noexcept(noexcept(*a < *b))
+      {
+         return *a < *b;
+      }
+   };
+
+   bool lock_enabled_ = false;
+   std::mutex mtx_;
+   using prio_queue = std::priority_queue<std::unique_ptr<queued_handler_base>, std::deque<std::unique_ptr<queued_handler_base>>, deref_less>;
+   prio_queue handlers_;
+};
+
+} // appbase

libraries/custom_appbase/tests/custom_appbase_tests.cpp

@@ -27,28 +27,28 @@ BOOST_AUTO_TEST_CASE( default_exec_window ) {
    // post functions
    std::map<int, int> rslts {};
    int seq_num = 0;
-   app->executor().post( priority::medium, exec_queue::read_only_trx_safe,  [&]() { rslts[0]=seq_num; ++seq_num; } );
-   app->executor().post( priority::medium, exec_queue::general, [&]() { rslts[1]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[2]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::read_only_trx_safe,  [&]() { rslts[3]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::read_only_trx_safe,  [&]() { rslts[4]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::general, [&]() { rslts[5]=seq_num; ++seq_num; } );
-   app->executor().post( priority::highest,exec_queue::read_only_trx_safe,  [&]() { rslts[6]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[7]=seq_num; ++seq_num; } );
+   app->executor().post( priority::medium, exec_queue::read_only,  [&]() { rslts[0]=seq_num; ++seq_num; } );
+   app->executor().post( priority::medium, exec_queue::read_write, [&]() { rslts[1]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[2]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_only,  [&]() { rslts[3]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_only,  [&]() { rslts[4]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_write, [&]() { rslts[5]=seq_num; ++seq_num; } );
+   app->executor().post( priority::highest,exec_queue::read_only,  [&]() { rslts[6]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[7]=seq_num; ++seq_num; } );
 
    // Stop app. Use the lowest priority to make sure this function to execute the last
-   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe, [&]() {
-      // read_only_trx_safe_queue should only contain the current lambda function,
-      // and general_queue should have execute all its functions
-      BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().size(), 1);
-      BOOST_REQUIRE_EQUAL( app->executor().general_queue().size(), 0 );
+   app->executor().post( priority::lowest, exec_queue::read_only, [&]() {
+      // read_only_queue should only contain the current lambda function,
+      // and read_write_queue should have executed all its functions
+      BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().size(), 1); // pop()s after execute
+      BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().size(), 0 );
       app->quit();
       } );
    app_thread.join();
 
    // both queues are cleared after execution 
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().general_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
 
    // exactly number of both queues' functions processed
    BOOST_REQUIRE_EQUAL( rslts.size(), 8 );
@@ -64,40 +64,40 @@ BOOST_AUTO_TEST_CASE( default_exec_window ) {
    BOOST_CHECK_LT( rslts[6], rslts[7] );
 }
 
-// verify functions only from read_only_trx_safe queue are processed during read window
+// verify functions only from read_only queue are processed during read window
 BOOST_AUTO_TEST_CASE( execute_from_read_queue ) {
    appbase::scoped_app app;
    auto app_thread = start_app_thread(app);
    
-   // set to run functions from read_only_trx_safe queue only
+   // set to run functions from read_only queue only
    app->executor().set_to_read_window();
 
    // post functions
    std::map<int, int> rslts {};
    int seq_num = 0;
-   app->executor().post( priority::medium, exec_queue::general, [&]() { rslts[0]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::read_only_trx_safe,  [&]() { rslts[1]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[2]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::read_only_trx_safe,  [&]() { rslts[3]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::read_only_trx_safe,  [&]() { rslts[4]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::general, [&]() { rslts[5]=seq_num; ++seq_num; } );
-   app->executor().post( priority::highest,exec_queue::read_only_trx_safe,  [&]() { rslts[6]=seq_num; ++seq_num; } );
-   app->executor().post( priority::highest,exec_queue::read_only_trx_safe,  [&]() { rslts[7]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::read_only_trx_safe,  [&]() { rslts[8]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[9]=seq_num; ++seq_num; } );
+   app->executor().post( priority::medium, exec_queue::read_write, [&]() { rslts[0]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_only,  [&]() { rslts[1]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[2]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_only,  [&]() { rslts[3]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_only,  [&]() { rslts[4]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_write, [&]() { rslts[5]=seq_num; ++seq_num; } );
+   app->executor().post( priority::highest,exec_queue::read_only,  [&]() { rslts[6]=seq_num; ++seq_num; } );
+   app->executor().post( priority::highest,exec_queue::read_only,  [&]() { rslts[7]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_only,  [&]() { rslts[8]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[9]=seq_num; ++seq_num; } );
 
    // stop application. Use lowest at the end to make sure this executes the last
-   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe, [&]() {
-      // read_queue should have current function and write_queue should have all its functions
-      BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().size(), 1);
-      BOOST_REQUIRE_EQUAL( app->executor().general_queue().size(), 4 );
+   app->executor().post( priority::lowest, exec_queue::read_only, [&]() {
+      // read_queue should be empty (read window pops before execute) and write_queue should have all its functions
+      BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().size(), 0); // pop()s before execute
+      BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().size(), 4 );
       app->quit();
       } );
    app_thread.join();
 
    // both queues are cleared after execution 
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().general_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
 
    // exactly number of posts processed
    BOOST_REQUIRE_EQUAL( rslts.size(), 6 );
@@ -109,40 +109,40 @@ BOOST_AUTO_TEST_CASE( execute_from_read_queue ) {
    BOOST_CHECK_LT( rslts[3], rslts[4] );
 }
 
-// verify no functions are executed during  read window if read_only_trx_safe queue is empty
+// verify no functions are executed during  read window if read_only queue is empty
 BOOST_AUTO_TEST_CASE( execute_from_empty_read_queue ) {
    appbase::scoped_app app;
    auto app_thread = start_app_thread(app);
    
-   // set to run functions from read_only_trx_safe queue only
+   // set to run functions from read_only queue only
    app->executor().set_to_read_window();
 
    // post functions
    std::map<int, int> rslts {};
    int seq_num = 0;
-   app->executor().post( priority::medium, exec_queue::general, [&]() { rslts[0]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[1]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[2]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[3]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::general, [&]() { rslts[4]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::general, [&]() { rslts[5]=seq_num; ++seq_num; } );
-   app->executor().post( priority::highest,exec_queue::general, [&]() { rslts[6]=seq_num; ++seq_num; } );
-   app->executor().post( priority::highest,exec_queue::general, [&]() { rslts[7]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[8]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[9]=seq_num; ++seq_num; } );
+   app->executor().post( priority::medium, exec_queue::read_write, [&]() { rslts[0]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[1]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[2]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[3]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_write, [&]() { rslts[4]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_write, [&]() { rslts[5]=seq_num; ++seq_num; } );
+   app->executor().post( priority::highest,exec_queue::read_write, [&]() { rslts[6]=seq_num; ++seq_num; } );
+   app->executor().post( priority::highest,exec_queue::read_write, [&]() { rslts[7]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[8]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[9]=seq_num; ++seq_num; } );
 
    // Stop application. Use lowest at the end to make sure this executes the last
-   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe, [&]() {
-      // read_queue should have current function and write_queue should have all its functions
-      BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().size(), 1);
-      BOOST_REQUIRE_EQUAL( app->executor().general_queue().size(), 10 );
+   app->executor().post( priority::lowest, exec_queue::read_only, [&]() {
+      // read_queue should be empty (read window pops before execute) and write_queue should have all its functions
+      BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().size(), 0); // pop()s before execute
+      BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().size(), 10 );
       app->quit();
       } );
    app_thread.join();
 
    // both queues are cleared after execution 
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().general_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
 
    // no results
    BOOST_REQUIRE_EQUAL( rslts.size(), 0 );
@@ -159,32 +159,32 @@ BOOST_AUTO_TEST_CASE( execute_from_both_queues ) {
    // post functions
    std::map<int, int> rslts {};
    int seq_num = 0;
-   app->executor().post( priority::medium, exec_queue::read_only_trx_safe,  [&]() { rslts[0]=seq_num; ++seq_num; } );
-   app->executor().post( priority::medium, exec_queue::general, [&]() { rslts[1]=seq_num; ++seq_num; } );
-   app->executor().post( priority::high,   exec_queue::general, [&]() { rslts[2]=seq_num; ++seq_num; } );
-   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe,  [&]() { rslts[3]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::read_only_trx_safe,  [&]() { rslts[4]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::general, [&]() { rslts[5]=seq_num; ++seq_num; } );
-   app->executor().post( priority::highest,exec_queue::read_only_trx_safe,  [&]() { rslts[6]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::general, [&]() { rslts[7]=seq_num; ++seq_num; } );
-   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe,  [&]() { rslts[8]=seq_num; ++seq_num; } );
-   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe,  [&]() { rslts[9]=seq_num; ++seq_num; } );
-   app->executor().post( priority::low,    exec_queue::general, [&]() { rslts[10]=seq_num; ++seq_num; } );
-   app->executor().post( priority::medium, exec_queue::general, [&]() { rslts[11]=seq_num; ++seq_num; } );
+   app->executor().post( priority::medium, exec_queue::read_only,  [&]() { rslts[0]=seq_num; ++seq_num; } );
+   app->executor().post( priority::medium, exec_queue::read_write, [&]() { rslts[1]=seq_num; ++seq_num; } );
+   app->executor().post( priority::high,   exec_queue::read_write, [&]() { rslts[2]=seq_num; ++seq_num; } );
+   app->executor().post( priority::lowest, exec_queue::read_only,  [&]() { rslts[3]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_only,  [&]() { rslts[4]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_write, [&]() { rslts[5]=seq_num; ++seq_num; } );
+   app->executor().post( priority::highest,exec_queue::read_only,  [&]() { rslts[6]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_write, [&]() { rslts[7]=seq_num; ++seq_num; } );
+   app->executor().post( priority::lowest, exec_queue::read_only,  [&]() { rslts[8]=seq_num; ++seq_num; } );
+   app->executor().post( priority::lowest, exec_queue::read_only,  [&]() { rslts[9]=seq_num; ++seq_num; } );
+   app->executor().post( priority::low,    exec_queue::read_write, [&]() { rslts[10]=seq_num; ++seq_num; } );
+   app->executor().post( priority::medium, exec_queue::read_write, [&]() { rslts[11]=seq_num; ++seq_num; } );
 
    // stop application. Use lowest at the end to make sure this executes the last
-   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe, [&]() {
+   app->executor().post( priority::lowest, exec_queue::read_only, [&]() {
       // read_queue should have current function and write_queue's functions are all executed 
-      BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().size(), 1);
-      BOOST_REQUIRE_EQUAL( app->executor().general_queue().size(), 0 );
+      BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().size(), 1); // pop()s after execute
+      BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().size(), 0 );
       app->quit();
       } );
 
    app_thread.join();
 
    // queues are emptied after quit
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().general_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
 
    // exactly number of posts processed
    BOOST_REQUIRE_EQUAL( rslts.size(), 12 );

plugins/chain_api_plugin/chain_api_plugin.cpp

@@ -75,7 +75,7 @@ void chain_api_plugin::plugin_startup() {
    ro_api.set_shorten_abi_errors( !http_plugin::verbose_errors() );
 
    _http_plugin.add_api( {
-      CHAIN_RO_CALL(get_info, 200, http_params_types::no_params)}, appbase::exec_queue::read_only_trx_safe, appbase::priority::medium_high);
+      CHAIN_RO_CALL(get_info, 200, http_params_types::no_params)}, appbase::exec_queue::read_only, appbase::priority::medium_high);
    _http_plugin.add_api({
       CHAIN_RO_CALL(get_activated_protocol_features, 200, http_params_types::possible_no_params),
       CHAIN_RO_CALL(get_block_info, 200, http_params_types::params_required),
@@ -103,12 +103,12 @@ void chain_api_plugin::plugin_startup() {
       CHAIN_RW_CALL_ASYNC(push_transactions, chain_apis::read_write::push_transactions_results, 202, http_params_types::params_required),
       CHAIN_RW_CALL_ASYNC(send_transaction, chain_apis::read_write::send_transaction_results, 202, http_params_types::params_required),
       CHAIN_RW_CALL_ASYNC(send_transaction2, chain_apis::read_write::send_transaction_results, 202, http_params_types::params_required)
-   }, appbase::exec_queue::read_only_trx_safe);
+   }, appbase::exec_queue::read_only);
 
    // Not safe to run in parallel with read-only transactions
    _http_plugin.add_api({
       CHAIN_RW_CALL_ASYNC(push_block, chain_apis::read_write::push_block_results, 202, http_params_types::params_required)
-   }, appbase::exec_queue::general, appbase::priority::medium_low );
+   }, appbase::exec_queue::read_write, appbase::priority::medium_low );
 
    if (chain.account_queries_enabled()) {
       _http_plugin.add_async_api({
@@ -124,7 +124,7 @@ void chain_api_plugin::plugin_startup() {
    if (chain.transaction_finality_status_enabled()) {
       _http_plugin.add_api({
          CHAIN_RO_CALL_WITH_400(get_transaction_status, 200, http_params_types::params_required),
-      }, appbase::exec_queue::read_only_trx_safe);
+      }, appbase::exec_queue::read_only);
    }
 
    _http_plugin.add_api({
@@ -160,7 +160,7 @@ void chain_api_plugin::plugin_startup() {
            }
         }
       }
-   }, appbase::exec_queue::read_only_trx_safe);
+   }, appbase::exec_queue::read_only);
 }
 
 void chain_api_plugin::plugin_shutdown() {}

plugins/db_size_api_plugin/db_size_api_plugin.cpp

@@ -28,7 +28,7 @@ using namespace eosio;
 void db_size_api_plugin::plugin_startup() {
    app().get_plugin<http_plugin>().add_api({
        CALL_WITH_400(db_size, this, get,  INVOKE_R_V(this, get), 200),
-   }, appbase::exec_queue::read_only_trx_safe);
+   }, appbase::exec_queue::read_only);
 }
 
 db_size_stats db_size_api_plugin::get() {

plugins/http_plugin/http_plugin.cpp

@@ -334,7 +334,7 @@ namespace eosio {
                      handle_exception("node", "get_supported_apis", body.empty() ? "{}" : body, cb);
                   }
                }
-            }}, appbase::exec_queue::read_only_trx_safe);
+            }}, appbase::exec_queue::read_only);
             
          } catch (...) {
             fc_elog(logger(), "http_plugin startup fails, shutting down");

plugins/http_plugin/include/eosio/http_plugin/http_plugin.hpp

@@ -87,8 +87,12 @@ namespace eosio {
         void plugin_shutdown();
         void handle_sighup() override;
 
-        void add_handler(const string& url, const url_handler&, appbase::exec_queue q, int priority = appbase::priority::medium_low, http_content_type content_type = http_content_type::json);
-        void add_api(const api_description& api, appbase::exec_queue q, int priority = appbase::priority::medium_low, http_content_type content_type = http_content_type::json) {
+        void add_handler(const string& url, const url_handler&, appbase::exec_queue q,
+                         int priority = appbase::priority::medium_low,
+                         http_content_type content_type = http_content_type::json);
+        void add_api(const api_description& api, appbase::exec_queue q,
+                     int priority = appbase::priority::medium_low,
+                     http_content_type content_type = http_content_type::json) {
            for (const auto& call : api)
               add_handler(call.first, call.second, q, priority, content_type);
         }

plugins/http_plugin/tests/unit_tests.cpp

@@ -52,7 +52,7 @@ public:
                   cb(200, fc::time_point::maximum(), fc::variant(ok ? string("yes") : string("no")));
                }
             },
-         }, appbase::exec_queue::general);
+         }, appbase::exec_queue::read_write);
    }
 
 private:

plugins/net_api_plugin/net_api_plugin.cpp

@@ -53,11 +53,7 @@ void net_api_plugin::plugin_startup() {
    ilog("starting net_api_plugin");
    // lifetime of plugin is lifetime of application
    auto& net_mgr = app().get_plugin<net_plugin>();
-   app().get_plugin<http_plugin>().add_api({
-    //   CALL(net, net_mgr, set_timeout,
-    //        INVOKE_V_R(net_mgr, set_timeout, int64_t), 200),
-    //   CALL(net, net_mgr, sign_transaction,
-    //        INVOKE_R_R_R_R(net_mgr, sign_transaction, chain::signed_transaction, flat_set<public_key_type>, chain::chain_id_type), 201),
+   app().get_plugin<http_plugin>().add_async_api({
        CALL_WITH_400(net, net_mgr, connect,
             INVOKE_R_R(net_mgr, connect, std::string), 201),
        CALL_WITH_400(net, net_mgr, disconnect,
@@ -66,9 +62,7 @@ void net_api_plugin::plugin_startup() {
             INVOKE_R_R(net_mgr, status, std::string), 201),
        CALL_WITH_400(net, net_mgr, connections,
             INVOKE_R_V(net_mgr, connections), 201),
-    //   CALL(net, net_mgr, open,
-    //        INVOKE_V_R(net_mgr, open, std::string), 200),
-   }, appbase::exec_queue::read_only_trx_safe, appbase::priority::medium_high);
+   } );
 }
 
 void net_api_plugin::plugin_initialize(const variables_map& options) {

plugins/net_plugin/net_plugin.cpp

@@ -3234,7 +3234,7 @@ namespace eosio {
             my_impl->dispatcher->bcast_block( bsp->block, bsp->id );
          }
 
-         app().executor().post(priority::medium, exec_queue::general, [ptr{std::move(ptr)}, bsp{std::move(bsp)}, id, c{std::move(c)}]() mutable {
+         app().executor().post(priority::medium, exec_queue::read_write, [ptr{std::move(ptr)}, bsp{std::move(bsp)}, id, c{std::move(c)}]() mutable {
             c->process_signed_block( id, std::move(ptr), std::move(bsp) );
          });
 

plugins/producer_api_plugin/producer_api_plugin.cpp

@@ -118,9 +118,11 @@ void producer_api_plugin::plugin_startup() {
             INVOKE_R_R(producer, get_account_ram_corrections, producer_plugin::get_account_ram_corrections_params), 201),
        CALL_WITH_400(producer, producer, get_unapplied_transactions,
                      INVOKE_R_R_D(producer, get_unapplied_transactions, producer_plugin::get_unapplied_transactions_params), 200),
-   }, appbase::exec_queue::read_only_trx_safe, appbase::priority::medium_high);
+       CALL_WITH_400(producer, producer, get_snapshot_requests,
+                     INVOKE_R_V(producer, get_snapshot_requests), 201),
+   }, appbase::exec_queue::read_only, appbase::priority::medium_high);
 
-   // Not safe to run in parallel with read-only transactions
+   // Not safe to run in parallel
    app().get_plugin<http_plugin>().add_api({
        CALL_WITH_400(producer, producer, pause,
             INVOKE_V_V(producer, pause), 201),
@@ -138,15 +140,13 @@ void producer_api_plugin::plugin_startup() {
             INVOKE_R_V_ASYNC(producer, create_snapshot), 201),
        CALL_WITH_400(producer, producer, schedule_snapshot,
             INVOKE_V_R_II(producer, schedule_snapshot, producer_plugin::snapshot_request_information), 201),
-       CALL_WITH_400(producer, producer, get_snapshot_requests,
-            INVOKE_R_V(producer, get_snapshot_requests), 201),
        CALL_WITH_400(producer, producer, unschedule_snapshot,
             INVOKE_V_R(producer, unschedule_snapshot, producer_plugin::snapshot_request_id_information), 201),
        CALL_WITH_400(producer, producer, get_integrity_hash,
             INVOKE_R_V(producer, get_integrity_hash), 201),
        CALL_WITH_400(producer, producer, schedule_protocol_feature_activations,
             INVOKE_V_R(producer, schedule_protocol_feature_activations, producer_plugin::scheduled_protocol_feature_activations), 201),
-   }, appbase::exec_queue::general, appbase::priority::medium_high);
+   }, appbase::exec_queue::read_write, appbase::priority::medium_high);
 }
 
 void producer_api_plugin::plugin_initialize(const variables_map& options) {

plugins/producer_plugin/producer_plugin.cpp

@@ -14,6 +14,7 @@
 #include <fc/io/json.hpp>
 #include <fc/log/logger_config.hpp>
 #include <fc/scoped_exit.hpp>
+#include <fc/time.hpp>
 
 #include <boost/asio.hpp>
 #include <boost/date_time/posix_time/posix_time.hpp>
@@ -324,9 +325,9 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
       push_result push_transaction( const fc::time_point& block_deadline,
                                     const transaction_metadata_ptr& trx,
                                     bool api_trx, bool return_failure_trace,
-                                    next_function<transaction_trace_ptr> next );
+                                    const next_function<transaction_trace_ptr>& next );
       push_result handle_push_result( const transaction_metadata_ptr& trx,
-                                      next_function<transaction_trace_ptr> next,
+                                      const next_function<transaction_trace_ptr>& next,
                                       const fc::time_point& start,
                                       const chain::controller& chain,
                                       const transaction_trace_ptr& trace,
@@ -417,7 +418,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          packed_transaction_ptr  trx;
          next_func_t             next;
       };
-      // The queue storing read-only transactions to be executed by read-only threads
+      // The queue storing previously exhausted read-only transactions to be re-executed by read-only threads
       struct ro_trx_queue_t {
          std::mutex              mtx;
          std::deque<ro_trx_t>    queue;
@@ -429,24 +430,26 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
       fc::microseconds                _ro_read_window_time_us{ 60000 };
       static constexpr fc::microseconds _ro_read_window_minimum_time_us{ 10000 };
       fc::microseconds                _ro_read_window_effective_time_us{ 0 }; // calculated during option initialization
+      fc::time_point                  _ro_window_deadline; // only modified on app thread
       boost::asio::deadline_timer     _ro_write_window_timer;
       boost::asio::deadline_timer     _ro_read_window_timer;
       fc::microseconds                _ro_max_trx_time_us{ 0 }; // calculated during option initialization
-      ro_trx_queue_t                  _ro_trx_queue;
-      std::atomic<uint32_t>            _ro_num_active_trx_exec_tasks{ 0 };
-      std::vector<std::future<bool>>  _ro_trx_exec_tasks_fut;
+      ro_trx_queue_t                  _ro_exhausted_trx_queue;
+      std::atomic<uint32_t>           _ro_num_active_exec_tasks{ 0 };
+      std::atomic<bool>               _ro_exiting_read_window{ false };
+      bool                            _ro_in_read_only_mode{false}; // only modified on app thread
+      std::vector<std::future<bool>>  _ro_exec_tasks_fut;
 
       void start_write_window();
       void switch_to_write_window();
       void switch_to_read_window();
-      bool read_only_trx_execution_task();
+      bool read_only_execution_task();
+      void process_read_only_transactions(const fc::time_point& deadline);
       bool process_read_only_transaction(const packed_transaction_ptr& trx,
-                                         const next_function<transaction_trace_ptr>& next,
-                                         const fc::time_point& read_window_start_time);
+                                         const next_function<transaction_trace_ptr>& next);
       bool push_read_only_transaction(const transaction_metadata_ptr& trx,
-                                      next_function<transaction_trace_ptr> next,
-                                      const fc::time_point& read_window_start_time);
-      
+                                      const next_function<transaction_trace_ptr>& next);
+
       void consider_new_watermark( account_name producer, uint32_t block_num, block_timestamp_type timestamp) {
          auto itr = _producer_watermarks.find( producer );
          if( itr != _producer_watermarks.end() ) {
@@ -639,12 +642,11 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
                                          transaction_metadata::trx_type trx_type,
                                          bool return_failure_traces,
                                          next_function<transaction_trace_ptr> next) {
-         if ( trx_type == transaction_metadata::trx_type::read_only && _ro_thread_pool_size > 0 ) {
-            // Parallel read-only trx execution enabled.
-            // Store the transaction in read-only-trx queue so that it is
-            // executed in read window
-            std::lock_guard<std::mutex> g( _ro_trx_queue.mtx );
-            _ro_trx_queue.queue.push_back({trx, std::move(next)});
+         if ( trx_type == transaction_metadata::trx_type::read_only ) {
+            // Post all read only trxs to read_only queue for execution.
+            app().executor().post(priority::low, exec_queue::read_only, [this, trx=trx, next{std::move(next)}]() mutable {
+               process_read_only_transaction( trx, next );
+            } );
             return;
          }
 
@@ -677,7 +679,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
                                                           next{std::move(next)}, trx=trx]() mutable {
             if( future.valid() ) {
                future.wait();
-               app().executor().post( priority::low, exec_queue::general, [self, future{std::move(future)}, api_trx, is_transient, next{std::move( next )}, trx{std::move(trx)}, return_failure_traces]() mutable {
+               app().executor().post( priority::low, exec_queue::read_write, [self, future{std::move(future)}, api_trx, is_transient, next{std::move( next )}, trx{std::move(trx)}, return_failure_traces]() mutable {
                   auto start = fc::time_point::now();
                   auto idle_time = start - self->_idle_trx_time;
                   self->_time_tracker.add_idle_time( idle_time );
@@ -709,7 +711,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
       bool process_incoming_transaction_async(const transaction_metadata_ptr& trx,
                                               bool api_trx,
                                               bool return_failure_trace,
-                                              next_function<transaction_trace_ptr> next) {
+                                              const next_function<transaction_trace_ptr>& next) {
          bool exhausted = false;
          chain::controller& chain = chain_plug->chain();
          try {
@@ -892,7 +894,7 @@ void producer_plugin::set_program_options(
           "Number of worker threads in producer thread pool")
          ("snapshots-dir", bpo::value<bfs::path>()->default_value("snapshots"),
           "the location of the snapshots directory (absolute path or relative to application data dir)")
-         ("read-only-threads", bpo::value<uint16_t>()->default_value(my->_ro_thread_pool_size),
+         ("read-only-threads", bpo::value<uint16_t>(),
           "Number of worker threads in read-only transaction execution thread pool")
          ("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")
@@ -1073,11 +1075,15 @@ void producer_plugin::plugin_initialize(const boost::program_options::variables_
       }
    }
 
-   my->_ro_thread_pool_size = options.at( "read-only-threads" ).as<uint16_t>();
+   if ( options.count( "read-only-threads" ) ) {
+      my->_ro_thread_pool_size = options.at( "read-only-threads" ).as<uint16_t>();
+      EOS_ASSERT( my->_producers.empty(), plugin_config_exception, "--read-only-threads not allowed on producer node" );
+   } else if ( my->_producers.empty() ) {
+      // default to 3 threads for non producer nodes if not specified
+      my->_ro_thread_pool_size = 3;
+   }
    // only initialize other read-only options when read-only thread pool is enabled
    if ( my->_ro_thread_pool_size > 0 ) {
-      EOS_ASSERT( my->_producers.empty(), plugin_config_exception, "--read-only-threads not allowed on producer node" );
-
 #ifdef EOSIO_EOS_VM_OC_RUNTIME_ENABLED
       if (chain.is_eos_vm_oc_enabled()) {
          // EOS VM OC requires 4.2TB Virtual for each executing thread. Make sure the memory
@@ -1109,26 +1115,27 @@ void producer_plugin::plugin_initialize(const boost::program_options::variables_
          }
       }
 #endif
-
-      my->_ro_write_window_time_us = fc::microseconds( options.at( "read-only-write-window-time-us" ).as<uint32_t>() );
-      my->_ro_read_window_time_us = fc::microseconds( options.at( "read-only-read-window-time-us" ).as<uint32_t>() );
-      EOS_ASSERT( my->_ro_read_window_time_us > my->_ro_read_window_minimum_time_us, plugin_config_exception, "minimum of --read-only-read-window-time-us (${read}) must be ${min} microseconds", ("read", my->_ro_read_window_time_us) ("min", my->_ro_read_window_minimum_time_us) );
-      my->_ro_read_window_effective_time_us = my->_ro_read_window_time_us - my->_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.
-      // Use _ro_read_window_effective_time_us instead of _ro_read_window_time_us
-      // for safety marging
-      if ( my->_max_transaction_time_ms.load() > 0 ) {
-         EOS_ASSERT( my->_ro_read_window_effective_time_us > fc::milliseconds(my->_max_transaction_time_ms.load()), plugin_config_exception, "--read-only-read-window-time-us (${read}) must be greater than --max-transaction-time ${trx_time} ms plus a margin of ${min} us", ("read", my->_ro_read_window_time_us) ("trx_time", my->_max_transaction_time_ms.load()) ("min", my->_ro_read_window_minimum_time_us) );
-         my->_ro_max_trx_time_us = fc::milliseconds(my->_max_transaction_time_ms.load());
-      } else {
-         // _max_transaction_time_ms can be set to negative in testing (for unlimited)
-         my->_ro_max_trx_time_us = my->_ro_read_window_effective_time_us;
-      }
-      ilog("_ro_thread_pool_size ${s},  _ro_write_window_time_us ${ww}, _ro_read_window_time_us ${rw}, _ro_max_trx_time_us ${t}, _ro_read_window_effective_time_us ${w}", ("s", my->_ro_thread_pool_size) ("ww", my->_ro_write_window_time_us) ("rw", my->_ro_read_window_time_us) ("t", my->_ro_max_trx_time_us) ("w", my->_ro_read_window_effective_time_us));
    }
 
+   my->_ro_write_window_time_us = fc::microseconds( options.at( "read-only-write-window-time-us" ).as<uint32_t>() );
+   my->_ro_read_window_time_us = fc::microseconds( options.at( "read-only-read-window-time-us" ).as<uint32_t>() );
+   EOS_ASSERT( my->_ro_read_window_time_us > my->_ro_read_window_minimum_time_us, plugin_config_exception, "minimum of --read-only-read-window-time-us (${read}) must be ${min} microseconds", ("read", my->_ro_read_window_time_us) ("min", my->_ro_read_window_minimum_time_us) );
+   my->_ro_read_window_effective_time_us = my->_ro_read_window_time_us - my->_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.
+   // Use _ro_read_window_effective_time_us instead of _ro_read_window_time_us
+   // for safety marging
+   if ( my->_max_transaction_time_ms.load() > 0 ) {
+      EOS_ASSERT( my->_ro_read_window_effective_time_us > fc::milliseconds(my->_max_transaction_time_ms.load()), plugin_config_exception, "--read-only-read-window-time-us (${read}) must be greater than --max-transaction-time ${trx_time} ms plus a margin of ${min} us", ("read", my->_ro_read_window_time_us) ("trx_time", my->_max_transaction_time_ms.load()) ("min", my->_ro_read_window_minimum_time_us) );
+      my->_ro_max_trx_time_us = fc::milliseconds(my->_max_transaction_time_ms.load());
+   } else {
+      // _max_transaction_time_ms can be set to negative in testing (for unlimited)
+      my->_ro_max_trx_time_us = my->_ro_read_window_effective_time_us;
+   }
+   ilog("ro_thread_pool_size ${s}, ro_write_window_time_us ${ww}, ro_read_window_time_us ${rw}, ro_max_trx_time_us ${t}, ro_read_window_effective_time_us ${w}",
+        ("s", my->_ro_thread_pool_size)("ww", my->_ro_write_window_time_us)("rw", my->_ro_read_window_time_us)("t", my->_ro_max_trx_time_us)("w", my->_ro_read_window_effective_time_us));
+
    my->_incoming_block_sync_provider = app().get_method<incoming::methods::block_sync>().register_provider(
          [this](const signed_block_ptr& block, const std::optional<block_id_type>& block_id, const block_state_ptr& bsp) {
       return my->on_incoming_block(block, block_id, bsp);
@@ -1994,6 +2001,8 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
             process_scheduled_and_incoming_trxs( scheduled_trx_deadline, incoming_itr );
          }
 
+         process_read_only_transactions( preprocess_deadline );
+
          if( app().is_quiting() ) // db guard exception above in LOG_AND_DROP could have called app().quit()
             return start_block_result::failed;
          if ( should_interrupt_start_block( preprocess_deadline ) || block_is_exhausted() ) {
@@ -2195,9 +2204,10 @@ producer_plugin_impl::push_transaction( const fc::time_point& block_deadline,
                                         const transaction_metadata_ptr& trx,
                                         bool api_trx,
                                         bool return_failure_trace,
-                                        next_function<transaction_trace_ptr> next )
+                                        const next_function<transaction_trace_ptr>& next )
 {
    auto start = fc::time_point::now();
+   EOS_ASSERT(!trx->is_read_only(), producer_exception, "Unexpected read-only trx");
 
    bool disable_subjective_enforcement = (api_trx && _disable_subjective_api_billing)
                                          || (!api_trx && _disable_subjective_p2p_billing)
@@ -2234,14 +2244,6 @@ producer_plugin_impl::push_transaction( const fc::time_point& block_deadline,
       }
    }
 
-   // Make sure db_read_only_mode always to be unset
-   auto db_read_only_mode_guard = fc::make_scoped_exit([trx, &chain]{
-      if( trx->is_read_only() )
-         chain.unset_db_read_only_mode();
-   });
-   if( trx->is_read_only() )
-      chain.set_db_read_only_mode();
-
    auto trace = chain.push_transaction( trx, block_deadline, max_trx_time, prev_billed_cpu_time_us, false, sub_bill );
 
    return handle_push_result(trx, next, start, chain, trace, return_failure_trace, disable_subjective_enforcement, first_auth, sub_bill, prev_billed_cpu_time_us);
@@ -2249,7 +2251,7 @@ producer_plugin_impl::push_transaction( const fc::time_point& block_deadline,
 
 producer_plugin_impl::push_result
 producer_plugin_impl::handle_push_result( const transaction_metadata_ptr& trx,
-                                          next_function<transaction_trace_ptr> next,
+                                          const next_function<transaction_trace_ptr>& next,
                                           const fc::time_point& start,
                                           const chain::controller& chain,
                                           const transaction_trace_ptr& trace,
@@ -2555,7 +2557,7 @@ void producer_plugin_impl::schedule_production_loop() {
       _timer.expires_from_now( boost::posix_time::microseconds( config::block_interval_us  / 10 ));
 
       // we failed to start a block, so try again later?
-      _timer.async_wait( app().executor().wrap( priority::high, exec_queue::general,
+      _timer.async_wait( app().executor().wrap( priority::high, exec_queue::read_write,
           [weak_this = weak_from_this(), cid = ++_timer_corelation_id]( const boost::system::error_code& ec ) {
              auto self = weak_this.lock();
              if( self && ec != boost::asio::error::operation_aborted && cid == self->_timer_corelation_id ) {
@@ -2608,7 +2610,7 @@ void producer_plugin_impl::schedule_maybe_produce_block( bool exhausted ) {
                ("num", chain.head_block_num() + 1)("desc", block_is_exhausted() ? "Exhausted" : "Deadline exceeded") );
    }
 
-   _timer.async_wait( app().executor().wrap( priority::high, exec_queue::general,
+   _timer.async_wait( app().executor().wrap( priority::high, exec_queue::read_write,
          [&chain, weak_this = weak_from_this(), cid=++_timer_corelation_id](const boost::system::error_code& ec) {
             auto self = weak_this.lock();
             if( self && ec != boost::asio::error::operation_aborted && cid == self->_timer_corelation_id ) {
@@ -2650,7 +2652,7 @@ void producer_plugin_impl::schedule_delayed_production_loop(const std::weak_ptr<
       fc_dlog(_log, "Scheduling Speculative/Production Change at ${time}", ("time", wake_up_time));
       static const boost::posix_time::ptime epoch(boost::gregorian::date(1970, 1, 1));
       _timer.expires_at(epoch + boost::posix_time::microseconds(wake_up_time->time_since_epoch().count()));
-      _timer.async_wait( app().executor().wrap( priority::high, exec_queue::general,
+      _timer.async_wait( app().executor().wrap( priority::high, exec_queue::read_write,
          [weak_this,cid=++_timer_corelation_id](const boost::system::error_code& ec) {
             auto self = weak_this.lock();
             if( self && ec != boost::asio::error::operation_aborted && cid == self->_timer_corelation_id ) {
@@ -2764,7 +2766,7 @@ void producer_plugin::log_failed_transaction(const transaction_id_type& trx_id,
             ("entire_trx", packed_trx_ptr ? my->chain_plug->get_log_trx(packed_trx_ptr->get_transaction()) : fc::variant{trx_id}));
 }
 
-// Called from app thread
+// Called from only one read_only thread
 void producer_plugin_impl::switch_to_write_window() {
    // this method can be called from multiple places. it is possible
    // we are already in write window.
@@ -2772,26 +2774,29 @@ void producer_plugin_impl::switch_to_write_window() {
       return;
    }
 
-   EOS_ASSERT(_ro_num_active_trx_exec_tasks.load() == 0 && _ro_trx_exec_tasks_fut.empty(), producer_exception, "no read-only tasks should be running before switching to write window");
+   EOS_ASSERT( _ro_num_active_exec_tasks.load() == 0 && _ro_exec_tasks_fut.empty(), producer_exception, "no read-only tasks should be running before switching to write window");
    _ro_read_window_timer.cancel();
    _ro_write_window_timer.cancel();
 
    start_write_window();
 }
 
-// Called from app thread
+// Called from app thread on plugin_startup
+// Called from only one read_only thread & called from app thread, but not concurrently
 void producer_plugin_impl::start_write_window() {
    chain::controller& chain = chain_plug->chain();
 
    app().executor().set_to_write_window();
    chain.unset_db_read_only_mode();
+   _ro_in_read_only_mode = false;
    _idle_trx_time = fc::time_point::now();
 
+   _ro_window_deadline = _idle_trx_time + _ro_write_window_time_us; // not allowed on block producers, so no need to limit to block deadline
    auto expire_time = boost::posix_time::microseconds(_ro_write_window_time_us.count());
    _ro_write_window_timer.expires_from_now( expire_time );
-   _ro_write_window_timer.async_wait( app().executor().wrap(  // stay on app thread
+   _ro_write_window_timer.async_wait( app().executor().wrap( // run from app thread
       priority::high,
-      exec_queue::read_only_trx_safe, // placed in read_only_trx_safe queue so it is ensured to be executed in either window
+      exec_queue::read_write, // placed in read_write so only called from main thread
       [weak_this = weak_from_this()]( const boost::system::error_code& ec ) {
          auto self = weak_this.lock();
          if( self && ec != boost::asio::error::operation_aborted ) {
@@ -2800,86 +2805,98 @@ void producer_plugin_impl::start_write_window() {
       }));
 }
 
-// Called from app thread
+// Called only from app thread
 void producer_plugin_impl::switch_to_read_window() {
    EOS_ASSERT(app().executor().is_write_window(),  producer_exception, "expected to be in write window");
-   EOS_ASSERT(_ro_num_active_trx_exec_tasks.load() == 0 && _ro_trx_exec_tasks_fut.empty(),  producer_exception, "_ro_trx_exec_tasks_fut expected to be empty" );
+   EOS_ASSERT( _ro_num_active_exec_tasks.load() == 0 && _ro_exec_tasks_fut.empty(), producer_exception, "_ro_exec_tasks_fut expected to be empty" );
 
    _ro_write_window_timer.cancel();
    _ro_read_window_timer.cancel();
    _time_tracker.add_idle_time( fc::time_point::now() - _idle_trx_time );
 
    // we are in write window, so no read-only trx threads are processing transactions.
-   // _ro_trx_queue is not being accessed. No need to lock.
-   if ( _ro_trx_queue.queue.empty() ) { // no read-only trxs to process. stay in write window
+   // _ro_exhausted_trx_queue is not being accessed by read-only threads, no need to lock.
+   if ( _ro_exhausted_trx_queue.queue.empty() && app().executor().read_only_queue().empty() ) { // no read-only trxs to process. stay in write window
       start_write_window(); // restart write window timer for next round
       return;
    }
 
    app().executor().set_to_read_window();
    chain_plug->chain().set_db_read_only_mode();
+   _ro_in_read_only_mode = true;
    _received_block = false;
+   _ro_exiting_read_window = false;
 
    // start a read-only transaction execution task in each thread in the thread pool
-   _ro_num_active_trx_exec_tasks = _ro_thread_pool_size;
+   _ro_num_active_exec_tasks = _ro_thread_pool_size;
+   _ro_window_deadline = fc::time_point::now() + _ro_read_window_effective_time_us;
    for (auto i = 0; i < _ro_thread_pool_size; ++i ) {
-      _ro_trx_exec_tasks_fut.emplace_back( post_async_task( _ro_thread_pool.get_executor(), [self = this] () {
-         return self->read_only_trx_execution_task();
+      _ro_exec_tasks_fut.emplace_back( post_async_task( _ro_thread_pool.get_executor(), [self = this] () {
+         return self->read_only_execution_task();
       }) );
    }
 
    auto expire_time = boost::posix_time::microseconds(_ro_read_window_time_us.count());
    _ro_read_window_timer.expires_from_now( expire_time );
-   _ro_read_window_timer.async_wait( app().executor().wrap(  // stay on app thread
+   // Needs to be on read_only because that is what is being processed until switch_to_write_window().
+   _ro_read_window_timer.async_wait( app().executor().wrap(
       priority::high,
-      exec_queue::read_only_trx_safe,
+      exec_queue::read_only,
       [weak_this = weak_from_this()]( const boost::system::error_code& ec ) {
          auto self = weak_this.lock();
          if( self && ec != boost::asio::error::operation_aborted ) {
             // use future to make sure all read-only tasks finished before switching to write window
-            for ( auto& task: self->_ro_trx_exec_tasks_fut ) {
+            for ( auto& task: self->_ro_exec_tasks_fut ) {
                task.get();
             }
-            self->_ro_trx_exec_tasks_fut.clear();
+            self->_ro_exec_tasks_fut.clear();
+            // will be executed from the main app thread because all read-only threads are idle now
             self->switch_to_write_window();
           } else if ( self ) {
-             self->_ro_trx_exec_tasks_fut.clear();
+             self->_ro_exec_tasks_fut.clear();
           }
        }));
 }
 
-// Called from a read only trx thread. Run in parallel with app and other read only trx threads
-bool producer_plugin_impl::read_only_trx_execution_task() {
-   auto start = fc::time_point::now();
-   auto read_window_deadline = start + _ro_read_window_effective_time_us;
+// Called from a read only thread. Run in parallel with app and other read only threads
+bool producer_plugin_impl::read_only_execution_task() {
    // We have 4 ways to break out the while loop:
    // 1. pass read window deadline
-   // 2. Net_plugin receives a block
-   // 3. No more transactions in the read-only trx queue
-   // 4. A transaction execution is exhaused
-   while ( fc::time_point::now() < read_window_deadline && !_received_block ) {
-      std::unique_lock<std::mutex> lck( _ro_trx_queue.mtx );
-      if ( _ro_trx_queue.queue.empty() ) {
-         break;
-      }
-      auto trx = _ro_trx_queue.queue.front();
-      _ro_trx_queue.queue.pop_front();
-      lck.unlock();
-      
-      auto retry = process_read_only_transaction( trx.trx, trx.next, start );
-      if ( retry ) {
-         lck.lock();
-         _ro_trx_queue.queue.push_front(trx);
-         // Do not schedule new execution
-         break;
+   // 2. net_plugin receives a block
+   // 3. No more transactions in the exhausted read-only queue and no read-only tasks to execute
+   // 4. A transaction execution is exhausted (means end of read window)
+   bool exhausted_trx_queue_empty = false;
+   while ( fc::time_point::now() < _ro_window_deadline && !_received_block && !_ro_exiting_read_window ) {
+      if (!exhausted_trx_queue_empty) {
+         std::unique_lock<std::mutex> lck( _ro_exhausted_trx_queue.mtx );
+         auto size = _ro_exhausted_trx_queue.queue.size();
+         if ( size > 0 ) { // execute exhausted read-only first
+            auto trx = _ro_exhausted_trx_queue.queue.front();
+            _ro_exhausted_trx_queue.queue.pop_front();
+            lck.unlock();
+
+            auto retry = process_read_only_transaction( trx.trx, trx.next );
+            if( retry ) {
+               _ro_exiting_read_window = true;
+               break;
+            }
+         }
+         exhausted_trx_queue_empty = (size <= 1); // one was executed above
+      } else {
+         bool more = app().executor().execute_highest_read_only();
+         if ( !more ) { // read_only queue empty, anything queued after this will be processed in the next window
+            _ro_exiting_read_window = true;
+            break;
+         }
       }
    }
 
    // If all tasks are finished, do not wait until end of read window; switch to write window now.
-   if ( --_ro_num_active_trx_exec_tasks == 0 ) {
-      // Do switching on app thread to serialize
-      app().executor().post( priority::high, exec_queue::read_only_trx_safe, [self=this]() {
-         self->_ro_trx_exec_tasks_fut.clear();
+   if ( --_ro_num_active_exec_tasks == 0 ) {
+      // Needs to be on read_only because that is what is being processed until switch_to_write_window().
+      app().executor().post( priority::high, exec_queue::read_only, [self=this]() {
+         self->_ro_exec_tasks_fut.clear();
+         // will be executed from the main app thread because all read-only threads are idle now
          self->switch_to_write_window();
       } );
    }
@@ -2887,9 +2904,28 @@ bool producer_plugin_impl::read_only_trx_execution_task() {
    return true;
 }
 
+// Called from app thread during start block.
+// Used for executing exhausted read-only trxs when no read only thread pool enabled.
+void producer_plugin_impl::process_read_only_transactions(const fc::time_point& deadline) {
+   // if the read only thread pool is enabled then exhausted read only transactions will be executed
+   // by read_only_execution_task, so don't execute them here.
+   if ( _ro_thread_pool_size == 0 ) {
+      _ro_window_deadline = std::min(deadline, fc::time_point::now() + _ro_read_window_effective_time_us);
+      // only app thread accessing _ro_exhausted_trx_queue since _ro_thread_pool_size == 0, no lock needed
+      while ( !_ro_exhausted_trx_queue.queue.empty() && fc::time_point::now() < _ro_window_deadline && !_received_block ) {
+         auto trx = _ro_exhausted_trx_queue.queue.front();
+         _ro_exhausted_trx_queue.queue.pop_front();
+
+         auto retry = process_read_only_transaction( trx.trx, trx.next );
+         if( retry )
+            break;
+      }
+   }
+}
+
 // Called from a read only trx thread. Run in parallel with app and other read only trx threads
 // Return whether the trx needs to be retried in next read window
-bool producer_plugin_impl::process_read_only_transaction(const packed_transaction_ptr& trx, const next_function<transaction_trace_ptr>& next, const fc::time_point& read_window_start_time) {
+bool producer_plugin_impl::process_read_only_transaction(const packed_transaction_ptr& trx, const next_function<transaction_trace_ptr>& next) {
    chain::controller& chain = chain_plug->chain();
    auto future = transaction_metadata::start_recover_keys( trx, _thread_pool.get_executor(),
                                                            chain.get_chain_id(), fc::microseconds::maximum(),
@@ -2902,17 +2938,19 @@ bool producer_plugin_impl::process_read_only_transaction(const packed_transactio
    future.wait();
    try {
       auto trx_metadata = future.get();
-      return push_read_only_transaction( trx_metadata, next, read_window_start_time );
+      bool retry = push_read_only_transaction( trx_metadata, next );
+      if( retry ) {
+         std::lock_guard g( _ro_exhausted_trx_queue.mtx );
+         _ro_exhausted_trx_queue.queue.push_front( {trx, next} );
+      }
+      return retry;
    } CATCH_AND_CALL(exception_handler);
    return false;
 }
 
 // Called from a read_only_trx execution thread
 // Return whether the trx needs to be retried in next read window
-bool producer_plugin_impl::push_read_only_transaction(
-            const transaction_metadata_ptr& trx,
-            next_function<transaction_trace_ptr> next,
-            const fc::time_point& read_window_start_time) {
+bool producer_plugin_impl::push_read_only_transaction(const transaction_metadata_ptr& trx, const next_function<transaction_trace_ptr>& next) {
    auto retry = false;
    chain::controller& chain = chain_plug->chain();
 
@@ -2922,18 +2960,24 @@ bool producer_plugin_impl::push_read_only_transaction(
    }
 
    try {
-      const auto block_deadline = calculate_block_deadline( chain.pending_block_time() );
       auto start = fc::time_point::now();
-      auto time_used_in_read_window_us = ( start - read_window_start_time );
-      if ( time_used_in_read_window_us >= _ro_read_window_time_us ) {
+      if ( start >= _ro_window_deadline ) {
          // already passed read-window deadline, try next round
          return true;
       }
 
-      auto remaining_time_in_read_window_us = _ro_read_window_time_us - time_used_in_read_window_us;
+      // when executing on the main thread while in the write window, need to switch db mode to read only
+      // _ro_in_read_only_mode can only be false if running on main thread as it is only modified from the main thread
+      const bool switch_read_mode = !_ro_in_read_only_mode;
+      auto db_read_only_mode_guard = fc::make_scoped_exit([&]{
+         if( switch_read_mode )
+            chain.unset_db_read_only_mode();
+      });
+      if( switch_read_mode )
+         chain.set_db_read_only_mode();
+
       // Ensure the trx to finish by the end of read-window.
-      auto window_deadline = std::min( start + remaining_time_in_read_window_us, block_deadline );
-      auto trace = chain.push_transaction( trx, window_deadline, _ro_max_trx_time_us, 0, false, 0 );
+      auto trace = chain.push_transaction( trx, _ro_window_deadline, _ro_max_trx_time_us, 0, false, 0 );
       auto pr = handle_push_result(trx, next, start, chain, trace, true /*return_failure_trace*/, true /*disable_subjective_enforcement*/, {} /*first_auth*/, 0 /*sub_bill*/, 0 /*prev_billed_cpu_time_us*/);
       // If a transaction was exhausted, that indicates we are close to
       // the end of read window. Retry in next round.

plugins/producer_plugin/test/test_read_only_trx.cpp

@@ -103,7 +103,7 @@ void test_trxs_common(std::vector<const char*>& specific_args) {
 
    for( size_t i = 1; i <= num_pushes; ++i ) {
       auto ptrx = make_unique_trx( chain_id );
-      app->executor().post( priority::low, exec_queue::general, [ptrx, &next_calls, &num_posts, &trace_with_except, &trx_match, &app]() {
+      app->executor().post( priority::low, exec_queue::read_write, [ptrx, &next_calls, &num_posts, &trace_with_except, &trx_match, &app]() {
          ++num_posts;
          bool return_failure_traces = true;
          app->get_method<plugin_interface::incoming::methods::transaction_async>()(ptrx,

plugins/producer_plugin/test/test_snapshot_scheduler.cpp

@@ -67,7 +67,7 @@ BOOST_AUTO_TEST_CASE(snapshot_scheduler_test) {
             fc::logger::get(DEFAULT_LOGGER).set_log_level(fc::log_level::debug);
             std::vector<const char*> argv =
                   {"test", "--data-dir", temp.c_str(), "--config-dir", temp.c_str(),
-                   "-p", "eosio", "-e", "--max-transaction-time", "475", "--disable-subjective-billing=true"};
+                   "-p", "eosio", "-e", "--disable-subjective-billing=true"};
             appbase::app().initialize<chain_plugin, producer_plugin>(argv.size(), (char**) &argv[0]);
             appbase::app().startup();
             plugin_promise.set_value(

plugins/producer_plugin/test/test_trx_full.cpp

@@ -112,7 +112,7 @@ BOOST_AUTO_TEST_CASE(producer) {
          fc::logger::get(DEFAULT_LOGGER).set_log_level(fc::log_level::debug);
          std::vector<const char*> argv =
                {"test", "--data-dir", temp_dir_str.c_str(), "--config-dir", temp_dir_str.c_str(),
-                "-p", "eosio", "-e", "--max-transaction-time", "475", "--disable-subjective-billing=true" };
+                "-p", "eosio", "-e", "--disable-subjective-billing=true" };
          app->initialize<chain_plugin, producer_plugin>( argv.size(), (char**) &argv[0] );
          app->startup();
          plugin_promise.set_value(

plugins/test_control_api_plugin/test_control_api_plugin.cpp

@@ -51,7 +51,7 @@ void test_control_api_plugin::plugin_startup() {
 
    app().get_plugin<http_plugin>().add_api({
       TEST_CONTROL_RW_CALL(kill_node_on_producer, 202, http_params_types::params_required)
-   }, appbase::exec_queue::general);
+   }, appbase::exec_queue::read_write);
 }
 
 void test_control_api_plugin::plugin_shutdown() {}

plugins/wallet_api_plugin/wallet_api_plugin.cpp

@@ -115,7 +115,7 @@ void wallet_api_plugin::plugin_startup() {
             INVOKE_R_R_R(wallet_mgr, list_keys, std::string, std::string), 200),
        CALL_WITH_400(wallet, wallet_mgr, get_public_keys,
             INVOKE_R_V(wallet_mgr, get_public_keys), 200)
-   }, appbase::exec_queue::general);
+   }, appbase::exec_queue::read_write);
 }
 
 void wallet_api_plugin::plugin_initialize(const variables_map& options) {

programs/keosd/main.cpp

@@ -111,7 +111,7 @@ int main(int argc, char** argv)
                        [&a=app](string, string, url_response_callback cb) {
          cb(200, fc::time_point::maximum(), fc::variant(fc::variant_object()));
          a->quit();
-      }, appbase::exec_queue::general );
+      }, appbase::exec_queue::read_write );
       app->startup();
       app->exec();
    } catch (const fc::exception& e) {

tests/CMakeLists.txt

@@ -125,11 +125,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 --clean-run ${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 --clean-run ${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 0 --clean-run ${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 3 --clean-run ${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 3 --num-test-runs 3 --clean-run ${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 --read-only-threads 3 --clean-run ${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 --read-only-threads 3 --num-test-runs 3 --clean-run ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST read-only-trx-parallel-eos-vm-oc-test PROPERTY LABELS nonparallelizable_tests)
 add_test(NAME subjective_billing_test COMMAND tests/subjective_billing_test.py -v -p 2 -n 4 --clean-run ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST subjective_billing_test PROPERTY LABELS nonparallelizable_tests)

tests/read_only_trx_test.py

@@ -21,6 +21,7 @@ errorExit=Utils.errorExit
 
 appArgs=AppArgs()
 appArgs.add(flag="--read-only-threads", type=int, help="number of read-only threads", default=0)
+appArgs.add(flag="--num-test-runs", type=int, help="number of times to run the tests", default=1)
 appArgs.add_bool(flag="--eos-vm-oc-enable", help="enable eos-vm-oc")
 appArgs.add(flag="--wasm-runtime", type=str, help="if set to eos-vm-oc, must compile with EOSIO_EOS_VM_OC_DEVELOPER", default="eos-vm-jit")
 
@@ -44,6 +45,7 @@ dontKill=args.leave_running
 dumpErrorDetails=args.dump_error_details
 killAll=args.clean_run
 keepLogs=args.keep_logs
+numTestRuns=args.num_test_runs
 
 killWallet=not dontKill
 killEosInstances=not dontKill
@@ -91,9 +93,8 @@ try:
     specificExtraNodeosArgs={}
     # producer nodes will be mapped to 0 through pnodes-1, so the number pnodes is the no-producing API node
     specificExtraNodeosArgs[pnodes]=" --plugin eosio::net_api_plugin"
-    if args.read_only_threads > 0:
-        specificExtraNodeosArgs[pnodes]+=" --read-only-threads "
-        specificExtraNodeosArgs[pnodes]+=str(args.read_only_threads)
+    specificExtraNodeosArgs[pnodes]+=" --read-only-threads "
+    specificExtraNodeosArgs[pnodes]+=str(args.read_only_threads)
     if args.eos_vm_oc_enable:
         specificExtraNodeosArgs[pnodes]+=" --eos-vm-oc-enable"
     if args.wasm_runtime:
@@ -259,7 +260,7 @@ try:
         runReadOnlyTrxAndRpcInParallel("chain", "get_raw_code_and_abi", "account_name", expectedValue=testAccountName, payload = {"account_name":testAccountName})
         runReadOnlyTrxAndRpcInParallel("chain", "get_raw_abi", "account_name", expectedValue=testAccountName, payload = {"account_name":testAccountName})
         runReadOnlyTrxAndRpcInParallel("chain", "get_producers", "rows", payload = {"json":"true","lower_bound":""})
-        runReadOnlyTrxAndRpcInParallel("chain", "get_table_rows", code=500, payload = {"json":"true","table":"noauth"})
+        runReadOnlyTrxAndRpcInParallel("chain", "get_table_rows", "rows", payload = {"json":"true","code":"eosio","scope":"eosio","table":"global"})
         runReadOnlyTrxAndRpcInParallel("chain", "get_table_by_scope", fieldIn="rows", payload = {"json":"true","table":"noauth"})
         runReadOnlyTrxAndRpcInParallel("chain", "get_currency_balance", code=200, payload = {"code":"eosio.token", "account":testAccountName})
         runReadOnlyTrxAndRpcInParallel("chain", "get_currency_stats", fieldIn="SYS", payload = {"code":"eosio.token", "symbol":"SYS"})
@@ -279,10 +280,11 @@ try:
     basicTests()
 
     if args.read_only_threads > 0: # Save test time. No need to run other tests if multi-threaded is not enabled
-        multiReadOnlyTests()
-        chainApiTests()
-        netApiTests()
-        mixedOpsTest()
+        for i in range(numTestRuns):
+            multiReadOnlyTests()
+            chainApiTests()
+            netApiTests()
+            mixedOpsTest()
 
     testSuccessful = True
 finally: