Merge branch 'fix_missing_node_py_changes' of https://github.com/AntelopeIO/leap into fix_missing_node_py_changes

fix noisy `ignoring packed attribute` warning on `IR::ControlStructureImm`

libraries/chain/block_log.cpp

@@ -1533,7 +1533,7 @@ namespace eosio { namespace chain {
       ilog("blocks.log and blocks.index agree on number of blocks");
 
       if (interval == 0) {
-         interval = std::max((log_bundle.log_index.num_blocks() + 7) >> 3, 1);
+         interval = std::max((log_bundle.log_index.num_blocks() + 7u) >> 3, 1u);
       }
       uint32_t expected_block_num = log_bundle.log_data.first_block_num();
 

libraries/chain/include/eosio/chain/log_index.hpp

@@ -32,7 +32,7 @@ class log_index {
    bool is_open() const { return file_.is_open(); }
 
    uint64_t back() { return nth_block_position(num_blocks()-1); }
-   int      num_blocks() const { return num_blocks_; }
+   unsigned num_blocks() const { return num_blocks_; }
    uint64_t nth_block_position(uint32_t n) {
       file_.seek(n*sizeof(uint64_t));
       uint64_t r;

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

@@ -1,33 +1,30 @@
 #pragma once
 
 #include <appbase/application_base.hpp>
-#include <eosio/chain/exec_pri_queue.hpp>
-#include <mutex>
+#include <appbase/execution_priority_queue.hpp>
 
 /*
- * Customize appbase to support two-queue executor.
+ * Custmomize appbase to support two-queue exector.
 */
 namespace appbase { 
 
 enum class exec_window {
-   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.
+   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.
 };
 
 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
-                       // 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.
+   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
 };
 
 class two_queue_executor {
@@ -35,64 +32,56 @@ public:
 
    template <typename Func>
    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)));
+      if ( q == exec_queue::general )
+         return boost::asio::post(io_serv_, general_queue_.wrap(priority, --order_, std::forward<Func>(func)));
       else
-         return boost::asio::post( io_serv_, read_only_queue_.wrap( priority, --order_, std::forward<Func>( func)));
+         return boost::asio::post(io_serv_, read_only_trx_safe_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 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)));
+      // 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)));
    }
 
    boost::asio::io_service& get_io_service() { return io_serv_; }
 
    bool execute_highest() {
       if ( exec_window_ == exec_window::write ) {
-         // 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();
+         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();
          }
-         return !read_only_queue_.empty() || !read_write_queue_.empty();
+         return !read_only_trx_safe_queue_.empty() || !general_queue_.empty();
       } else {
-         // When in read window, multiple threads including main app thread are accessing two_queue_executor, locking required
-         return read_only_queue_.execute_highest_locked();
+         return read_only_trx_safe_queue_.execute_highest();
       }
    }
 
-   bool execute_highest_read_only() {
-      return read_only_queue_.execute_highest_locked();
-   }
-
    template <typename Function>
-   boost::asio::executor_binder<Function, appbase::exec_pri_queue::executor>
+   boost::asio::executor_binder<Function, appbase::execution_priority_queue::executor>
    wrap(int priority, exec_queue q, Function&& func ) {
-      if ( q == exec_queue::read_write )
-         return read_write_queue_.wrap(priority, --order_, std::forward<Function>(func));
+      if ( q == exec_queue::general )
+         return general_queue_.wrap(priority, --order_, std::forward<Function>(func));
       else
-         return read_only_queue_.wrap( priority, --order_, std::forward<Function>( func));
+         return read_only_trx_safe_queue_.wrap(priority, --order_, std::forward<Function>(func));
    }
      
    void clear() {
-      read_only_queue_.clear();
-      read_write_queue_.clear();
+      read_only_trx_safe_queue_.clear();
+      general_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 {
@@ -103,15 +92,15 @@ public:
       return exec_window_ == exec_window::write;
    }
 
-   auto& read_only_queue() { return read_only_queue_; }
+   auto& read_only_trx_safe_queue() { return read_only_trx_safe_queue_; }
 
-   auto& read_write_queue() { return read_write_queue_; }
+   auto& general_queue() { return general_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::execution_priority_queue  read_only_trx_safe_queue_;
+   appbase::execution_priority_queue  general_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

@@ -1,203 +0,0 @@
-#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,  [&]() { 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; } );
+   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; } );
 
    // Stop app. Use the lowest priority to make sure this function to execute the last
