Associate worker thread pool creation to HTTP server (#99)

Fix eth_syncing integration test
Update README

Author: canepat

README.md

@@ -139,11 +139,11 @@ silkrpcdaemon: C++ implementation of ETH JSON Remote Procedure Call (RPC) daemon
 
   Flags from main.cpp:
     --chaindata (chain data path as string); default: "";
-    --local (HTTP JSON local binding as string <address>:<port>);
-      default: "localhost:8545";
+    --local (HTTP JSON local binding as string <address>:<port>); default: "localhost:8545";
     --logLevel (logging level); default: c;
-    --target (Erigon Core gRPC service location as string <address>:<port>);
-      default: "localhost:9090";
+    --numContexts (number of running I/O contexts as 32-bit integer); default: number of hardware thread contexts / 2;
+    --numWorkers (number of worker threads as 32-bit integer); default: number of hardware thread contexts;
+    --target (Erigon Core gRPC service location as string <address>:<port>); default: "localhost:9090";
     --timeout (gRPC call timeout as 32-bit integer); default: 10000;
 ```
 

silkrpc/commands/eth_api.hpp

@@ -17,9 +17,7 @@
 #ifndef SILKRPC_COMMANDS_ETH_API_HPP_
 #define SILKRPC_COMMANDS_ETH_API_HPP_
 
-#include <cstddef>
 #include <memory>
-#include <thread>
 #include <vector>
 
 #include <silkrpc/config.hpp> // NOLINT(build/include_order)
@@ -46,8 +44,8 @@ namespace silkrpc::commands {
 
 class EthereumRpcApi {
 public:
-    explicit EthereumRpcApi(Context& context, std::size_t workers_count = std::thread::hardware_concurrency())
-    : context_(context), database_(context.database), backend_(context.backend), workers_{workers_count} {}
+    explicit EthereumRpcApi(Context& context, asio::thread_pool& workers)
+    : context_(context), database_(context.database), backend_(context.backend), workers_{workers} {}
     virtual ~EthereumRpcApi() {}
 
     EthereumRpcApi(const EthereumRpcApi&) = delete;
@@ -103,7 +101,7 @@ class EthereumRpcApi {
     Context& context_;
     std::unique_ptr<ethdb::Database>& database_;
     std::unique_ptr<ethbackend::BackEnd>& backend_;
-    asio::thread_pool workers_;
+    asio::thread_pool& workers_;
 
     friend class silkrpc::http::RequestHandler;
 };

silkrpc/commands/rpc_api.hpp

@@ -19,6 +19,8 @@
 
 #include <memory>
 
+#include <asio/thread_pool.hpp>
+
 #include <silkrpc/commands/debug_api.hpp>
 #include <silkrpc/commands/eth_api.hpp>
 #include <silkrpc/commands/net_api.hpp>
@@ -34,8 +36,8 @@ namespace silkrpc::commands {
 
 class RpcApi : protected EthereumRpcApi, NetRpcApi, Web3RpcApi, DebugRpcApi, ParityRpcApi, TurboGethRpcApi, TraceRpcApi {
 public:
-    explicit RpcApi(Context& context) :
-        EthereumRpcApi{context}, NetRpcApi{context.backend}, Web3RpcApi{context}, DebugRpcApi{context.database},
+    explicit RpcApi(Context& context, asio::thread_pool& workers) :
+        EthereumRpcApi{context, workers}, NetRpcApi{context.backend}, Web3RpcApi{context}, DebugRpcApi{context.database},
         ParityRpcApi{context.database}, TurboGethRpcApi{context.database}, TraceRpcApi{context.database} {}
     virtual ~RpcApi() {}
 

silkrpc/http/connection.cpp

@@ -38,7 +38,7 @@
 
 namespace silkrpc::http {
 
-Connection::Connection(Context& context) : socket_{*context.io_context}, request_handler_{context} {
+Connection::Connection(Context& context, asio::thread_pool& workers) : socket_{*context.io_context}, request_handler_{context, workers} {
     request_.content.reserve(1024);
     request_.headers.reserve(8);
     request_.method.reserve(64);

silkrpc/http/connection.hpp

@@ -30,6 +30,7 @@
 
 #include <asio/awaitable.hpp>
 #include <asio/ip/tcp.hpp>
+#include <asio/thread_pool.hpp>
 
 #include <silkrpc/context_pool.hpp>
 #include "reply.hpp"
@@ -48,7 +49,7 @@ class Connection : public std::enable_shared_from_this<Connection> {
     Connection& operator=(const Connection&) = delete;
 
     /// Construct a connection running within the given execution context.
-    explicit Connection(Context& context);
+    explicit Connection(Context& context, asio::thread_pool& workers);
 
     ~Connection();
 

silkrpc/http/request_handler.hpp

@@ -30,6 +30,7 @@
 #include <silkrpc/config.hpp>
 
 #include <asio/awaitable.hpp>
+#include <asio/thread_pool.hpp>
 
 #include <silkrpc/commands/rpc_api.hpp>
 #include <silkrpc/context_pool.hpp>
@@ -44,7 +45,7 @@ class RequestHandler {
     RequestHandler(const RequestHandler&) = delete;
     RequestHandler& operator=(const RequestHandler&) = delete;
 
-    explicit RequestHandler(Context& context) : rpc_api_{context} {}
+    explicit RequestHandler(Context& context, asio::thread_pool& workers) : rpc_api_{context, workers} {}
 
     virtual ~RequestHandler() {}
 

silkrpc/http/server.cpp

@@ -35,8 +35,8 @@
 
 namespace silkrpc::http {
 
-Server::Server(const std::string& address, const std::string& port, ContextPool& context_pool)
-: context_pool_(context_pool), acceptor_{context_pool.get_io_context()} {
+Server::Server(const std::string& address, const std::string& port, ContextPool& context_pool, std::size_t num_workers)
+: context_pool_(context_pool), acceptor_{context_pool.get_io_context()}, workers_{num_workers} {
     // Open the acceptor with the option to reuse the address (i.e. SO_REUSEADDR).
     asio::ip::tcp::resolver resolver{acceptor_.get_executor()};
     asio::ip::tcp::endpoint endpoint = *resolver.resolve(address, port).begin();
@@ -62,7 +62,7 @@ asio::awaitable<void> Server::run() {
 
             SILKRPC_DEBUG << "Server::start accepting using io_context " << io_context << "...\n" << std::flush;
 
-            auto new_connection = std::make_shared<Connection>(context);
+            auto new_connection = std::make_shared<Connection>(context, workers_);
             co_await acceptor_.async_accept(new_connection->socket(), asio::use_awaitable);
             if (!acceptor_.is_open()) {
                 SILKRPC_TRACE << "Server::start returning...\n";

silkrpc/http/server.hpp

@@ -23,12 +23,14 @@
 #ifndef SILKRPC_HTTP_SERVER_HPP_
 #define SILKRPC_HTTP_SERVER_HPP_
 
+#include <cstddef>
 #include <string>
 
 #include <silkrpc/config.hpp>
 
 #include <asio/awaitable.hpp>
 #include <asio/ip/tcp.hpp>
+#include <asio/thread_pool.hpp>
 
 #include <silkrpc/context_pool.hpp>
 
@@ -41,7 +43,7 @@ class Server {
     Server& operator=(const Server&) = delete;
 
     // Construct the server to listen on the specified TCP address and port
-    explicit Server(const std::string& address, const std::string& port, ContextPool& context_pool);
+    explicit Server(const std::string& address, const std::string& port, ContextPool& context_pool, std::size_t num_workers);
 
     void start();
 
@@ -55,6 +57,8 @@ class Server {
 
     // The acceptor used to listen for incoming TCP connections
     asio::ip::tcp::acceptor acceptor_;
+
+    asio::thread_pool workers_;
 };
 
 } // namespace silkrpc::http

silkrpc/main.cpp

@@ -45,6 +45,7 @@ ABSL_FLAG(std::string, chaindata, silkrpc::common::kEmptyChainData, "chain data
 ABSL_FLAG(std::string, local, silkrpc::common::kDefaultLocal, "HTTP JSON local binding as string <address>:<port>");
 ABSL_FLAG(std::string, target, silkrpc::common::kDefaultTarget, "TG Core gRPC service location as string <address>:<port>");
 ABSL_FLAG(uint32_t, numContexts, std::thread::hardware_concurrency() / 2, "number of running I/O contexts as 32-bit integer");
+ABSL_FLAG(uint32_t, numWorkers, std::thread::hardware_concurrency(), "number of worker threads as 32-bit integer");
 ABSL_FLAG(uint32_t, timeout, silkrpc::common::kDefaultTimeout.count(), "gRPC call timeout as 32-bit integer");
 ABSL_FLAG(silkrpc::LogLevel, logLevel, silkrpc::LogLevel::Critical, "logging level");
 
@@ -74,7 +75,7 @@ int main(int argc, char* argv[]) {
         auto chaindata{absl::GetFlag(FLAGS_chaindata)};
         if (!chaindata.empty() && !std::filesystem::exists(chaindata)) {
             SILKRPC_ERROR << "Parameter chaindata is invalid: [" << chaindata << "]\n";
-            SILKRPC_ERROR << "Use --chaindata flag to specify the path of Turbo-Geth database\n";
+            SILKRPC_ERROR << "Use --chaindata flag to specify the path of Erigon database\n";
             return -1;
         }
 
@@ -88,20 +89,20 @@ int main(int argc, char* argv[]) {
         auto target{absl::GetFlag(FLAGS_target)};
         if (!target.empty() && target.find(":") == std::string::npos) {
             SILKRPC_ERROR << "Parameter target is invalid: [" << target << "]\n";
-            SILKRPC_ERROR << "Use --target flag to specify the location of Turbo-Geth running instance\n";
+            SILKRPC_ERROR << "Use --target flag to specify the location of Erigon running instance\n";
             return -1;
         }
 
         if (chaindata.empty() && target.empty()) {
             SILKRPC_ERROR << "Parameters chaindata and target cannot be both empty, specify one of them\n";
-            SILKRPC_ERROR << "Use --chaindata or --target flag to specify the path or the location of Turbo-Geth instance\n";
+            SILKRPC_ERROR << "Use --chaindata or --target flag to specify the path or the location of Erigon instance\n";
             return -1;
         }
 
         auto timeout{absl::GetFlag(FLAGS_timeout)};
         if (timeout < 0) {
             SILKRPC_ERROR << "Parameter timeout is invalid: [" << timeout << "]\n";
-            SILKRPC_ERROR << "Use --timeout flag to specify the timeout in msecs for Turbo-Geth KV gRPC I/F\n";
+            SILKRPC_ERROR << "Use --timeout flag to specify the timeout in msecs for Erigon KV gRPC I/F\n";
             return -1;
         }
 
@@ -112,6 +113,13 @@ int main(int argc, char* argv[]) {
             return -1;
         }
 
+        auto numWorkers{absl::GetFlag(FLAGS_numWorkers)};
+        if (numWorkers < 0) {
+            SILKRPC_ERROR << "Parameter numWorkers is invalid: [" << numWorkers << "]\n";
+            SILKRPC_ERROR << "Use --numWorkers flag to specify the number of worker threads executing long-run operations\n";
+            return -1;
+        }
+
         if (chaindata.empty()) {
             SILKRPC_LOG << "Silkrpc launched with target " << target << " using " << numContexts << " contexts\n";
         } else {
@@ -139,7 +147,7 @@ int main(int argc, char* argv[]) {
 
         const auto http_host = local.substr(0, local.find(kAddressPortSeparator));
         const auto http_port = local.substr(local.find(kAddressPortSeparator) + 1, std::string::npos);
-        silkrpc::http::Server http_server{http_host, http_port, context_pool};
+        silkrpc::http::Server http_server{http_host, http_port, context_pool, numWorkers};
 
         auto& io_context = context_pool.get_io_context();
         asio::signal_set signals{io_context, SIGINT, SIGTERM};

tests/integration/jsonrpc_commands_goerli.json

@@ -113,7 +113,7 @@
         "response":{
             "jsonrpc":"2.0",
             "id":1,
-            "result":false
+            "result":null
         }
     },
     {