Merge pull request #1447 from AntelopeIO/GH-1441-oc-compile-queue-1.1

[1.1.5] Fix OC scheduling of compiles

Author: heifner

libraries/chain/controller.cpp

@@ -4920,8 +4920,10 @@ struct controller_impl {
       return wasmif;
    }
 
-   void code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version, uint32_t block_num) {
-      wasmif.code_block_num_last_used(code_hash, vm_type, vm_version, block_num);
+   void code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version,
+                                 block_num_type first_used_block_num, block_num_type block_num_last_used)
+   {
+      wasmif.code_block_num_last_used(code_hash, vm_type, vm_version, first_used_block_num, block_num_last_used);
    }
 
    void set_node_finalizer_keys(const bls_pub_priv_key_map_t& finalizer_keys) {
@@ -6127,8 +6129,9 @@ bool controller::is_write_window() const {
    return my->is_write_window();
 }
 
-void controller::code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version, uint32_t block_num) {
-   return my->code_block_num_last_used(code_hash, vm_type, vm_version, block_num);
+void controller::code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version,
+                                          block_num_type first_used_block_num, block_num_type block_num_last_used) {
+   return my->code_block_num_last_used(code_hash, vm_type, vm_version, first_used_block_num, block_num_last_used);
 }
 
 void controller::set_node_finalizer_keys(const bls_pub_priv_key_map_t& finalizer_keys) {

libraries/chain/eosio_contract.cpp

@@ -160,7 +160,8 @@ void apply_eosio_setcode(apply_context& context) {
       old_size  = (int64_t)old_code_entry.code.size() * config::setcode_ram_bytes_multiplier;
       if( old_code_entry.code_ref_count == 1 ) {
          db.remove(old_code_entry);
-         context.control.code_block_num_last_used(account.code_hash, account.vm_type, account.vm_version, context.control.head().block_num() + 1);
+         context.control.code_block_num_last_used(account.code_hash, account.vm_type, account.vm_version,
+                                                  old_code_entry.first_block_used, context.control.head().block_num() + 1);
       } else {
          db.modify(old_code_entry, [](code_object& o) {
             --o.code_ref_count;

libraries/chain/include/eosio/chain/controller.hpp

@@ -489,7 +489,8 @@ namespace eosio::chain {
       void set_to_write_window();
       void set_to_read_window();
       bool is_write_window() const;
-      void code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version, uint32_t block_num);
+      void code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version,
+                                    block_num_type first_used_block_num, block_num_type block_num_last_used);
       void set_node_finalizer_keys(const bls_pub_priv_key_map_t& finalizer_keys);
 
       // is the bls key a registered finalizer key of this node, thread safe

libraries/chain/include/eosio/chain/wasm_interface.hpp

@@ -68,7 +68,8 @@ namespace eosio { namespace chain {
          static void validate(const controller& control, const bytes& code);
 
          //indicate that a particular code probably won't be used after given block_num
-         void code_block_num_last_used(const digest_type& code_hash, const uint8_t& vm_type, const uint8_t& vm_version, const uint32_t& block_num);
+         void code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version,
+                                       block_num_type first_used_block_num, block_num_type last_used_block_num);
 
          //indicate the current LIB. evicts old cache entries
          void current_lib(const uint32_t lib);

libraries/chain/include/eosio/chain/wasm_interface_private.hpp

@@ -207,16 +207,32 @@ struct eosvmoc_tier {
          return it != wasm_instantiation_cache.end();
       }
 
-      void code_block_num_last_used(const digest_type& code_hash, const uint8_t& vm_type, const uint8_t& vm_version, const uint32_t& block_num) {
+      void code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version,
+                                    block_num_type first_used_block_num, block_num_type block_num_last_used)
+      {
          // The caller of this method apply_eosio_setcode has asserted that
          // the transaction is not read-only, implying we are
          // in write window. Read-only threads are not running.
          // Safe to update the cache without locking.
          wasm_cache_index::iterator it = wasm_instantiation_cache.find(boost::make_tuple(code_hash, vm_type, vm_version));
-         if(it != wasm_instantiation_cache.end())
-            wasm_instantiation_cache.modify(it, [block_num](wasm_cache_entry& e) {
-               e.last_block_num_used = block_num;
-            });
+         if(it != wasm_instantiation_cache.end()) {
+            if (first_used_block_num == block_num_last_used) {
+               // First used and no longer needed in the same block, erase immediately, do not wait for LIB.
+               // Since created and destroyed in the same block, likely will not be needed in a forked block.
+               // Prevents many setcodes in the same block using up space in the cache.
+               wasm_instantiation_cache.erase(it);
+            } else {
+               wasm_instantiation_cache.modify(it, [block_num_last_used](wasm_cache_entry& e) {
+                  e.last_block_num_used = block_num_last_used;
+               });
+            }
+         }
+
+#ifdef EOSIO_EOS_VM_OC_RUNTIME_ENABLED
+         // see comment above
+         if (first_used_block_num == block_num_last_used && eosvmoc)
+            eosvmoc->cc.free_code(code_hash, vm_version);
+#endif
       }
 
       // reports each code_hash and vm_version that will be erased to callback

libraries/chain/include/eosio/chain/webassembly/eos-vm-oc/code_cache.hpp

@@ -5,17 +5,21 @@
 #include <boost/multi_index_container.hpp>
 #include <boost/multi_index/hashed_index.hpp>
 #include <boost/multi_index/sequenced_index.hpp>
-#include <boost/multi_index/composite_key.hpp>
 #include <boost/multi_index/key_extractors.hpp>
 #include <boost/multi_index/member.hpp>
-#include <boost/lockfree/spsc_queue.hpp>
+#include <boost/lockfree/queue.hpp>
 
 #include <boost/interprocess/mem_algo/rbtree_best_fit.hpp>
 #include <boost/asio/local/datagram_protocol.hpp>
 
 #include <fc/crypto/sha256.hpp>
 
+#include <atomic>
 #include <thread>
+#include <tuple>
+#include <unordered_map>
+#include <unordered_set>
+#include <filesystem>
 
 namespace eosio { namespace chain { namespace eosvmoc {
 
@@ -77,12 +81,12 @@ class code_cache_base {
 
       struct queued_compile_entry {
          compile_wasm_message    msg;
-         std::vector<wrapped_fd> fds_to_pass;
+         std::vector<char>       code;
 
          const digest_type&      code_id() const { return msg.code.code_id; }
       };
       //these are really only useful to the async code cache, but keep them here so free_code can be shared
-      using queued_compilies_t = boost::multi_index_container<
+      using queued_compiles_t = boost::multi_index_container<
          queued_compile_entry,
          indexed_by<
             sequenced<>,
@@ -91,7 +95,7 @@ class code_cache_base {
          >
       >;
       std::mutex                             _mtx;
-      queued_compilies_t                     _queued_compiles;                  // protected by _mtx
+      queued_compiles_t                      _queued_compiles;                  // protected by _mtx
       std::unordered_map<digest_type, bool>  _outstanding_compiles_and_poison;  // protected by _mtx
       std::atomic<size_t>                    _outstanding_compiles{0};
 
@@ -123,14 +127,14 @@ class code_cache_async : public code_cache_base {
    private:
       compile_complete_callback _compile_complete_func; // called from async thread, provides executing_action_id
       std::thread _monitor_reply_thread;
-      boost::lockfree::spsc_queue<wasm_compilation_result_message> _result_queue;
+      boost::lockfree::queue<wasm_compilation_result_message> _result_queue;
       std::unordered_set<digest_type> _blacklist;
       size_t _threads;
 
       void wait_on_compile_monitor_message();
       std::tuple<size_t, size_t> consume_compile_thread_queue();
       void process_queued_compiles();
-      void write_message(const digest_type& code_id, const eosvmoc_message& message, const std::vector<wrapped_fd>& fds);
+      void write_message(const digest_type& code_id, const eosvmoc_message& message, std::span<wrapped_fd> fds);
 
 };
 

libraries/chain/include/eosio/chain/webassembly/eos-vm-oc/ipc_helpers.hpp

@@ -47,8 +47,8 @@ class wrapped_fd {
 
 std::tuple<bool, eosvmoc_message, std::vector<wrapped_fd>> read_message_with_fds(boost::asio::local::datagram_protocol::socket& s);
 std::tuple<bool, eosvmoc_message, std::vector<wrapped_fd>> read_message_with_fds(int fd);
-bool write_message_with_fds(boost::asio::local::datagram_protocol::socket& s, const eosvmoc_message& message, const std::vector<wrapped_fd>& fds = std::vector<wrapped_fd>());
-bool write_message_with_fds(int fd_to_send_to, const eosvmoc_message& message, const std::vector<wrapped_fd>& fds = std::vector<wrapped_fd>());
+bool write_message_with_fds(boost::asio::local::datagram_protocol::socket& s, const eosvmoc_message& message, std::span<wrapped_fd> fds = {});
+bool write_message_with_fds(int fd_to_send_to, const eosvmoc_message& message, std::span<wrapped_fd> fds = {});
 
 template<typename T>
 wrapped_fd memfd_for_bytearray(const T& bytes) {

libraries/chain/wasm_interface.cpp

@@ -77,8 +77,10 @@ namespace eosio { namespace chain {
       //Hard: Kick off instantiation in a separate thread at this location
    }
 
-   void wasm_interface::code_block_num_last_used(const digest_type& code_hash, const uint8_t& vm_type, const uint8_t& vm_version, const uint32_t& block_num) {
-      my->code_block_num_last_used(code_hash, vm_type, vm_version, block_num);
+   void wasm_interface::code_block_num_last_used(const digest_type& code_hash, uint8_t vm_type, uint8_t vm_version,
+                                                 block_num_type first_used_block_num, block_num_type block_num_last_used)
+   {
+      my->code_block_num_last_used(code_hash, vm_type, vm_version, first_used_block_num, block_num_last_used);
    }
 
    void wasm_interface::current_lib(const uint32_t lib) {

libraries/chain/webassembly/runtimes/eos-vm-oc/code_cache.cpp

@@ -1,5 +1,3 @@
-#include <fc/log/logger_config.hpp> //set_thread_name
-
 #include <eosio/chain/webassembly/eos-vm-oc/code_cache.hpp>
 #include <eosio/chain/webassembly/eos-vm-oc/config.hpp>
 #include <eosio/chain/webassembly/common.hpp>
@@ -9,17 +7,11 @@
 #include <eosio/chain/webassembly/eos-vm-oc/compile_monitor.hpp>
 #include <eosio/chain/exceptions.hpp>
 
+#include <fc/log/logger_config.hpp> //set_thread_name
+
+#include <fstream>
 #include <unistd.h>
-#include <sys/syscall.h>
 #include <sys/mman.h>
-#include <linux/memfd.h>
-
-#include "IR/Module.h"
-#include "IR/Validate.h"
-#include "WASM/WASM.h"
-#include "LLVMJIT.h"
-
-using namespace IR;
 
 namespace eosio { namespace chain { namespace eosvmoc {
 
@@ -81,7 +73,8 @@ void code_cache_async::wait_on_compile_monitor_message() {
       --_outstanding_compiles;
 
       const auto& msg = std::get<wasm_compilation_result_message>(message);
-      _result_queue.push(msg);
+      bool p = _result_queue.push(msg);
+      assert(p);
 
       _compile_complete_func(_ctx, msg.code.code_id, msg.queued_time);
 
@@ -92,7 +85,7 @@ void code_cache_async::wait_on_compile_monitor_message() {
 }
 
 //call with _mtx locked
-void code_cache_async::write_message(const digest_type& code_id, const eosvmoc_message& message, const std::vector<wrapped_fd>& fds) {
+void code_cache_async::write_message(const digest_type& code_id, const eosvmoc_message& message, std::span<wrapped_fd> fds) {
    _outstanding_compiles_and_poison.emplace(code_id, false);
    ++_outstanding_compiles;
    if (!write_message_with_fds(_compile_monitor_write_socket, message, fds)) {
@@ -106,7 +99,8 @@ void code_cache_async::process_queued_compiles() {
    while (_outstanding_compiles < _threads && !_queued_compiles.empty()) {
       auto nextup = _queued_compiles.begin();
 
-      write_message(nextup->code_id(), nextup->msg, nextup->fds_to_pass);
+      auto fd = memfd_for_bytearray(nextup->code);
+      write_message(nextup->code_id(), nextup->msg, std::span<wrapped_fd>{&fd, 1});
 
       _queued_compiles.erase(nextup);
    }
@@ -116,7 +110,9 @@ void code_cache_async::process_queued_compiles() {
 //number processed, bytes available (only if number processed > 0)
 std::tuple<size_t, size_t> code_cache_async::consume_compile_thread_queue() {
    std::unique_lock g(_mtx);
-   auto outstanding_compiles = _outstanding_compiles_and_poison; // will always be small, <= _threads
+   // will be relatively small, ~ _threads. Can be larger than _threads if multiple compiles finish before
+   // consume_compile_thread_queue() is called on the main thread
+   auto outstanding_compiles = _outstanding_compiles_and_poison;
    g.unlock();
 
    std::vector<digest_type> erased;
@@ -142,8 +138,10 @@ std::tuple<size_t, size_t> code_cache_async::consume_compile_thread_queue() {
    });
 
    g.lock();
-   for (const auto& e : erased)
-      _outstanding_compiles_and_poison.erase(e);
+   for (const auto& e : erased) {
+      auto c = _outstanding_compiles_and_poison.erase(e);
+      assert(c > 0);
+   }
    g.unlock();
 
    return {gotsome, bytes_remaining};
@@ -163,8 +161,7 @@ code_cache_async::get_descriptor_for_code(mode m, const digest_type& code_id, co
    }
 
    //check for entry in cache
-   code_cache_index::index<by_hash>::type::iterator it = _cache_index.get<by_hash>().find(code_id);
-   if(it != _cache_index.get<by_hash>().end()) {
+   if(auto it = _cache_index.get<by_hash>().find(code_id); it != _cache_index.get<by_hash>().end()) {
       if (m.write_window)
          _cache_index.relocate(_cache_index.begin(), _cache_index.project<0>(it));
       return &*it;
@@ -205,20 +202,20 @@ code_cache_async::get_descriptor_for_code(mode m, const digest_type& code_id, co
       .queued_time = fc::time_point::now(),
       .limits = !m.whitelisted ? _eosvmoc_config.non_whitelisted_limits : std::optional<subjective_compile_limits>{}
    };
-   std::vector<wrapped_fd> fds_to_pass;
-   fds_to_pass.emplace_back(memfd_for_bytearray(codeobject->code));
 
    g.lock();
-   if(_outstanding_compiles_and_poison.size() >= _threads) {
+   if(_outstanding_compiles >= _threads) {
+      std::vector<char> code{codeobject->code.begin(), codeobject->code.end()};
       if (m.high_priority)
-         _queued_compiles.emplace_front(std::move(msg), std::move(fds_to_pass));
+         _queued_compiles.emplace_front(std::move(msg), std::move(code));
       else
-         _queued_compiles.emplace_back(std::move(msg), std::move(fds_to_pass));
+         _queued_compiles.emplace_back(std::move(msg), std::move(code));
       failure = get_cd_failure::temporary; // Compile might not be done yet
       return nullptr;
    }
 
-   write_message(code_id, msg, fds_to_pass);
+   auto fd = memfd_for_bytearray(codeobject->code);
+   write_message(code_id, msg, std::span<wrapped_fd>{&fd, 1});
    failure = get_cd_failure::temporary; // Compile might not be done yet
    return nullptr;
 }
@@ -247,15 +244,13 @@ const code_descriptor* const code_cache_sync::get_descriptor_for_code_sync(mode
    if(!codeobject) //should be impossible right?
       return nullptr;
 
-   std::vector<wrapped_fd> fds_to_pass;
-   fds_to_pass.emplace_back(memfd_for_bytearray(codeobject->code));
-
    auto msg = compile_wasm_message{
       .code = { code_id, vm_version },
       .queued_time = fc::time_point{}, // could use now() if compile time measurement desired
       .limits = !m.whitelisted ? _eosvmoc_config.non_whitelisted_limits : std::optional<subjective_compile_limits>{}
    };
-   write_message_with_fds(_compile_monitor_write_socket, msg, fds_to_pass);
+   auto fd = memfd_for_bytearray(codeobject->code);
+   write_message_with_fds(_compile_monitor_write_socket, msg, std::span<wrapped_fd>{&fd, 1});
    auto [success, message, fds] = read_message_with_fds(_compile_monitor_read_socket);
    EOS_ASSERT(success, wasm_execution_error, "failed to read response from monitor process");
    EOS_ASSERT(std::holds_alternative<wasm_compilation_result_message>(message), wasm_execution_error, "unexpected response from monitor process");

libraries/chain/webassembly/runtimes/eos-vm-oc/compile_monitor.cpp

@@ -96,9 +96,7 @@ struct compile_monitor_session {
       socketpair(AF_UNIX, SOCK_SEQPACKET | SOCK_CLOEXEC, 0, socks);
       local::datagram_protocol::socket response_socket(_ctx);
       response_socket.assign(local::datagram_protocol(), socks[0]);
-      std::vector<wrapped_fd> fds_pass_to_trampoline;
-      fds_pass_to_trampoline.emplace_back(socks[1]);
-      fds_pass_to_trampoline.emplace_back(std::move(wasm_code));
+      std::array<wrapped_fd, 2> fds_pass_to_trampoline { socks[1], std::move(wasm_code) };
 
       eosvmoc_message trampoline_compile_request = msg;
       if(write_message_with_fds(_trampoline_socket, trampoline_compile_request, fds_pass_to_trampoline) == false) {
@@ -324,9 +322,7 @@ wrapped_fd get_connection_to_compile_monitor(int cache_fd) {
    FC_ASSERT(dup_of_cache_fd != -1, "failed to dup cache_fd");
    wrapped_fd dup_cache_fd(dup_of_cache_fd);
 
-   std::vector<wrapped_fd> fds_to_pass; 
-   fds_to_pass.emplace_back(std::move(socket_to_hand_to_monitor_session));
-   fds_to_pass.emplace_back(std::move(dup_cache_fd));
+   std::array<wrapped_fd, 2> fds_to_pass { std::move(socket_to_hand_to_monitor_session), std::move(dup_cache_fd) };
    write_message_with_fds(the_compile_monitor_trampoline.compile_manager_fd, initialize_message(), fds_to_pass);
 
    auto [success, message, fds] = read_message_with_fds(the_compile_monitor_trampoline.compile_manager_fd);