-   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->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->quit();
       } );
    app_thread.join();
 
    // both queues are cleared after execution 
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().general_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 queue are processed during read window
+// verify functions only from read_only_trx_safe 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 queue only
+   // set to run functions from read_only_trx_safe 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::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; } );
+   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; } );
 
    // stop application. Use lowest at the end to make sure this executes the last
-   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->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->quit();
       } );
    app_thread.join();
 
    // both queues are cleared after execution 
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().general_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 queue is empty
+// verify no functions are executed during  read window if read_only_trx_safe 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 queue only
+   // set to run functions from read_only_trx_safe 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::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; } );
+   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; } );
 
    // Stop application. Use lowest at the end to make sure this executes the last
-   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->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->quit();
       } );
    app_thread.join();
 
    // both queues are cleared after execution 
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().general_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,  [&]() { 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; } );
+   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; } );
 
    // stop application. Use lowest at the end to make sure this executes the last
-   app->executor().post( priority::lowest, exec_queue::read_only, [&]() {
+   app->executor().post( priority::lowest, exec_queue::read_only_trx_safe, [&]() {
       // read_queue should have current function and write_queue's functions are all executed 
-      BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().size(), 1); // pop()s after execute
-      BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().size(), 0 );
+      BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().size(), 1);
+      BOOST_REQUIRE_EQUAL( app->executor().general_queue().size(), 0 );
       app->quit();
       } );
 
    app_thread.join();
 
    // queues are emptied after quit
-   BOOST_REQUIRE_EQUAL( app->executor().read_only_queue().empty(), true);
-   BOOST_REQUIRE_EQUAL( app->executor().read_write_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().read_only_trx_safe_queue().empty(), true);
+   BOOST_REQUIRE_EQUAL( app->executor().general_queue().empty(), true);
 
    // exactly number of posts processed
    BOOST_REQUIRE_EQUAL( rslts.size(), 12 );

libraries/wasm-jit/Include/IR/Operators.h

@@ -12,10 +12,11 @@ namespace IR
 	struct NoImm {};
 	struct MemoryImm {};
 
+	PACKED_STRUCT(
 	struct ControlStructureImm
 	{
 		ResultType resultType{};
-	};
+	});
 
 	struct BranchImm
 	{
@@ -675,4 +676,19 @@ namespace IR
 	};
 
 	IR_API const char* getOpcodeName(Opcode opcode);
-}
+}
+
+//paranoia for future platforms
+static_assert(sizeof(IR::OpcodeAndImm<IR::NoImm>) == 2);
+static_assert(sizeof(IR::OpcodeAndImm<IR::MemoryImm>) == 2);
+static_assert(sizeof(IR::OpcodeAndImm<IR::ControlStructureImm>) == 3);
+static_assert(sizeof(IR::OpcodeAndImm<IR::BranchImm>) == 6);
+static_assert(sizeof(IR::OpcodeAndImm<IR::BranchTableImm>) == 18);
+static_assert(sizeof(IR::OpcodeAndImm<IR::LiteralImm<I32>>) == 6);
+static_assert(sizeof(IR::OpcodeAndImm<IR::LiteralImm<I64>>) == 10);
+static_assert(sizeof(IR::OpcodeAndImm<IR::LiteralImm<F32>>) == 6);
+static_assert(sizeof(IR::OpcodeAndImm<IR::LiteralImm<F64>>) == 10);
+static_assert(sizeof(IR::OpcodeAndImm<IR::GetOrSetVariableImm<false>>) == 6);
+static_assert(sizeof(IR::OpcodeAndImm<IR::CallImm>) == 6);
+static_assert(sizeof(IR::OpcodeAndImm<IR::CallIndirectImm>) == 6);
+static_assert(sizeof(IR::OpcodeAndImm<IR::LoadOrStoreImm<0>>) == 10);

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, appbase::priority::medium_high);
+      CHAIN_RO_CALL(get_info, 200, http_params_types::no_params)}, appbase::exec_queue::read_only_trx_safe, 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);
+   }, appbase::exec_queue::read_only_trx_safe);
 
    // 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::read_write, appbase::priority::medium_low );
