Merge branch 'feature-3.6.0' into feature-3.6.0

JimmyShi22 · Nov 15, 2023 · df62b07 · df62b07
2 parents 1acd840 + 5d9163d
commit df62b07
Show file tree

Hide file tree

Showing 19 changed files with 379 additions and 324 deletions.
diff --git a/bcos-framework/bcos-framework/storage2/MemoryStorage.h b/bcos-framework/bcos-framework/storage2/MemoryStorage.h
@@ -147,7 +147,7 @@ class MemoryStorage
     }
 
     friend auto tag_invoke(
-        bcos::storage2::tag_t<storage2::range> /*unused*/, MemoryStorage& storage)
+        bcos::storage2::tag_t<storage2::range> /*unused*/, MemoryStorage const& storage)
         requires(!withConcurrent)
     {
         auto range = RANGES::views::transform(storage.m_buckets[0].container,
@@ -327,7 +327,7 @@ class MemoryStorage
     {
         task::AwaitableValue<ReadIterator> outputAwaitable(ReadIterator{});
         ReadIterator& output = outputAwaitable.value();
-        if constexpr (RANGES::sized_range<std::remove_cvref_t<decltype(keys)>>)
+        if constexpr (RANGES::sized_range<decltype(keys)>)
         {
             output.m_iterators.reserve(RANGES::size(keys));
         }
@@ -440,7 +440,8 @@ class MemoryStorage
             else
             {
                 it = bucket.get().container.emplace_hint(
-                    it, Data{.key = KeyType(key), .value = std::forward<decltype(value)>(value)});
+                    it, Data{.key = KeyType(std::forward<decltype(key)>(key)),
+                            .value = std::forward<decltype(value)>(value)});
             }
 
             if constexpr (withMRU)

diff --git a/bcos-framework/bcos-framework/storage2/Storage.h b/bcos-framework/bcos-framework/storage2/Storage.h
@@ -13,16 +13,24 @@ struct STORAGE_BEGIN_TYPE
 };
 inline constexpr STORAGE_BEGIN_TYPE STORAGE_BEGIN{};
 
+struct READ_FRONT_TYPE
+{
+};
+inline constexpr READ_FRONT_TYPE READ_FRONT{};
+
 template <class Invoke>
 using ReturnType = typename task::AwaitableReturnType<Invoke>;
 
 struct ReadSome
 {
-    auto operator()(auto& storage, RANGES::input_range auto const& keys) const
-        -> task::Task<ReturnType<decltype(tag_invoke(*this, storage, keys))>>
-        requires RANGES::range<ReturnType<decltype(tag_invoke(*this, storage, keys))>>
+    auto operator()(auto& storage, RANGES::input_range auto&& keys, auto&&... args) const
+        -> task::Task<ReturnType<decltype(tag_invoke(*this, storage,
+            std::forward<decltype(keys)>(keys), std::forward<decltype(args)>(args)...))>>
+        requires RANGES::range<ReturnType<decltype(tag_invoke(*this, storage,
+            std::forward<decltype(keys)>(keys), std::forward<decltype(args)>(args)...))>>
     {
-        co_return co_await tag_invoke(*this, storage, keys);
+        co_return co_await tag_invoke(*this, storage, std::forward<decltype(keys)>(keys),
+            std::forward<decltype(args)>(args)...);
     }
 };
 inline constexpr ReadSome readSome{};
@@ -53,10 +61,11 @@ struct RemoveSome
 inline constexpr RemoveSome removeSome{};
 struct ReadOne
 {
-    auto operator()(auto& storage, auto const& key) const
-        -> task::Task<ReturnType<decltype(tag_invoke(*this, storage, key))>>
+    auto operator()(auto& storage, auto const& key, auto&&... args) const
+        -> task::Task<ReturnType<decltype(tag_invoke(
+            *this, storage, key, std::forward<decltype(args)>(args)...))>>
     {
-        co_return co_await tag_invoke(*this, storage, key);
+        co_return co_await tag_invoke(*this, storage, key, std::forward<decltype(args)>(args)...);
     }
 };
 inline constexpr ReadOne readOne{};
@@ -152,4 +161,17 @@ auto tag_invoke(bcos::storage2::tag_t<existsOne> /*unused*/, auto& storage, auto
     co_return result.has_value();
 }
 
+task::Task<void> tag_invoke(
+    bcos::storage2::tag_t<merge> /*unused*/, auto const& fromStorage, auto& toStorage)
+{
+    auto range = co_await storage2::range(fromStorage);
+    for (auto [key, value] : range)
+    {
+        if (value)
+        {
+            co_await storage2::writeOne(toStorage, *key, *value);
+        }
+    }
+}
+
 }  // namespace bcos::storage2
