Add initial KV Tx RPC implementation

cmd/backend_kv_server.cpp

@@ -81,10 +81,13 @@ int parse_command_line(int argc, char* argv[], BackEndKvSettings& settings) {
         std::string data_dir{silkworm::DataDirectory::get_default_storage_path().string()};
         std::string etherbase_address{""};
         uint32_t num_contexts{std::thread::hardware_concurrency() / 2};
+        uint32_t max_readers{silkworm::db::EnvConfig{}.max_readers};
         app.add_option("--datadir", data_dir, "The path to data directory", true);
         app.add_option("--etherbase", etherbase_address, "The chain identifier as string", true);
         // TODO(canepat) add check on etherbase using EthAddressValidator [TBD]
-        app.add_option("--numContexts", num_contexts, "The number of running contexts", true);
+        app.add_option("--contexts", num_contexts, "The number of running contexts", true);
+        app.add_option("--mdbx.max.readers", max_readers, "The maximum number of MDBX readers", true)
+            ->check(CLI::Range(1, 32767));
 
         // RPC Server options
         app.add_option("--private.api.addr", node_settings.private_api_addr,
@@ -147,6 +150,7 @@ int parse_command_line(int argc, char* argv[], BackEndKvSettings& settings) {
             /*create=*/false,
             /*readonly=*/true
         };
+        node_settings.chaindata_env_config.max_readers = max_readers;
 
         server_settings.set_address_uri(node_settings.private_api_addr);
         server_settings.set_num_contexts(num_contexts);
@@ -187,6 +191,8 @@ int main(int argc, char* argv[]) {
         silkworm::EthereumBackEnd backend{node_settings, &database_env};
         backend.set_node_name(node_name);
 
+        SILK_INFO << "BackEndKvServer MDBX max readers: " << database_env.max_readers();
+
         silkworm::rpc::BackEndKvServer server{server_settings, backend};
         server.build_and_start();
 

node/silkworm/backend/ethereum_backend.cpp

@@ -20,7 +20,7 @@
 
 namespace silkworm {
 
-EthereumBackEnd::EthereumBackEnd(const NodeSettings& node_settings, mdbx::env_managed* chaindata_env)
+EthereumBackEnd::EthereumBackEnd(const NodeSettings& node_settings, mdbx::env* chaindata_env)
     : node_settings_(node_settings), chaindata_env_(chaindata_env) {
     // Get the numeric chain identifier from node settings
     if (node_settings_.chain_config) {

node/silkworm/backend/ethereum_backend.hpp

@@ -32,12 +32,12 @@ constexpr const char kSentryAddressDelimiter{','};
 
 class EthereumBackEnd {
   public:
-    explicit EthereumBackEnd(const NodeSettings& node_settings, mdbx::env_managed* chaindata_env);
+    explicit EthereumBackEnd(const NodeSettings& node_settings, mdbx::env* chaindata_env);
 
     EthereumBackEnd(const EthereumBackEnd&) = delete;
     EthereumBackEnd& operator=(const EthereumBackEnd&) = delete;
 
-    mdbx::env_managed* chaindata_env() const noexcept { return chaindata_env_; }
+    mdbx::env* chaindata_env() const noexcept { return chaindata_env_; }
     const std::string& node_name() const noexcept { return node_name_; }
     std::optional<uint64_t> chain_id() const noexcept { return chain_id_; }
     std::optional<evmc::address> etherbase() const noexcept { return node_settings_.etherbase; }
@@ -47,7 +47,7 @@ class EthereumBackEnd {
 
   private:
     const NodeSettings& node_settings_;
-    mdbx::env_managed* chaindata_env_;
+    mdbx::env* chaindata_env_;
     std::string node_name_{kDefaultNodeName};
     std::optional<uint64_t> chain_id_{std::nullopt};
     std::vector<std::string> sentry_addresses_;

node/silkworm/rpc/backend_calls.cpp

@@ -50,8 +50,8 @@ void EtherbaseCall::fill_predefined_reply(const EthereumBackEnd& backend) {
     }
 }
 
-EtherbaseCall::EtherbaseCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::EtherbaseRequest, remote::EtherbaseReply>(service, queue, handlers) {
+EtherbaseCall::EtherbaseCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::EtherbaseRequest, remote::EtherbaseReply>(scheduler, service, queue, handlers) {
 }
 
 void EtherbaseCall::process(const remote::EtherbaseRequest* request) {
@@ -82,8 +82,8 @@ void NetVersionCall::fill_predefined_reply(const EthereumBackEnd& backend) {
     }
 }
 
-NetVersionCall::NetVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::NetVersionRequest, remote::NetVersionReply>(service, queue, handlers) {
+NetVersionCall::NetVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::NetVersionRequest, remote::NetVersionReply>(scheduler, service, queue, handlers) {
 }
 
 void NetVersionCall::process(const remote::NetVersionRequest* request) {
@@ -109,8 +109,8 @@ void NetPeerCountCall::remove_sentry(SentryClient* sentry) {
     NetPeerCountCall::sentries_.erase(sentry);
 }
 
-NetPeerCountCall::NetPeerCountCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::NetPeerCountRequest, remote::NetPeerCountReply>(service, queue, handlers) {
+NetPeerCountCall::NetPeerCountCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::NetPeerCountRequest, remote::NetPeerCountReply>(scheduler, service, queue, handlers) {
 }
 
 void NetPeerCountCall::process(const remote::NetPeerCountRequest* request) {
@@ -169,8 +169,8 @@ void BackEndVersionCall::fill_predefined_reply() {
     BackEndVersionCall::response_.set_patch(std::get<2>(kEthBackEndApiVersion));
 }
 
-BackEndVersionCall::BackEndVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : UnaryRpc<remote::ETHBACKEND::AsyncService, google::protobuf::Empty, types::VersionReply>(service, queue, handlers) {
+BackEndVersionCall::BackEndVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : UnaryRpc<remote::ETHBACKEND::AsyncService, google::protobuf::Empty, types::VersionReply>(scheduler, service, queue, handlers) {
 }
 
 void BackEndVersionCall::process(const google::protobuf::Empty* request) {
@@ -192,8 +192,8 @@ void ProtocolVersionCall::fill_predefined_reply() {
     ProtocolVersionCall::response_.set_id(kEthDevp2pProtocolVersion);
 }
 
-ProtocolVersionCall::ProtocolVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::ProtocolVersionRequest, remote::ProtocolVersionReply>(service, queue, handlers) {
+ProtocolVersionCall::ProtocolVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::ProtocolVersionRequest, remote::ProtocolVersionReply>(scheduler, service, queue, handlers) {
 }
 
 void ProtocolVersionCall::process(const remote::ProtocolVersionRequest* request) {
@@ -215,8 +215,8 @@ void ClientVersionCall::fill_predefined_reply(const EthereumBackEnd& backend) {
     ClientVersionCall::response_.set_nodename(backend.node_name());
 }
 
-ClientVersionCall::ClientVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::ClientVersionRequest, remote::ClientVersionReply>(service, queue, handlers) {
+ClientVersionCall::ClientVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::ClientVersionRequest, remote::ClientVersionReply>(scheduler, service, queue, handlers) {
 }
 
 void ClientVersionCall::process(const remote::ClientVersionRequest* request) {
@@ -232,8 +232,8 @@ ClientVersionCallFactory::ClientVersionCallFactory(const EthereumBackEnd& backen
     ClientVersionCall::fill_predefined_reply(backend);
 }
 
-SubscribeCall::SubscribeCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : ServerStreamingRpc<remote::ETHBACKEND::AsyncService, remote::SubscribeRequest, remote::SubscribeReply>(service, queue, handlers) {
+SubscribeCall::SubscribeCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : ServerStreamingRpc<remote::ETHBACKEND::AsyncService, remote::SubscribeRequest, remote::SubscribeReply>(scheduler, service, queue, handlers) {
 }
 
 void SubscribeCall::process(const remote::SubscribeRequest* request) {
@@ -268,8 +268,8 @@ void NodeInfoCall::remove_sentry(SentryClient* sentry) {
     NodeInfoCall::sentries_.erase(sentry);
 }
 
-NodeInfoCall::NodeInfoCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
-    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::NodesInfoRequest, remote::NodesInfoReply>(service, queue, handlers) {
+NodeInfoCall::NodeInfoCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers)
+    : UnaryRpc<remote::ETHBACKEND::AsyncService, remote::NodesInfoRequest, remote::NodesInfoReply>(scheduler, service, queue, handlers) {
 }
 
 void NodeInfoCall::process(const remote::NodesInfoRequest* request) {
@@ -316,16 +316,16 @@ BackEndService::BackEndService(const EthereumBackEnd& backend)
     : etherbase_factory_{backend}, net_version_factory_{backend}, client_version_factory_{backend} {
 }
 
-void BackEndService::register_backend_request_calls(remote::ETHBACKEND::AsyncService* async_service, grpc::ServerCompletionQueue* queue) {
+void BackEndService::register_backend_request_calls(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* async_service, grpc::ServerCompletionQueue* queue) {
     // Register one requested call for each RPC factory
-    etherbase_factory_.create_rpc(async_service, queue);
-    net_version_factory_.create_rpc(async_service, queue);
-    net_peer_count_factory_.create_rpc(async_service, queue);
-    backend_version_factory_.create_rpc(async_service, queue);
-    protocol_version_factory_.create_rpc(async_service, queue);
-    client_version_factory_.create_rpc(async_service, queue);
-    subscribe_factory_.create_rpc(async_service, queue);
-    node_info_factory_.create_rpc(async_service, queue);
+    etherbase_factory_.create_rpc(scheduler, async_service, queue);
+    net_version_factory_.create_rpc(scheduler, async_service, queue);
+    net_peer_count_factory_.create_rpc(scheduler, async_service, queue);
+    backend_version_factory_.create_rpc(scheduler, async_service, queue);
+    protocol_version_factory_.create_rpc(scheduler, async_service, queue);
+    client_version_factory_.create_rpc(scheduler, async_service, queue);
+    subscribe_factory_.create_rpc(scheduler, async_service, queue);
+    node_info_factory_.create_rpc(scheduler, async_service, queue);
 }
 
 void BackEndService::add_sentry(std::unique_ptr<SentryClient>&& sentry) {

node/silkworm/rpc/backend_calls.hpp

@@ -49,7 +49,7 @@ class EtherbaseCall : public UnaryRpc<remote::ETHBACKEND::AsyncService, remote::
   public:
     static void fill_predefined_reply(const EthereumBackEnd& backend);
 
-    EtherbaseCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    EtherbaseCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const remote::EtherbaseRequest* request) override;
 
@@ -68,7 +68,7 @@ class NetVersionCall : public UnaryRpc<remote::ETHBACKEND::AsyncService, remote:
   public:
     static void fill_predefined_reply(const EthereumBackEnd& backend);
 
-    NetVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    NetVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const remote::NetVersionRequest* request) override;
 
@@ -88,7 +88,7 @@ class NetPeerCountCall : public UnaryRpc<remote::ETHBACKEND::AsyncService, remot
     static void add_sentry(SentryClient* sentry);
     static void remove_sentry(SentryClient* sentry);
 
-    NetPeerCountCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    NetPeerCountCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const remote::NetPeerCountRequest* request) override;
 
@@ -111,7 +111,7 @@ class BackEndVersionCall : public UnaryRpc<remote::ETHBACKEND::AsyncService, goo
   public:
     static void fill_predefined_reply();
 
-    BackEndVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    BackEndVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const google::protobuf::Empty* request) override;
 
@@ -130,7 +130,7 @@ class ProtocolVersionCall : public UnaryRpc<remote::ETHBACKEND::AsyncService, re
   public:
     static void fill_predefined_reply();
 
-    ProtocolVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    ProtocolVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const remote::ProtocolVersionRequest* request) override;
 
@@ -149,7 +149,7 @@ class ClientVersionCall : public UnaryRpc<remote::ETHBACKEND::AsyncService, remo
   public:
     static void fill_predefined_reply(const EthereumBackEnd& backend);
 
-    ClientVersionCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    ClientVersionCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const remote::ClientVersionRequest* request) override;
 
@@ -166,7 +166,7 @@ class ClientVersionCallFactory : public CallFactory<remote::ETHBACKEND::AsyncSer
 //! Server-streaming RPC for Subscribe method of 'ethbackend' gRPC protocol.
 class SubscribeCall : public ServerStreamingRpc<remote::ETHBACKEND::AsyncService, remote::SubscribeRequest, remote::SubscribeReply> {
   public:
-    SubscribeCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    SubscribeCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const remote::SubscribeRequest* request) override;
 };
@@ -183,7 +183,7 @@ class NodeInfoCall : public UnaryRpc<remote::ETHBACKEND::AsyncService, remote::N
     static void add_sentry(SentryClient* sentry);
     static void remove_sentry(SentryClient* sentry);
 
-    NodeInfoCall(remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
+    NodeInfoCall(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* service, grpc::ServerCompletionQueue* queue, Handlers handlers);
 
     void process(const remote::NodesInfoRequest* request) override;
 
@@ -207,7 +207,7 @@ struct BackEndService {
     explicit BackEndService(const EthereumBackEnd& backend);
     ~BackEndService();
 
-    void register_backend_request_calls(remote::ETHBACKEND::AsyncService* async_service, grpc::ServerCompletionQueue* queue);
+    void register_backend_request_calls(boost::asio::io_context& scheduler, remote::ETHBACKEND::AsyncService* async_service, grpc::ServerCompletionQueue* queue);
 
     void add_sentry(std::unique_ptr<SentryClient>&& sentry);
 

node/silkworm/rpc/backend_kv_server.cpp

@@ -45,7 +45,6 @@ void BackEndKvServer::register_request_calls() {
     // Start one server-side RPC request for each available server context
     for (auto& backend_kv_svc : backend_kv_services_) {
         const auto& server_context = next_context();
-        const auto server_queue = server_context.server_queue();
         const auto client_queue = server_context.client_queue();
 
         // Complete the service initialization
@@ -55,8 +54,10 @@ void BackEndKvServer::register_request_calls() {
         }
 
         // Register initial requested calls for ETHBACKEND and KV services
-        backend_kv_svc->register_backend_request_calls(&backend_async_service_, server_queue);
-        backend_kv_svc->register_kv_request_calls(&kv_async_service_, server_queue);
+        const auto io_context = server_context.io_context();
+        const auto server_queue = server_context.server_queue();
+        backend_kv_svc->register_backend_request_calls(*io_context, &backend_async_service_, server_queue);
+        backend_kv_svc->register_kv_request_calls(*io_context, &kv_async_service_, server_queue);
     }
 }