+   }, appbase::exec_queue::general, 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);
+      }, appbase::exec_queue::read_only_trx_safe);
    }
 
    _http_plugin.add_api({
@@ -160,7 +160,7 @@ void chain_api_plugin::plugin_startup() {
            }
         }
       }
-   }, appbase::exec_queue::read_only);
+   }, appbase::exec_queue::read_only_trx_safe);
 }
 
 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);
+   }, appbase::exec_queue::read_only_trx_safe);
 }
 
 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);
+            }}, appbase::exec_queue::read_only_trx_safe);
             
          } catch (...) {
             fc_elog(logger(), "http_plugin startup fails, shutting down");

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

@@ -87,12 +87,8 @@ 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::read_write);
+         }, appbase::exec_queue::general);
    }
 
 private:

plugins/net_api_plugin/net_api_plugin.cpp

@@ -53,7 +53,11 @@ 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_async_api({
+   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),
        CALL_WITH_400(net, net_mgr, connect,
             INVOKE_R_R(net_mgr, connect, std::string), 201),
        CALL_WITH_400(net, net_mgr, disconnect,
@@ -62,7 +66,9 @@ 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::read_write, [ptr{std::move(ptr)}, bsp{std::move(bsp)}, id, c{std::move(c)}]() mutable {
+         app().executor().post(priority::medium, exec_queue::general, [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,11 +118,9 @@ 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),
-       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);
+   }, appbase::exec_queue::read_only_trx_safe, appbase::priority::medium_high);
 
-   // Not safe to run in parallel
+   // Not safe to run in parallel with read-only transactions
    app().get_plugin<http_plugin>().add_api({
        CALL_WITH_400(producer, producer, pause,
             INVOKE_V_V(producer, pause), 201),
@@ -140,13 +138,15 @@ 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::read_write, appbase::priority::medium_high);
+   }, appbase::exec_queue::general, appbase::priority::medium_high);
 }
 
 void producer_api_plugin::plugin_initialize(const variables_map& options) {

plugins/producer_plugin/producer_plugin.cpp

@@ -14,7 +14,6 @@
 #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>
@@ -325,9 +324,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,
-                                    const next_function<transaction_trace_ptr>& next );
+                                    next_function<transaction_trace_ptr> next );
       push_result handle_push_result( const transaction_metadata_ptr& trx,
-                                      const next_function<transaction_trace_ptr>& next,
+                                      next_function<transaction_trace_ptr> next,
                                       const fc::time_point& start,
                                       const chain::controller& chain,
                                       const transaction_trace_ptr& trace,
@@ -418,7 +417,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          packed_transaction_ptr  trx;
          next_func_t             next;
       };
-      // The queue storing previously exhausted read-only transactions to be re-executed by read-only threads
+      // The queue storing read-only transactions to be executed by read-only threads
       struct ro_trx_queue_t {
          std::mutex              mtx;
          std::deque<ro_trx_t>    queue;
@@ -430,26 +429,24 @@ 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_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;
+      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;
 
       void start_write_window();
       void switch_to_write_window();
       void switch_to_read_window();
-      bool read_only_execution_task();
-      void process_read_only_transactions(const fc::time_point& deadline);
+      bool read_only_trx_execution_task();
       bool process_read_only_transaction(const packed_transaction_ptr& trx,
-                                         const next_function<transaction_trace_ptr>& next);
+                                         const next_function<transaction_trace_ptr>& next,
+                                         const fc::time_point& read_window_start_time);
       bool push_read_only_transaction(const transaction_metadata_ptr& trx,
-                                      const next_function<transaction_trace_ptr>& next);
-
+                                      next_function<transaction_trace_ptr> next,
+                                      const fc::time_point& read_window_start_time);
+      
       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() ) {
@@ -642,11 +639,12 @@ 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 ) {
-            // 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 );
-            } );
+         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)});
             return;
          }
 
