sentry: serial node db execution in a strand (#1302)

silkworm/sentry/discovery/discovery.cpp

@@ -36,6 +36,7 @@ class DiscoveryImpl {
         std::vector<EnodeUrl> peer_urls,
         bool with_dynamic_discovery,
         const std::filesystem::path& data_dir_path,
+        boost::asio::any_io_executor node_db_executor,
         std::function<EccKeyPair()> node_key,
         uint16_t disc_v4_port);
 
@@ -64,11 +65,13 @@ DiscoveryImpl::DiscoveryImpl(
     std::vector<EnodeUrl> peer_urls,
     bool with_dynamic_discovery,
     const std::filesystem::path& data_dir_path,
+    boost::asio::any_io_executor node_db_executor,
     std::function<EccKeyPair()> node_key,
     uint16_t disc_v4_port)
     : peer_urls_(std::move(peer_urls)),
       with_dynamic_discovery_(with_dynamic_discovery),
       data_dir_path_(data_dir_path),
+      node_db_(std::move(node_db_executor)),
       disc_v4_discovery_(disc_v4_port, node_key, node_db_.interface()) {
 }
 
@@ -116,12 +119,14 @@ Discovery::Discovery(
     std::vector<EnodeUrl> peer_urls,
     bool with_dynamic_discovery,
     const std::filesystem::path& data_dir_path,
+    boost::asio::any_io_executor node_db_executor,
     std::function<EccKeyPair()> node_key,
     uint16_t disc_v4_port)
     : p_impl_(std::make_unique<DiscoveryImpl>(
           std::move(peer_urls),
           with_dynamic_discovery,
           data_dir_path,
+          std::move(node_db_executor),
           std::move(node_key),
           disc_v4_port)) {}
 

silkworm/sentry/discovery/discovery.hpp

@@ -23,6 +23,8 @@
 
 #include <silkworm/infra/concurrency/task.hpp>
 
+#include <boost/asio/any_io_executor.hpp>
+
 #include <silkworm/sentry/common/ecc_key_pair.hpp>
 #include <silkworm/sentry/common/enode_url.hpp>
 
@@ -36,6 +38,7 @@ class Discovery {
         std::vector<EnodeUrl> peer_urls,
         bool with_dynamic_discovery,
         const std::filesystem::path& data_dir_path,
+        boost::asio::any_io_executor node_db_executor,
         std::function<EccKeyPair()> node_key,
         uint16_t disc_v4_port);
     ~Discovery();

silkworm/sentry/discovery/node_db/node_db_sqlite.cpp

@@ -25,6 +25,8 @@
 #include <silkworm/infra/common/log.hpp>
 #include <silkworm/infra/common/unix_timestamp.hpp>
 
+#include "serial_node_db.hpp"
+
 namespace silkworm::sentry::discovery::node_db {
 
 static const char* kSqlCreateSchema = R"sql(
@@ -197,7 +199,9 @@ class NodeDbSqliteImpl : public NodeDb {
     std::unique_ptr<SQLite::Database> db_;
 };
 
-NodeDbSqlite::NodeDbSqlite() : p_impl_(std::make_unique<NodeDbSqliteImpl>()) {
+NodeDbSqlite::NodeDbSqlite(boost::asio::any_io_executor executor)
+    : p_impl_(std::make_unique<NodeDbSqliteImpl>()),
+      interface_(std::make_unique<SerialNodeDb>(*p_impl_, std::move(executor))) {
 }
 
 NodeDbSqlite::~NodeDbSqlite() {
@@ -213,7 +217,7 @@ void NodeDbSqlite::setup_in_memory() {
 }
 
 NodeDb& NodeDbSqlite::interface() {
-    return *p_impl_;
+    return *interface_;
 }
 
 }  // namespace silkworm::sentry::discovery::node_db

silkworm/sentry/discovery/node_db/node_db_sqlite.hpp

@@ -19,6 +19,8 @@
 #include <filesystem>
 #include <memory>
 
+#include <boost/asio/any_io_executor.hpp>
+
 #include "node_db.hpp"
 
 namespace silkworm::sentry::discovery::node_db {
@@ -27,7 +29,7 @@ class NodeDbSqliteImpl;
 
 class NodeDbSqlite {
   public:
-    NodeDbSqlite();
+    explicit NodeDbSqlite(boost::asio::any_io_executor executor);
     ~NodeDbSqlite();
 
     void setup(const std::filesystem::path& db_dir_path);
@@ -37,6 +39,7 @@ class NodeDbSqlite {
 
   private:
     std::unique_ptr<NodeDbSqliteImpl> p_impl_;
+    std::unique_ptr<NodeDb> interface_;
 };
 
 }  // namespace silkworm::sentry::discovery::node_db

silkworm/sentry/discovery/node_db/node_db_sqlite_test.cpp

@@ -34,7 +34,7 @@ bool operator==(const NodeAddress& lhs, const NodeAddress& rhs) {
 TEST_CASE("NodeDbSqlite") {
     test_util::TaskRunner runner;
 
-    NodeDbSqlite db_sqlite;
+    NodeDbSqlite db_sqlite{any_io_executor{runner.context().get_executor()}};
     db_sqlite.setup_in_memory();
     NodeDb& db = db_sqlite.interface();
 

silkworm/sentry/discovery/node_db/serial_node_db.cpp

@@ -0,0 +1,46 @@
+/*
+   Copyright 2023 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#include "serial_node_db.hpp"
+
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/use_awaitable.hpp>
+
+namespace silkworm::sentry::discovery::node_db {
+
+using namespace boost::asio;
+
+Task<void> SerialNodeDb::upsert_node_address(NodeId id, NodeAddress address) {
+    return co_spawn(strand_, db_.upsert_node_address(std::move(id), std::move(address)), use_awaitable);
+}
+
+Task<std::optional<NodeAddress>> SerialNodeDb::find_node_address_v4(NodeId id) {
+    return co_spawn(strand_, db_.find_node_address_v4(std::move(id)), use_awaitable);
+}
+
+Task<std::optional<NodeAddress>> SerialNodeDb::find_node_address_v6(NodeId id) {
+    return co_spawn(strand_, db_.find_node_address_v6(std::move(id)), use_awaitable);
+}
+
+Task<void> SerialNodeDb::update_last_pong_time(NodeId id, Time value) {
+    return co_spawn(strand_, db_.update_last_pong_time(std::move(id), std::move(value)), use_awaitable);
+}
+
+Task<void> SerialNodeDb::delete_node(NodeId id) {
+    return co_spawn(strand_, db_.delete_node(std::move(id)), use_awaitable);
+}
+
+}  // namespace silkworm::sentry::discovery::node_db

silkworm/sentry/discovery/node_db/serial_node_db.hpp

@@ -0,0 +1,46 @@
+/*
+   Copyright 2023 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#pragma once
+
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/strand.hpp>
+
+#include "node_db.hpp"
+
+namespace silkworm::sentry::discovery::node_db {
+
+class SerialNodeDb : public NodeDb {
+  public:
+    SerialNodeDb(
+        NodeDb& db,
+        boost::asio::any_io_executor executor)
+        : db_(db),
+          strand_(boost::asio::make_strand(std::move(executor))) {}
+    ~SerialNodeDb() override = default;
+
+    Task<void> upsert_node_address(NodeId id, NodeAddress address) override;
+    Task<std::optional<NodeAddress>> find_node_address_v4(NodeId id) override;
+    Task<std::optional<NodeAddress>> find_node_address_v6(NodeId id) override;
+    Task<void> update_last_pong_time(NodeId id, Time value) override;
+    Task<void> delete_node(NodeId id) override;
+
+  private:
+    NodeDb& db_;
+    boost::asio::strand<boost::asio::any_io_executor> strand_;
+};
+
+}  // namespace silkworm::sentry::discovery::node_db

silkworm/sentry/sentry.cpp

@@ -137,7 +137,13 @@ SentryImpl::SentryImpl(Settings settings, silkworm::rpc::ServerContextPool& cont
       context_pool_(context_pool),
       status_manager_(context_pool_.next_io_context()),
       rlpx_server_(context_pool_.next_io_context(), settings_.port),
-      discovery_(settings_.static_peers, !settings_.no_discover, settings_.data_dir_path, node_key_provider(), settings_.port),
+      discovery_(
+          settings_.static_peers,
+          !settings_.no_discover,
+          settings_.data_dir_path,
+          boost::asio::any_io_executor(context_pool_.next_io_context().get_executor()),
+          node_key_provider(),
+          settings_.port),
       peer_manager_(context_pool_.next_io_context(), settings_.max_peers, context_pool_),
       message_sender_(context_pool_.next_io_context()),
       message_receiver_(std::make_shared<MessageReceiver>(context_pool_.next_io_context(), settings_.max_peers)),