diff --git a/bcos-framework/bcos-framework/storage2/StorageMethods.h b/bcos-framework/bcos-framework/storage2/StorageMethods.h
@@ -114,32 +114,30 @@ task::Task<void> tag_invoke(bcos::storage2::tag_t<merge> /*unused*/,
     }
 }
 
-template <ErasableStorage Storage>
-auto tag_invoke(bcos::storage2::tag_t<removeSome> /*unused*/, Storage& storage,
+auto tag_invoke(bcos::storage2::tag_t<removeSome> /*unused*/, ErasableStorage auto& storage,
     RANGES::input_range auto const& keys) -> task::Task<void>
 {
     co_await storage.remove(keys);
     co_return;
 }
 
-template <ReadableStorage Storage>
-auto tag_invoke(bcos::storage2::tag_t<readSome> /*unused*/, Storage& storage,
-    RANGES::input_range auto const& keys)
+auto tag_invoke(bcos::storage2::tag_t<readSome> /*unused*/, ReadableStorage auto& storage,
+    RANGES::input_range auto&& keys)
     -> task::Task<boost::container::small_vector<
-        std::optional<std::remove_cvref_t<
-            typename task::AwaitableReturnType<decltype(storage.read(keys))>::Value>>,
+        std::optional<std::remove_cvref_t<typename task::AwaitableReturnType<decltype(storage.read(
+            std::forward<decltype(keys)>(keys)))>::Value>>,
         1>>
 {
-    using ValueType = std::remove_cvref_t<
-        typename task::AwaitableReturnType<decltype(storage.read(keys))>::Value>;
+    using ValueType = std::remove_cvref_t<typename task::AwaitableReturnType<decltype(storage.read(
+        std::forward<decltype(keys)>(keys)))>::Value>;
     static_assert(std::is_copy_assignable_v<ValueType>);
 
     boost::container::small_vector<std::optional<ValueType>, 1> values;
     if constexpr (RANGES::sized_range<decltype(keys)>)
     {
         values.reserve(RANGES::size(keys));
     }
-    auto it = co_await storage.read(keys);
+    auto it = co_await storage.read(std::forward<decltype(keys)>(keys));
     while (co_await it.next())
     {
         if (co_await it.hasValue())

diff --git a/bcos-framework/bcos-framework/transaction-executor/TransactionExecutor.h b/bcos-framework/bcos-framework/transaction-executor/TransactionExecutor.h
@@ -83,6 +83,7 @@ inline constexpr ExecuteTransaction executeTransaction{};
 
 template <auto& Tag>
 using tag_t = std::decay_t<decltype(Tag)>;
+
 }  // namespace bcos::transaction_executor
 
 template <>

diff --git a/bcos-storage/bcos-storage/RocksDBStorage2.h b/bcos-storage/bcos-storage/RocksDBStorage2.h
@@ -92,7 +92,7 @@ class RocksDBStorage2
         }
     };
 