@@ -679,7 +677,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::read_write, [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::general, [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 );
@@ -711,7 +709,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,
-                                              const next_function<transaction_trace_ptr>& next) {
+                                              next_function<transaction_trace_ptr> next) {
          bool exhausted = false;
          chain::controller& chain = chain_plug->chain();
          try {
@@ -894,7 +892,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>(),
+         ("read-only-threads", bpo::value<uint16_t>()->default_value(my->_ro_thread_pool_size),
           "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")
@@ -1075,15 +1073,11 @@ void producer_plugin::plugin_initialize(const boost::program_options::variables_
       }
    }
 
-   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;
-   }
+   my->_ro_thread_pool_size = options.at( "read-only-threads" ).as<uint16_t>();
    // 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
@@ -1115,26 +1109,25 @@ 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;
+      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;
+      // 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));
    }
-   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) {
@@ -2001,8 +1994,6 @@ 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() ) {
@@ -2204,10 +2195,9 @@ producer_plugin_impl::push_transaction( const fc::time_point& block_deadline,
                                         const transaction_metadata_ptr& trx,
                                         bool api_trx,
                                         bool return_failure_trace,
-                                        const next_function<transaction_trace_ptr>& next )
+                                        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)
@@ -2244,6 +2234,14 @@ 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);
@@ -2251,7 +2249,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,
-                                          const next_function<transaction_trace_ptr>& next,
+                                          next_function<transaction_trace_ptr> next,
                                           const fc::time_point& start,
                                           const chain::controller& chain,
                                           const transaction_trace_ptr& trace,
@@ -2557,7 +2555,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::read_write,
+      _timer.async_wait( app().executor().wrap( priority::high, exec_queue::general,
           [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 ) {
@@ -2610,7 +2608,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::read_write,
+   _timer.async_wait( app().executor().wrap( priority::high, exec_queue::general,
          [&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 ) {
@@ -2652,7 +2650,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::read_write,
+      _timer.async_wait( app().executor().wrap( priority::high, exec_queue::general,
          [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 ) {
@@ -2766,7 +2764,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 only one read_only thread
+// Called from app 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.
@@ -2774,29 +2772,26 @@ void producer_plugin_impl::switch_to_write_window() {
       return;
    }
 
-   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");
+   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");
    _ro_read_window_timer.cancel();
    _ro_write_window_timer.cancel();
 
    start_write_window();
 }
 
-// Called from app thread on plugin_startup
-// Called from only one read_only thread & called from app thread, but not concurrently
+// Called from app thread
 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( // run from app thread
+   _ro_write_window_timer.async_wait( app().executor().wrap(  // stay on app thread
       priority::high,
-      exec_queue::read_write, // placed in read_write so only called from main thread
+      exec_queue::read_only_trx_safe, // placed in read_only_trx_safe queue so it is ensured to be executed in either window
       [weak_this = weak_from_this()]( const boost::system::error_code& ec ) {
          auto self = weak_this.lock();
          if( self && ec != boost::asio::error::operation_aborted ) {
@@ -2805,98 +2800,86 @@ void producer_plugin_impl::start_write_window() {
       }));
 }
 
-// Called only from app thread
+// Called 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_exec_tasks.load() == 0 && _ro_exec_tasks_fut.empty(), producer_exception, "_ro_exec_tasks_fut expected to be empty" );
+   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" );
 
    _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_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
+   // _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
       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_exec_tasks = _ro_thread_pool_size;
-   _ro_window_deadline = fc::time_point::now() + _ro_read_window_effective_time_us;
+   _ro_num_active_trx_exec_tasks = _ro_thread_pool_size;
    for (auto i = 0; i < _ro_thread_pool_size; ++i ) {
-      _ro_exec_tasks_fut.emplace_back( post_async_task( _ro_thread_pool.get_executor(), [self = this] () {
-         return self->read_only_execution_task();
+      _ro_trx_exec_tasks_fut.emplace_back( post_async_task( _ro_thread_pool.get_executor(), [self = this] () {
+         return self->read_only_trx_execution_task();
       }) );
    }
 
    auto expire_time = boost::posix_time::microseconds(_ro_read_window_time_us.count());
    _ro_read_window_timer.expires_from_now( expire_time );
-   // 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(
+   _ro_read_window_timer.async_wait( app().executor().wrap(  // stay on app thread
       priority::high,
-      exec_queue::read_only,
+      exec_queue::read_only_trx_safe,
       [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_exec_tasks_fut ) {
+            for ( auto& task: self->_ro_trx_exec_tasks_fut ) {
                task.get();
             }
-            self->_ro_exec_tasks_fut.clear();
-            // will be executed from the main app thread because all read-only threads are idle now
+            self->_ro_trx_exec_tasks_fut.clear();
             self->switch_to_write_window();
           } else if ( self ) {
-             self->_ro_exec_tasks_fut.clear();
+             self->_ro_trx_exec_tasks_fut.clear();
           }
        }));
 }
 
-// Called from a read only thread. Run in parallel with app and other read only threads
-bool producer_plugin_impl::read_only_execution_task() {
+// 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;
    // 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 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;
-         }
+   // 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;
       }
    }
 
    // If all tasks are finished, do not wait until end of read window; switch to write window now.
-   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
+   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();
          self->switch_to_write_window();
       } );
    }
@@ -2904,28 +2887,9 @@ bool producer_plugin_impl::read_only_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) {
+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) {
    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(),
@@ -2938,19 +2902,17 @@ bool producer_plugin_impl::process_read_only_transaction(const packed_transactio
    future.wait();
    try {
       auto trx_metadata = future.get();
-      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;
+      return push_read_only_transaction( trx_metadata, next, read_window_start_time );
    } 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, const next_function<transaction_trace_ptr>& next) {
+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) {
    auto retry = false;
    chain::controller& chain = chain_plug->chain();
 
@@ -2960,24 +2922,18 @@ bool producer_plugin_impl::push_read_only_transaction(const transaction_metadata
    }
 
    try {
+      const auto block_deadline = calculate_block_deadline( chain.pending_block_time() );
       auto start = fc::time_point::now();
-      if ( start >= _ro_window_deadline ) {
+      auto time_used_in_read_window_us = ( start - read_window_start_time );
+      if ( time_used_in_read_window_us >= _ro_read_window_time_us ) {
          // already passed read-window deadline, try next round
          return true;
       }
 
-      // 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();
-
+      auto remaining_time_in_read_window_us = _ro_read_window_time_us - time_used_in_read_window_us;
       // Ensure the trx to finish by the end of read-window.
-      auto trace = chain.push_transaction( trx, _ro_window_deadline, _ro_max_trx_time_us, 0, false, 0 );
+      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 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::read_write, [ptrx, &next_calls, &num_posts, &trace_with_except, &trx_match, &app]() {
+      app->executor().post( priority::low, exec_queue::general, [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", "--disable-subjective-billing=true"};
+                   "-p", "eosio", "-e", "--max-transaction-time", "475", "--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", "--disable-subjective-billing=true" };
+                "-p", "eosio", "-e", "--max-transaction-time", "475", "--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::read_write);
+   }, appbase::exec_queue::general);
 }
 
 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::read_write);
+   }, appbase::exec_queue::general);
 }
 
 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::read_write );
+      }, appbase::exec_queue::general );
       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 --read-only-threads 0 --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 --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 --num-test-runs 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 --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 --num-test-runs 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 --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/TestHarness/Cluster.py

@@ -1727,4 +1727,7 @@ class Cluster(object):
                 for line in f:
                     firstTrxs.append(line.rstrip('\n'))
         Utils.Print(f"first transactions: {firstTrxs}")
-        node.waitForTransactionsInBlock(firstTrxs)
+        status = node.waitForTransactionsInBlock(firstTrxs)
+        if status is None:
+            Utils.Print('ERROR: Failed to spin up transaction generators: never received first transactions')
+        return status

tests/TestHarness/queries.py

@@ -118,10 +118,10 @@ class NodeosQueries:
         # could be a transaction response
         if cntxt.hasKey("processed"):
             cntxt.add("processed")
-            cntxt.add("action_traces")
-            cntxt.index(0)
             if not cntxt.isSectionNull("except"):
                 return "no_block"
+            cntxt.add("action_traces")
+            cntxt.index(0)
             return cntxt.add("block_num")
 
         # or what the trace api plugin returns
@@ -242,7 +242,7 @@ class NodeosQueries:
         assert(isinstance(transId, str))
         exitOnErrorForDelayed=not delayedRetry and exitOnError
         timeout=3
-        cmdDesc="get transaction_trace"
+        cmdDesc=self.fetchTransactionCommand()
         cmd="%s %s" % (cmdDesc, transId)
         msg="(transaction id=%s)" % (transId);
         for i in range(0,(int(60/timeout) - 1)):
@@ -295,8 +295,8 @@ class NodeosQueries:
         refBlockNum=None
         key=""
         try:
-            key="[transaction][transaction_header][ref_block_num]"
-            refBlockNum=trans["transaction_header"]["ref_block_num"]
+            key = self.fetchKeyCommand()
+            refBlockNum = self.fetchRefBlock(trans)
             refBlockNum=int(refBlockNum)+1
         except (TypeError, ValueError, KeyError) as _:
             Utils.Print("transaction%s not found. Transaction: %s" % (key, trans))
@@ -347,8 +347,8 @@ class NodeosQueries:
 
     def getTable(self, contract, scope, table, exitOnError=False):
         cmdDesc = "get table"
-        cmd="%s %s %s %s" % (cmdDesc, contract, scope, table)
-        msg="contract=%s, scope=%s, table=%s" % (contract, scope, table);
+        cmd=f"{cmdDesc} {self.cleosLimit} {contract} {scope} {table}"
+        msg=f"contract={contract}, scope={scope}, table={table}"
         return self.processCleosCmd(cmd, cmdDesc, exitOnError=exitOnError, exitMsg=msg)
 
     def getTableAccountBalance(self, contract, scope):
@@ -529,7 +529,7 @@ class NodeosQueries:
             return m.group(1)
         except subprocess.CalledProcessError as ex:
             end=time.perf_counter()
-            msg=ex.output.decode("utf-8")
+            msg=ex.stderr.decode("utf-8")
             Utils.Print("ERROR: Exception during code hash retrieval.  cmd Duration: %.3f sec.  %s" % (end-start, msg))
             return None
 
@@ -580,8 +580,9 @@ class NodeosQueries:
         except subprocess.CalledProcessError as ex:
             if not silentErrors:
                 end=time.perf_counter()
-                msg=ex.output.decode("utf-8")
-                errorMsg="Exception during \"%s\". Exception message: %s.  cmd Duration=%.3f sec. %s" % (cmdDesc, msg, end-start, exitMsg)
+                out=ex.output.decode("utf-8")
+                msg=ex.stderr.decode("utf-8")
+                errorMsg="Exception during \"%s\". Exception message: %s.  stdout: %s.  cmd Duration=%.3f sec. %s" % (cmdDesc, msg, out, end-start, exitMsg)
                 if exitOnError:
                     Utils.cmdError(errorMsg)
                     Utils.errorExit(errorMsg)

tests/TestHarness/transactions.py

@@ -107,7 +107,7 @@ class Transactions(NodeosQueries):
                 self.trackCmdTransaction(trans, reportStatus=reportStatus)
         except subprocess.CalledProcessError as ex:
             end=time.perf_counter()
-            msg=ex.output.decode("utf-8")
+            msg=ex.stderr.decode("utf-8")
             Utils.Print("ERROR: Exception during funds transfer.  cmd Duration: %.3f sec.  %s" % (end-start, msg))
             if exitOnError:
                 Utils.cmdError("could not transfer \"%s\" from %s to %s" % (amountStr, source, destination))
@@ -131,7 +131,7 @@ class Transactions(NodeosQueries):
                 Utils.Print("cmd Duration: %.3f sec" % (end-start))
         except subprocess.CalledProcessError as ex:
             end=time.perf_counter()
-            msg=ex.output.decode("utf-8")
+            msg=ex.stderr.decode("utf-8")
             Utils.Print("ERROR: Exception during spawn of funds transfer.  cmd Duration: %.3f sec.  %s" % (end-start, msg))
             if exitOnError:
                 Utils.cmdError("could not transfer \"%s\" from %s to %s" % (amountStr, source, destination))
@@ -158,15 +158,15 @@ class Transactions(NodeosQueries):
         except subprocess.CalledProcessError as ex:
             if not shouldFail:
                 end=time.perf_counter()
-                msg=ex.output.decode("utf-8")
-                Utils.Print("ERROR: Exception during set contract.  cmd Duration: %.3f sec.  %s" % (end-start, msg))
+                out=ex.output.decode("utf-8")
+                msg=ex.stderr.decode("utf-8")
+                Utils.Print("ERROR: Exception during set contract. stderr: %s.  stdout: %s.  cmd Duration: %.3f sec." % (msg, out, end-start))
                 return None
             else:
                 retMap={}
                 retMap["returncode"]=ex.returncode
                 retMap["cmd"]=ex.cmd
                 retMap["output"]=ex.output
-                retMap["stdout"]=ex.stdout
                 retMap["stderr"]=ex.stderr
                 return retMap
 
@@ -213,7 +213,7 @@ class Transactions(NodeosQueries):
                 Utils.Print("cmd Duration: %.3f sec" % (end-start))
             return (NodeosQueries.getTransStatus(retTrans) == 'executed', retTrans)
         except subprocess.CalledProcessError as ex:
-            msg=ex.output.decode("utf-8")
+            msg=ex.stderr.decode("utf-8")
             if not silentErrors:
                 end=time.perf_counter()
                 Utils.Print("ERROR: Exception during push transaction.  cmd Duration=%.3f sec.  %s" % (end - start, msg))
@@ -245,7 +245,6 @@ class Transactions(NodeosQueries):
         except subprocess.CalledProcessError as ex:
             msg=ex.stderr.decode("utf-8")
             output=ex.output.decode("utf-8")
-            msg=ex.output.decode("utf-8")
             if not silentErrors:
                 end=time.perf_counter()
                 Utils.Print("ERROR: Exception during push message. stderr: %s. stdout: %s.  cmd Duration=%.3f sec." % (msg, output, end - start))

tests/nodeos_snapshot_diff_test.py

@@ -126,13 +126,15 @@ try:
     waitForBlock(node0, blockNum, blockType=BlockType.lib)
 
     Print("Configure and launch txn generators")
+    
     targetTpsPerGenerator = 10
     testTrxGenDurationSec=60*30
     cluster.launchTrxGenerators(contractOwnerAcctName=cluster.eosioAccount.name, acctNamesList=[account1Name, account2Name],
                                 acctPrivKeysList=[account1PrivKey,account2PrivKey], nodeId=snapshotNodeId, tpsPerGenerator=targetTpsPerGenerator,
                                 numGenerators=trxGeneratorCnt, durationSec=testTrxGenDurationSec, waitToComplete=False)
 
-    cluster.waitForTrxGeneratorsSpinup(nodeId=snapshotNodeId, numGenerators=trxGeneratorCnt)
+    status = cluster.waitForTrxGeneratorsSpinup(nodeId=snapshotNodeId, numGenerators=trxGeneratorCnt)
+    assert status is not None, "ERROR: Failed to spinup Transaction Generators"
 
     blockNum=node0.getBlockNum(BlockType.head)
     timePerBlock=500
@@ -156,12 +158,8 @@ try:
     minReqPctLeeway=0.60
     minRequiredTransactions=minReqPctLeeway*transactionsPerBlock
     assert steadyStateAvg>=minRequiredTransactions, "Expected to at least receive %s transactions per block, but only getting %s" % (minRequiredTransactions, steadyStateAvg)
-
-    Print("Create snapshot")
-    blockNumSnap=nodeSnap.getBlockNum(BlockType.head)
-    ret = nodeProg.scheduleSnapshotAt(blockNumSnap)
-    assert ret is not None, "Snapshot creation failed"
     
+    Print("Create snapshot (node 0)")
     ret = nodeSnap.createSnapshot()
     assert ret is not None, "Snapshot creation failed"
     ret_head_block_num = ret["payload"]["head_block_num"]
@@ -178,6 +176,29 @@ try:
     Utils.processLeapUtilCmd("snapshot to-json --input-file {}".format(snapshotFile), "snapshot to-json", silentErrors=False)
     snapshotFile = snapshotFile + ".json"
 
+    Print("Trim programmable blocklog to snapshot head block num and relaunch programmable node")
+    nodeProg.kill(signal.SIGTERM)
+    output=cluster.getBlockLog(progNodeId, blockLogAction=BlockLogAction.trim, first=0, last=ret_head_block_num, throwException=True)
+    removeState(progNodeId)
+    Utils.rmFromFile(Utils.getNodeConfigDir(progNodeId, "config.ini"), "p2p-peer-address")
+    isRelaunchSuccess = nodeProg.relaunch(chainArg="--replay", addSwapFlags={}, timeout=relaunchTimeout, cachePopen=True)
+    assert isRelaunchSuccess, "Failed to relaunch programmable node"
+
+    Print("Schedule snapshot (node 2)")
+    ret = nodeProg.scheduleSnapshotAt(ret_head_block_num)
+    assert ret is not None, "Snapshot scheduling failed"
+
+    Print("Wait for programmable node lib to advance")
+    waitForBlock(nodeProg, ret_head_block_num+1, blockType=BlockType.lib)
+
+    Print("Kill programmable node")
+    nodeProg.kill(signal.SIGTERM)
+
+    Print("Convert snapshot to JSON")
+    progSnapshotFile = getLatestSnapshot(progNodeId)
+    Utils.processLeapUtilCmd("snapshot to-json --input-file {}".format(progSnapshotFile), "snapshot to-json", silentErrors=False)
+    progSnapshotFile = progSnapshotFile + ".json"
+
     Print("Trim irreversible blocklog to snapshot head block num")
     nodeIrr.kill(signal.SIGTERM)
     output=cluster.getBlockLog(irrNodeId, blockLogAction=BlockLogAction.trim, first=0, last=ret_head_block_num, throwException=True)
@@ -205,15 +226,6 @@ try:
     irrSnapshotFile = irrSnapshotFile + ".json"
 
     assert Utils.compareFiles(snapshotFile, irrSnapshotFile), f"Snapshot files differ {snapshotFile} != {irrSnapshotFile}"
-  
-    Print("Kill programmable node")
-    nodeProg.kill(signal.SIGTERM)
-
-    Print("Convert snapshot to JSON")
-    progSnapshotFile = getLatestSnapshot(progNodeId)
-    Utils.processLeapUtilCmd("snapshot to-json --input-file {}".format(progSnapshotFile), "snapshot to-json", silentErrors=False)
-    progSnapshotFile = progSnapshotFile + ".json"
-
     assert Utils.compareFiles(progSnapshotFile, irrSnapshotFile), f"Snapshot files differ {progSnapshotFile} != {irrSnapshotFile}"
 
     testSuccessful=True

tests/nodeos_startup_catchup.py

@@ -118,7 +118,8 @@ try:
                                 tpsPerGenerator=targetTpsPerGenerator, numGenerators=trxGeneratorCnt, durationSec=testTrxGenDurationSec,
                                 waitToComplete=False)
 