-    auto read(RANGES::input_range auto const& keys) & -> task::AwaitableValue<ReadIterator>
+    auto read(RANGES::input_range auto&& keys) & -> task::AwaitableValue<ReadIterator>
     {
         task::AwaitableValue<ReadIterator> readIteratorAwaitable(ReadIterator{m_valueResolver});
         auto& readIterator = readIteratorAwaitable.value();

diff --git a/bcos-txpool/bcos-txpool/txpool/storage/MemoryStorage.cpp b/bcos-txpool/bcos-txpool/txpool/storage/MemoryStorage.cpp
@@ -782,23 +782,12 @@ void MemoryStorage::batchFetchTxs(Block::Ptr _txsList, Block::Ptr _sysTxsList, s
 
     if (_avoidDuplicate)
     {
-        size_t eachBucketTxsLimit = 0;
-        // After performance testing, 0.25 had the best performance.
-        eachBucketTxsLimit = (_txsLimit / (0.25 * CPU_CORES)) + 1;
-        if (c_fileLogLevel == LogLevel::TRACE) [[unlikely]]
-        {
-            TXPOOL_LOG(TRACE) << LOG_DESC("batchFetchTxs")
-                              << LOG_KV("pendingTxs", m_txsTable.size())
-                              << LOG_KV("limit", _txsLimit)
-                              << LOG_KV("eachBucketTxsLimit", eachBucketTxsLimit);
-        }
         m_txsTable.forEach<TxsMap::ReadAccessor>(
-            m_knownLatestSealedTxHash, eachBucketTxsLimit, [&](TxsMap::ReadAccessor::Ptr accessor) {
+            m_knownLatestSealedTxHash, [&](TxsMap::ReadAccessor::Ptr accessor) {
                 const auto& tx = accessor->value();
-                bool isTxValid = handleTx(tx);
-                bool needContinue = (_txsList->transactionsMetaDataSize() +
-                                        _sysTxsList->transactionsMetaDataSize()) < _txsLimit;
-                return std::pair<bool, bool>(needContinue, isTxValid);
+                handleTx(tx);
+                return (_txsList->transactionsMetaDataSize() +
+                           _sysTxsList->transactionsMetaDataSize()) < _txsLimit;
             });
     }
     else

diff --git a/bcos-txpool/bcos-txpool/txpool/storage/MemoryStorage.h b/bcos-txpool/bcos-txpool/txpool/storage/MemoryStorage.h
@@ -169,7 +169,7 @@ class MemoryStorage : public TxPoolStorageInterface,
     TxPoolConfig::Ptr m_config;
 
     using TxsMap = BucketMap<bcos::crypto::HashType, bcos::protocol::Transaction::Ptr,
-        std::hash<bcos::crypto::HashType>, TransactionBucket>;
+        std::hash<bcos::crypto::HashType>>;
     TxsMap m_txsTable, m_invalidTxs;
 
     using HashSet = BucketSet<bcos::crypto::HashType, std::hash<bcos::crypto::HashType>>;

diff --git a/transaction-executor/bcos-transaction-executor/RollbackableStorage.h b/transaction-executor/bcos-transaction-executor/RollbackableStorage.h
@@ -5,19 +5,20 @@
 #include "bcos-task/Trait.h"
 #include <bcos-framework/transaction-executor/TransactionExecutor.h>
 #include <boost/container/small_vector.hpp>
+#include <type_traits>
 
 namespace bcos::transaction_executor
 {
 
 template <class Storage>
-concept HasReadDirectly =
-    requires(Storage&& storage) {
-        requires storage2::ReadableStorage<Storage>;
-        requires storage2::ReadIterator<task::AwaitableReturnType<decltype(storage.readDirect(
-            std::declval<std::vector<typename Storage::Key>>()))>>;
+concept HasReadOneDirect =
+    requires(Storage& storage) {
+        requires RANGES::range<task::AwaitableReturnType<decltype(storage2::readSome(
+            storage, std::declval<std::vector<typename Storage::Key>>(), storage2::READ_FRONT))>>;
     };
 
 template <class Storage>
+    requires HasReadOneDirect<Storage>
 class Rollbackable
 {
 private:
@@ -39,6 +40,7 @@ class Rollbackable
 
     Storage& storage() { return m_storage; }
     Savepoint current() const { return static_cast<int64_t>(m_records.size()); }
+
     task::Task<void> rollback(Savepoint savepoint)
     {
         for (auto index = static_cast<int64_t>(m_records.size()); index > savepoint; --index)
@@ -47,77 +49,66 @@ class Rollbackable
             auto& record = m_records[index - 1];
             if (record.oldValue)
             {
-                co_await m_storage.write(RANGES::single_view(record.key),
-                    RANGES::single_view(std::move(*record.oldValue)));
+                co_await storage2::writeOne(
+                    m_storage, std::move(record.key), std::move(*record.oldValue));
             }
             else
             {
-                co_await m_storage.remove(RANGES::single_view(record.key));
+                co_await storage2::removeOne(m_storage, record.key);
             }
             m_records.pop_back();
         }
         co_return;
     }
 
-    auto read(RANGES::input_range auto const& keys)
-        -> task::Task<task::AwaitableReturnType<decltype(m_storage.read(keys))>>
+    friend auto tag_invoke(storage2::tag_t<storage2::readSome> /*unused*/, Rollbackable& storage,
+        RANGES::input_range auto&& keys)
+        -> task::Task<task::AwaitableReturnType<decltype(storage2::readSome(
+            (Storage&)std::declval<Storage>(), std::forward<decltype(keys)>(keys)))>>
     {
-        co_return co_await m_storage.read(keys);
+        co_return co_await storage2::readSome(
+            storage.m_storage, std::forward<decltype(keys)>(keys));
     }
 
-    auto write(RANGES::input_range auto&& keys, RANGES::input_range auto&& values)
-        -> task::Task<task::AwaitableReturnType<decltype(m_storage.write(
-            std::forward<decltype(keys)>(keys), std::forward<decltype(values)>(values)))>>
+    friend auto tag_invoke(storage2::tag_t<storage2::writeSome> /*unused*/, Rollbackable& storage,
+        RANGES::input_range auto&& keys, RANGES::input_range auto&& values)
+        -> task::Task<task::AwaitableReturnType<decltype(storage2::writeSome(
+            (Storage&)std::declval<Storage>(), std::forward<decltype(keys)>(keys),
+            std::forward<decltype(values)>(values)))>>
     {
         // Store values to history
+        auto oldValues = co_await storage2::readSome(storage.m_storage, keys, storage2::READ_FRONT);
+        for (auto&& [key, value] : RANGES::views::zip(keys, oldValues))
         {
-            std::optional<task::AwaitableReturnType<decltype(m_storage.read(keys))>> storageIt;
-            if constexpr (HasReadDirectly<Storage>)
-            {
-                storageIt = co_await m_storage.readDirect(keys);
-            }
-            else
-            {
-                storageIt = co_await m_storage.read(keys);
-            }
-
-            auto keyIt = RANGES::begin(keys);
-            while (co_await storageIt->next())
+            auto& record =
+                storage.m_records.emplace_back(Record{.key = StateKey{key}, .oldValue = {}});
+            if (value)
             {
-                auto& record =
-                    m_records.emplace_back(Record{.key = StateKey{*(keyIt++)}, .oldValue = {}});
-                if (co_await storageIt->hasValue())
-                {
-                    // Update exists value, store the old value
-                    record.oldValue.emplace(co_await storageIt->value());
-                }
+                record.oldValue.emplace(std::move(*value));
             }
         }
 
-        co_return co_await m_storage.write(
+        co_return co_await storage2::writeSome(storage.m_storage,
             std::forward<decltype(keys)>(keys), std::forward<decltype(values)>(values));
     }
 
-    auto remove(RANGES::input_range auto const& keys)
-        -> task::Task<task::AwaitableReturnType<decltype(m_storage.remove(keys))>>
+    friend auto tag_invoke(storage2::tag_t<storage2::removeSome> /*unused*/, Rollbackable& storage,
+        RANGES::input_range auto const& keys)
+        -> task::Task<task::AwaitableReturnType<decltype(storage2::removeSome(
+            (Storage&)std::declval<Storage>(), keys))>>
     {
         // Store values to history
+        auto oldValues = co_await storage2::readSome(storage.m_storage, keys, storage2::READ_FRONT);
+        for (auto&& [key, value] : RANGES::views::zip(keys, oldValues))
         {
-            auto storageIt = co_await m_storage.read(keys);
-            auto keyIt = RANGES::begin(keys);
-            while (co_await storageIt.next())
+            if (value)
             {
-                auto& record = m_records.emplace_back();
-                record.key = *(keyIt++);
-                if (co_await storageIt.hasValue())
-                {
-                    // Update exists value, store the old value
-                    record.oldValue.emplace(co_await storageIt.value());
-                }
+                auto& record = storage.m_records.emplace_back(
+                    Record{.key = StateKey{key}, .oldValue = std::move(*value)});
             }
         }
 
-        co_return co_await m_storage.remove(keys);
+        co_return co_await storage2::removeSome(storage.m_storage, keys);
     }
 };
 

diff --git a/transaction-executor/bcos-transaction-executor/TransactionExecutorImpl.h b/transaction-executor/bcos-transaction-executor/TransactionExecutorImpl.h
@@ -52,7 +52,7 @@ class TransactionExecutorImpl
                     << "Execte transaction: " << transaction.hash().hex();
             }
 
-            Rollbackable<std::remove_reference_t<decltype(storage)>> rollbackableStorage(storage);
+            Rollbackable<std::decay_t<decltype(storage)>> rollbackableStorage(storage);
 
             auto toAddress = unhexAddress(transaction.to());
             evmc_message evmcMessage = {.kind = transaction.to().empty() ? EVMC_CREATE : EVMC_CALL,

diff --git a/transaction-executor/benchmark/benchmakrExecutor.cpp b/transaction-executor/benchmark/benchmakrExecutor.cpp
@@ -1,5 +1,6 @@
 #include "../bcos-transaction-executor/TransactionExecutorImpl.h"
 #include "../tests/TestBytecode.h"
+#include "../tests/TestMemoryStorage.h"
 #include "bcos-codec/bcos-codec/abi/ContractABICodec.h"
 #include "bcos-crypto/interfaces/crypto/CryptoSuite.h"
 #include "bcos-framework/ledger/LedgerConfig.h"
@@ -18,8 +19,8 @@ using namespace bcos;
 using namespace bcos::storage2::memory_storage;
 using namespace bcos::transaction_executor;
 
-using MutableStorage = MemoryStorage<StateKey, StateValue, ORDERED>;
 using ReceiptFactory = bcostars::protocol::TransactionReceiptFactoryImpl;
+static_assert(HasReadOneDirect<MutableStorage>);
 
 struct Fixture
 {

diff --git a/transaction-executor/tests/TestHostContext.cpp b/transaction-executor/tests/TestHostContext.cpp
@@ -2,6 +2,7 @@
 #include "../bcos-transaction-executor/vm/HostContext.h"
 #include "../bcos-transaction-executor/vm/VMInstance.h"
 #include "TestBytecode.h"
+#include "TestMemoryStorage.h"
 #include "bcos-codec/bcos-codec/abi/ContractABICodec.h"
 #include "bcos-crypto/interfaces/crypto/CryptoSuite.h"
 #include "bcos-crypto/interfaces/crypto/Hash.h"
@@ -36,7 +37,7 @@ class TestHostContextFixture
 {
 public:
     bcos::crypto::Hash::Ptr hashImpl = std::make_shared<bcos::crypto::Keccak256>();
-    memory_storage::MemoryStorage<StateKey, StateValue, memory_storage::ORDERED> storage;
+    MutableStorage storage;
     Rollbackable<decltype(storage)> rollbackableStorage;
     evmc_address helloworldAddress;
     VMFactory vmFactory;

diff --git a/transaction-executor/tests/TestMemoryStorage.h b/transaction-executor/tests/TestMemoryStorage.h
@@ -0,0 +1,17 @@
+#pragma once
+#include "bcos-framework/storage2/MemoryStorage.h"
+#include "bcos-framework/storage2/Storage.h"
+#include "bcos-framework/transaction-executor/TransactionExecutor.h"
+
+namespace bcos::transaction_executor
+{
+using MutableStorage = storage2::memory_storage::MemoryStorage<StateKey, StateValue,
+    storage2::memory_storage::ORDERED>;
+
+auto tag_invoke(storage2::tag_t<storage2::readSome> /*unused*/, MutableStorage& storage,
+    RANGES::input_range auto&& keys, storage2::READ_FRONT_TYPE const& /*unused*/)
+    -> task::Task<task::AwaitableReturnType<decltype(storage2::readSome(storage, keys))>>
+{
+    co_return co_await storage2::readSome(storage, std::forward<decltype(keys)>(keys));
+}
+}  // namespace bcos::transaction_executor