-    cluster.waitForTrxGeneratorsSpinup(nodeId=node0.nodeId, numGenerators=trxGeneratorCnt)
+    status = cluster.waitForTrxGeneratorsSpinup(nodeId=node0.nodeId, numGenerators=trxGeneratorCnt)
+    assert status is not None, "ERROR: Failed to spinup Transaction Generators"
 
     blockNum=head(node0)
     timePerBlock=500

tests/read_only_trx_test.py

@@ -21,7 +21,6 @@ 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")
 
@@ -45,7 +44,6 @@ 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
@@ -93,8 +91,9 @@ 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"
-    specificExtraNodeosArgs[pnodes]+=" --read-only-threads "
-    specificExtraNodeosArgs[pnodes]+=str(args.read_only_threads)
+    if args.read_only_threads > 0:
+        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:
@@ -260,7 +259,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", "rows", payload = {"json":"true","code":"eosio","scope":"eosio","table":"global"})
+        runReadOnlyTrxAndRpcInParallel("chain", "get_table_rows", code=500, payload = {"json":"true","table":"noauth"})
         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"})
@@ -280,11 +279,10 @@ try:
     basicTests()
 
     if args.read_only_threads > 0: # Save test time. No need to run other tests if multi-threaded is not enabled
-        for i in range(numTestRuns):
-            multiReadOnlyTests()
-            chainApiTests()
-            netApiTests()
-            mixedOpsTest()
+        multiReadOnlyTests()
+        chainApiTests()
+        netApiTests()
+        mixedOpsTest()
 
     testSuccessful = True
 finally: